hermes-agent/hermes_cli/cli_modal_mixin.py

"""Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command palette,
slash-confirm, external editor.

Mixin split out of ``cli.py``; bound onto ``HermesCLI`` via the MRO. cli.py-internal symbols are
imported LAZILY inside each method — the mixin never imports ``cli`` at module load time (cycle).
"""

from __future__ import annotations

import json
import queue
import sys
import threading
import time as _time
import webbrowser

from agent.i18n import t
from hermes_cli.callbacks import prompt_for_secret
from typing import Optional

_TIMED_OUT = object()  # sentinel returned by _poll_modal_queue when the deadline passes

# Typed answers accepted by the slash-confirm modal, mapped onto the canonical choice values.
_CONFIRM_ALIASES = {
    "1": "once", "once": "once", "approve": "once", "yes": "once", "y": "once", "ok": "once",
    "2": "always", "always": "always", "remember": "always",
    "3": "cancel", "cancel": "cancel", "nevermind": "cancel", "no": "cancel", "n": "cancel"}

# Approval result -> catalog key of the outcome label shown in the persisted prompt summary.
# Resolved at call time (``_approval_outcome_label``) so the active language applies.
_APPROVAL_OUTCOME_KEYS = {
    "once": "cli.approval.outcome_once",
    "session": "cli.approval.outcome_session",
    "always": "cli.approval.outcome_always",
    "deny": "cli.approval.outcome_deny"}


def _approval_outcome_label(result) -> str:
    key = _APPROVAL_OUTCOME_KEYS.get(result)
    return t(key) if key else str(result)


def _approval_gate_on(key: str) -> bool:
    """Read ``approvals.<key>`` (default on); any load failure keeps the prompt enabled."""
    from cli import load_cli_config
    try:
        cfg = load_cli_config()
        approvals = cfg.get("approvals") if isinstance(cfg, dict) else None
        if isinstance(approvals, dict):
            return bool(approvals.get(key, True))
    except Exception:
        pass
    return True


def _gated_confirm(self, command, key, *, title, detail, choices, unchanged, always_msg, once_verb=None,
                   persist_failed_msg=None):
    """Shared once/always/cancel confirm behind ``approvals.<key>`` (destructive slash, /reload-mcp).

    Returns ``"once"`` without prompting when the gate is off; ``None`` on cancel / no input /
    unrecognized answer (already reported to the user). Picking "always" persists the opt-out.
    ``persist_failed_msg`` is the full sentence printed when the opt-out could not be saved;
    ``once_verb`` is the legacy fragment spliced into the default sentence.
    """
    from cli import save_config_value
    if not _approval_gate_on(key):
        return "once"
    raw = self._prompt_text_input_modal(title=title, detail=detail, choices=choices)
    if raw is None:
        print(t("cli.modal.cancelled_no_input", command=command))
        return None
    choice = self._normalize_slash_confirm_choice(raw, choices)
    if choice is None:
        print(t("cli.modal.unrecognized_choice", raw=raw, command=command))
        return None
    if choice == "cancel":
        print(t("cli.modal.cancelled", command=command, unchanged=unchanged))
        return None
    if choice == "always":
        if save_config_value(f"approvals.{key}", False):
            print(always_msg)
            print(t("cli.modal.re_enable_hint", setting=key))
        else:
            print(persist_failed_msg or t("cli.modal.persist_opt_out_failed", once_verb=once_verb or ""))
    return choice


