app.py

"""Textual ``App`` subclass that hosts the Claw Codex TUI.

The app owns everything that must outlive a single screen push: the
``Session`` / ``Conversation``, the provider instance, the tool registry,
the tool context, the :class:`AppState`, and the
:class:`AgentBridge` that shuttles events between the agent-loop worker
thread and the UI.

Phase 1 boots the :class:`REPLScreen` on mount and delegates user
submissions to :class:`AgentBridge.submit`. Permission requests from
tools land here as :class:`PermissionRequested` messages, which the
screen then materialises as a modal.
"""

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from src.services.bash_mode import BashModeOutcome

from textual.app import App

from src import __version__ as CLAW_VERSION
from src.agent import Session
from src.tool_system.context import ToolContext
from src.tool_system.defaults import build_default_registry
from src.tool_system.registry import ToolRegistry

from .a11y import Announcer, describe_status
from .agent_bridge import AgentBridge
from .commands import (
    CommandDispatchResult,
    CommandSuggestion,
    build_command_suggestions,
    build_command_words,
    dispatch_local_command,
    dispatch_registry_command,
)
from .history_store import HistoryStore  # noqa: F401 (re-exported for tests)
from .messages import CancelRequested
from .screens.cost_threshold import CostThresholdScreen
from .screens.diff_dialog import DiffDialogScreen, FileDiff
from .screens.effort_picker import EffortPickerScreen
from .screens.exit_flow import ExitFlowScreen
from .screens.history_search import HistoryEntry, HistorySearchScreen
from .screens.idle_return import IdleReturnScreen
from .screens.mcp_dialogs import McpListScreen, McpServer
from .screens.message_selector import MessageSelectorScreen, TranscriptMessage
from .screens.model_picker import ModelPickerScreen
from .screens.repl import REPLScreen
from .screens.theme_picker import ThemePickerScreen
from .state import AppState
from .terminal_chrome import (
    clear_terminal_title,
    disable_focus_events,
    enable_focus_events,
    ring_bell,
    set_tab_status,
    set_terminal_title,
)
from .theme import (
    get_palette,
    list_theme_names,
    resolve_auto_theme,
    textual_css_overrides,
)
from .widgets.transcript_view import Transcript


def _flatten_message_text(content: Any) -> str:
    """Normalise ``Message.content`` (string or block list) to text."""

    if content is None:
        return ""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, dict):
                kind = item.get("type")
                if kind in (None, "text"):
                    parts.append(str(item.get("text") or ""))
                elif kind == "tool_use":
                    parts.append(f"[tool:{item.get('name') or ''}]")
                else:
                    parts.append("")
        return "\n".join(p for p in parts if p).strip()
    return str(content)


