core / tui_interface

core.tui_interface

Textual-based terminal user interface for interacting with the agent.

TUIInterface

Asynchronous Textual TUI driver that feeds user prompts to the agent.

Source code in core\tui_interface.py

class TUIInterface:
    """Asynchronous Textual TUI driver that feeds user prompts to the agent."""

    _STYLE_COLORS = {
        "user": "bold plum1",
        "agent": "bold gold1",
        "action": "bold deep_sky_blue1",
        "task": "bold dark_orange",
        "error": "bold red",
        "info": "bold grey70",
        "system": "bold medium_orchid",
    }

    _CHAT_LABEL_WIDTH = 7
    _ACTION_LABEL_WIDTH = 7

    def __init__(
        self, agent: "AgentBase", *, default_provider: str, default_api_key: str
    ) -> None:
        self._agent = agent
        self._running: bool = False
        self._tracked_sessions: set[str] = set()
        self._seen_events: set[Tuple[str, str, str, str]] = set()
        self._status_message: str = "Idle"
        self._app: _CraftApp | None = None
        self._event_task: asyncio.Task[None] | None = None

        self._command_handlers: dict[str, Callable[[], Awaitable[None]]] = {}

        self.chat_updates: Queue[TimelineEntry] = Queue()
        self.action_updates: Queue[_ActionEntry] = Queue()
        self.status_updates: Queue[str] = Queue()

        self._default_provider = default_provider
        self._default_api_key = default_api_key

        self._register_commands()

    def _register_commands(self) -> None:
        self._command_handlers = {
            "/exit": self._handle_exit_command,
            "/clear": self._handle_clear_command,
            "/reset": self._handle_reset_command,
            "/menu": self._handle_menu_command,
            "/help": self._handle_help_command,
        }

    async def _maybe_handle_command(self, message: str) -> bool:
        command = message.split()[0].lower()

        handler = self._command_handlers.get(command)
        if handler:
            await handler()
            return True

        agent_command = self._agent.get_commands().get(command)
        if agent_command:
            result = await agent_command.handler()
            await self.chat_updates.put(
                (
                    "System",
                    result or f"Command '{command}' executed.",
                    "system",
                )
            )
            return True

        return False

    async def start(self) -> None:
        """Start the Textual TUI session and background consumers."""
        if self._running:
            return

        self._running = True
        logger.debug("Starting Textual TUI interface. Press Ctrl+C to exit.")

        await self.chat_updates.put(
            (
                "System",
                "White Collar Agent TUI ready. Type /help for more info and /exit to quit.",
                "system",
            )
        )
        await self.status_updates.put(self._status_message)

        trigger_consumer = asyncio.create_task(self._consume_triggers())
        self._event_task = asyncio.create_task(self._watch_events())

        self._app = _CraftApp(self, self._default_provider, self._default_api_key)

        try:
            await self._app.run_async()
        finally:
            self._running = False
            self._agent.is_running = False

            trigger_consumer.cancel()
            try:
                await trigger_consumer
            except asyncio.CancelledError:  # pragma: no cover - event loop teardown
                pass

            if self._event_task:
                self._event_task.cancel()
                try:
                    await self._event_task
                except asyncio.CancelledError:  # pragma: no cover - event loop teardown
                    pass

    async def submit_user_message(self, message: str) -> None:
        """Handle chat input captured by the Textual app."""
        if not message:
            return

        if await self._maybe_handle_command(message):
            return

        await self.chat_updates.put(("You", message, "user"))
        await self.status_updates.put("Awaiting agent response…")

        payload = {
            "text": message,
            "sender": {"id": "cli_user", "type": "user"},
            "gui_mode": False,
        }
        await self._agent._handle_chat_message(payload)

    def configure_provider(self, provider: str, api_key: str) -> None:
        key_lookup = {
            "openai": "OPENAI_API_KEY",
            "gemini": "GOOGLE_API_KEY",
            "byteplus": "BYTEPLUS_API_KEY",
        }
        key_name = key_lookup.get(provider)
        if key_name and api_key:
            os.environ[key_name] = api_key
        os.environ["LLM_PROVIDER"] = provider
        self._agent.llm.provider = provider

    def notify_provider(self, provider: str) -> None:
        self.chat_updates.put_nowait(
            (
                "System",
                f"Launching agent with provider: {provider}",
                "system",
            )
        )

    async def request_shutdown(self) -> None:
        """Stop the interface and close the Textual application."""
        if not self._running:
            return

        self._running = False
        self._agent.is_running = False

        if self._app and self._app.is_running:
            self._app.exit()

    async def _handle_exit_command(self) -> None:
        await self.chat_updates.put(("System", "Session terminated by user.", "system"))
        await self.status_updates.put("Idle")
        await self.request_shutdown()

    async def _handle_menu_command(self) -> None:
        # Switch UI back to menu layer if the app is running
        if self._app:
            self._app.show_settings = False
            self._app.show_menu = True

        await self.chat_updates.put(("System", "Returned to menu.", "system"))
        await self.status_updates.put("Idle")

    async def _handle_help_command(self) -> None:
        help_text = self._build_help_text()
        await self.chat_updates.put(("System", help_text, "system"))

    def _build_help_text(self) -> str:
        intro = (
            "I am a computer-use AI agent., I can perform computer-based task autonomously "
            "for you with simple instruction."
        )

        builtin = {
            "/help": "Show this help message.",
            "/menu": "Return to the main menu.",
            "/clear": "Clear chat and action timelines from the display.",
            "/reset": "Reset the agent and clear interface state.",
            "/exit": "Exit the session.",
        }

        lines: list[str] = [intro, "", "Available commands:"]

        # Built-in commands first
        for cmd in sorted(builtin.keys()):
            lines.append(f"  {cmd}  - {builtin[cmd]}")

        # Agent-provided commands (if any)
        agent_cmds = self._agent.get_commands() if self._agent else {}
        extra = [c for c in agent_cmds.keys() if c not in builtin]

        if extra:
            lines.append("")
            lines.append("Agent commands:")
            for cmd in sorted(extra):
                obj = agent_cmds[cmd]
                desc = (
                    getattr(obj, "description", None)
                    or getattr(obj, "help", None)
                    or getattr(obj, "doc", None)
                )
                if not desc and getattr(obj, "handler", None):
                    desc = getattr(obj.handler, "__doc__", None)

                desc = (desc or "Agent command.").strip()
                lines.append(f"  {cmd}  - {desc}")

        return "\n".join(lines)

    def _clear_display_logs(self) -> None:
        if self._app:
            self._app.clear_logs()

    async def _handle_clear_command(self) -> None:
        self._clear_display_logs()
        self.chat_updates = Queue()
        self.action_updates = Queue()
        await self.chat_updates.put(("System", "Cleared chat and action timelines.", "system"))

    async def _handle_reset_command(self) -> None:
        response: str | None = None
        reset_command = self._agent.get_commands().get("/reset")
        if reset_command:
            response = await reset_command.handler()

        await self._reset_interface_state()
        await self.chat_updates.put(("System", response or "Agent reset. Starting fresh.", "system"))

    async def _reset_interface_state(self) -> None:
        self._tracked_sessions.clear()
        self._seen_events.clear()
        self.chat_updates = Queue()
        self.action_updates = Queue()
        self.status_updates = Queue()
        self._status_message = "Idle"
        self._clear_display_logs()
        await self.status_updates.put(self._status_message)

    async def _consume_triggers(self) -> None:
        """Continuously consume triggers and hand them to the agent."""
        try:
            while self._agent.is_running:
                trigger = await self._agent.triggers.get()
                if trigger.session_id:
                    self._tracked_sessions.add(trigger.session_id)
                await self._agent.react(trigger)
        except asyncio.CancelledError:  # pragma: no cover
            raise

    async def _watch_events(self) -> None:
        """Refresh the conversation timeline with agent actions."""
        try:
            while self._running and self._agent.is_running:
                stream = self._agent.event_stream_manager.get_stream()
                if not stream:
                    await asyncio.sleep(0.05)
                    continue

                for event in stream.as_list():
                    key = (event.iso_ts, event.kind, event.message)
                    if key in self._seen_events:
                        continue
                    self._seen_events.add(key)

                    if event.kind == "screen":
                        continue

                    style = self._style_for_event(event.kind, event.severity)
                    label = self._label_for_style(style, event.kind)
                    display_text = event.display_text()

                    if style in {"action", "task"}:
                        await self._handle_action_event(
                            event.kind,
                            display_text,
                            style=style,
                        )
                        continue

                    if style not in {"agent", "system", "user", "error", "info"}:
                        continue

                    if display_text is not None:
                        await self.chat_updates.put((label, display_text, style))

                await asyncio.sleep(0.05)

        except asyncio.CancelledError:  # pragma: no cover
            raise

    async def _handle_action_event(self, kind: str, message: str, *, style: str = "action") -> None:
        """Record an action update and refresh the status bar."""
        await self.action_updates.put(_ActionEntry(kind=kind, message=message, style=style))
        if style == "action":
            status = self._derive_status(kind, message)
            if status != self._status_message:
                self._status_message = status
                await self.status_updates.put(status)

    def _derive_status(self, kind: str, message: str) -> str:
        normalized = message.strip() or ""
        if kind == "action_start":
            return f"Running: {normalized or 'action in progress'}"
        if kind == "action_end":
            return f"Completed: {normalized or 'last action'}"
        if kind == "action":
            return normalized or "Action in progress"
        return normalized or self._status_message or "Idle"

    def _format_labelled_entry(
        self,
        label_text: str,
        message: Text | str,
        *,
        colour: str,
        label_width: int,
    ) -> Table:
        table = Table.grid(padding=(0, 1))
        table.expand = True
        table.add_column(
            "label",
            width=label_width,
            min_width=label_width,
            max_width=label_width,
            style=colour,
            no_wrap=True,
            justify="left",
        )
        table.add_column("message", ratio=1)

        label_cell = Text(label_text, style=colour, no_wrap=True)
        message_text = message if isinstance(message, Text) else Text(str(message))
        message_text.no_wrap = False
        message_text.overflow = "fold"

        table.add_row(label_cell, message_text)
        return table

    def format_chat_entry(self, label: str, message: str, style: str) -> RenderableType:
        colour = self._STYLE_COLORS.get(style, self._STYLE_COLORS["info"])
        label_text = f"{label}:"
        return self._format_labelled_entry(
            label_text,
            message,
            colour=colour,
            label_width=self._CHAT_LABEL_WIDTH,
        )

    def format_action_entry(self, entry: _ActionEntry) -> RenderableType:
        kind = entry.kind.replace("_", " ").title()
        colour = "bold deep_sky_blue1" if entry.style == "action" else "bold dark_orange"
        label_text = f"{kind}:"
        return self._format_labelled_entry(
            label_text,
            entry.message,
            colour=colour,
            label_width=self._ACTION_LABEL_WIDTH,
        )

    def _style_for_event(self, kind: str, severity: str) -> str:
        if severity.upper() == "ERROR":
            return "error"
        if kind == "system":
            return "system"
        if kind.startswith("task"):
            return "task"
        if kind in {"action", "action_start", "action_end"}:
            return "action"
        if kind in {"screen", "info", "note"}:
            return "info"
        if kind == "user":
            return "user"
        return "agent"

    @staticmethod
    def _label_for_style(style: str, kind: str) -> str:
        if style == "agent":
            return "Agent"
        if style == "system":
            return "System"
        if style == "user":
            return "You"
        if style == "error":
            return "Error"
        if style == "task":
            return kind.replace("_", " ").title()
        if style == "info":
            return kind.replace("_", " ").title()
        return kind.title()