class CLIModalMixin:
    """Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command
    palette, slash-confirm, external editor."""

    _connection_state = None

    def _open_external_editor(self, buffer=None) -> bool:
        """Open the active input buffer in an external editor."""
        from cli import _DIM, _RST, _cprint
        app = getattr(self, "_app", None)
        if not app:
            _cprint(f"{_DIM}{t('cli.editor.only_interactive')}{_RST}")
            return False
        if self._command_running:
            _cprint(f"{_DIM}{t('cli.editor.wait_for_command')}{_RST}")
            return False
        if (self._sudo_state or self._secret_state or self._approval_state
                or getattr(self, "_slash_confirm_state", None) or self._clarify_state
                or self._connection_state):
            _cprint(f"{_DIM}{t('cli.editor.finish_active_prompt')}{_RST}")
            return False
        target_buffer = buffer or getattr(app, "current_buffer", None)
        if target_buffer is None:
            _cprint(f"{_DIM}{t('cli.editor.no_input_buffer')}{_RST}")
            return False
        try:
            # Inline pastes so the editor sees real content; set the skip flag unconditionally so
            # the editor-close text-change doesn't re-collapse it.
            self._inline_pastes(target_buffer)
            self._skip_paste_collapse = True
            # Submission here is driven by the custom `enter` keybinding, NOT the buffer's
            # accept_handler, so validate_and_handle can't route through it; chain a done-callback
            # that re-uses the real submit pipeline (TUI Ctrl+G parity: save == send).
            task = target_buffer.open_in_editor(validate_and_handle=False)
            if task is not None and hasattr(task, "add_done_callback"):
                task.add_done_callback(lambda _t, b=target_buffer: self._submit_editor_buffer(b))
            return True
        except Exception as exc:
            _cprint(f"{_DIM}{t('cli.editor.open_failed', error=exc)}{_RST}")
            return False

    def _submit_editor_buffer(self, buffer) -> None:
        """Submit the draft an external editor left in ``buffer`` (Ctrl+G done-callback), mirroring
        the `enter` keybinding: empty save ignored, bang/slash dispatched, else queued. Runs on the
        prompt_toolkit loop, so it must stay cheap/non-blocking."""
        from cli import _DIM, _RST, _cprint, _looks_like_slash_command
        try:
            text = (getattr(buffer, "text", "") or "").strip()
        except Exception:
            return
        if not text:
            return

        app = getattr(self, "_app", None)

        def _done() -> None:
            self._reset_input_buffer(buffer)
            if app is not None:
                app.invalidate()

        # `!<command>` shell mode is checked before slash dispatch, matching the Enter path.
        try:
            if self.handle_bang_shell(text):
                _done()
                return
        except Exception as exc:
            _cprint(f"  {_DIM}{t('cli.editor.shell_command_failed', error=exc)}{_RST}")
            _done()
            return

        if _looks_like_slash_command(text):
            try:
                if not self.process_command(text):
                    self._should_exit = True
                    if app is not None and app.is_running:
                        app.exit()
            except Exception as exc:
                _cprint(f"  {_DIM}{t('cli.editor.command_failed', error=exc)}{_RST}")
            finally:
                _done()
            return

        if self._agent_running:
            # Agent busy → honour the configured busy-input behaviour (interrupt/steer remain
            # reachable via the normal Enter path).
            if self.busy_input_mode == "interrupt":
                self._interrupt_queue.put(text)
            else:
                self._pending_input.put(text)
            preview = text[:80] + ("..." if len(text) > 80 else "")
            _cprint(f"  {t('cli.editor.queued_next_turn', preview=preview)}")
        else:
            self._pending_input.put(text)
        _done()

    def _inline_pastes(self, buffer) -> None:
        """Replace collapsed ``[Pasted text #N -> file]`` placeholders in ``buffer`` with real text.

        History recall and the external editor need the content (the file may be gone or on another
        machine); inlining before ``reset(append_to_history=True)`` also lets prompt_toolkit persist
        it. Sets ``_skip_paste_collapse`` so the ensuing text-change doesn't re-collapse it.
        """
        from cli import logger
        try:
            existing = getattr(buffer, "text", "")
            expanded = self._expand_paste_references(existing)
            if expanded != existing and hasattr(buffer, "text"):
                self._skip_paste_collapse = True
                buffer.text = expanded
                if hasattr(buffer, "cursor_position"):
                    buffer.cursor_position = len(expanded)
        except Exception:
            logger.debug("Failed to inline paste placeholders", exc_info=True)

    def _reset_input_buffer(self, buffer) -> None:
        """Clear an input buffer after a programmatic submit (best-effort)."""
        try:
            buffer.reset(append_to_history=True)
        except Exception:
            try:
                buffer.text = ""
            except Exception:
                pass

    def _prefill_input_buffer(self, text: str) -> None:
        """Place ``text`` in the active prompt_toolkit buffer, editable."""
        from cli import logger
        app = getattr(self, "_app", None)
        if app is None:
            return
        try:
            buf = app.current_buffer
            buf.text = text
            if hasattr(buf, "cursor_position"):
                buf.cursor_position = len(text)
            app.invalidate()
        except Exception as e:
            logger.debug("undo: prefill buffer failed: %s", e)

    def _prompt_text_input(self, prompt_text: str) -> str | None:
        """Prompt for free-text input safely inside or outside prompt_toolkit.

        ``run_in_terminal`` only works on the main-thread loop; on the ``process_loop`` daemon
        thread a bare ``input()`` would block forever on loop-owned stdin, so with an app running
        off-main we cancel cleanly (None) — mirroring ``_stdin_fallback`` in the modal prompt.

        Mirrors the thread-aware guard in ``_run_curses_picker``: ``run_in_terminal`` returns a coroutine
        that must be awaited by the prompt_toolkit event loop, which only exists on the main thread. Slash
        commands are dispatched from the ``process_loop`` daemon thread (see issue #23185), so calling
        ``run_in_terminal`` from there orphans the coroutine — ``_ask`` never runs, and user keystrokes leak
        into the composer instead. Fall back to a direct ``input()`` when we're off the main thread.
        """
        result = [None]

        def _ask():
            try:
                result[0] = input(prompt_text).strip() or None
            except (KeyboardInterrupt, EOFError):
                pass

        in_main_thread = threading.current_thread() is threading.main_thread()
        # Slash-worker guard (#23185 / billing auto-reload hang): when a prompt_toolkit app is running but
        # we're on a non-main thread (the process_loop / TUI slash-worker daemon thread), stdin is owned by
        # the event loop / JSON-RPC pipe. A bare input() there blocks forever until the worker's 45s timeout
        # fires. We cannot safely prompt off the main thread, so cancel cleanly (None) instead of hanging —
        # mirrors the _stdin_fallback discipline in _prompt_text_input_modal.
        if self._app and not in_main_thread:
            self._invalidate()
            return None

        if self._app and in_main_thread:
            from prompt_toolkit.application import run_in_terminal
            was_visible = self._status_bar_visible
            self._status_bar_visible = False
            self._app.invalidate()
            try:
                run_in_terminal(_ask)
            except Exception:
                # WSL / Warp / some emulators silently drop the scheduled coroutine — fall back to
                # a direct input() so keystrokes don't leak into the agent buffer.
                try:
                    _ask()
                except Exception:
                    pass
            finally:
                self._status_bar_visible = was_visible
                self._app.invalidate()
        else:
            _ask()
        return result[0]

    def _poll_modal_queue(self, response_queue, deadline_attr, *, refresh=1.0, paint=None):
        """Block until a value lands on ``response_queue`` or ``self.<deadline_attr>`` passes
        (``None`` deadline = unlimited). Returns the value or ``_TIMED_OUT``.

        Repaints every ``refresh`` seconds (``0`` = on every idle tick) so countdown hints stay
        live; ``paint`` defaults to ``_paint_now`` — modal prompts must bypass the ``_invalidate``
        throttle/resize guard or the panel can be dropped and time out unseen.
        """
        paint = paint or self._paint_now
        last = _time.monotonic()
        while True:
            try:
                return response_queue.get(timeout=1)
            except queue.Empty:
                deadline = getattr(self, deadline_attr)
                if deadline is not None and deadline - _time.monotonic() <= 0:
                    return _TIMED_OUT
                now = _time.monotonic()
                if now - last >= refresh:
                    last = now
                    paint()

    def _prompt_text_input_modal(
        self, *, title: str, detail: str, choices: list[tuple[str, str, str]], timeout: float = 120
    ) -> str | None:
        """Slash-command confirmation through the prompt_toolkit composer (raw input() fought
        prompt_toolkit's stdin ownership: prompt above the TUI, Enter read as EOF). All platforms
        drive the modal via ``self._app.loop`` + ``call_soon_threadsafe``; raw ``input()`` is kept
        only for the safe cases (no app, no loop, scheduling failure) — on Windows a non-main-thread
        input() deadlocks against prompt_toolkit, so that case cancels instead.

        **Platform note (Windows — issue #33961):** Earlier code bypassed the modal on ``sys.platform ==
        "win32"`` and fell back to a raw ``input()`` prompt. When the confirm was triggered from the
        ``process_loop`` daemon thread (the normal case) that ``input()`` ran off the main thread and
        deadlocked against prompt_toolkit's stdin ownership — the user saw a frozen cursor and Ctrl-C was
        swallowed (bare ``/reset`` froze; ``/reset now`` worked only because it skips the prompt entirely).
        """
        if not choices:
            return None
        if not getattr(self, "_app", None):
            return self._prompt_text_input(t("cli.modal.choice_prompt"))

        try:
            app_loop = self._app.loop
        except Exception:
            app_loop = None
        in_main_thread = threading.current_thread() is threading.main_thread()

        def _stdin_fallback() -> str | None:
            # On native Windows a raw input() from a non-main thread deadlocks against prompt_toolkit's
            # stdin ownership (#33961). With an app running we cannot safely prompt off the main thread, so
            # cancel cleanly (None) rather than hang the terminal.
            if sys.platform == "win32" and not in_main_thread:
                self._invalidate()
                return None
            return self._prompt_text_input(t("cli.modal.choice_prompt"))

        if not in_main_thread and app_loop is None:
            return _stdin_fallback()

        response_queue = queue.Queue()

        def _setup_modal() -> None:
            self._capture_modal_input_snapshot()
            self._slash_confirm_state = {
                "title": title,
                "detail": detail,
                "choices": choices,
                "selected": 0,
                "response_queue": response_queue}
            self._slash_confirm_deadline = _time.monotonic() + timeout
            self._invalidate()

        def _teardown_modal() -> None:
            self._slash_confirm_state = None
            self._slash_confirm_deadline = 0
            self._restore_modal_input_snapshot()
            self._invalidate()

        def _run_on_app_loop(fn) -> bool:
            if in_main_thread or app_loop is None:
                fn()
                return True
            ready = threading.Event()

            def _wrapped() -> None:
                try:
                    fn()
                finally:
                    ready.set()

            try:
                app_loop.call_soon_threadsafe(_wrapped)
            except Exception:
                return False
            return ready.wait(timeout=5)

        if not _run_on_app_loop(_setup_modal):
            return _stdin_fallback()
        try:
            result = self._poll_modal_queue(
                response_queue, "_slash_confirm_deadline", refresh=5.0, paint=self._invalidate)
            if result is not _TIMED_OUT:
                _run_on_app_loop(_teardown_modal)
                return result
        finally:
            if self._slash_confirm_state is not None:
                _run_on_app_loop(_teardown_modal)
        return None

    def _submit_slash_confirm_response(self, value: str | None) -> None:
        state = self._slash_confirm_state
        if not state:
            return
        state["response_queue"].put(value)
        self._slash_confirm_state = None
        self._slash_confirm_deadline = 0
        self._invalidate()

    def _normalize_slash_confirm_choice(
        self, raw: str | None, choices: list[tuple[str, str, str]]) -> str | None:
        if raw is None:
            return None
        choice_raw = raw.strip().lower()
        if not choice_raw:
            return None
        allowed = {choice[0] for choice in choices}
        normalized = _CONFIRM_ALIASES.get(choice_raw)
        if normalized in allowed:
            return normalized
        if choice_raw in allowed:
            return choice_raw
        return None

    def _build_command_palette_entries(self) -> list:
        """Flat (command, category, desc) rows for the Ctrl+P palette: the COMMAND_REGISTRY behind
        /help filtered to this surface, plus installed skill commands. Selecting inserts the exact
        command string — never a fuzzy resolution."""
        from cli import _ensure_skill_commands
        from hermes_cli.commands import COMMANDS_BY_CATEGORY

        entries: list[tuple[str, str, str]] = []
        for category, commands in COMMANDS_BY_CATEGORY.items():
            for cmd, desc in commands.items():
                if self._command_available(cmd):
                    entries.append((cmd, category, desc))
        try:
            for cmd, info in sorted(_ensure_skill_commands().items()):
                entries.append((cmd, t("cli.palette.category_skill"), info.get("description", "")))
        except Exception:
            pass
        return entries

    def _open_command_palette(self) -> None:
        """Open the Ctrl+P fuzzy command palette modal (never stacked over another modal)."""
        if getattr(self, "_command_palette_state", None):
            return
        if (self._model_picker_state or self._clarify_state or self._approval_state
                or self._slash_confirm_state or self._sudo_state or self._secret_state
                or self._connection_state):
            return
        self._capture_modal_input_snapshot()
        self._command_palette_state = {
            "entries": self._build_command_palette_entries(),
            "filter": "",
            "selected": 0,
            "_scroll_offset": 0}
        self._invalidate(min_interval=0.0)

    def _close_command_palette(self) -> None:
        self._command_palette_state = None
        self._restore_modal_input_snapshot()
        self._invalidate(min_interval=0.0)

    def _command_palette_visible_entries(self) -> list:
        """Rows matching the active filter, ranked command-name-first (a bare subsequence over
        "cmd category desc" is uselessly permissive — "steer" would match 130+ rows via text):
        0 exact command, 1 command startswith, 2 substring in command, 3 subsequence in command,
        4 substring in description. Non-matches are dropped; ties keep registry order."""
        state = self._command_palette_state or {}
        entries = state.get("entries") or []
        q = (state.get("filter", "") or "").strip().lower()
        if not q:
            return list(entries)

        def _subseq(needle: str, hay: str) -> bool:
            it = iter(hay)
            return all(ch in it for ch in needle)

        qn = q.lstrip("/")
        ranked = []
        for order, row in enumerate(entries):
            cmd, _cat, desc = row
            name = cmd.lower().lstrip("/")
            desc_l = (desc or "").lower()
            if name == qn:
                rank = 0
            elif name.startswith(qn):
                rank = 1
            elif qn in name:
                rank = 2
            elif _subseq(qn, name):
                rank = 3
            elif q in desc_l:
                rank = 4
            else:
                continue
            ranked.append((rank, order, row))
        ranked.sort(key=lambda t: (t[0], t[1]))
        return [row for (_r, _o, row) in ranked]

    def _handle_command_palette_selection(self) -> None:
        """Prefill the selected command into the composer — never auto-run (many take args)."""
        from cli import logger
        state = self._command_palette_state
        if not state:
            return
        rows = self._command_palette_visible_entries()
        selected = state.get("selected", 0)
        if not (0 <= selected < len(rows)):
            self._close_command_palette()
            return
        cmd = rows[selected][0]
        self._close_command_palette()
        try:
            app = getattr(self, "_app", None)
            if app is not None:
                buf = app.current_buffer
                buf.text = cmd + " "
                buf.cursor_position = len(buf.text)
                self._invalidate(min_interval=0.0)
        except Exception:
            logger.debug("command palette prefill failed", exc_info=True)

    @classmethod
    def _split_destructive_skip(cls, cmd_text: Optional[str]) -> tuple[str, bool]:
        """Split inline-skip tokens out of a destructive slash command → ``(remainder, skip)``.

        ``remainder`` is the text minus the leading "/cmd" word and any skip tokens; ``skip`` is
        True iff one was found: "/reset now" -> ("", True); "/reset --yes My title" ->
        ("My title", True); "/new My title" -> ("My title", False).
        """
        tokens = (cmd_text or "").strip().split()
        if not tokens:
            return "", False
        if tokens[0].startswith("/"):
            tokens = tokens[1:]
        kept = [tok for tok in tokens if tok.lower() not in cls._DESTRUCTIVE_SKIP_TOKENS]
        return " ".join(kept), len(kept) != len(tokens)

    def _confirm_destructive_slash(
        self, command: str, detail: str, cmd_original: Optional[str] = None) -> Optional[str]:
        """Confirm a destructive slash command (``/clear``, ``/new``/``/reset``, ``/undo``): returns
        ``"once"``, ``"always"`` (persists the opt-out) or ``None`` (cancelled). Gate off → "once"
        silently; ``now`` / ``--yes`` / ``-y`` in ``cmd_original`` bypasses the modal (callers strip
        the tokens via :meth:`_split_destructive_skip`).

        Inline-skip: if ``cmd_original`` contains ``now``, ``--yes``, or ``-y`` as an argument (e.g.
        ``/reset now``, ``/new --yes My title``), the modal is bypassed and ``"once"`` is returned
        immediately. This is an escape hatch for non-interactive use and for the degraded path where the
        modal can't be marshaled onto the app loop (native Windows itself now drives the modal normally —
        see #33961). Callers are responsible for stripping the skip tokens from any remaining argument
        parsing (see :meth:`_split_destructive_skip`).
        """
        if cmd_original and self._split_destructive_skip(cmd_original)[1]:
            return "once"
        return _gated_confirm(
            self, command, "destructive_slash_confirm",
            title=t("cli.modal.destructive_title", command=command),
            detail=detail,
            choices=[
                ("once", t("cli.modal.approve_once"), t("cli.modal.approve_once_detail")),
                ("always", t("cli.modal.always_approve"), t("cli.modal.always_approve_detail")),
                ("cancel", t("cli.modal.cancel"), t("cli.modal.cancel_detail"))],
            unchanged=t("cli.modal.conversation_unchanged"),
            always_msg=t("cli.modal.destructive_always_msg"),
            persist_failed_msg=t("cli.modal.persist_failed_proceeding"))

    def _ring_bell(self, prompt: bool = False, context: str = "", detail: str = "") -> None:
        """Terminal bell (\\a) gated by ``display.bell_on_prompt`` (``prompt=True``, blocking modals)
        or ``display.bell_on_complete`` (end of turn); works over SSH. The same flag also emits the
        OSC 9 / Warp OSC 777 desktop notification; ``context`` is the short notification body."""
        flag = "bell_on_prompt" if prompt else "bell_on_complete"
        if not getattr(self, flag, False) or getattr(self, "_terminal_io_broken", False):
            return
        from hermes_cli.cli_terminal_mixin import _run_on_app_loop, _write_terminal_sequence
        from hermes_cli.terminal_notify import notification_sequence, write_tty
        body = context or (t("cli.modal.bell_input_needed") if prompt else t("cli.modal.bell_turn_complete"))
        try:
            seq = "\a" + notification_sequence(
                body, prompt=prompt, session_id=getattr(self, "session_id", "") or "", detail=detail)
        except Exception:
            return
        app = getattr(self, "_app", None)
        if app is None or not getattr(app, "_is_running", False):
            write_tty(seq)
            return

        # Agent thread. The loop thread may be mid-write of a 12 KB kitty pet frame that the tty
        # drains ~1 KB at a time; a second writer on the same tty (/dev/tty, sys.stdout) splices
        # in, the foreign ESC aborts the APC, and the terminal paints the rest of the payload as
        # base64 at the input cursor. Serialize behind the renderer instead.
        def _emit() -> None:
            try:
                _write_terminal_sequence(app, seq)
            except (OSError, ValueError):
                pass  # dead tty: same fail-quiet as _pet_flush_kitty_frame

        _run_on_app_loop(app, _emit)

    def _clarify_teardown(self) -> None:
        self._clarify_state = None
        self._clarify_freetext = False
        self._clarify_deadline = None
        self._clarify_multi_base = None
        self._paint_now()

    # --- Connection setup --------------------------------------------------
    def _connection_operation(self, payload):
        from tools.connectors import live

        session_key = getattr(self, "session_id", "") or ""
        op_id = str(payload.get("op_id") or "")
        return live.get(session_key, op_id) or live.current(session_key)

    def _connection_install_hook(self) -> bool:
        from tools.connectors.operation import ConnectionOperation

        if ConnectionOperation.on_change is not None:
            return False
        ConnectionOperation.on_change = self._connection_on_change
        return True

    def _connection_restore_hook(self) -> None:
        from tools.connectors.operation import ConnectionOperation

        if ConnectionOperation.on_change == self._connection_on_change:
            ConnectionOperation.on_change = None

    def _connection_close(self) -> None:
        self._connection_state = None
        self._connection_restore_hook()
        self._restore_modal_input_snapshot()
        self._paint_now()

    @staticmethod
    def _connection_fields(target) -> list[dict]:
        return [dict(field) for field in target.get("required_env") or () if isinstance(field, dict)]

    @staticmethod
    def _connection_opening_phase(target) -> str:
        """The phase a target opens in. A pending install or enable waits for the user's Connect
        even with no fields to fill; a pending authorize is the backend still minting the link."""
        target_state = target.get("state")
        if target_state == "initiated" and target.get("connect_url"):
            return "url"
        if target_state == "failed":
            return "failed"
        if target_state == "pending" and target.get("action") != "authorize":
            return "form"
        return "waiting"

    def _connection_show_target(self, payload, index: int) -> None:
        targets = payload.get("targets") or []
        if not targets:
            self._connection_close()
            return
        index = max(0, min(index, len(targets) - 1))
        target = targets[index]
        fields = self._connection_fields(target)
        previous = self._connection_state or {}
        drafts = previous.setdefault("drafts", {})
        target_draft = drafts.setdefault(target.get("name", ""), {})
        for field in fields:
            if "type" not in field:
                field["type"] = "secret" if field.get("secret") else "plain"
            if field.get("type") != "secret" and field.get("name") not in target_draft:
                target_draft[field.get("name")] = str(field.get("default") or "")
        self._connection_state = {
            **previous,
            "payload": payload,
            "target_index": index,
            "target": target,
            "fields": fields,
            "field_index": 0,
            "selected": 0,
            "phase": self._connection_opening_phase(target),
            "drafts": drafts,
        }
        self._connection_sync_input_buffer()
        self._paint_now()

    def _connection_active_field_is_secret(self) -> bool:
        state = self._connection_state
        if not state or state.get("phase") not in {"form", "failed"}:
            return False
        fields = state.get("fields") or []
        index = state.get("field_index", 0)
        return 0 <= index < len(fields) and fields[index].get("type") == "secret"

    def _connection_prefill_text(self) -> str:
        state = self._connection_state
        if not state or self._connection_active_field_is_secret():
            return ""
        fields = state.get("fields") or []
        index = state.get("field_index", 0)
        if not (0 <= index < len(fields)):
            return ""
        field = fields[index]
        target_name = state.get("target", {}).get("name", "")
        return str(state.get("drafts", {}).get(target_name, {}).get(field.get("name"), ""))

    def _connection_sync_input_buffer(self) -> None:
        app = getattr(self, "_app", None)
        if app is None:
            return

        def _apply() -> None:
            try:
                buf = app.current_buffer
                buf.text = self._connection_prefill_text()
                buf.cursor_position = len(buf.text)
            except Exception:
                pass

        loop = getattr(app, "loop", None)
        if loop is not None and threading.current_thread() is not threading.main_thread():
            try:
                loop.call_soon_threadsafe(_apply)
                return
            except Exception:
                pass
        _apply()

    def _connection_callback(self, payload):
        if not isinstance(payload, dict):
            return None
        self._capture_modal_input_snapshot()
        installed = self._connection_install_hook()
        self._connection_show_target(payload, 0)
        state = self._connection_state
        if state is None:
            return None
        state["owns_hook"] = installed
        state["tool_thread_id"] = threading.current_thread().ident
        if state["phase"] != "waiting":
            self._ring_bell(prompt=True, context=t("cli.connect.bell_context"))
        return None

    def _connection_answer(self, *, approve: bool) -> None:
        state = self._connection_state
        if not state:
            return
        target = state["target"]
        operation = self._connection_operation(state["payload"])
        if operation is None:
            self._connection_close()
            return
        name = str(target.get("name") or "")
        answer = {"name": name, "status": "approved" if approve else "skipped"}
        if approve:
            answer["env"] = dict(state.get("drafts", {}).get(name, {}))
        from tools.connectors.mcp import apply_answer

        # The backend applies the answer on this thread, and its change hook sets the next phase
        # (the URL step, the form again for a missing field) before apply_answer returns. Set the
        # waiting phase first so it cannot overwrite that.
        state["phase"] = "waiting"
        apply_answer(operation, json.dumps({"targets": [answer]}))
        self._paint_now()

    def _connection_retry(self) -> None:
        state = self._connection_state
        if not state:
            return
        operation = self._connection_operation(state["payload"])
        if operation is None:
            self._connection_close()
            return
        target = state["target"]
        if target.get("kind") == "connector" and target.get("state") in {"failed", "expired"}:
            from tools.connectors.run import reissue

            state["phase"] = "waiting"
            if reissue(operation, [str(target.get("name") or "")]) is not None:
                target["detail"] = t("cli.connect.cannot_restart")
                state["phase"] = "failed"
            self._paint_now()
            return
        if target.get("state") in {"pending", "failed", "expired"}:
            # Connect on a failed row is the same attempt with the values now in the draft.
            self._connection_answer(approve=True)
            return
        from tools.connectors.mcp import retry

        state["phase"] = "waiting"
        retry(operation, [str(target.get("name") or "")])
        self._paint_now()

    def _connection_continue(self) -> None:
        state = self._connection_state
        if not state:
            return
        operation = self._connection_operation(state["payload"])
        if operation is not None:
            from tools.connectors.mcp import apply_answer

            apply_answer(operation, json.dumps({"settled_by": "continue"}))
        self._connection_close()

    def _connection_open_url(self) -> None:
        state = self._connection_state
        if state and state.get("phase") == "url" and state["target"].get("connect_url"):
            webbrowser.open(state["target"]["connect_url"])

    def _connection_cancel(self) -> None:
        self._connection_answer(approve=False)

    def _connection_interrupt(self) -> None:
        state = self._connection_state
        if state:
            from tools.interrupt import set_interrupt

            operation = self._connection_operation(state["payload"])
            set_interrupt(True, thread_id=state.get("tool_thread_id"))
            if operation is not None:
                operation.wake.set()
        self._connection_close()

    def _connection_on_change(self, operation, _change, snapshot) -> None:
        state = self._connection_state
        if not state or operation.op_id != state["payload"].get("op_id"):
            return
        if snapshot.get("settled_at") is not None:
            self._connection_close()
            return
        active_name = state["target"].get("name")
        target = next((item for item in snapshot.get("targets") or () if item.get("name") == active_name), None)
        if target is None:
            return
        state["payload"] = snapshot
        state["target"] = target
        target_state = target.get("state")
        if target_state == "initiated" and target.get("connect_url"):
            state["phase"] = "url"
        elif target_state == "failed":
            state["phase"] = "failed"
        elif target_state == "pending" and self._connection_fields(target):
            # The backend refused the answer because a required field is still empty: reopen the
            # form on the first one it named, over the draft the panel kept.
            missing = {field.get("name") for field in self._connection_fields(target)}
            names = [field.get("name") for field in state.get("fields") or []]
            state["field_index"] = next((i for i, name in enumerate(names) if name in missing), 0)
            state["phase"] = "form"
            self._connection_sync_input_buffer()
        elif target_state == "connected" and target.get("discovery_error"):
            state["phase"] = "authorized"
        elif target_state in {"connected", "skipped"}:
            unresolved = [
                item for item in snapshot.get("targets") or ()
                if item.get("state") not in {"connected", "skipped"}
            ]
            if unresolved:
                self._connection_show_target(snapshot, (snapshot.get("targets") or []).index(unresolved[0]))
                return
            state["phase"] = "connected"
        else:
            state["phase"] = "waiting"
        self._paint_now()

    def _connection_set_field(self, value: str) -> None:
        """Save the active input row in the private draft and advance to the next row/action."""
        state = self._connection_state
        if not state or state.get("phase") not in {"form", "failed"}:
            return
        fields = state.get("fields") or []
        index = state.get("field_index", 0)
        if not (0 <= index < len(fields)):
            return
        field = fields[index]
        state["drafts"][state["target"].get("name", "")][field.get("name")] = value
        state["field_index"] = min(index + 1, len(fields))
        self._connection_sync_input_buffer()
        self._paint_now()

    def _connection_submit(self) -> None:
        """Enter action for prompt_toolkit bindings: URL open, selected action, or field advance."""
        state = self._connection_state
        if not state:
            return
        phase = state.get("phase")
        if phase == "url":
            self._connection_open_url()
        elif phase == "authorized":
            self._connection_continue()
        elif phase in {"form", "failed"} and state.get("field_index", 0) >= len(state.get("fields") or []):
            (self._connection_retry if state.get("selected", 0) == 0 else self._connection_cancel)()

    def _connection_field_lines(self, state, target) -> list[str]:
        draft = state.get("drafts", {}).get(target.get("name", ""), {})
        lines = []
        for field in state.get("fields") or ():
            marker = "*" if field.get("required") else ""
            value = t("cli.connect.secret_set") if field.get("type") == "secret" and draft.get(field.get("name")) else draft.get(field.get("name"), "")
            lines.append(f"{field.get('prompt') or field.get('name')}{marker}: {value}")
        return lines

    def _connection_render_lines(self) -> list[str]:
        """Panel copy without exposing secret drafts; the prompt_toolkit renderer consumes these rows."""
        state = self._connection_state
        if not state:
            return []
        target = state["target"]
        phase = state.get("phase")
        buttons = f"{t('cli.connect.button_connect')}    {t('cli.connect.button_cancel')}"
        lines = [t("cli.connect.set_up", name=target.get("name", ""))]
        if target.get("instructions"):
            lines.append(str(target["instructions"]))
        if phase == "form":
            lines.extend(self._connection_field_lines(state, target))
            lines.append(buttons)
        elif phase == "url":
            lines.extend([str(target.get("connect_url") or ""), str(target.get("detail") or ""), t("cli.connect.press_enter_open")])
        elif phase == "failed":
            lines.append(str(target.get("detail") or t("cli.connect.failed")))
            lines.extend(self._connection_field_lines(state, target))
            lines.append(buttons)
        elif phase == "authorized":
            # A connected row cannot be re-run inside this operation; the agent retries discovery
            # with its next manage_connections call, which needs no new consent.
            lines.extend([t("cli.connect.authorized_no_tools"), t("cli.connect.button_continue")])
        elif phase == "connected":
            lines.append(t("cli.connect.connected"))
        else:
            lines.append(str(target.get("detail") or t("cli.connect.waiting")))
        return [line for line in lines if line]

    # --- Batch clarify (multi-question, issue #18450) -----------------------
    def _clarify_batch_set_active(self, state, index) -> None:
        """Point the batch clarify panel at question ``index``: mirror it into the flat keys the
        keybindings/renderer read so ↑/↓/Space/number keys work unchanged;
        open-ended drops into freetext; re-visiting restores the earlier cursor/checkboxes."""
        questions_list = state["questions"]
        index = max(0, min(index, len(questions_list) - 1))
        entry = questions_list[index]
        choices = entry["choices"] or []
        state["active"] = index
        state["question"] = entry["question"]
        state["choices"] = choices
        state["selected"] = 0
        state["multi_select"] = bool(entry["multi_select"])
        state["selected_indices"] = set() if entry["multi_select"] else None
        self._clarify_freetext = not entry["choices"]
        self._clarify_multi_base = None
        meta = (state.get("answer_meta") or {}).get(entry["qid"])
        if meta is None:
            return
        kind = meta.get("kind")
        if kind == "choice":
            answer = state["answers"].get(entry["qid"])
            if answer in choices:
                state["selected"] = choices.index(answer)
        elif kind == "other":
            state["selected"] = len(choices)
        elif kind == "multi":
            checked = {choices.index(c) for c in meta.get("choices") or [] if c in choices}
            if meta.get("other_text"):
                checked.add(len(choices))
            state["selected_indices"] = checked

    def _clarify_batch_lock(self, state, answer, meta=None) -> None:
        """Lock ``answer`` for the active batch question (overwriting an earlier one) and advance to
        the next unanswered; ``meta`` ({"kind": "choice"|"other"|"multi", ...}) lets a re-visit
        restore the cursor / prefill an "Other" edit. All answered → resolve the queue, tear down."""
        entry = state["questions"][state["active"]]
        state["answers"][entry["qid"]] = answer
        state.setdefault("answer_meta", {})[entry["qid"]] = meta or {"kind": "choice"}
        self._persist_prompt_summary("?", t("cli.clarify.label"), entry["question"], "" if answer is None else str(answer))
        total = len(state["questions"])
        for offset in range(1, total + 1):
            candidate = (state["active"] + offset) % total
            if state["questions"][candidate]["qid"] not in state["answers"]:
                self._clarify_batch_set_active(state, candidate)
                return
        try:
            state["response_queue"].put(dict(state["answers"]))
        except Exception:
            pass
        self._clarify_state = None
        self._clarify_freetext = False
        self._clarify_multi_base = None

    def _clarify_batch_enter(self, state) -> None:
        """Enter in batch choice mode: lock the active selection. Multi-select locks a JSON array of
        checked labels (parsed by the tool core); "Other" switches to freetext, prefilled with an
        earlier typed answer so Enter on an answered Other edits instead of retyping."""
        choices = state.get("choices") or []
        selected = state.get("selected", 0)
        entry = state["questions"][state["active"]]
        meta = (state.get("answer_meta") or {}).get(entry["qid"]) or {}
        if state.get("multi_select"):
            sorted_idx = sorted(state.get("selected_indices") or set())
            selected_choices = [choices[i] for i in sorted_idx if i < len(choices)]
            if len(choices) in sorted_idx:
                # Stash the checked real choices so the freetext submit appends the typed answer.
                self._clarify_multi_base = selected_choices
                self._clarify_freetext = True
                self._clarify_prefill = meta.get("other_text") or ""
                return
            self._clarify_batch_lock(
                state,
                json.dumps(selected_choices, ensure_ascii=False),
                meta={"kind": "multi", "choices": selected_choices, "other_text": ""})
            return
        if selected < len(choices):
            self._clarify_batch_lock(state, choices[selected], meta={"kind": "choice"})
            return
        self._clarify_freetext = True
        self._clarify_prefill = meta.get("other_text") or "" if meta.get("kind") == "other" else ""

    def _clarify_callback(self, questions):
        """Clarify-tool platform callback (agent thread, #18450): the batch panel (A-compact) shows
        all questions, one active, and blocks until the key bindings lock every answer. Returns
        ``{"answers": {qid: raw | None}, "outcome"}`` (None = skipped): ``submitted`` when every
        question is locked, ``timed_out`` when the deadline expires, ``cancelled`` on an interrupt."""
        from cli import CLI_CONFIG, _DIM, _RST, _cprint
        from tools.clarify_gateway import resolve_clarify_timeout

        timeout = resolve_clarify_timeout(CLI_CONFIG)
        response_queue = queue.Queue()
        state = {
            "questions": list(questions),
            "answers": {},
            "answer_meta": {},
            "active": 0,
            "response_queue": response_queue,
            # Flat keys mirroring the active question — filled by _clarify_batch_set_active.
            "question": "",
            "choices": [],
            "selected": 0,
            "multi_select": False,
            "selected_indices": None}
        self._clarify_state = state
        self._clarify_batch_set_active(state, 0)
        self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout
        self._ring_bell(prompt=True, context=t("cli.clarify.bell_context"))
        self._paint_now()

        result = self._poll_modal_queue(response_queue, "_clarify_deadline")
        if result is not _TIMED_OUT:
            self._clarify_deadline = None
            outcome = "cancelled" if result is None else "submitted"
            return {"answers": dict(state["answers"]), "outcome": outcome}
        partial = dict(state["answers"])
        self._clarify_teardown()
        _cprint(f"\n{_DIM}{t('cli.clarify.timed_out_locked_answers', timeout=timeout)}{_RST}")
        return {"answers": partial, "outcome": "timed_out"}

    def _sudo_password_callback(self) -> str:
        """Prompt for a sudo password through the prompt_toolkit UI (agent thread); clarify-style
        state + queue answered by the Enter binding."""
        from cli import _DIM, _RST, _cprint

        response_queue = queue.Queue()
        self._capture_modal_input_snapshot()
        self._sudo_state = {"response_queue": response_queue}
        self._sudo_deadline = _time.monotonic() + 45
        self._ring_bell(prompt=True, context=t("cli.secret.bell_sudo"))
        self._paint_now()

        result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
        self._sudo_state = None
        self._sudo_deadline = 0
        self._restore_modal_input_snapshot()
        self._paint_now()
        if result is _TIMED_OUT:
            _cprint(f"\n{_DIM}  {t('cli.secret.sudo_timeout')}{_RST}")
            return ""
        if result:
            _cprint(f"\n{_DIM}  {t('cli.secret.sudo_received')}{_RST}")
        else:
            _cprint(f"\n{_DIM}  {t('cli.secret.sudo_skipped')}{_RST}")
        return result

    def _approval_callback(self, command: str, description: str,
                           *, allow_permanent: bool = True,
                           allow_session: bool = True,
                           smart_denied: bool = False) -> str:
        """Dangerous-command approval through the prompt_toolkit UI (agent thread).

        Choices: once / session / always / deny (see ``_approval_choices``), plus 'view' for long
        commands. ``_approval_lock`` serializes concurrent requests (parallel delegation subtasks)
        so the shared ``_approval_state`` / ``_approval_deadline`` aren't clobbered.
        """
        from cli import CLI_CONFIG, _DIM, _RST, _cprint

        with self._approval_lock:
            timeout = int(CLI_CONFIG.get("approvals", {}).get("timeout", 300))
            response_queue = queue.Queue()
            self._approval_state = {
                "command": command,
                "description": description,
                "choices": self._approval_choices(
                    command,
                    allow_permanent=allow_permanent,
                    allow_session=allow_session,
                    smart_denied=smart_denied),
                "selected": 0,
                "response_queue": response_queue}
            self._approval_deadline = _time.monotonic() + timeout
            self._ring_bell(prompt=True, context=t("cli.approval.bell_context"), detail=command)
            self._paint_now()

            result = self._poll_modal_queue(response_queue, "_approval_deadline")
            self._approval_state = None
            self._approval_deadline = 0
            self._paint_now()
            if result is _TIMED_OUT:
                _cprint(f"\n{_DIM}  {t('cli.approval.timeout_denying')}{_RST}")
                self._persist_prompt_summary(
                    "⚠", t("cli.approval.label"), command, t("cli.approval.timed_out_no_response"))
                return "timeout"
            self._persist_prompt_summary(
                "⚠", t("cli.approval.label"), command, _approval_outcome_label(result))
            return result

    def _approval_choices(self, command: str, *, allow_permanent: bool = True,
                          allow_session: bool = True,
                          smart_denied: bool = False) -> list[str]:
        """Smart-DENY overrides and re-ask-every-time gates (allow_session=False) show only
        once/deny; ``allow_permanent=False`` for another reason hides only 'always'."""
        if smart_denied or not allow_session:
            choices = ["once", "deny"]
        elif allow_permanent:
            choices = ["once", "session", "always", "deny"]
        else:
            choices = ["once", "session", "deny"]
        if len(command) > 70:
            choices.append("view")
        return choices

    def _handle_approval_selection(self) -> None:
        """Process the currently selected dangerous-command approval choice."""
        state = self._approval_state
        if not state:
            return
        selected = state.get("selected", 0)
        choices = state.get("choices")
        if not isinstance(choices, list):
            choices = []
        if not (0 <= selected < len(choices)):
            return
        chosen = choices[selected]
        if chosen == "view":
            state["show_full"] = True
            state["choices"] = [choice for choice in choices if choice != "view"]
            if state["selected"] >= len(state["choices"]):
                state["selected"] = max(0, len(state["choices"]) - 1)
            self._invalidate()
            return
        state["response_queue"].put(chosen)
        self._approval_state = None
        self._invalidate()

    def _vault_unlock_callback(self, backend_name: str, display_name: str) -> str:
        """Masked master-password prompt for an external password manager (agent thread).
        Reuses the sudo panel state so rendering, Enter/ESC handling and interrupt cleanup are shared."""
        from cli import _DIM, _RST, _cprint

        response_queue = queue.Queue()
        self._capture_modal_input_snapshot()
        self._sudo_state = {"response_queue": response_queue, "vault_backend": display_name}
        self._sudo_deadline = _time.monotonic() + 120
        self._ring_bell(prompt=True, context=t("cli.secret.bell_unlock", name=display_name))
        self._paint_now()

        result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
        self._sudo_state = None
        self._sudo_deadline = 0
        self._restore_modal_input_snapshot()
        self._paint_now()
        if result is _TIMED_OUT or not result:
            _cprint(f"\n{_DIM}  {t('cli.secret.stays_locked', name=display_name)}{_RST}")
            return ""
        _cprint(f"\n{_DIM}  {t('cli.secret.unlocking_for_session', name=display_name)}{_RST}")
        return result

    def _vault_save_login_callback(self, origin: str, site: str):
        """Two-step "save this login" prompt (identifier shown, password masked) on the sudo panel; the
        answer goes to the vault store, never to the model. None = declined."""
        from cli import _DIM, _RST, _cprint

        answer: dict = {}
        for step in ("identifier", "password"):
            response_queue = queue.Queue()
            self._capture_modal_input_snapshot()
            self._sudo_state = {"response_queue": response_queue, "vault_save": {"site": site, "origin": origin,
                                                                                  "step": step}}
            self._sudo_deadline = _time.monotonic() + 180
            if step == "identifier":
                self._ring_bell(prompt=True, context=t("cli.secret.bell_save_login", site=site))
            self._paint_now()
            result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
            self._sudo_state = None
            self._sudo_deadline = 0
            self._restore_modal_input_snapshot()
            self._paint_now()
            if result is _TIMED_OUT or not result:
                _cprint(f"\n{_DIM}  {t('cli.secret.login_not_saved', site=site)}{_RST}")
                return None
            answer[step] = result
        _cprint(f"\n{_DIM}  {t('cli.secret.login_saved', site=site)}{_RST}")
        return answer

    def _vault_code_callback(self, site: str, hint: str) -> str:
        """One-time-code prompt (shown as typed; a 6-digit code is not a secret worth masking and users
        need to see typos) on the sudo panel. "" = declined/timed out."""
        from cli import _DIM, _RST, _cprint

        response_queue = queue.Queue()
        self._capture_modal_input_snapshot()
        self._sudo_state = {"response_queue": response_queue, "vault_code": {"site": site, "hint": hint}}
        self._sudo_deadline = _time.monotonic() + 180
        self._ring_bell(prompt=True, context=t("cli.secret.bell_verification_code", site=site))
        self._paint_now()
        result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
        self._sudo_state = None
        self._sudo_deadline = 0
        self._restore_modal_input_snapshot()
        self._paint_now()
        if result is _TIMED_OUT or not result:
            _cprint(f"\n{_DIM}  {t('cli.secret.code_not_entered', site=site)}{_RST}")
            return ""
        _cprint(f"\n{_DIM}  {t('cli.secret.code_entered', site=site)}{_RST}")
        return result

    def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict:
        self._capture_modal_input_snapshot()
        try:
            return prompt_for_secret(self, var_name, prompt, metadata)
        finally:
            self._restore_modal_input_snapshot()
            self._paint_now()

    def _capture_modal_input_snapshot(self) -> None:
        """Temporarily clear the input buffer and save the user's in-progress draft."""
        if getattr(self, "_modal_input_snapshot", None) is not None or not getattr(self, "_app", None):
            return
        try:
            buf = self._app.current_buffer
            self._modal_input_snapshot = {"text": buf.text, "cursor_position": buf.cursor_position}
            buf.reset()
        except Exception:
            self._modal_input_snapshot = None

    def _restore_modal_input_snapshot(self) -> None:
        """Restore any draft text that was present before a modal prompt opened."""
        snapshot = getattr(self, "_modal_input_snapshot", None)
        self._modal_input_snapshot = None
        if not snapshot or not getattr(self, "_app", None):
            return
        try:
            buf = self._app.current_buffer
            buf.text = snapshot.get("text", "")
            buf.cursor_position = min(snapshot.get("cursor_position", 0), len(buf.text))
        except Exception:
            pass

    def _clear_active_overlays_for_interrupt(self) -> None:
        """Drain and clear every input-blocking overlay left by an interrupted agent: the worker
        thread is gone but the state dict still gates input (frozen terminal until its timeout).
        Push a safe value onto each queue (approval -> "deny", others -> cancel), nil the state,
        restore the draft; each step is wrapped so a dead queue can't block the others."""
        def _put(state, value) -> None:
            try:
                state["response_queue"].put(value)
            except Exception:
                pass

        if self._approval_state:
            _put(self._approval_state, "deny")
            self._approval_state = None
        if self._connection_state:
            self._connection_interrupt()
        if self._clarify_state:
            _put(self._clarify_state, None)
            self._clarify_state = None
            self._clarify_freetext = False
            self._clarify_multi_base = None
        if self._sudo_state:
            _put(self._sudo_state, "")
            self._sudo_state = None
            self._sudo_deadline = 0
            self._restore_modal_input_snapshot()
        if self._secret_state:
            try:
                self._cancel_secret_capture()
            except Exception:
                self._secret_state = None

    def _submit_secret_response(self, value: str) -> None:
        if not self._secret_state:
            return
        self._secret_state["response_queue"].put(value)
        self._secret_state = None
        self._secret_deadline = 0
        self._paint_now()  # direct paint so the secret panel clears at once (no throttle)

    def _cancel_secret_capture(self) -> None:
        self._submit_secret_response("")

    def _clear_secret_input_buffer(self) -> None:
        if getattr(self, "_app", None):
            try:
                self._app.current_buffer.reset()
            except Exception:
                pass