OpenSpace/openspace/core/tui_bridge.py

"""
IPC bridge between Python Core (this process) and TS TUI (child process).

Communication is bidirectional NDJSON over the child's stdin/stdout.
The Python side is the *parent*: it spawns the TS TUI via
``asyncio.create_subprocess_exec`` and talks through pipes.
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import shutil
import sys
from typing import Any, AsyncGenerator

from openspace.protocol import CoreToTuiEvent, StreamEvent

logger = logging.getLogger(__name__)

# U+2028 / U+2029 must be escaped in NDJSON to prevent line-split issues
_JS_LINE_TERMINATORS = str.maketrans({"\u2028": "\\u2028", "\u2029": "\\u2029"})


def _ndjson_dumps(obj: Any) -> str:
    """Serialize *obj* to a single NDJSON line, safe for JS receivers."""
    return json.dumps(obj, ensure_ascii=False).translate(_JS_LINE_TERMINATORS)


class TUIBridge:
    """Manages the lifecycle of the TS TUI child process and IPC messaging."""

    def __init__(
        self,
        tui_entry: str | None = None,
        node_bin: str | None = None,
        tui_args: list[str] | None = None,
    ) -> None:
        self._tui_entry = tui_entry or self._default_tui_entry()
        self._node_bin = node_bin or self._find_node()
        self._tui_args = list(tui_args or [])
        self._process: asyncio.subprocess.Process | None = None
        self._pending_permissions: dict[str, asyncio.Future[dict[str, Any]]] = {}
        self._pending_prompts: dict[str, asyncio.Future[dict[str, Any]]] = {}
        self._tool_permission_tasks: dict[str, asyncio.Task[None]] = {}
        self._cancelled = False
        self._shutdown_event = asyncio.Event()

    # ── Lifecycle ────────────────────────────────────────────────

    async def start(self) -> None:
        """Spawn the TUI child process."""
        if not os.path.exists(self._tui_entry):
            searched = "\n".join(
                f"  - {path}" for path in self.default_tui_entry_candidates()
            )
            raise FileNotFoundError(
                f"TUI entry not found: {self._tui_entry}. "
                f"{self.default_tui_missing_hint()}\n"
                "Searched default locations:\n"
                f"{searched}"
            )

        self._shutdown_event.clear()
        env = dict(os.environ)
        env["OPENSPACE_TUI_IPC"] = "1"
        self._process = await asyncio.create_subprocess_exec(
            self._node_bin,
            self._tui_entry,
            *self._tui_args,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env=env,
        )
        logger.info("TUI process started (pid=%s)", self._process.pid)

    async def shutdown(self) -> None:
        """Gracefully shut down the TUI child process."""
        self._shutdown_event.set()
        active_tool_permission_ids = self._active_tool_permission_ask_ids()
        tool_permission_tasks = self._reject_active_tool_permission_asks(
            "TUI bridge shut down"
        )
        self._reject_pending_permissions("TUI bridge shut down")
        self._reject_pending_prompts("TUI bridge shut down")
        await self._notify_tool_permission_cancelled(
            active_tool_permission_ids,
            "TUI bridge shut down",
        )
        if tool_permission_tasks:
            await asyncio.gather(*tool_permission_tasks, return_exceptions=True)
        proc = self._process
        if proc is None:
            return

        if proc.stdin and not proc.stdin.is_closing():
            proc.stdin.close()

        try:
            await asyncio.wait_for(proc.wait(), timeout=5.0)
        except asyncio.TimeoutError:
            logger.warning("TUI process did not exit in time, killing")
            try:
                proc.kill()
            except ProcessLookupError:
                pass
            await proc.wait()

        logger.info("TUI process exited (code=%s)", proc.returncode)
        self._process = None

    # ── Sending (Core → TUI) ────────────────────────────────────

    async def send(self, event_type: str, data: dict[str, Any] | None = None) -> None:
        """Send an IPC event to the TUI child process."""
        await self.send_event(StreamEvent(type=event_type, data=data or {}))

    async def send_event(self, event: StreamEvent) -> None:
        """Send a pre-built StreamEvent."""
        if event.type == "tool_permission_ask":
            await self._forward_tool_permission_ask(event)
            return
        if event.type == "tool_permission_cancel":
            tool_use_id = str(event.data.get("tool_use_id") or "").strip()
            if tool_use_id:
                self._cancel_tool_permission_prompt(
                    tool_use_id,
                    str(event.data.get("reason") or "Permission prompt cancelled."),
                )
            await self._write_event_to_child(event)
            return

        await self._write_event_to_child(event)

    async def _write_event_to_child(self, event: StreamEvent) -> None:
        """Write an event directly to the child TUI process when available."""
        proc = self._process
        if proc is None or proc.stdin is None or proc.stdin.is_closing():
            logger.debug("send() skipped: TUI process not available")
            return

        line = _ndjson_dumps(event.to_dict()) + "\n"
        proc.stdin.write(line.encode("utf-8"))
        await proc.stdin.drain()

    # ── Receiving (TUI → Core) ───────────────────────────────────

    async def receive(self) -> AsyncGenerator[StreamEvent, None]:
        """Async generator that yields events from the TUI child process."""
        proc = self._process
        if proc is None or proc.stdout is None:
            return

        while not self._shutdown_event.is_set():
            try:
                raw_line = await proc.stdout.readline()
            except (asyncio.CancelledError, ConnectionError):
                break

            if not raw_line:
                break

            line = raw_line.decode("utf-8", errors="replace").strip()
            if not line:
                continue

            try:
                parsed = json.loads(line)
            except json.JSONDecodeError:
                logger.warning("Ignoring non-JSON line from TUI: %.100s", line)
                continue

            event = StreamEvent.from_dict(parsed)

            if event.type == "permission_response":
                self._handle_permission_response(event)
            elif event.type == "tool_permission_response":
                self._handle_tool_permission_response(event)
            elif event.type == "prompt_response":
                self._handle_prompt_response(event)
            elif event.type == "cancel":
                self._cancelled = True
                self._reject_active_tool_permission_asks("Cancelled by TUI")
                self._reject_pending_permissions("Cancelled by TUI")
                self._reject_pending_prompts("Cancelled by TUI")

            yield event

        if self._active_tool_permission_ask_ids():
            self._reject_active_tool_permission_asks(
                "TUI process exited before permission prompt completed"
            )
        if self._pending_permissions:
            self._reject_pending_permissions(
                "TUI process exited before permission responses were received"
            )
        if self._pending_prompts:
            self._reject_pending_prompts(
                "TUI process exited before prompt responses were received"
            )

    # ── Permission round-trip ────────────────────────────────────

    async def request_permission(
        self,
        tool_name: str,
        tool_input: dict[str, Any],
        tool_use_id: str,
        risk_level: str = "medium",
        description: str | None = None,
        request_kind: str | None = None,
        host: str | None = None,
        origin: str | None = None,
        agent_id: str | None = None,
        agent_name: str | None = None,
        agent_color: str | None = None,
        allow_always_pattern: str | None = None,
    ) -> dict[str, Any]:
        """
        Send a permission_request to TUI and wait for the user's decision.

        Returns the permission_response data dict with keys:
        ``tool_use_id``, ``decision``, and optionally ``pattern``.
        """
        loop = asyncio.get_running_loop()
        future: asyncio.Future[dict[str, Any]] = loop.create_future()
        self._pending_permissions[tool_use_id] = future

        payload = {
            "tool_use_id": tool_use_id,
            "tool_name": tool_name,
            "tool_input": tool_input,
            "risk_level": risk_level,
            "description": description or f"Allow {tool_name}?",
            "request_kind": request_kind or "tool",
            "origin": origin or "primary",
        }
        if host is not None:
            payload["host"] = host
        if agent_id is not None:
            payload["agent_id"] = agent_id
        if agent_name is not None:
            payload["agent_name"] = agent_name
        if agent_color is not None:
            payload["agent_color"] = agent_color
        if allow_always_pattern is not None:
            payload["allow_always_pattern"] = allow_always_pattern

        await self.send(
            CoreToTuiEvent.PERMISSION_REQUEST.value,
            payload,
        )

        try:
            return await future
        finally:
            self._pending_permissions.pop(tool_use_id, None)

    def _handle_permission_response(self, event: StreamEvent) -> None:
        tool_use_id = event.data.get("tool_use_id")
        if not tool_use_id:
            return
        future = self._pending_permissions.get(tool_use_id)
        if future and not future.done():
            future.set_result(event.data)

    def _handle_tool_permission_response(self, event: StreamEvent) -> None:
        if not isinstance(event.data, dict):
            self._fail_malformed_tool_permission_response(
                "Malformed tool_permission_response had non-object data."
            )
            return

        tool_use_id = str(event.data.get("tool_use_id") or "").strip()
        if not tool_use_id:
            self._fail_malformed_tool_permission_response(
                "Malformed tool_permission_response missing tool_use_id."
            )
            return

        if (
            tool_use_id not in self._tool_permission_tasks
            and not self._is_tool_permission_ask_pending(tool_use_id)
        ):
            self._fail_malformed_tool_permission_response(
                f"Malformed tool_permission_response referenced unknown tool_use_id={tool_use_id!r}.",
                preferred_tool_use_id=tool_use_id,
            )
            return

        self._resolve_tool_permission_ask(tool_use_id, dict(event.data))
        self._cancel_tool_permission_task(tool_use_id)

    def _reject_pending_permissions(self, reason: str) -> None:
        for tool_use_id, future in list(self._pending_permissions.items()):
            if not future.done():
                future.set_exception(RuntimeError(reason))
            self._pending_permissions.pop(tool_use_id, None)

    async def request_prompt(
        self,
        prompt_id: str,
        *,
        title: str | None = None,
        description: str | None = None,
        default_value: str | None = None,
        placeholder: str | None = None,
        multiline: bool = False,
    ) -> dict[str, Any]:
        """Send a prompt_request to TUI and wait for the user's response."""
        loop = asyncio.get_running_loop()
        future: asyncio.Future[dict[str, Any]] = loop.create_future()
        self._pending_prompts[prompt_id] = future

        await self.send(
            CoreToTuiEvent.PROMPT_REQUEST.value,
            {
                "prompt_id": prompt_id,
                "title": title,
                "description": description,
                "default_value": default_value,
                "placeholder": placeholder,
                "multiline": multiline,
            },
        )

        try:
            return await future
        finally:
            self._pending_prompts.pop(prompt_id, None)

    def _handle_prompt_response(self, event: StreamEvent) -> None:
        prompt_id = event.data.get("prompt_id")
        if not prompt_id:
            return
        future = self._pending_prompts.get(prompt_id)
        if future and not future.done():
            future.set_result(event.data)

    def _reject_pending_prompts(self, reason: str) -> None:
        for prompt_id, future in list(self._pending_prompts.items()):
            if not future.done():
                future.set_exception(RuntimeError(reason))
            self._pending_prompts.pop(prompt_id, None)

    def _cancel_tool_permission_prompt(
        self,
        tool_use_id: str,
        reason: str,
    ) -> asyncio.Task[None] | None:
        logger.debug("Cancelling tool permission prompt %s: %s", tool_use_id, reason)
        return self._cancel_tool_permission_task(tool_use_id)

    def _cancel_tool_permission_prompt_waiters(self, tool_use_id: str) -> None:
        prompt_ids = {
            self._tool_permission_prompt_id(tool_use_id, "choice"),
            self._tool_permission_prompt_id(tool_use_id, "edit"),
        }
        for prompt_id, future in list(self._pending_prompts.items()):
            is_ask_prompt = (
                prompt_id.startswith(
                    self._tool_permission_prompt_id(tool_use_id, "ask-")
                )
                or prompt_id.startswith(
                    self._tool_permission_prompt_id(tool_use_id, "ask-other-")
                )
            )
            if prompt_id not in prompt_ids and not is_ask_prompt:
                continue
            if not future.done():
                future.cancel()
            self._pending_prompts.pop(prompt_id, None)

    def _fail_malformed_tool_permission_response(
        self,
        message: str,
        *,
        preferred_tool_use_id: str | None = None,
    ) -> None:
        active_ids = self._active_tool_permission_ask_ids()

        # Deterministic fail-closed policy for malformed permission responses:
        # if a usable tool_use_id maps to exactly one live ask, deny only that
        # ask so unrelated concurrent prompts survive. If the malformed event
        # cannot be tied to one ask, deny every active ask instead of letting
        # any wait for the full permission timeout.
        if preferred_tool_use_id and preferred_tool_use_id in active_ids:
            target_ids = [preferred_tool_use_id]
        else:
            target_ids = active_ids

        if not target_ids:
            logger.warning("%s", message)
            return

        logger.warning("%s", message)
        for pending_tool_use_id in target_ids:
            self._resolve_tool_permission_ask(
                pending_tool_use_id,
                {
                    "option_id": "deny",
                    "message": message,
                },
            )
            self._cancel_tool_permission_task(pending_tool_use_id)

    def _active_tool_permission_ask_ids(self) -> list[str]:
        from openspace.tool_runtime.permissions import pending_permission_ask_ids

        active_ids = list(self._tool_permission_tasks)
        for tool_use_id in pending_permission_ask_ids():
            if tool_use_id not in active_ids:
                active_ids.append(tool_use_id)
        return active_ids

    def _start_tool_permission_prompt(self, event: StreamEvent) -> None:
        tool_use_id = str(event.data.get("tool_use_id") or "").strip()
        if not tool_use_id:
            logger.warning("Ignoring tool_permission_ask without tool_use_id")
            return
        permission_ask_id = str(event.data.get("permission_ask_id") or "").strip()

        proc = self._process
        if (
            self._shutdown_event.is_set()
            or proc is None
            or proc.stdin is None
            or proc.stdin.is_closing()
        ):
            self._resolve_tool_permission_ask(
                tool_use_id,
                {
                    "option_id": "deny",
                    "permission_ask_id": permission_ask_id,
                    "message": "Interactive permission prompt unavailable because the TUI bridge is not running.",
                },
            )
            return

        existing = self._tool_permission_tasks.get(tool_use_id)
        if existing is not None and not existing.done():
            self._resolve_tool_permission_ask(
                tool_use_id,
                {
                    "option_id": "deny",
                    "permission_ask_id": permission_ask_id,
                    "message": "Permission prompt was superseded by a newer request.",
                },
            )
            self._cancel_tool_permission_task(tool_use_id)

        task = asyncio.create_task(self._drive_tool_permission_prompt(dict(event.data)))
        self._tool_permission_tasks[tool_use_id] = task

        def _discard(done_task: asyncio.Task[None], ask_id: str = tool_use_id) -> None:
            if self._tool_permission_tasks.get(ask_id) is done_task:
                self._tool_permission_tasks.pop(ask_id, None)

        task.add_done_callback(_discard)

    async def _forward_tool_permission_ask(self, event: StreamEvent) -> None:
        tool_use_id = str(event.data.get("tool_use_id") or "").strip()
        if not tool_use_id:
            logger.warning("Ignoring tool_permission_ask without tool_use_id")
            return
        permission_ask_id = str(event.data.get("permission_ask_id") or "").strip()

        proc = self._process
        if (
            self._shutdown_event.is_set()
            or proc is None
            or proc.stdin is None
            or proc.stdin.is_closing()
        ):
            self._resolve_tool_permission_ask(
                tool_use_id,
                {
                    "option_id": "deny",
                    "permission_ask_id": permission_ask_id,
                    "message": "Interactive permission prompt unavailable because the TUI bridge is not running.",
                },
            )
            return

        payload = dict(event.data)
        payload.setdefault("response_channel", "tool_permission_response")
        payload.setdefault("request_kind", "tool")
        await self._write_event_to_child(
            StreamEvent(type=event.type, data=payload)
        )

    async def _drive_tool_permission_prompt(self, payload: dict[str, Any]) -> None:
        tool_use_id = str(payload.get("tool_use_id") or "").strip()
        if not tool_use_id:
            return
        permission_ask_id = str(payload.get("permission_ask_id") or "").strip()

        try:
            response = await self._collect_tool_permission_response(payload)
        except asyncio.CancelledError:
            self._resolve_tool_permission_ask(
                tool_use_id,
                {
                    "option_id": "deny",
                    "permission_ask_id": permission_ask_id,
                    "message": "Permission prompt was cancelled.",
                },
            )
            raise
        except Exception as exc:
            logger.warning(
                "Tool permission prompt failed for %s: %s",
                tool_use_id,
                exc,
            )
            response = {
                "option_id": "deny",
                "permission_ask_id": permission_ask_id,
                "message": f"Interactive permission prompt failed: {exc}",
            }

        if permission_ask_id:
            response.setdefault("permission_ask_id", permission_ask_id)
        self._resolve_tool_permission_ask(tool_use_id, response)

    def _reject_active_tool_permission_asks(
        self,
        reason: str,
    ) -> list[asyncio.Task[None]]:
        task_map = dict(self._tool_permission_tasks)
        active_ids = self._active_tool_permission_ask_ids()
        for tool_use_id in active_ids:
            self._resolve_tool_permission_ask(
                tool_use_id,
                {
                    "option_id": "deny",
                    "message": reason,
                },
            )
            self._cancel_tool_permission_prompt_waiters(tool_use_id)
            task = self._tool_permission_tasks.pop(tool_use_id, None)
            if task is not None and not task.done():
                task.cancel()
        return [task for tool_use_id, task in task_map.items() if tool_use_id in active_ids]

    async def _notify_tool_permission_cancelled(
        self,
        tool_use_ids: list[str],
        reason: str,
    ) -> None:
        """Best-effort child notification to dismiss rendered permission prompts."""
        for tool_use_id in tool_use_ids:
            try:
                await self._write_event_to_child(
                    StreamEvent(
                        type="tool_permission_cancel",
                        data={
                            "tool_use_id": tool_use_id,
                            "reason": reason,
                        },
                    )
                )
            except Exception:
                logger.debug(
                    "Failed to notify TUI about cancelled tool permission prompt %s",
                    tool_use_id,
                    exc_info=True,
                )

    def _cancel_tool_permission_task(
        self,
        tool_use_id: str,
    ) -> asyncio.Task[None] | None:
        self._cancel_tool_permission_prompt_waiters(tool_use_id)
        task = self._tool_permission_tasks.pop(tool_use_id, None)
        if task is not None and not task.done():
            task.cancel()
        return task

    async def _collect_tool_permission_response(
        self,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        tool_use_id = str(payload.get("tool_use_id") or "").strip()
        tool_name = str(payload.get("tool_name") or "tool").strip() or "tool"
        tool_input = payload.get("tool_input")
        if payload.get("interaction") == "ask_user_question":
            return await self._collect_ask_user_question_response(payload)

        options_raw = payload.get("options")
        options = [
            option
            for option in options_raw
            if isinstance(option, dict) and option.get("option_id")
        ] if isinstance(options_raw, list) else []

        if not options:
            options = [
                {"option_id": "allow_once", "label": "Allow once"},
                {"option_id": "deny", "label": "Deny"},
            ]

        choice_error: str | None = None
        while self._is_tool_permission_ask_pending(tool_use_id):
            prompt_response = await self.request_prompt(
                self._tool_permission_prompt_id(tool_use_id, "choice"),
                title=f"Permission Required: {tool_name}",
                description=self._build_tool_permission_choice_description(
                    payload,
                    options,
                    choice_error,
                ),
                default_value="1",
                placeholder="Enter option number or name",
                multiline=False,
            )
            if prompt_response.get("decision") != "submit":
                return {
                    "option_id": "deny",
                    "message": "Permission prompt cancelled by user.",
                }

            selection = str(prompt_response.get("value") or "").strip()
            selected_option = self._select_tool_permission_option(selection, options)
            if selected_option is None:
                choice_error = f"Invalid selection: {selection or '<empty>'}"
                continue

            option_id = str(selected_option.get("option_id") or "").strip()
            if option_id != "provide_input":
                return {
                    "option_id": option_id,
                    "suggestion_index": selected_option.get("suggestion_index"),
                }

            edit_error: str | None = None
            while self._is_tool_permission_ask_pending(tool_use_id):
                edit_response = await self.request_prompt(
                    self._tool_permission_prompt_id(tool_use_id, "edit"),
                    title=f"Edit Tool Input: {tool_name}",
                    description=self._build_tool_permission_edit_description(
                        payload,
                        edit_error,
                    ),
                    default_value=self._format_tool_input_json(tool_input),
                    placeholder='{"key": "value"}',
                    multiline=True,
                )
                if edit_response.get("decision") != "submit":
                    return {
                        "option_id": "deny",
                        "message": "Input edit cancelled by user.",
                    }

                try:
                    edited_input = self._parse_edited_tool_input(
                        edit_response.get("value"),
                    )
                except ValueError as exc:
                    edit_error = str(exc)
                    continue

                return {
                    "option_id": "provide_input",
                    "edited_input": edited_input,
                }

        return {"option_id": "deny", "message": "Permission prompt expired."}

    async def _collect_ask_user_question_response(
        self,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        tool_use_id = str(payload.get("tool_use_id") or "").strip()
        tool_input = payload.get("tool_input")
        if not isinstance(tool_input, dict):
            tool_input = {}
        questions_raw = payload.get("questions")
        if not isinstance(questions_raw, list):
            questions_raw = tool_input.get("questions")
        if not isinstance(questions_raw, list) or not questions_raw:
            return {
                "option_id": "deny",
                "message": "AskUserQuestion payload did not include questions.",
            }

        answers: dict[str, str] = {}
        existing_annotations = tool_input.get("annotations")
        annotations: dict[str, Any] = (
            dict(existing_annotations) if isinstance(existing_annotations, dict) else {}
        )
        total = len(questions_raw)
        for index, question in enumerate(questions_raw, start=1):
            if not isinstance(question, dict):
                return {
                    "option_id": "deny",
                    "message": "AskUserQuestion question payload was malformed.",
                }
            question_text = str(question.get("question") or "").strip()
            options = self._ask_user_question_options(question)
            if not question_text or len(options) < 2:
                return {
                    "option_id": "deny",
                    "message": "AskUserQuestion question payload was incomplete.",
                }

            answer_error: str | None = None
            while self._is_tool_permission_ask_pending(tool_use_id):
                prompt_response = await self.request_prompt(
                    self._tool_permission_prompt_id(tool_use_id, f"ask-{index}"),
                    title=self._ask_user_question_title(question, index, total),
                    description=self._build_ask_user_question_description(
                        question,
                        index,
                        total,
                        answer_error,
                    ),
                    default_value="",
                    placeholder=(
                        "Enter option numbers separated by commas, or custom text"
                        if self._ask_user_question_multi_select(question)
                        else "Enter option number, label, or custom text"
                    ),
                    multiline=False,
                )
                if prompt_response.get("decision") != "submit":
                    return {
                        "option_id": "deny",
                        "message": "Question prompt cancelled by user.",
                    }

                try:
                    answer, annotation = self._parse_ask_user_question_answer(
                        prompt_response.get("value"),
                        question,
                    )
                except ValueError as exc:
                    answer_error = str(exc)
                    continue

                if "__OPENSPACE_OTHER_ANSWER__" in answer:
                    other_response = await self.request_prompt(
                        self._tool_permission_prompt_id(
                            tool_use_id,
                            f"ask-other-{index}",
                        ),
                        title=self._ask_user_question_title(question, index, total),
                        description="Enter a custom answer.",
                        default_value="",
                        placeholder="Custom answer",
                        multiline=False,
                    )
                    if other_response.get("decision") != "submit":
                        return {
                            "option_id": "deny",
                            "message": "Custom answer prompt cancelled by user.",
                        }
                    custom_answer = str(other_response.get("value") or "").strip()
                    if not custom_answer:
                        answer_error = "Custom answer cannot be empty."
                        continue
                    answer = answer.replace(
                        "__OPENSPACE_OTHER_ANSWER__",
                        custom_answer,
                    )

                answers[question_text] = answer
                if annotation:
                    current_annotation = annotations.get(question_text)
                    if isinstance(current_annotation, dict):
                        annotations[question_text] = {
                            **current_annotation,
                            **annotation,
                        }
                    elif current_annotation is not None:
                        annotations[question_text] = {
                            "value": current_annotation,
                            **annotation,
                        }
                    else:
                        annotations[question_text] = annotation
                break

        if len(answers) != total:
            return {"option_id": "deny", "message": "Question prompt expired."}

        updated_input = dict(tool_input)
        updated_input["questions"] = questions_raw
        updated_input["answers"] = answers
        if annotations:
            updated_input["annotations"] = annotations
        return {
            "option_id": "allow_once",
            "updated_input": updated_input,
        }

    def _resolve_tool_permission_ask(
        self,
        tool_use_id: str,
        response: dict[str, Any],
    ) -> None:
        from openspace.tool_runtime.permissions import resolve_permission_ask

        if not resolve_permission_ask(tool_use_id, response):
            logger.debug(
                "tool_permission_ask response dropped for %s; no pending ask",
                tool_use_id,
            )

    @staticmethod
    def _ask_user_question_options(question: dict[str, Any]) -> list[dict[str, Any]]:
        options_raw = question.get("options")
        return [
            option
            for option in options_raw
            if isinstance(option, dict) and option.get("label")
        ] if isinstance(options_raw, list) else []

    @staticmethod
    def _ask_user_question_multi_select(question: dict[str, Any]) -> bool:
        return bool(question.get("multiSelect", question.get("multi_select", False)))

    @staticmethod
    def _ask_user_question_title(
        question: dict[str, Any],
        index: int,
        total: int,
    ) -> str:
        header = str(question.get("header") or "").strip()
        prefix = f"Question {index}/{total}"
        return f"{prefix}: {header}" if header else prefix

    @classmethod
    def _build_ask_user_question_description(
        cls,
        question: dict[str, Any],
        index: int,
        total: int,
        error_message: str | None,
    ) -> str:
        options = cls._ask_user_question_options(question)
        multi_select = cls._ask_user_question_multi_select(question)
        lines: list[str] = []
        if error_message:
            lines.append(error_message)
        lines.append(str(question.get("question") or f"Question {index}/{total}"))
        if multi_select:
            lines.append(
                "Select one or more options by number or label. Use commas "
                "for multiple selections."
            )
        else:
            lines.append(
                "Select one option by number or label. Type custom text for Other."
            )
        for option_index, option in enumerate(options, start=1):
            label = str(option.get("label") or option_index)
            description = str(option.get("description") or "").strip()
            suffix = f" - {description}" if description else ""
            lines.append(f"{option_index}. {label}{suffix}")
            preview = option.get("preview")
            if isinstance(preview, str) and preview.strip():
                lines.append(
                    "   Preview:\n"
                    + cls._indent_preview(cls._truncate_preview(preview.strip()))
                )
        lines.append(f"{len(options) + 1}. Other - Provide a custom answer")
        return "\n".join(lines)

    @classmethod
    def _parse_ask_user_question_answer(
        cls,
        raw_value: Any,
        question: dict[str, Any],
    ) -> tuple[str, dict[str, str] | None]:
        raw_text = str(raw_value or "").strip()
        if not raw_text:
            raise ValueError("Answer cannot be empty.")

        options = cls._ask_user_question_options(question)
        if cls._ask_user_question_multi_select(question):
            parts = [part.strip() for part in raw_text.split(",") if part.strip()]
            if not parts:
                raise ValueError("Answer cannot be empty.")
            labels: list[str] = []
            for part in parts:
                selected = cls._select_ask_user_question_option(part, options)
                if selected == "__other__":
                    labels.append("__OPENSPACE_OTHER_ANSWER__")
                elif isinstance(selected, dict):
                    labels.append(str(selected.get("label") or part))
                else:
                    labels.append(part)
            return ", ".join(labels), None

        selected = cls._select_ask_user_question_option(raw_text, options)
        if selected == "__other__":
            return "__OPENSPACE_OTHER_ANSWER__", None
        if isinstance(selected, dict):
            label = str(selected.get("label") or raw_text)
            preview = selected.get("preview")
            annotation = (
                {"preview": preview}
                if isinstance(preview, str) and preview
                else None
            )
            return label, annotation
        return raw_text, None

    @staticmethod
    def _select_ask_user_question_option(
        selection: str,
        options: list[dict[str, Any]],
    ) -> dict[str, Any] | str | None:
        normalized = selection.strip().lower()
        if normalized.isdigit():
            index = int(normalized) - 1
            if 0 <= index < len(options):
                return options[index]
            if index == len(options):
                return "__other__"
            return None
        if normalized == "other":
            return "__other__"
        for option in options:
            label = str(option.get("label") or "").strip().lower()
            if normalized == label:
                return option
        return None

    @staticmethod
    def _truncate_preview(preview: str, limit: int = 1200) -> str:
        if len(preview) <= limit:
            return preview
        return preview[:limit] + "\n[preview truncated]"

    @staticmethod
    def _indent_preview(preview: str) -> str:
        return "\n".join(f"   {line}" for line in preview.splitlines())

    @staticmethod
    def _tool_permission_prompt_id(tool_use_id: str, stage: str) -> str:
        return f"tool-permission-{tool_use_id}-{stage}"

    @staticmethod
    def _select_tool_permission_option(
        selection: str,
        options: list[dict[str, Any]],
    ) -> dict[str, Any] | None:
        normalized = selection.strip().lower()
        if normalized.isdigit():
            index = int(normalized) - 1
            if 0 <= index < len(options):
                return options[index]

        aliases = {
            "allow": "allow_once",
            "always": "allow_always",
            "edit": "provide_input",
            "input": "provide_input",
            "no": "deny",
        }
        normalized = aliases.get(normalized, normalized)
        for option in options:
            option_id = str(option.get("option_id") or "").strip().lower()
            label = str(option.get("label") or "").strip().lower()
            if normalized in {option_id, label}:
                return option
        return None

    @staticmethod
    def _build_tool_permission_choice_description(
        payload: dict[str, Any],
        options: list[dict[str, Any]],
        error_message: str | None,
    ) -> str:
        lines: list[str] = []
        message = str(payload.get("message") or "").strip()
        if message:
            lines.append(message)
        blocked_path = str(payload.get("blocked_path") or "").strip()
        if blocked_path:
            lines.append(f"Blocked path: {blocked_path}")
        if error_message:
            lines.append(error_message)
        lines.append("Choose an option by number or option_id:")
        for index, option in enumerate(options, start=1):
            label = str(option.get("label") or option.get("option_id") or index)
            lines.append(f"{index}. {label}")
        return "\n".join(lines)

    @staticmethod
    def _build_tool_permission_edit_description(
        payload: dict[str, Any],
        error_message: str | None,
    ) -> str:
        lines = [
            "Edit the JSON tool input and submit a JSON object to continue.",
        ]
        message = str(payload.get("message") or "").strip()
        if message:
            lines.insert(0, message)
        if error_message:
            lines.append(error_message)
        return "\n".join(lines)

    @staticmethod
    def _format_tool_input_json(tool_input: Any) -> str:
        return json.dumps(tool_input, ensure_ascii=False, indent=2, sort_keys=True)

    @staticmethod
    def _parse_edited_tool_input(raw_value: Any) -> dict[str, Any]:
        if not isinstance(raw_value, str):
            raise ValueError("Edited input must be submitted as JSON text.")
        try:
            parsed = json.loads(raw_value)
        except json.JSONDecodeError as exc:
            raise ValueError(f"Invalid JSON: {exc.msg}") from exc
        if not isinstance(parsed, dict):
            raise ValueError("Edited input must be a JSON object.")
        return parsed

    @staticmethod
    def _is_tool_permission_ask_pending(tool_use_id: str) -> bool:
        from openspace.tool_runtime.permissions import is_permission_ask_pending

        return is_permission_ask_pending(tool_use_id)

    # ── Cancel / interrupt ───────────────────────────────────────

    async def cancel(self) -> None:
        """Send a cancel event to the TUI (triggered by Ctrl+C in Core)."""
        self._cancelled = True
        active_tool_permission_ids = self._active_tool_permission_ask_ids()
        tool_permission_tasks = self._reject_active_tool_permission_asks(
            "Cancelled by user interrupt"
        )
        self._reject_pending_permissions("Cancelled by user interrupt")
        self._reject_pending_prompts("Cancelled by user interrupt")
        await self._notify_tool_permission_cancelled(
            active_tool_permission_ids,
            "Cancelled by user interrupt",
        )
        if tool_permission_tasks:
            await asyncio.gather(*tool_permission_tasks, return_exceptions=True)
        await self.send("cancel", {"reason": "user_interrupt"})

    def reset_cancel(self) -> None:
        self._cancelled = False

    @property
    def is_cancelled(self) -> bool:
        return self._cancelled

    # ── Stderr monitoring ────────────────────────────────────────

    async def drain_stderr(self) -> None:
        """Read and log TUI stderr until the process exits."""
        proc = self._process
        if proc is None or proc.stderr is None:
            return
        while True:
            line = await proc.stderr.readline()
            if not line:
                break
            logger.info("[TUI stderr] %s", line.decode("utf-8", errors="replace").rstrip())

    # ── Helpers ──────────────────────────────────────────────────

    @staticmethod
    def _default_tui_entry() -> str:
        candidates = TUIBridge.default_tui_entry_candidates()
        for entry in candidates:
            if os.path.exists(entry):
                return entry
        return candidates[0]

    @staticmethod
    def _tui_entry_paths() -> tuple[str, str]:
        package_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        repo_root = os.path.dirname(package_root)
        source_entry = os.path.join(
            repo_root,
            "apps",
            "tui",
            "dist",
            "index.js",
        )
        packaged_entry = os.path.join(
            package_root,
            "packaged",
            "tui",
            "index.js",
        )
        return source_entry, packaged_entry

    @staticmethod
    def _source_checkout_root() -> str:
        package_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        return os.path.dirname(package_root)

    @staticmethod
    def running_from_source_checkout() -> bool:
        """Return True when the imported package is inside this repo checkout."""
        repo_root = TUIBridge._source_checkout_root()
        return os.path.isfile(
            os.path.join(repo_root, "pyproject.toml")
        ) and os.path.isdir(
            os.path.join(repo_root, "apps", "tui")
        )

    @staticmethod
    def default_tui_entry_candidates() -> list[str]:
        source_entry, packaged_entry = TUIBridge._tui_entry_paths()
        if TUIBridge.running_from_source_checkout():
            return [source_entry]
        return [packaged_entry]

    @staticmethod
    def default_tui_available() -> bool:
        return any(
            os.path.exists(entry)
            for entry in TUIBridge.default_tui_entry_candidates()
        )

    @staticmethod
    def interactive_terminal_available() -> bool:
        """Return whether the TUI can attach to a user terminal.

        The TUI reserves stdin/stdout for NDJSON IPC with Python, so the
        rendered Ink UI must open the controlling terminal separately.
        """
        if os.name == "nt":
            return bool(sys.stdin.isatty() and sys.stderr.isatty())

        try:
            fd = os.open("/dev/tty", os.O_RDWR)
        except OSError:
            return False
        else:
            os.close(fd)
            return True

    @staticmethod
    def default_tui_missing_hint() -> str:
        if TUIBridge.running_from_source_checkout():
            return (
                "Build the source TUI with `npm --prefix apps/tui run build`, "
                "or run without --tui."
            )
        return (
            "The packaged TUI artifact is missing. Reinstall OpenSpace from a "
            "package built with `npm --prefix apps/tui run build:packaged`, or "
            "run without --tui."
        )

    @staticmethod
    def _find_node() -> str:
        node = shutil.which("node")
        if node is None:
            raise FileNotFoundError(
                "Node.js not found on PATH. The TS TUI requires Node.js to run."
            )
        return node