start() async

Start the Textual TUI session and background consumers.

Source code in core\tui_interface.py

async def start(self) -> None:
    """Start the Textual TUI session and background consumers."""
    if self._running:
        return

    self._running = True
    logger.debug("Starting Textual TUI interface. Press Ctrl+C to exit.")

    await self.chat_updates.put(
        (
            "System",
            "White Collar Agent TUI ready. Type /help for more info and /exit to quit.",
            "system",
        )
    )
    await self.status_updates.put(self._status_message)

    trigger_consumer = asyncio.create_task(self._consume_triggers())
    self._event_task = asyncio.create_task(self._watch_events())

    self._app = _CraftApp(self, self._default_provider, self._default_api_key)

    try:
        await self._app.run_async()
    finally:
        self._running = False
        self._agent.is_running = False

        trigger_consumer.cancel()
        try:
            await trigger_consumer
        except asyncio.CancelledError:  # pragma: no cover - event loop teardown
            pass

        if self._event_task:
            self._event_task.cancel()
            try:
                await self._event_task
            except asyncio.CancelledError:  # pragma: no cover - event loop teardown
                pass

submit_user_message(message) async

Handle chat input captured by the Textual app.

Source code in core\tui_interface.py

async def submit_user_message(self, message: str) -> None:
    """Handle chat input captured by the Textual app."""
    if not message:
        return

    if await self._maybe_handle_command(message):
        return

    await self.chat_updates.put(("You", message, "user"))
    await self.status_updates.put("Awaiting agent response…")

    payload = {
        "text": message,
        "sender": {"id": "cli_user", "type": "user"},
        "gui_mode": False,
    }
    await self._agent._handle_chat_message(payload)

request_shutdown() async

Stop the interface and close the Textual application.

Source code in core\tui_interface.py

async def request_shutdown(self) -> None:
    """Stop the interface and close the Textual application."""
    if not self._running:
        return

    self._running = False
    self._agent.is_running = False

    if self._app and self._app.is_running:
        self._app.exit()