class ClawCodexTUI(App):
    """Top-level Textual application for Claw Codex."""

    TITLE = "ClawCodex"
    SUB_TITLE = "interactive terminal"

    BINDINGS = [
        ("ctrl+c", "cancel_or_quit", "Cancel / Quit"),
        ("ctrl+d", "quit", "Quit"),
    ]

    def __init__(
        self,
        *,
        provider,
        provider_name: str,
        workspace_root: Path,
        tool_registry: ToolRegistry | None = None,
        tool_context: ToolContext | None = None,
        session: Session | None = None,
        max_turns: int = 20,
        stream: bool = True,
        theme_name: str | None = None,
    ) -> None:
        super().__init__()
        self.provider = provider
        self.provider_name = provider_name
        self.workspace_root = Path(workspace_root)
        self.max_turns = max_turns
        self.stream = stream
        self.model = getattr(provider, "model", "unknown")
        self.session = session or Session.create(provider_name, self.model or "")
        self.tool_registry = tool_registry or build_default_registry(provider=provider)
        self.tool_context = tool_context or self._build_default_tool_context()
        # Theme is resolved once on boot; ``/theme`` can switch it
        # live via :meth:`apply_theme`.
        self.palette = get_palette(theme_name or self._resolve_theme_name())
        self.app_state = AppState(
            model=self.model or "",
            provider=provider_name,
        )
        self._repl_screen: REPLScreen | None = None
        self._command_context: Any | None = None
        # P0-6 Option B: guards one-time global registration of builtins +
        # user-invocable skills (see _ensure_command_context).
        self._skills_registered: bool = False
        # Reactive AppState store backing interactive commands like
        # /permissions. Created once on first command-context build and
        # held here so the chosen mode persists across invocations.
        self._app_state_store: Any | None = None
        # Transcript renderables captured at exit time so entry points
        # can dump them back to the main terminal scrollback after the
        # alt-screen tears down. Mirrors the TS ink behaviour where the
        # conversation the user saw stays on-screen after ``/exit``.
        self.exit_snapshot: list[Any] = []
        # Persistent prompt history used by the PromptInput (↑/↓) and
        # the /history slash-command dialog. The store is append-only
        # per turn and auto-rotates past ``max_entries``.
        self.history_store = HistoryStore()
        self._theme_name = (theme_name or self._resolve_theme_name())
        # Screen-reader announcer. The :class:`LiveRegion` widget is
        # bound in :meth:`on_mount` once the REPL screen is composed.
        self.announcer = Announcer(self)
        self._agent_bridge = AgentBridge(
            post_message=self._post_to_screen,
            session=self.session,
            provider=self.provider,
            tool_registry=self.tool_registry,
            tool_context=self.tool_context,
            app_state=self.app_state,
            run_worker=self.run_worker,
            max_turns=self.max_turns,
            stream=self.stream,
        )

    # The base CSS for the REPL; Phase 1 uses Textual's default theme
    # variables ($primary, $surface, …) — palette overrides sit in
    # ``textual_css_overrides`` and are appended at class build time.
    CSS = ""

    def _resolve_theme_name(self) -> str:
        try:
            from src.config import load_config

            cfg = load_config() or {}
            return cfg.get("theme") or "dark"
        except Exception:
            return "dark"

    # ---- lifecycle ----
    def on_mount(self) -> None:
        # Apply palette-derived CSS on top of the component defaults so
        # the chrome picks up the correct background / foreground even
        # when Textual's internal theme doesn't cover every slot.
        try:
            self.stylesheet.add_source(
                textual_css_overrides(self.palette),
                path="palette-overrides",
            )
            self.stylesheet.parse()
        except Exception:
            pass

        self._repl_screen = REPLScreen(
            version=CLAW_VERSION,
            provider=self.provider_name,
            model=self.model,
            workspace_root=self.workspace_root,
            words_provider=self._slash_command_words,
            suggestions_provider=self._slash_command_suggestions,
            # Pass the live BaseProvider so the status line's advisor
            # segment can call ``decide_advisor_mode(provider, ...)``
            # and show the correct mode label (server/client/inactive).
            provider_instance=self.provider,
        )
        self.push_screen(self._repl_screen)

        # Terminal chrome: set a descriptive title, enable DEC 1004
        # focus reporting, and mark the tab idle. The app-state
        # observer below keeps title + tab status in sync with agent
        # activity.
        self._last_thinking: bool = self.app_state.is_thinking
        self._sync_terminal_title()
        set_tab_status("idle")
        try:
            enable_focus_events()
        except Exception:
            pass
        self._state_unsub = self.app_state.subscribe(self._on_state_change)
        # C8 → C6 → C7 boot chain, strictly sequential: the security
        # gates (trust → external includes → bypass acceptance; TS
        # interactiveHelpers order) must resolve BEFORE the .mcp.json
        # approval prompts — a user who hasn't trusted the folder yet
        # must not be asked to enable its MCP servers first.
        self.call_after_refresh(self._run_startup_chain)

    # ---- C8 startup security gates ---------------------------------------
    def _run_startup_chain(self) -> None:
        self._startup_gate_trust()

    def _startup_gate_trust(self) -> None:
        try:
            from src.services.startup_gates import check_trust_accepted

            needed = not check_trust_accepted()
        except Exception:
            # Fail CLOSED: a consent gate that errors must ask, not
            # silently wave the folder through.
            needed = True
        if not needed:
            self._startup_gate_includes()
            return
        from src.tui.screens.startup_gates import TrustFolderScreen

        try:
            from src.services.startup_gates import collect_trust_warnings

            warnings = collect_trust_warnings()
        except Exception:
            warnings = []
        self.push_screen(
            TrustFolderScreen(str(self.workspace_root), warnings),
            callback=self._on_trust_choice,
        )

    def _on_trust_choice(self, choice: str | None) -> None:
        if choice != "trust":
            # TS TrustDialog "No, exit" → gracefulShutdownSync(1).
            self.exit(return_code=1)
            return
        persisted = False
        try:
            from src.services.startup_gates import record_trust_accepted

            persisted = record_trust_accepted()
        except Exception:
            persisted = False
        # The tool context was built before the dialog ran (untrusted seed);
        # propagate acceptance so hooks stop being trust-skipped (#275).
        # record_trust_accepted already synced the bootstrap flag for any
        # contexts constructed after this point.
        self.tool_context.workspace_trusted = True
        if not persisted and self._repl_screen is not None:
            self._repl_screen.transcript.append_system(
                "Folder trusted for this session only — could not write "
                "the global config; you'll be asked again next launch.",
                style="muted",
            )
        self._startup_gate_includes()

    def _startup_gate_includes(self) -> None:
        try:
            from src.services.startup_gates import (
                get_external_includes_state,
            )

            state = get_external_includes_state()
        except Exception:
            state = "declined"  # fail safe: no prompt, no external load
        if state != "unset":
            self._startup_gate_bypass()
            return
        self.run_worker(
            self._detect_external_includes(),
            exclusive=False,
            name="c8-external-includes",
        )

    async def _detect_external_includes(self) -> None:
        try:
            from src.services.startup_gates import list_external_includes

            externals = await list_external_includes()
        except Exception:
            externals = []
        if not externals:
            # Nothing external to approve — the question never arises
            # (TS shows the dialog only when externals exist).
            self._startup_gate_bypass()
            return
        from src.tui.screens.startup_gates import ExternalIncludesScreen

        self.push_screen(
            ExternalIncludesScreen(externals),
            callback=self._on_includes_choice,
        )

    def _on_includes_choice(self, choice: str | None) -> None:
        approved = choice == "yes"
        persisted = False
        try:
            from src.services.startup_gates import (
                record_external_includes_choice,
            )

            persisted = record_external_includes_choice(approved)
        except Exception:
            persisted = False
        if self._repl_screen is not None:
            transcript = self._repl_screen.transcript
            if not persisted:
                transcript.append_system(
                    "Could not save the external-imports choice — "
                    "you'll be asked again next launch.",
                    style="muted",
                )
            elif approved:
                transcript.append_system(
                    "External CLAUDE.md imports enabled for this project.",
                    style="muted",
                )
            else:
                transcript.append_system(
                    "External CLAUDE.md imports disabled for this project.",
                    style="muted",
                )
        self._startup_gate_bypass()

    def _startup_gate_bypass(self) -> None:
        mode = getattr(
            getattr(self.tool_context, "permission_context", None),
            "mode",
            "default",
        )
        if mode != "bypassPermissions":
            self._finish_startup_gates()
            return
        try:
            from src.services.startup_gates import (
                has_skip_dangerous_mode_permission_prompt,
            )

            already_accepted = has_skip_dangerous_mode_permission_prompt()
        except Exception:
            already_accepted = False
        if already_accepted:
            self._finish_startup_gates()
            return
        from src.tui.screens.startup_gates import BypassPermissionsScreen

        self.push_screen(
            BypassPermissionsScreen(), callback=self._on_bypass_choice
        )

    def _on_bypass_choice(self, choice: str | None) -> None:
        if choice == "accept":
            persisted = False
            try:
                from src.services.startup_gates import record_bypass_accepted

                persisted = record_bypass_accepted()
            except Exception:
                persisted = False
            if not persisted and self._repl_screen is not None:
                self._repl_screen.transcript.append_system(
                    "Could not persist the bypass acceptance — you'll be "
                    "asked again next launch.",
                    style="muted",
                )
            self._finish_startup_gates()
            return
        # TS: explicit "No, exit" → exit 1; Esc (or any non-answer
        # dismissal) → exit 0.
        self.exit(return_code=1 if choice == "decline" else 0)

    def _finish_startup_gates(self) -> None:
        # C6: surface ignored/malformed config files as startup rows
        # (TS StatusNotices / InvalidSettingsDialog family — Python's
        # loader silently falls back to {}, so warn honestly instead).
        self._show_config_warnings()
        # C7: prompt for any pending .mcp.json servers (TS
        # handleMcpjsonServerApprovals).
        self._run_mcp_approvals()

    # ---- C7 .mcp.json server approvals ----------------------------------
    # cwd note: approvals deliberately key off the process cwd (the
    # default of every mcp_approval call), NOT self.workspace_root —
    # enumeration and enforcement in services/mcp/config both resolve
    # from the process cwd, and splitting the two would persist
    # approvals to a file enforcement never reads.
    def _run_mcp_approvals(self) -> None:
        try:
            from src.services.mcp_approval import list_pending_mcpjson_servers

            pending = list_pending_mcpjson_servers()
        except Exception:
            return
        if pending:
            self._prompt_next_mcp_approval(list(pending))

    def _prompt_next_mcp_approval(self, queue: list[str]) -> None:
        from src.services.mcp_approval import get_mcpjson_server_status
        from src.tui.screens.mcp_approval import McpApprovalScreen

        while queue:
            name = queue.pop(0)
            try:
                still_pending = get_mcpjson_server_status(name) == "pending"
            except Exception:
                still_pending = False
            if not still_pending:
                # An earlier answer (enable_all, or a normalization-equal
                # name) already decided this one — don't re-ask.
                continue
            self.push_screen(
                McpApprovalScreen(name),
                callback=lambda choice, _n=name, _q=queue: (
                    self._on_mcp_approval(_n, choice, _q)
                ),
            )
            return

    def _on_mcp_approval(
        self, name: str, choice: str | None, queue: list[str]
    ) -> None:
        transcript = (
            self._repl_screen.transcript if self._repl_screen else None
        )
        if choice:
            persisted = False
            try:
                from src.services.mcp_approval import record_mcpjson_choice

                persisted = record_mcpjson_choice(name, choice)
            except Exception:
                persisted = False
            if not persisted:
                if transcript is not None:
                    transcript.append_system(
                        f"Could not save the choice for MCP server "
                        f"'{name}' (settings file not writable) — it "
                        "stays pending.",
                        style="muted",
                    )
                self._prompt_next_mcp_approval(queue)
                return
        if transcript is not None:
            if choice == "enable":
                transcript.append_system(
                    f"MCP server '{name}' enabled.", style="muted"
                )
            elif choice == "enable_all":
                transcript.append_system(
                    "All .mcp.json servers from this project enabled.",
                    style="muted",
                )
            elif choice == "disable":
                transcript.append_system(
                    f"MCP server '{name}' disabled.", style="muted"
                )
            else:
                transcript.append_system(
                    f"MCP server '{name}' left pending — you'll be asked "
                    "again next launch.",
                    style="muted",
                )
        if choice == "enable_all":
            queue.clear()  # everything else is now approved too
        self._prompt_next_mcp_approval(queue)

    def _show_config_warnings(self) -> None:
        if self._repl_screen is None:
            return
        try:
            from src.services.config_health import (
                collect_config_warnings,
                collect_rule_warnings,
            )

            warnings = collect_config_warnings(str(self.workspace_root))
            rule_warnings = collect_rule_warnings(str(self.workspace_root))
            transcript = self._repl_screen.transcript
            for warning in warnings:
                transcript.append_system(
                    f"⚠ {warning.message()}", style="warning"
                )
            for text in rule_warnings:
                transcript.append_system(f"⚠ {text}", style="warning")
            if warnings or rule_warnings:
                transcript.append_system(
                    "Run /doctor for details.", style="muted"
                )
        except Exception:
            # A health-check failure must never crash app mount (the
            # call_after_refresh callback would take the app down).
            return

    def on_unmount(self) -> None:
        # Best-effort cleanup so we don't leave stale chrome on the host.
        try:
            self._state_unsub()  # type: ignore[attr-defined]
        except Exception:
            pass
        try:
            set_tab_status(None)
            clear_terminal_title()
            disable_focus_events()
        except Exception:
            pass
        # Fallback capture in case ``exit()`` wasn't the path out (e.g.
        # Ctrl+C / SIGTERM). Entry points will print whatever landed
        # here to the host shell after the alt-screen exits.
        if not self.exit_snapshot:
            self._capture_exit_snapshot()

    # ---- exit / snapshot ----------------------------------------------
    def _capture_exit_snapshot(self) -> None:
        """Collect the transcript's renderables into :attr:`exit_snapshot`.

        Called from :meth:`exit` and (as a fallback) :meth:`on_unmount`
        so no matter which shutdown path fires we preserve what the
        user saw. Failures are swallowed — a blank snapshot is fine,
        but raising would mask a normal exit.
        """

        if self.exit_snapshot or self._repl_screen is None:
            return
        try:
            self.exit_snapshot = list(self._repl_screen.transcript.snapshot())
        except Exception:
            self.exit_snapshot = []

    def exit(self, result=None, return_code=0, message=None):  # type: ignore[override]
        """Capture transcript before handing control back to Textual.

        Overriding ``exit()`` lets the entry-point reprint the
        conversation to the host terminal once the alt-screen unwinds,
        matching the TS ink reference's non-fullscreen UX where
        `/exit` leaves the printed text intact in scrollback.
        """

        self._capture_exit_snapshot()
        return super().exit(result, return_code=return_code, message=message)

    def _on_state_change(self) -> None:
        """React to :class:`AppState` changes to refresh terminal chrome.

        Title reflects the active verb; tab status flips between
        ``busy`` (agent thinking) and ``idle`` (prompt ready); a
        terminal bell rings on the idle→thinking→idle edge to
        announce turn completion, matching the TS reference's
        idle-notification.
        """

        thinking = self.app_state.is_thinking
        if thinking != self._last_thinking:
            set_tab_status("busy" if thinking else "idle")
            if not thinking:
                # Turn completed — poke the host so the user notices
                # even when they tabbed away.
                try:
                    ring_bell()
                except Exception:
                    pass
                self.announcer.announce(
                    describe_status("idle"), level="polite", notify=False
                )
            else:
                self.announcer.announce(
                    describe_status("busy", verb=self.app_state.verb),
                    level="polite",
                    notify=False,
                )
            self._last_thinking = thinking
        self._sync_terminal_title()

    def _sync_terminal_title(self) -> None:
        try:
            state = self.app_state
            verb = state.verb if state.is_thinking else "Ready"
            title = f"ClawCodex — {state.model or self.provider_name}: {verb}"
            set_terminal_title(title)
        except Exception:
            pass

    # ---- bindings ----
    def action_cancel_or_quit(self) -> None:
        # Phase 1 keeps Ctrl+C as exit. Real cancellation (interrupt the
        # in-flight agent loop) lands in Phase 2 alongside the cost /
        # idle dialogs.
        self.exit()

    # ---- local command dispatcher ----
    def handle_local_slash_command(self, text: str, transcript: Transcript) -> bool:
        """Return ``True`` if the command was handled without hitting the agent.

        The dispatcher tries the local built-ins first (``/exit``,
        ``/help``, …), then falls through to the shared
        :mod:`src.command_system` registry. Commands that produce a
        prompt (``/init``) forward the prompt back to the agent bridge.
        """

        result = dispatch_local_command(
            text,
            session=self.session,
            workspace_root=self.workspace_root,
            tool_registry=self.tool_registry,
        )
        if result.handled:
            self._apply_command_result(result, transcript)
            return True

        # Fall through to the async command registry. We run it via the
        # asyncio loop that Textual already runs on.
        async def _run() -> CommandDispatchResult:
            return await dispatch_registry_command(
                text,
                command_context=self._ensure_command_context(),
            )

        # Schedule the async work on the Textual loop; if it comes back
        # handled we emit the appropriate UI response.
        self.run_worker(self._dispatch_registry_async(text, transcript), exclusive=False, name="slash-cmd")
        return True

    async def _dispatch_registry_async(self, text: str, transcript: Transcript) -> None:
        result = await dispatch_registry_command(
            text,
            command_context=self._ensure_command_context(),
        )
        if not result.handled:
            # Unknown command — show the raw text as a user prompt so
            # the agent can react to it, matching legacy REPL behavior.
            transcript.append_user(text)
            self.submit_to_agent(text)
            return
        self._apply_command_result(result, transcript)

    def _apply_command_result(
        self,
        result: CommandDispatchResult,
        transcript: Transcript,
    ) -> None:
        if result.error:
            transcript.append_system(result.error, style="error")
            return
        if result.open_dialog:
            self._open_phase2_dialog(result.open_dialog, transcript)
            return
        if result.system_text == "__exit__":
            self._confirm_exit(transcript)
            return
        if result.system_text == "__clear__":
            transcript.clear_transcript()
            self._agent_bridge.reset_advisor_dedup()
            # C3a: the context-% segment must not keep showing the
            # pre-clear context (TS shows nothing for an empty list).
            self.app_state.last_turn_input_tokens = 0
            return
        if result.system_text == "__thinking__":
            # C3b /thinking: flip the bridge's session override. First use
            # DISABLES (auto → off, matching the TS toggle's "respond
            # without extended thinking" default action), then alternates.
            bridge = self._agent_bridge
            new_value = bridge.extended_thinking is False
            if new_value:
                # Honest enable (review M4): an explicit True bypasses the
                # query layer's model gate and a non-Anthropic provider
                # ignores it silently — refuse instead of lying.
                try:
                    from src.providers.anthropic_provider import (
                        AnthropicProvider,
                    )
                    from src.providers.minimax_provider import MinimaxProvider
                    from src.query.query import (
                        _model_supports_extended_thinking,
                    )

                    provider_obj = bridge._provider
                    supported = isinstance(
                        provider_obj, (AnthropicProvider, MinimaxProvider)
                    ) and _model_supports_extended_thinking(
                        getattr(provider_obj, "model", None)
                    )
                except Exception:
                    supported = False
                if not supported:
                    transcript.append_system(
                        "Extended thinking is not supported by the current "
                        "model/provider — leaving it disabled.",
                        style="muted",
                    )
                    return
            bridge.extended_thinking = new_value
            transcript.append_system(
                "Extended thinking enabled for this session."
                if new_value
                else "Extended thinking disabled for this session — "
                "the model will respond without extended thinking.",
                style="muted",
            )
            return
        if result.compact and result.system_text:
            transcript.append_compact_boundary(result.system_text)
            return
        if result.system_text:
            transcript.append_system(result.system_text, style="muted")
        if result.prompt_text:
            transcript.append_user(f"(from slash command) {result.prompt_text[:80]}…")
            self.submit_to_agent(result.prompt_text)

    # ---- C4 bash-mode (`!` prefix) --------------------------------------
    # ---- C9 `#` memory-append shortcut -----------------------------------
    def run_memory_shortcut(self, note: str, transcript: Transcript) -> None:
        # History first, like bash-mode does for `!` — a saved (or
        # cancelled) note must stay reachable via up-arrow.
        try:
            self.history_store.append(f"#{note}")
        except Exception:
            pass
        self.run_worker(
            self._memory_shortcut_flow(note, transcript),
            exclusive=False,
            name="c9-memory-shortcut",
        )

    async def _memory_shortcut_flow(
        self, note: str, transcript: Transcript
    ) -> None:
        try:
            from src.command_system.memory_command import build_memory_options

            options = await build_memory_options(str(self.workspace_root))
        except Exception:
            options = []
        if not options:
            transcript.append_system(
                "Could not enumerate memory files — note not saved.",
                style="muted",
            )
            self._insert_into_prompt(f"#{note}")
            return
        from src.tui.screens.memory_save import MemorySaveScreen

        self.push_screen(
            MemorySaveScreen(note, options),
            callback=lambda path, _n=note, _t=transcript: (
                self._on_memory_target(_n, path, _t)
            ),
        )

    def _on_memory_target(
        self, note: str, path: str | None, transcript: Transcript
    ) -> None:
        from src.services.memory_append import (
            append_memory_note,
            pick_saving_message,
        )

        if path is None:
            # Verbatim TS cancel string (memory.tsx:65); the note goes
            # back into the prompt so nothing typed is lost.
            transcript.append_system("Cancelled memory editing", style="muted")
            self._insert_into_prompt(f"#{note}")
            return
        ok = False
        try:
            ok = append_memory_note(path, note)
        except Exception:
            ok = False
        if not ok:
            transcript.append_system(
                f"Could not write {path} — memory not saved.", style="muted"
            )
            self._insert_into_prompt(f"#{note}")
            return
        # TS UserMemoryInputMessage: the `#`-styled note row plus a
        # random acknowledgement line.
        transcript.append_system(f"# {note.strip()}", style="muted")
        transcript.append_system(pick_saving_message(), style="muted")

    def run_bash_mode(self, command: str, transcript: Transcript) -> None:
        """Execute a user-typed ``!command`` directly (no agent turn).

        Divergences from TS (documented; see services/bash_mode):
        sequential only (a second ``!`` while one runs is refused — TS
        queues), and no live progress streaming / ESC cancel yet (the
        echo row shows "running…" until completion; cancel is a
        follow-up). If the agent starts mid-flight, the conversation
        texts DEFER via the bridge and land after the run (review B1).
        """

        command = (command or "").strip()
        if not command:
            transcript.append_system("Usage: !<shell command>", style="muted")
            return
        # History first: TS records queued/refused commands too (m11).
        self.history_store.append(f"!{command}")
        if getattr(self, "_bash_inflight", False):
            transcript.append_system(
                "A bash command is already running — wait for it to finish.",
                style="muted",
            )
            return
        if self._agent_bridge.busy:
            transcript.append_system(
                "Agent is working — bash mode is unavailable until idle.",
                style="muted",
            )
            return
        self._bash_inflight = True
        row = transcript.append_bash_running(command)

        def _work() -> None:
            from src.services.bash_mode import run_bash_mode_command

            try:
                outcome = run_bash_mode_command(command, self.tool_context)
            except Exception as exc:  # service contract: never raises
                try:
                    self.call_from_thread(
                        self._bash_mode_error, command, str(exc), transcript
                    )
                except Exception:
                    pass
                return
            try:
                self.call_from_thread(
                    self._finish_bash_mode, outcome, row, transcript
                )
            except Exception:
                # App shutting down mid-flight; nothing to render.
                self._bash_inflight = False

        self.run_worker(
            _work, thread=True, exit_on_error=False, group="bash-mode"
        )

    def _bash_mode_error(
        self, command: str, error: str, transcript: Transcript
    ) -> None:
        self._bash_inflight = False
        transcript.append_system(f"!{command}: {error}", style="error")

    def _finish_bash_mode(
        self,
        outcome: "BashModeOutcome",
        row: Any,
        transcript: Transcript,
    ) -> None:
        self._bash_inflight = False
        transcript.finish_bash_io(
            row,
            command=outcome.command,
            stdout=outcome.stdout,
            stderr=outcome.stderr,
            exit_code=outcome.exit_code,
            ok=outcome.ok,
        )
        # TS parity: both messages enter the CONVERSATION (the model sees
        # them next turn) — via the bridge, which defers while a run is
        # in flight so the texts can never interleave a tool_use/result
        # pair (review B1); no agent turn fires (shouldQuery: false).
        self._agent_bridge.append_user_texts(outcome.conversation_texts)

    # ---- Phase 2 dialog dispatcher -------------------------------------
    def _open_phase2_dialog(self, name: str, transcript: Transcript) -> None:
        """Push the modal screen for ``name`` from the slash command.

        ``name`` is one of the values produced by
        :func:`dispatch_local_command`; unknown names degrade to a
        muted system message.
        """

        if name == "model":
            self._open_model_picker(transcript)
        elif name == "effort":
            self._open_effort_picker(transcript)
        elif name == "history":
            self._open_history_search(transcript)
        elif name == "cost":
            self._open_cost_threshold(transcript)
        elif name == "idle":
            self._open_idle_return(transcript)
        elif name == "theme":
            self._open_theme_picker(transcript)
        elif name == "diff":
            self._open_diff_dialog(transcript)
        elif name == "mcp":
            self._open_mcp_list(transcript)
        elif name in ("rewind", "messages"):
            self._open_message_selector(transcript)
        elif name == "tasks":
            self._open_tasks_dialog(transcript)
        elif name == "workflows":
            self._open_workflows_dialog(transcript)
        elif name == "resume":
            self._open_resume_picker(transcript)
        elif name == "search" or name.startswith("search:"):
            self._open_global_search(
                transcript, name.partition(":")[2] if ":" in name else ""
            )
        elif name == "quickopen":
            self._open_quick_open(transcript)
        elif name == "doctor":
            from src.tui.screens.doctor import DoctorScreen

            self.push_screen(
                DoctorScreen(
                    app_state=self.app_state,
                    workspace_root=self.workspace_root,
                )
            )
        else:
            transcript.append_system(f"Dialog '{name}' not available.", style="muted")

    # ---- C5 workspace search dialogs -----------------------------------
    def _insert_into_prompt(self, insertion: str | None) -> None:
        if not insertion:
            return
        if self._repl_screen is not None:
            self._repl_screen.prompt_input.append_value(insertion)

    def _open_global_search(
        self, transcript: Transcript, initial_query: str = ""
    ) -> None:
        from src.tui.screens.workspace_search import GlobalSearchScreen

        self.push_screen(
            GlobalSearchScreen(
                cwd=str(self.workspace_root), initial_query=initial_query
            ),
            callback=self._insert_into_prompt,
        )

    def _open_quick_open(self, transcript: Transcript) -> None:
        from src.tui.screens.workspace_search import QuickOpenScreen

        self.push_screen(
            QuickOpenScreen(cwd=str(self.workspace_root)),
            callback=self._insert_into_prompt,
        )

    def _open_resume_picker(self, transcript: Transcript) -> None:
        """C2: list persisted sessions; selection swaps the live session.

        Refuses while the agent is busy (the bridge would also refuse —
        this just gives the honest message before pushing a modal).
        """

        if self._agent_bridge.busy:
            transcript.append_system(
                "Cannot resume while the agent is working — try again when idle.",
                style="muted",
            )
            return

        from src.bootstrap.state import get_session_id
        from src.services.session_storage import SessionStorage
        from src.tui.screens.resume_conversation import (
            ResumeConversation,
            build_resume_entries,
        )

        try:
            metas = SessionStorage.list_sessions()
        except Exception:
            metas = []
        entries, hidden = build_resume_entries(
            metas, exclude_session_id=str(get_session_id())
        )

        def _on_selected(session_id: str | None) -> None:
            if not session_id:
                return
            loaded = self._agent_bridge.resume_session(session_id)
            if loaded is None:
                transcript.append_system(
                    f"Could not resume {session_id}: no stored messages "
                    "(or the agent became busy).",
                    style="muted",
                )
                return
            # The resumed context size is unknown until its first turn —
            # zero the live-context measure instead of showing the
            # PREVIOUS session's percentage (C3a review F4).
            self.app_state.last_turn_input_tokens = 0
            self._render_resumed_messages(transcript, session_id, loaded)

        self.push_screen(
            ResumeConversation(entries=entries, hidden_count=hidden),
            callback=_on_selected,
        )

    def _render_resumed_messages(
        self, transcript: Transcript, session_id: str, messages: list[Any]
    ) -> None:
        """Replay a resumed transcript into the view, degraded-but-honest.

        Text content renders as normal user/assistant rows; non-text
        blocks (tool calls/results, thinking) are counted and summarized
        in one system row rather than fully re-rendered — the live
        renderers are event-driven and replaying tool events would
        re-trigger activity widgets for work that is not running.
        """

        transcript.clear_transcript()
        skipped_blocks = 0
        for message in messages:
            role = getattr(message, "role", "")
            content = getattr(message, "content", "")
            if isinstance(content, str):
                text = content
            else:
                parts: list[str] = []
                for block in content or []:
                    # The resume reader yields DATACLASS blocks (TextBlock
                    # et al., types/content_blocks.py) for known types and
                    # plain dicts for unknown ones — handle both shapes.
                    if isinstance(block, dict):
                        block_type = block.get("type")
                        block_text = block.get("text", "")
                    else:
                        block_type = getattr(block, "type", None)
                        block_text = getattr(block, "text", "")
                    if block_type == "text":
                        parts.append(str(block_text))
                    else:
                        skipped_blocks += 1
                text = "\n".join(p for p in parts if p)
            if not text:
                continue
            if role == "user":
                transcript.append_user(text)
            elif role == "assistant":
                transcript.append_assistant(text)
        note = f"Resumed session {session_id} ({len(messages)} messages)"
        if skipped_blocks:
            note += f"; {skipped_blocks} tool/attachment block(s) not re-rendered"
        transcript.append_system(note, style="muted")

    def _open_workflows_dialog(self, transcript: Transcript) -> None:
        registry = getattr(self.tool_context, "runtime_tasks", None)
        if registry is None:
            transcript.append_system("Workflows are unavailable.", style="muted")
            return
        from src.tui.screens.workflow_dialog import WorkflowsScreen

        self.push_screen(WorkflowsScreen(registry=registry))

    def _open_model_picker(self, transcript: Transcript) -> None:
        models = self._list_available_models()

        def _on_selected(model_id: str | None) -> None:
            if not model_id or model_id == self.model:
                self._restore_prompt_focus()
                return
            self.model = model_id
            try:
                if hasattr(self.provider, "model"):
                    setattr(self.provider, "model", model_id)
            except Exception:
                pass
            self.app_state.model = model_id
            transcript.append_system(f"Model switched to {model_id}.", style="muted")
            if self._repl_screen is not None:
                self._repl_screen.status_bar.refresh_identity(model=model_id)
            self.announcer.announce(f"Model switched to {model_id}.")
            self._restore_prompt_focus()

        self.announcer.announce("Opened model picker.", notify=False)
        self.push_screen(
            ModelPickerScreen(
                models=models,
                current_model=self.model,
            ),
            callback=_on_selected,
        )

    def _open_effort_picker(self, transcript: Transcript) -> None:
        current = getattr(self.app_state, "effort", None) or None

        def _on_selected(result: tuple[str | None, bool]) -> None:
            effort, persisted = result
            self._restore_prompt_focus()
            if not persisted:
                return
            # Persistence is handled by on_persist=set_effort below (the screen fires
            # it on selection only, not on cancel). The previous
            # ``setattr(self.app_state, "effort", effort)`` raised on the frozen
            # AppState (and set a field nothing reads), so it is removed.
            transcript.append_system(
                f"Reasoning effort set to {effort or 'auto'}.", style="muted"
            )
            self.announcer.announce(
                f"Reasoning effort set to {effort or 'auto'}."
            )

        self.announcer.announce("Opened effort picker.", notify=False)
        # D2: persist the pick like TS (effort.tsx always writes settings.effort). The
        # screen fires on_persist only on selection, not on cancel
        # (effort_picker.py:78-91), so Esc leaves the saved value untouched — matching TS.
        from src.config import set_effort

        self.push_screen(
            EffortPickerScreen(current=current, on_persist=set_effort),
            callback=_on_selected,
        )

    def _open_history_search(self, transcript: Transcript) -> None:
        records = self.history_store.recent(limit=500)
        entries = [
            HistoryEntry(prompt=r.prompt, timestamp=r.timestamp) for r in records
        ]
        if not entries:
            transcript.append_system(
                "History is empty — run some prompts first.", style="muted"
            )
            return

        def _on_selected(result: str | None) -> None:
            self._restore_prompt_focus()
            if not result:
                return
            if self._repl_screen is not None:
                self._repl_screen.prompt_input.set_value(result)
                self._repl_screen.prompt_input.focus_input()
            self.announcer.announce("Prompt restored from history.", notify=False)

        self.announcer.announce("Opened history search.", notify=False)
        self.push_screen(HistorySearchScreen(entries=entries), callback=_on_selected)

    def _open_cost_threshold(self, transcript: Transcript) -> None:
        tokens = self.app_state.usage.get("input_tokens", 0) + self.app_state.usage.get(
            "output_tokens", 0
        )
        # Rough estimate: $5 per 1M tokens. Phase 2 keeps this simple;
        # real per-model rates land with /cost refactor in Phase 3.
        estimate = (tokens / 1_000_000) * 5.0
        self.announcer.announce(
            f"Session cost estimate ${estimate:.2f}.", notify=False
        )
        self.push_screen(
            CostThresholdScreen(provider=self.provider_name, amount_usd=estimate),
            callback=lambda _=None: self._restore_prompt_focus(),
        )

    def _open_idle_return(self, transcript: Transcript) -> None:
        tokens = self.app_state.usage.get("input_tokens", 0)

        def _on_choice(action: str) -> None:
            if action == "clear":
                transcript.clear_transcript()
                self._agent_bridge.reset_advisor_dedup()
                transcript.append_system("Conversation cleared.", style="muted")
                self.announcer.announce("Conversation cleared.")
            elif action == "never":
                transcript.append_system(
                    "Idle-return prompts disabled for this session.", style="muted"
                )

        self.announcer.announce("Idle return prompt open.", notify=False)
        self.push_screen(
            IdleReturnScreen(
                idle_minutes=0,
                total_input_tokens=tokens,
                on_choice=_on_choice,
            ),
            callback=lambda _=None: self._restore_prompt_focus(),
        )

    def _open_theme_picker(self, transcript: Transcript) -> None:
        original_theme = self._theme_name

        def _on_preview(name: str | None) -> None:
            # Live-preview the highlighted theme; Esc restores the
            # original one so we don't leak an unintended swap.
            target = name or original_theme
            if target and target != self._theme_name:
                self.apply_theme(target, transcript=None)

        def _on_selected(name: str | None) -> None:
            self._restore_prompt_focus()
            if not name:
                # User cancelled — restore the starting theme.
                if self._theme_name != original_theme:
                    self.apply_theme(original_theme, transcript=None)
                return
            self.apply_theme(name, transcript=transcript)

        self.announcer.announce("Opened theme picker.", notify=False)
        # D2: persist the pick like TS (setTheme always saves to the global config).
        # The screen fires on_persist only on selection, not on cancel
        # (theme_picker.py:76-93), so Esc leaves the saved theme untouched — matching
        # TS (cancel ⇒ no setTheme). The live hot-swap still runs via _on_selected;
        # persistence (disk) and apply_theme (live palette) are complementary.
        from src.config import set_theme

        self.push_screen(
            ThemePickerScreen(
                themes=list_theme_names(),
                current=self._theme_name,
                on_persist=set_theme,
                on_preview=_on_preview,
            ),
            callback=_on_selected,
        )

    # ---- Phase 3 dialogs ----
    def _open_diff_dialog(self, transcript: Transcript) -> None:
        """Show pending file diffs, if the provider can surface them.

        We collect diffs from ``app_state.pending_diffs`` (populated by
        the file-edit tools) or the conversation's most-recent tool
        results. If nothing is available we drop a muted note instead
        of opening an empty dialog.
        """

        files: list[FileDiff] = []
        pending = getattr(self.app_state, "pending_diffs", None) or []
        for entry in pending:
            if isinstance(entry, FileDiff):
                files.append(entry)
            elif isinstance(entry, dict) and "patch" in entry and "path" in entry:
                files.append(FileDiff(path=str(entry["path"]), patch=str(entry["patch"])))

        if not files:
            transcript.append_system(
                "No pending diffs to display.", style="muted"
            )
            return

        self.announcer.announce(
            f"Diff dialog open. {len(files)} file(s) changed.", notify=False
        )
        self.push_screen(
            DiffDialogScreen(files=files),
            callback=lambda _=None: self._restore_prompt_focus(),
        )

    def _open_message_selector(self, transcript: Transcript) -> None:
        messages = self._collect_transcript_messages()
        if not messages:
            transcript.append_system(
                "Nothing to rewind — the transcript is empty.", style="muted"
            )
            return

        def _on_choice(result: tuple[int, str]) -> None:
            index, action = result
            self._restore_prompt_focus()
            if action == "cancel" or index < 0:
                return
            selected = next((m for m in messages if m.index == index), None)
            if selected is None:
                return
            if action == "restore" and self._repl_screen is not None:
                self._repl_screen.prompt_input.set_value(selected.text)
                self._repl_screen.prompt_input.focus_input()
                transcript.append_system(
                    f"Restored prompt from message #{index}.", style="muted"
                )
                self.announcer.announce(
                    f"Restored prompt from message {index}.", notify=False
                )
            elif action == "summarize":
                transcript.append_system(
                    f"Summarise-from-here requested for message #{index}.",
                    style="muted",
                )
                self.announcer.announce(
                    f"Summarise requested for message {index}."
                )

        self.announcer.announce(
            f"Message selector open. {len(messages)} message(s).", notify=False
        )
        self.push_screen(
            MessageSelectorScreen(messages=messages, on_choice=None),
            callback=_on_choice,
        )

    def _collect_transcript_messages(self) -> list[TranscriptMessage]:
        out: list[TranscriptMessage] = []
        try:
            conversation = self.session.conversation
            history = getattr(conversation, "messages", None) or []
        except Exception:
            return out

        idx = 0
        for msg in history:
            role = getattr(msg, "role", None)
            if role is None and isinstance(msg, dict):
                role = msg.get("role")
            content = getattr(msg, "content", None)
            if content is None and isinstance(msg, dict):
                content = msg.get("content")
            if not role or role not in ("user", "assistant"):
                continue
            text = _flatten_message_text(content)
            if not text.strip():
                continue
            out.append(TranscriptMessage(index=idx, kind=role, text=text))
            idx += 1
        return out

    def _open_mcp_list(self, transcript: Transcript) -> None:
        servers = self._collect_mcp_servers()
        if not servers:
            transcript.append_system(
                "No MCP servers configured.", style="muted"
            )
            return
        self.announcer.announce(
            f"MCP servers list open. {len(servers)} server(s).", notify=False
        )
        self.push_screen(
            McpListScreen(servers=servers),
            callback=lambda _=None: self._restore_prompt_focus(),
        )

    def _collect_mcp_servers(self) -> list[McpServer]:
        try:
            from src.config import load_config

            cfg = load_config() or {}
            raw = cfg.get("mcp_servers") or cfg.get("mcpServers") or {}
        except Exception:
            raw = {}
        servers: list[McpServer] = []
        if isinstance(raw, dict):
            for server_id, entry in raw.items():
                name = entry.get("name", server_id) if isinstance(entry, dict) else server_id
                status = entry.get("status", "disconnected") if isinstance(entry, dict) else "disconnected"
                tools = entry.get("tools", []) if isinstance(entry, dict) else []
                servers.append(
                    McpServer(
                        id=str(server_id),
                        name=str(name),
                        status=status,  # type: ignore[arg-type]
                        tools=list(tools) if isinstance(tools, list) else [],
                    )
                )
        return servers

    def _open_tasks_dialog(self, transcript: Transcript) -> None:
        # The background task panel lives on the REPL screen; the slash
        # command just routes focus to it rather than stacking a modal.
        if self._repl_screen is not None and hasattr(
            self._repl_screen, "focus_task_panel"
        ):
            try:
                self._repl_screen.focus_task_panel()
                return
            except Exception:
                pass
        transcript.append_system(
            "Task panel focus is not available in this build.", style="muted"
        )

    def _confirm_exit(self, transcript: Transcript) -> None:
        """Push :class:`ExitFlowScreen` instead of quitting immediately."""

        def _on_choice(action: str) -> None:
            if action == "quit":
                self.exit()
            elif action == "quit-clear":
                transcript.clear_transcript()
                self.exit()
            else:  # "cancel"
                self._restore_prompt_focus()

        self.announcer.announce(
            "Exit confirmation open.", level="assertive", notify=False
        )
        self.push_screen(
            ExitFlowScreen(
                has_inflight_work=self.app_state.is_thinking,
                on_choice=_on_choice,
            )
        )

    # ---- focus helpers ----
    def _restore_prompt_focus(self) -> None:
        """Return keyboard focus to the prompt input after a modal closes.

        Modals that dismiss via the ``callback=`` path don't
        automatically restore focus on the previous screen, so we do
        it explicitly for every dialog close. No-op when the REPL
        screen hasn't mounted yet.
        """

        if self._repl_screen is None:
            return
        try:
            self._repl_screen.prompt_input.focus_input()
        except Exception:
            pass

    # ---- theme live-switch ----
    def apply_theme(self, name: str, *, transcript: Transcript | None = None) -> None:
        """Hot-swap the palette and refresh the stylesheet overrides."""

        self.palette = get_palette(name)
        self._theme_name = name
        try:
            self.stylesheet.add_source(
                textual_css_overrides(self.palette),
                path="palette-overrides",
            )
            self.stylesheet.parse()
            self.refresh(layout=True)
        except Exception:
            pass
        if transcript is not None:
            transcript.append_system(f"Theme set to {name}.", style="muted")
            self.announcer.announce(f"Theme set to {name}.", notify=False)

    # ---- model discovery ----
    def _list_available_models(self) -> list[str]:
        """Return a best-effort list of models for the active provider."""

        try:
            if hasattr(self.provider, "list_models"):
                models = list(self.provider.list_models() or [])  # type: ignore[attr-defined]
                if models:
                    return [str(m) for m in models]
        except Exception:
            pass
        try:
            from src.config import get_provider_config

            cfg = get_provider_config(self.provider_name) or {}
            models = cfg.get("models")
            if isinstance(models, list) and models:
                return [str(m) for m in models]
            default = cfg.get("default_model")
            if default:
                return [str(default)]
        except Exception:
            pass
        # Fallback: just the active model.
        return [self.model or "default"]

    def _register_skills_once(self) -> None:
        """Register builtins + user-invocable skills into the GLOBAL registry,
        exactly once (P0-6 Option B / Phase 3.5).

        ``dispatch_registry_command`` executes through
        ``execute_command_async``, which resolves names from the GLOBAL
        registry — so skills must land there, not in a TUI-local registry.
        Builtins are registered first so the shadowing guard in
        ``load_and_register_skills`` lets them win on a name collision. Idempotent
        via ``self._skills_registered``; failures must never block command
        dispatch (builtins still resolve from the suggestion-builder path)."""
        if self._skills_registered:
            return
        try:
            from src.command_system import (
                load_and_register_skills,
                register_builtin_commands,
            )

            register_builtin_commands(None)
            load_and_register_skills(registry=None)
            # Saved workflows (.claude/workflows/*.py) as /<name> commands —
            # same global-only registration + shadowing guard as skills.
            from src.command_system.workflows_integration import (
                load_and_register_workflows,
            )

            load_and_register_workflows(registry=None)
        except Exception:
            pass
        self._skills_registered = True

    def _ensure_command_context(self) -> Any:
        self._register_skills_once()
        if self._command_context is not None:
            return self._command_context
        try:
            from src.command_system.engine import create_command_context
            from src.cost_tracker import CostTracker
            from src.history import HistoryLog
            from src.state.app_state import create_app_state_store
            from src.tui.ui_host import TextualUIHost

            if self._app_state_store is None:
                self._app_state_store = create_app_state_store()

            self._command_context = create_command_context(
                workspace_root=self.workspace_root,
                conversation=self.session.conversation,
                cost_tracker=CostTracker(),
                history=HistoryLog(),
                provider=self.provider,
                app_state_store=self._app_state_store,
                ui=TextualUIHost(self),
                # P0-6 Option B: lets a SkillPromptCommand render via the same
                # _run_markdown_skill path the Skill tool uses (session id +
                # gated shell-exec) rather than lossy bare substitution.
                tool_context=self.tool_context,
            )
        except Exception:
            self._command_context = None
        return self._command_context

    # ---- agent loop plumbing ----
    def submit_to_agent(self, prompt: str) -> None:
        try:
            self.history_store.append(prompt)
        except Exception:
            pass
        submitted = self._agent_bridge.submit(prompt)
        if not submitted:
            # If the bridge is busy we queue the prompt for the next
            # turn so the user can keep typing. Phase 2 adds a visible
            # queued-prompts pill in the status line.
            self.app_state.queued_prompts.append(prompt)

    def on_cancel_requested(self, _: CancelRequested) -> None:
        """ESC from the prompt — cancel the in-flight agent run, if any."""

        if self._agent_bridge.cancel():
            self.announcer.announce("Cancelling…", level="assertive", notify=False)

    # ---- helpers ----
    def _build_default_tool_context(self) -> ToolContext:
        ctx = ToolContext(workspace_root=self.workspace_root)
        ctx.ask_user = lambda questions: {
            q["id"]: "" for q in questions if isinstance(q, dict) and "id" in q
        }
        return ctx

    def _slash_command_words(self) -> list[str]:
        return build_command_words(self.workspace_root, self.tool_context)

    def _slash_command_suggestions(self) -> list[CommandSuggestion]:
        return build_command_suggestions(self.workspace_root, self.tool_context)

    def _post_to_screen(self, message: Any) -> None:
        target = self._repl_screen or self
        try:
            target.post_message(message)
        except Exception:
            pass