diff --git a/src/deep_code_agent/tui/__init__.py b/src/deep_code_agent/tui/__init__.py index c8c866f..6f78387 100644 --- a/src/deep_code_agent/tui/__init__.py +++ b/src/deep_code_agent/tui/__init__.py @@ -1,7 +1,7 @@ """Deep Code Agent TUI - Textual-based terminal user interface. This module provides a rich terminal interface for the Deep Code Agent, -featuring a chat-style conversation view, HITL approval modals, and +featuring a chat-style conversation view, inline HITL approvals, and real-time streaming responses. Example: diff --git a/src/deep_code_agent/tui/app.py b/src/deep_code_agent/tui/app.py index aa4f1b5..822e27f 100644 --- a/src/deep_code_agent/tui/app.py +++ b/src/deep_code_agent/tui/app.py @@ -158,12 +158,8 @@ def action_toggle_dark(self) -> None: def action_help(self) -> None: """Show help.""" - self.notify( - "Enter send prompt\nCtrl+L clear chat\nCtrl+D theme\nTab navigate", - title="Help", - severity="information", - timeout=10, - ) + if self._main_screen is not None: + self._main_screen.action_help() def update_session_info(self, session_info: dict) -> None: """Update session information.""" diff --git a/src/deep_code_agent/tui/bridge/agent_bridge.py b/src/deep_code_agent/tui/bridge/agent_bridge.py index 3e04bd2..1eec5a3 100644 --- a/src/deep_code_agent/tui/bridge/agent_bridge.py +++ b/src/deep_code_agent/tui/bridge/agent_bridge.py @@ -263,7 +263,6 @@ async def process_request(self, message: str) -> None: pass except Exception as e: # Dispatch error event - self._run_on_app(self.app.notify, f"[ERROR] {e}", title="Error", severity="error") import traceback traceback.print_exc() @@ -272,11 +271,11 @@ async def process_request(self, message: str) -> None: async def resume_with_decision(self, decision: dict) -> None: """Resume after HITL decision. - Called when the user has made a decision in the ApprovalModal. + Called when the user has made a HITL approval decision. Resumes the agent stream with the decision. Args: - decision: User decision dict from ApprovalModal + decision: User decision dict from the approval UI """ if self.stream_handler is None or self.app is None: return @@ -537,9 +536,6 @@ def handle_event() -> None: asyncio.create_task(self.resume_with_decision({"decisions": decisions})) return - # Show approval modal - from deep_code_agent.tui.screens.approval_modal import ApprovalModal - def on_decision(decision: dict) -> None: # Check if tool should be added to auto-approve if decision.get("add_to_auto_approve", False): @@ -547,11 +543,6 @@ def on_decision(decision: dict) -> None: if tool_to_add and tool_to_add not in auto_approve_tools: # We're already in the main thread, so we can set directly setattr(app, "auto_approve_tools", auto_approve_tools + [tool_to_add]) - app.notify( - f"Auto-approve enabled for: {tool_to_add}", - title="Auto-Approve", - severity="information", - ) if "decisions" in decision and isinstance(decision["decisions"], list): decisions = decision["decisions"] @@ -567,8 +558,7 @@ def on_decision(decision: dict) -> None: decisions = [base for _ in range(self._pending_hitl_action_count)] asyncio.create_task(self.resume_with_decision({"decisions": decisions})) - modal = ApprovalModal(interrupt_data, callback=on_decision) - app.push_screen(modal) + chat_log.add_approval_request(interrupt_data, callback=on_decision) elif event.type == EventType.ERROR: self._reset_streaming_state() diff --git a/src/deep_code_agent/tui/screens/approval_modal.py b/src/deep_code_agent/tui/screens/approval_modal.py deleted file mode 100644 index 0f74c05..0000000 --- a/src/deep_code_agent/tui/screens/approval_modal.py +++ /dev/null @@ -1,307 +0,0 @@ -"""Approval modal for HITL (Human-in-the-Loop) approval.""" - -import json -from typing import Callable - -from textual.app import ComposeResult -from textual.containers import Vertical -from textual.screen import ModalScreen -from textual.widgets import Static - -from deep_code_agent.tui.widgets.selectable_option import SelectableOption - - -class ApprovalModal(ModalScreen): - """Modal screen for HITL action approval. - - Displays tool call details and provides keyboard-navigable options - for approve, reject, or cancel the action. - - Args: - interrupt_data: Data from LangGraph interrupt - callback: Function to call with user decision - """ - - BINDINGS = [ - ("up", "navigate_up", "Previous Option"), - ("down", "navigate_down", "Next Option"), - ("k", "navigate_up", "Previous Option"), - ("j", "navigate_down", "Next Option"), - ("1", "select_index(0)", "Select 1"), - ("2", "select_index(1)", "Select 2"), - ("3", "select_index(2)", "Select 3"), - ("4", "select_index(3)", "Select 4"), - ("enter", "confirm_selection", "Confirm"), - ("escape", "cancel", "Cancel"), - ] - - DEFAULT_CSS = """ - ApprovalModal { - align: center middle; - } - - ApprovalModal > Vertical { - width: 80; - height: auto; - max-height: 90%; - background: #111411; - border: thick #7dc4a4; - padding: 2; - } - - ApprovalModal #dialog-title { - content-align: center middle; - text-style: bold; - color: #f0f4ef; - height: 3; - margin-bottom: 1; - border-bottom: solid #343832; - } - - ApprovalModal #tool-info { - margin: 1 0; - padding: 1; - background: #171b17; - border: solid #2e342e; - height: auto; - max-height: 20; - overflow: auto; - } - - ApprovalModal #tool-info Static { - margin: 0; - } - - ApprovalModal #tool-name { - text-style: bold; - color: #a7cdbd; - margin-bottom: 1; - } - - ApprovalModal #options-list { - margin-top: 1; - } - """ - - def __init__(self, interrupt_data: dict, callback: Callable[[dict], None], **kwargs): - super().__init__(**kwargs) - self.interrupt_data = interrupt_data - self.callback = callback - self.selected_index = 0 - self._extract_tool_call() - self._setup_options() - - def _extract_tool_call(self) -> None: - """Extract tool call info from interrupt data.""" - self.tool_name = "unknown" - self.tool_args = {} - self.action_requests = [] - - try: - # Handle different interrupt data structures - if isinstance(self.interrupt_data, (list, tuple)) and len(self.interrupt_data) > 0: - item = self.interrupt_data[0] - if hasattr(item, "value"): - value = item.value - elif isinstance(item, dict) and "value" in item: - value = item.get("value") - else: - value = item - elif isinstance(self.interrupt_data, dict): - value = self.interrupt_data.get("value") if "value" in self.interrupt_data else self.interrupt_data - elif not isinstance(self.interrupt_data, (list, dict)) and hasattr(self.interrupt_data, "value"): - value = self.interrupt_data.value - else: - value = {} - - if not isinstance(value, dict): - value = {} - - # Try multiple paths to find tool call information - # Path 1: action_requests (common in deepagents) - action_requests = value.get("action_requests", []) - if action_requests: - self.action_requests = action_requests - action = action_requests[0] - action_data = action.action if hasattr(action, "action") else action - if isinstance(action_data, dict) and isinstance(action_data.get("action"), dict): - action_data = action_data["action"] - if hasattr(action_data, "model_dump"): - action_data = action_data.model_dump() - if not isinstance(action_data, dict): - action_data = {} - self.tool_name = action_data.get("name", "unknown") - self.tool_args = action_data.get("args", {}) - return - - # Path 2: tool_calls (common in LangGraph) - tool_calls = value.get("tool_calls", []) - if tool_calls: - tc = tool_calls[0] - tc_data = tc if isinstance(tc, dict) else getattr(tc, "model_dump", lambda: {})() - self.tool_name = tc_data.get("name", "unknown") - self.tool_args = tc_data.get("args", {}) - return - - # Path 3: Check for nested "action" key at top level - if "action" in value: - action = value["action"] - action_data = action if isinstance(action, dict) else getattr(action, "model_dump", lambda: {})() - self.tool_name = action_data.get("name", "unknown") - self.tool_args = action_data.get("args", {}) - return - - # Path 4: Look in messages for tool calls - if "messages" in value: - messages = value["messages"] - if messages: - msg = messages[0] - if hasattr(msg, "tool_calls") and msg.tool_calls: - tc = msg.tool_calls[0] - self.tool_name = tc.get("name", "unknown") - self.tool_args = tc.get("args", {}) - return - - # Path 5: Check if value has name directly (for Interrupt objects) - if isinstance(value, dict) and "name" in value: - self.tool_name = value.get("name", "unknown") - self.tool_args = value.get("args", {}) - return - - # Path 6: Check for __interrupt__ structure - if "__interrupt__" in value: - interrupt = value["__interrupt__"] - if isinstance(interrupt, list) and len(interrupt) > 0: - item = interrupt[0] - if hasattr(item, "value"): - item = item.value - if isinstance(item, dict): - self.tool_name = item.get("name", "unknown") - self.tool_args = item.get("args", {}) - return - - except Exception as e: - self.tool_name = "error" - self.tool_args = {"error": str(e), "debug": str(self.interrupt_data)[:200]} - - # Store raw data for debugging - self._debug_data = str(self.interrupt_data)[:500] - - def _setup_options(self) -> None: - """Setup options for the modal.""" - self.options = [ - {"key": "1", "label": "Approve", "description": "Allow this once", "action": "approve"}, - { - "key": "2", - "label": "Always Approve", - "description": "Trust this tool in this session", - "action": "approve_all", - }, - {"key": "3", "label": "Reject", "description": "Block execution", "action": "reject"}, - {"key": "4", "label": "Cancel", "description": "Dismiss dialog", "action": "cancel"}, - ] - - def compose(self) -> ComposeResult: - """Compose the approval modal.""" - with Vertical(): - yield Static("Action Requires Approval", id="dialog-title", markup=False) - - with Static(id="tool-info"): - tool_display = f"Tool: {self.tool_name}" - if self.tool_name == "unknown": - debug_data = getattr(self, "_debug_data", "N/A") - tool_display += f"\nDebug: {debug_data}" - yield Static(tool_display, id="tool-name", markup=False) - yield Static(self._format_args(self.tool_args), markup=False) - - with Vertical(id="options-list"): - yield Static("Choose an action:", markup=False) - for i, option in enumerate(self.options): - yield SelectableOption( - key=option["key"], - label=option["label"], - description=option["description"], - selected=(i == self.selected_index), - ) - - def _format_args(self, args: dict) -> str: - """Format tool arguments for display.""" - if not args: - return "No arguments" - - try: - formatted = json.dumps(args, indent=2, ensure_ascii=False, default=str) - except TypeError: - formatted = str(args) - if len(formatted) > 900: - formatted = formatted[:897] + "..." - return f"Arguments:\n{formatted}" - - def _update_selection(self) -> None: - """Update visual selection state.""" - try: - options_container = self.query_one("#options-list", Vertical) - selectable_options = options_container.query(SelectableOption) - for i, widget in enumerate(selectable_options): - widget.set_selected(i == self.selected_index) - except Exception: - pass - - def action_navigate_up(self) -> None: - """Navigate to previous option.""" - if self.selected_index > 0: - self.selected_index -= 1 - self._update_selection() - - def action_navigate_down(self) -> None: - """Navigate to next option.""" - if self.selected_index < len(self.options) - 1: - self.selected_index += 1 - self._update_selection() - - def action_select_index(self, index: int) -> None: - """Select option by index.""" - if 0 <= index < len(self.options): - self.selected_index = index - self._update_selection() - - def action_confirm_selection(self) -> None: - """Confirm current selection.""" - if 0 <= self.selected_index < len(self.options): - action = self.options[self.selected_index]["action"] - if action == "approve": - self._approve() - elif action == "approve_all": - self._approve_all() - elif action == "reject": - self._reject() - elif action == "cancel": - self._cancel() - - def action_cancel(self) -> None: - """Cancel the modal.""" - self._cancel() - - def _approve(self) -> None: - """Approve the action.""" - decision = {"type": "approve"} - self.callback(decision) - self.dismiss() - - def _approve_all(self) -> None: - """Approve and add tool to auto-approve list.""" - decision = {"type": "approve", "add_to_auto_approve": True, "tool_name": self.tool_name} - self.callback(decision) - self.dismiss() - - def _reject(self) -> None: - """Reject the action.""" - decision = {"type": "reject", "message": "Action rejected by user"} - self.callback(decision) - self.dismiss() - - def _cancel(self) -> None: - """Cancel the modal.""" - decision = {"type": "reject", "message": "Action cancelled by user"} - self.callback(decision) - self.dismiss() diff --git a/src/deep_code_agent/tui/screens/main_screen.py b/src/deep_code_agent/tui/screens/main_screen.py index 47fe469..9d7a8dc 100644 --- a/src/deep_code_agent/tui/screens/main_screen.py +++ b/src/deep_code_agent/tui/screens/main_screen.py @@ -72,13 +72,7 @@ def action_toggle_dark(self) -> None: def action_help(self) -> None: """Show help screen.""" - commands = "\n".join(f"{command.name} - {command.description}" for command in SLASH_COMMANDS) - self.notify( - f"Enter send\nCtrl+L clear chat\nCtrl+D theme\n\n{commands}", - title="Shortcuts", - severity="information", - timeout=8, - ) + self.get_chat_log().add_system_message(self._format_help_message()) def action_clear_chat(self) -> None: """Clear the conversation stream.""" @@ -130,7 +124,7 @@ async def process_agent_request(self, content: str) -> None: await bridge.process_request(content) except Exception as e: app = cast("DeepCodeAgentApp", self.app) - app.call_from_thread(self.notify, f"[ERROR] Worker error: {e}", title="ERROR", severity="error") + app.call_from_thread(self._show_worker_error, str(e)) import traceback traceback.print_exc() @@ -169,6 +163,25 @@ def _handle_local_command(self, content: str) -> bool: return True return False + def _show_worker_error(self, message: str) -> None: + """Render worker failures in the main UI instead of a transient popup.""" + try: + self.get_status_bar().set_error(message) + except Exception: + pass + try: + self.get_input_box().set_disabled(False) + except Exception: + pass + try: + self.get_chat_log().add_system_message(f"Error: {message}") + except Exception: + pass + + def _format_help_message(self) -> str: + commands = "\n".join(f"- {command.name}: {command.description}" for command in SLASH_COMMANDS) + return f"Shortcuts:\n- Enter: send\n- Ctrl+L: clear chat\n- Ctrl+D: theme\n\nCommands:\n{commands}" + def _format_skills_message(self) -> str: skills = self.session_info.get("skills") or [] names: list[str] = [] diff --git a/src/deep_code_agent/tui/utils/approval.py b/src/deep_code_agent/tui/utils/approval.py new file mode 100644 index 0000000..84c1824 --- /dev/null +++ b/src/deep_code_agent/tui/utils/approval.py @@ -0,0 +1,129 @@ +"""Helpers for rendering human-in-the-loop approval requests.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class ApprovalToolCall: + """Tool call details extracted from a LangGraph interrupt.""" + + tool_name: str = "unknown" + tool_args: dict[str, Any] = field(default_factory=dict) + action_requests: list[Any] = field(default_factory=list) + debug_data: str = "" + + +def extract_approval_tool_call(interrupt_data: Any) -> ApprovalToolCall: + """Extract the first tool call from supported interrupt shapes.""" + debug_data = str(interrupt_data)[:500] + + try: + value = _unwrap_interrupt_value(interrupt_data) + if not isinstance(value, dict): + return ApprovalToolCall(debug_data=debug_data) + + action_requests = value.get("action_requests", []) + if action_requests: + action = action_requests[0] + action_data = _normalize_action_data(action) + if isinstance(action_data, dict): + return ApprovalToolCall( + tool_name=str(action_data.get("name", "unknown")), + tool_args=_normalize_args(action_data.get("args", {})), + action_requests=list(action_requests), + debug_data=debug_data, + ) + + tool_calls = value.get("tool_calls", []) + if tool_calls: + tool_call = _dump_model(tool_calls[0]) + if isinstance(tool_call, dict): + return ApprovalToolCall( + tool_name=str(tool_call.get("name", "unknown")), + tool_args=_normalize_args(tool_call.get("args", {})), + debug_data=debug_data, + ) + + if "action" in value: + action_data = _dump_model(value["action"]) + if isinstance(action_data, dict): + return ApprovalToolCall( + tool_name=str(action_data.get("name", "unknown")), + tool_args=_normalize_args(action_data.get("args", {})), + debug_data=debug_data, + ) + + messages = value.get("messages", []) + if messages: + message = messages[0] + tool_calls = getattr(message, "tool_calls", None) + if tool_calls: + tool_call = _dump_model(tool_calls[0]) + if isinstance(tool_call, dict): + return ApprovalToolCall( + tool_name=str(tool_call.get("name", "unknown")), + tool_args=_normalize_args(tool_call.get("args", {})), + debug_data=debug_data, + ) + + if "name" in value: + return ApprovalToolCall( + tool_name=str(value.get("name", "unknown")), + tool_args=_normalize_args(value.get("args", {})), + debug_data=debug_data, + ) + + nested_interrupt = value.get("__interrupt__") + if isinstance(nested_interrupt, list) and nested_interrupt: + nested_value = _unwrap_interrupt_value(nested_interrupt[0]) + if isinstance(nested_value, dict): + return ApprovalToolCall( + tool_name=str(nested_value.get("name", "unknown")), + tool_args=_normalize_args(nested_value.get("args", {})), + debug_data=debug_data, + ) + + except Exception as exc: + return ApprovalToolCall(tool_name="error", tool_args={"error": str(exc), "debug": debug_data}) + + return ApprovalToolCall(debug_data=debug_data) + + +def _unwrap_interrupt_value(interrupt_data: Any) -> Any: + if isinstance(interrupt_data, (list, tuple)) and interrupt_data: + item = interrupt_data[0] + if hasattr(item, "value"): + return item.value + if isinstance(item, dict) and "value" in item: + return item.get("value") + return item + if isinstance(interrupt_data, dict): + return interrupt_data.get("value") if "value" in interrupt_data else interrupt_data + if not isinstance(interrupt_data, (list, dict)) and hasattr(interrupt_data, "value"): + return interrupt_data.value + return {} + + +def _normalize_action_data(action: Any) -> dict[str, Any] | None: + action_data = action.action if hasattr(action, "action") else action + if isinstance(action_data, dict) and isinstance(action_data.get("action"), dict): + action_data = action_data["action"] + action_data = _dump_model(action_data) + return action_data if isinstance(action_data, dict) else None + + +def _dump_model(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump() + return value + + +def _normalize_args(args: Any) -> dict[str, Any]: + if args is None: + return {} + if isinstance(args, dict): + return args + return {"value": args} diff --git a/src/deep_code_agent/tui/widgets/__init__.py b/src/deep_code_agent/tui/widgets/__init__.py index 7e97d94..833169a 100644 --- a/src/deep_code_agent/tui/widgets/__init__.py +++ b/src/deep_code_agent/tui/widgets/__init__.py @@ -1,9 +1,9 @@ """TUI widgets for Deep Code Agent.""" +from deep_code_agent.tui.widgets.approval_request import ApprovalRequest from deep_code_agent.tui.widgets.chat_log import ChatLog from deep_code_agent.tui.widgets.input_box import InputBox from deep_code_agent.tui.widgets.message_bubble import MessageBubble -from deep_code_agent.tui.widgets.selectable_option import SelectableOption from deep_code_agent.tui.widgets.session_header import SessionHeader from deep_code_agent.tui.widgets.side_panel import SidePanel from deep_code_agent.tui.widgets.status_bar import StatusBar @@ -12,9 +12,9 @@ __all__ = [ "ChatLog", + "ApprovalRequest", "InputBox", "MessageBubble", - "SelectableOption", "SessionHeader", "SidePanel", "StatusBar", diff --git a/src/deep_code_agent/tui/widgets/approval_request.py b/src/deep_code_agent/tui/widgets/approval_request.py new file mode 100644 index 0000000..bc8a4af --- /dev/null +++ b/src/deep_code_agent/tui/widgets/approval_request.py @@ -0,0 +1,326 @@ +"""Inline approval request widget for HITL tool calls.""" + +from __future__ import annotations + +import json +from collections.abc import Callable + +from textual.app import ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal, Vertical +from textual.reactive import reactive +from textual.widgets import Static + +from deep_code_agent.tui.utils.approval import extract_approval_tool_call + + +class ApprovalChoice(Horizontal): + """A compact selectable row inside an approval request.""" + + DEFAULT_CSS = """ + ApprovalChoice { + width: 100%; + height: 1; + padding: 0 1; + background: transparent; + } + + ApprovalChoice.selected { + background: #303030; + color: #f2f2f2; + } + + ApprovalChoice #approval-choice-marker { + width: 2; + color: #7dc4a4; + text-style: bold; + } + + ApprovalChoice #approval-choice-label { + width: 24; + text-style: bold; + } + + ApprovalChoice #approval-choice-description { + width: 1fr; + color: #a5a5a5; + text-align: right; + } + + Screen:light ApprovalChoice.selected { + background: #d9efe6; + color: #202020; + } + + Screen:light ApprovalChoice #approval-choice-description { + color: #5d665f; + } + """ + + selected = reactive(False) + + def __init__( + self, + index: int, + key: str, + label: str, + description: str, + selected: bool = False, + **kwargs, + ): + super().__init__(**kwargs) + self.index = index + self.key = key + self.label = label + self.description = description + self.selected = selected + if selected: + self.add_class("selected") + + def compose(self) -> ComposeResult: + marker = "›" if self.selected else " " + yield Static(marker, id="approval-choice-marker", markup=False) + yield Static(f"{self.key}. {self.label}", id="approval-choice-label", markup=False) + yield Static(self.description, id="approval-choice-description", markup=False) + + def on_click(self, event) -> None: + event.stop() + parent = self.parent + while parent is not None and not isinstance(parent, ApprovalRequest): + parent = parent.parent + if parent is not None: + parent.focus() + + def watch_selected(self, selected: bool) -> None: + try: + self.query_one("#approval-choice-marker", Static).update("›" if selected else " ") + except Exception: + pass + self.set_class(selected, "selected") + + def set_selected(self, selected: bool) -> None: + self.selected = selected + + +class ApprovalRequest(Vertical, can_focus=True): + """Inline approval card shown in the main transcript.""" + + BINDINGS = [ + Binding("up", "navigate_up", "Previous Option", show=False, priority=True), + Binding("down", "navigate_down", "Next Option", show=False, priority=True), + Binding("k", "navigate_up", "Previous Option", show=False, priority=True), + Binding("j", "navigate_down", "Next Option", show=False, priority=True), + Binding("1", "select_index(0)", "Select 1", show=False, priority=True), + Binding("2", "select_index(1)", "Select 2", show=False, priority=True), + Binding("3", "select_index(2)", "Select 3", show=False, priority=True), + Binding("4", "select_index(3)", "Select 4", show=False, priority=True), + Binding("enter", "confirm_selection", "Confirm", show=False, priority=True), + Binding("escape", "cancel", "Cancel", show=False, priority=True), + ] + + DEFAULT_CSS = """ + ApprovalRequest { + width: 100%; + height: auto; + margin: 0 0 1 0; + padding: 1 1; + background: #202020; + border: solid #444444; + } + + ApprovalRequest:focus { + border: tall #7dc4a4; + } + + ApprovalRequest.resolved { + border: solid #383838; + background: transparent; + } + + ApprovalRequest #approval-title { + height: 1; + color: #ffcf6b; + text-style: bold; + } + + ApprovalRequest #approval-summary { + height: auto; + color: #eeeeee; + margin-top: 1; + text-wrap: wrap; + } + + ApprovalRequest #approval-args { + height: auto; + margin-top: 1; + color: #b8b8b8; + text-wrap: wrap; + } + + ApprovalRequest #approval-options { + height: auto; + margin-top: 1; + } + + ApprovalRequest #approval-help { + height: 1; + margin-top: 1; + color: #8f8f8f; + } + + Screen:light ApprovalRequest { + background: #ffffff; + border: solid #c6c6c6; + } + + Screen:light ApprovalRequest:focus { + border: tall #40916c; + } + + Screen:light ApprovalRequest #approval-summary { + color: #222222; + } + + Screen:light ApprovalRequest #approval-args, + Screen:light ApprovalRequest #approval-help { + color: #666666; + } + """ + + OPTIONS = [ + {"key": "1", "label": "Approve", "description": "allow once", "action": "approve"}, + {"key": "2", "label": "Always Approve", "description": "trust this tool", "action": "approve_all"}, + {"key": "3", "label": "Reject", "description": "block execution", "action": "reject"}, + {"key": "4", "label": "Cancel", "description": "reject and stop waiting", "action": "cancel"}, + ] + + selected_index = reactive(0) + + def __init__(self, interrupt_data: object, callback: Callable[[dict], None], **kwargs): + super().__init__(**kwargs) + self.interrupt_data = interrupt_data + self.callback = callback + tool_call = extract_approval_tool_call(interrupt_data) + self.tool_name = tool_call.tool_name + self.tool_args = tool_call.tool_args + self.action_requests = tool_call.action_requests + self._debug_data = tool_call.debug_data + self._resolved = False + + def compose(self) -> ComposeResult: + yield Static("Tool approval required", id="approval-title", markup=False) + yield Static(self._summary_text(), id="approval-summary", markup=False) + yield Static(self._format_args(self.tool_args), id="approval-args", markup=False) + with Vertical(id="approval-options"): + for index, option in enumerate(self.OPTIONS): + yield ApprovalChoice( + index=index, + key=option["key"], + label=option["label"], + description=option["description"], + selected=(index == self.selected_index), + ) + yield Static("↑/↓ choose Enter confirm 1-4 jump Esc cancel", id="approval-help", markup=False) + + def _summary_text(self) -> str: + if self.tool_name == "unknown": + return f"An unknown tool wants to run.\nDebug: {self._debug_data}" + return f"{self.tool_name} wants to run." + + def _format_args(self, args: dict) -> str: + if not args: + return "Arguments: none" + + try: + formatted = json.dumps(args, indent=2, ensure_ascii=False, default=str) + except TypeError: + formatted = str(args) + + lines = formatted.splitlines() + if len(lines) > 8: + lines = [*lines[:7], "..."] + clipped = [] + for line in lines: + clipped.append(line if len(line) <= 150 else line[:147] + "...") + return "Arguments:\n" + "\n".join(clipped) + + def watch_selected_index(self, selected_index: int) -> None: + try: + choices = self.query(ApprovalChoice) + for index, choice in enumerate(choices): + choice.set_selected(index == selected_index) + except Exception: + pass + + def action_navigate_up(self) -> None: + if self._resolved: + return + if self.selected_index > 0: + self.selected_index -= 1 + + def action_navigate_down(self) -> None: + if self._resolved: + return + if self.selected_index < len(self.OPTIONS) - 1: + self.selected_index += 1 + + def action_select_index(self, index: int) -> None: + if self._resolved: + return + if 0 <= index < len(self.OPTIONS): + self.selected_index = index + + def action_confirm_selection(self) -> None: + if self._resolved or not 0 <= self.selected_index < len(self.OPTIONS): + return + action = self.OPTIONS[self.selected_index]["action"] + if action == "approve": + self._approve() + elif action == "approve_all": + self._approve_all() + elif action == "reject": + self._reject() + elif action == "cancel": + self._cancel() + + def action_cancel(self) -> None: + self._cancel() + + def _approve(self) -> None: + self._resolve({"type": "approve"}, f"Approved {self._display_tool_name()}.") + + def _approve_all(self) -> None: + decision = {"type": "approve", "add_to_auto_approve": True, "tool_name": self.tool_name} + self._resolve(decision, f"Always approving {self._display_tool_name()} in this session.") + + def _reject(self) -> None: + decision = {"type": "reject", "message": "Action rejected by user"} + self._resolve(decision, f"Rejected {self._display_tool_name()}.") + + def _cancel(self) -> None: + decision = {"type": "reject", "message": "Action cancelled by user"} + self._resolve(decision, f"Cancelled {self._display_tool_name()}.") + + def _resolve(self, decision: dict, summary: str) -> None: + if self._resolved: + return + self._resolved = True + self.add_class("resolved") + self._collapse(summary) + self.callback(decision) + + def _collapse(self, summary: str) -> None: + try: + self.query_one("#approval-title", Static).update("Tool approval resolved") + self.query_one("#approval-summary", Static).update(summary) + except Exception: + pass + + for selector in ("#approval-args", "#approval-options", "#approval-help"): + try: + self.query_one(selector).remove() + except Exception: + pass + + def _display_tool_name(self) -> str: + return self.tool_name if self.tool_name and self.tool_name != "unknown" else "tool" diff --git a/src/deep_code_agent/tui/widgets/chat_log.py b/src/deep_code_agent/tui/widgets/chat_log.py index c7e3e05..0d89441 100644 --- a/src/deep_code_agent/tui/widgets/chat_log.py +++ b/src/deep_code_agent/tui/widgets/chat_log.py @@ -161,6 +161,17 @@ def add_tool_call_widget( self._scroll_to_bottom() return widget + def add_approval_request(self, interrupt_data, callback, focus: bool = True): + """Add an inline HITL approval request to the chat log.""" + from deep_code_agent.tui.widgets.approval_request import ApprovalRequest + + widget = ApprovalRequest(interrupt_data, callback=callback) + self._mount_above_todos_card(widget) + self._scroll_to_bottom() + if focus: + self.call_after_refresh(widget.focus) + return widget + def upsert_todos_card(self, todos: list[dict[str, str]]) -> TodosProgressCard: """Create or update the singleton todos progress card. diff --git a/src/deep_code_agent/tui/widgets/selectable_option.py b/src/deep_code_agent/tui/widgets/selectable_option.py deleted file mode 100644 index 563ae6d..0000000 --- a/src/deep_code_agent/tui/widgets/selectable_option.py +++ /dev/null @@ -1,89 +0,0 @@ -"""Selectable option widget for approval modal.""" - -from textual.containers import Horizontal -from textual.reactive import reactive -from textual.widgets import Static - - -class SelectableOption(Horizontal): - """A selectable option widget. - - Args: - key: The option key (e.g., "1", "2") - label: The option label - description: Short description - selected: Whether this option is currently selected - """ - - DEFAULT_CSS = """ - SelectableOption { - width: 100%; - height: auto; - padding: 1; - margin-bottom: 1; - border: solid #30362f; - background: #151817; - } - - SelectableOption.selected { - border: tall #7dc4a4; - background: #18231d; - } - - SelectableOption #option-marker { - width: 3; - text-style: bold; - color: #7dc4a4; - } - - SelectableOption #option-label { - width: 1fr; - text-style: bold; - } - - SelectableOption #option-description { - width: auto; - color: #90998f; - } - """ - - selected = reactive(False) - key = reactive("") - label = reactive("") - description = reactive("") - - def __init__( - self, - key: str = "", - label: str = "", - description: str = "", - selected: bool = False, - **kwargs - ): - super().__init__(**kwargs) - self.key = key - self.label = label - self.description = description - self.selected = selected - if selected: - self.add_class("selected") - - def compose(self): - """Compose the option widget.""" - marker = ">" if self.selected else " " - yield Static(marker, id="option-marker", markup=False) - yield Static(f"{self.key}. {self.label}", id="option-label", markup=False) - yield Static(self.description, id="option-description", markup=False) - - def watch_selected(self, selected: bool) -> None: - """Update marker when selection changes.""" - try: - marker_widget = self.query_one("#option-marker", Static) - marker_widget.update(">" if selected else " ") - except Exception: - pass - self.set_class(selected, "selected") - - def set_selected(self, selected: bool) -> None: - """Set selection state.""" - self.selected = selected diff --git a/tests/tui/test_approval_modal.py b/tests/tui/test_approval_modal.py deleted file mode 100644 index 5e85158..0000000 --- a/tests/tui/test_approval_modal.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Tests for ApprovalModal screen.""" - -import pytest -from textual.app import App - - -def test_approval_modal_keyboard_navigation(): - """Test that modal can navigate options with keyboard.""" - import asyncio - from deep_code_agent.tui.screens.approval_modal import ApprovalModal - - async def run_test(): - app = App() - async with app.run_test() as pilot: - interrupt_data = { - "action_requests": [ - {"action": {"name": "read_file", "args": {"path": "test.txt"}}} - ] - } - - results = {} - def callback(decision): - results["decision"] = decision - - modal = ApprovalModal(interrupt_data, callback=callback) - await app.push_screen(modal) - - # Test initial state - assert modal.selected_index == 0 - - asyncio.run(run_test()) - - -def test_approval_modal_up_navigation(): - """Test up navigation wraps or stops at top.""" - import asyncio - from deep_code_agent.tui.screens.approval_modal import ApprovalModal - - async def run_test(): - app = App() - async with app.run_test() as pilot: - interrupt_data = { - "action_requests": [ - {"action": {"name": "read_file", "args": {}}} - ] - } - - modal = ApprovalModal(interrupt_data, callback=lambda d: None) - await app.push_screen(modal) - - # Start at index 0 - assert modal.selected_index == 0 - - # Navigate up (should stay at 0) - modal.action_navigate_up() - assert modal.selected_index == 0 - - asyncio.run(run_test()) - - -def test_approval_modal_down_navigation(): - """Test down navigation moves to next option.""" - import asyncio - from deep_code_agent.tui.screens.approval_modal import ApprovalModal - - async def run_test(): - app = App() - async with app.run_test() as pilot: - interrupt_data = { - "action_requests": [ - {"action": {"name": "read_file", "args": {}}} - ] - } - - modal = ApprovalModal(interrupt_data, callback=lambda d: None) - await app.push_screen(modal) - - # Navigate down - modal.action_navigate_down() - assert modal.selected_index == 1 - - modal.action_navigate_down() - assert modal.selected_index == 2 - - modal.action_navigate_down() - assert modal.selected_index == 3 - - # Navigate down at last index (should stay at 3) - modal.action_navigate_down() - assert modal.selected_index == 3 - - asyncio.run(run_test()) - - -def test_approval_modal_cancel_emits_rejection_decision(): - """Cancel should resolve the interrupt instead of leaving it hanging.""" - import asyncio - from deep_code_agent.tui.screens.approval_modal import ApprovalModal - - async def run_test(): - app = App() - async with app.run_test() as pilot: - interrupt_data = { - "action_requests": [ - {"action": {"name": "read_file", "args": {}}} - ] - } - - results = {} - - def callback(decision): - results["decision"] = decision - - modal = ApprovalModal(interrupt_data, callback=callback) - await app.push_screen(modal) - - modal.action_cancel() - - assert results["decision"] == { - "type": "reject", - "message": "Action cancelled by user", - } - - asyncio.run(run_test()) - - -def test_approval_modal_extracts_nested_action_request(): - """DeepAgents action request wrappers should render the real tool name.""" - from deep_code_agent.tui.screens.approval_modal import ApprovalModal - - interrupt_data = { - "action_requests": [ - {"action": {"name": "read_file", "args": {"path": "test.txt"}}} - ] - } - - modal = ApprovalModal(interrupt_data, callback=lambda d: None) - - assert modal.tool_name == "read_file" - assert modal.tool_args == {"path": "test.txt"} diff --git a/tests/tui/test_main_screen_commands.py b/tests/tui/test_main_screen_commands.py index 2acf6f6..119051f 100644 --- a/tests/tui/test_main_screen_commands.py +++ b/tests/tui/test_main_screen_commands.py @@ -63,3 +63,37 @@ async def run_test(): assert "gpt-test" in chat_text asyncio.run(run_test()) + + +def test_tui_help_command_renders_in_chat_log(): + """Slash help should stay in the transcript instead of showing a popup.""" + from deep_code_agent.tui.app import DeepCodeAgentApp + from deep_code_agent.tui.widgets.input_box import InputBox + + class ExplodingBridge: + async def process_request(self, content: str) -> None: + raise AssertionError(f"agent should not receive local command: {content}") + + async def run_test(): + app = DeepCodeAgentApp(agent=object()) + async with app.run_test() as pilot: + await pilot.pause() + screen = app._main_screen + assert screen is not None + app.bridge = ExplodingBridge() + + notifications = [] + + def capture_notify(*args, **kwargs): + notifications.append((args, kwargs)) + + app.notify = capture_notify + screen.on_input_box_user_input(InputBox.UserInput("/help")) + await pilot.pause() + + chat_text = "\n".join(str(getattr(child, "content", "")) for child in app._chat_log.children) + assert "Shortcuts:" in chat_text + assert "/skills" in chat_text + assert notifications == [] + + asyncio.run(run_test()) diff --git a/tests/tui/test_widgets.py b/tests/tui/test_widgets.py index 643beac..4d19e62 100644 --- a/tests/tui/test_widgets.py +++ b/tests/tui/test_widgets.py @@ -4,23 +4,6 @@ from textual.widgets import Input, Static -def test_selectable_option_creation(): - """Test that SelectableOption can be created and rendered.""" - from deep_code_agent.tui.widgets.selectable_option import SelectableOption - - option = SelectableOption( - key="1", - label="Approve", - description="Allow execution", - selected=True - ) - - assert option.key == "1" - assert option.label == "Approve" - assert option.description == "Allow execution" - assert option.selected is True - - def test_message_bubble_update_before_compose_is_safe(): """Streaming can update a newly mounted bubble before compose runs.""" from deep_code_agent.tui.widgets.message_bubble import MessageBubble @@ -232,3 +215,119 @@ def test_side_panel_compacts_long_codebase_path(): assert compact.startswith("...") assert len(compact) <= 26 + + +def test_approval_request_resolves_inline_decision(): + """The inline approval card should emit the same decision shape as the modal.""" + import asyncio + + from deep_code_agent.tui.widgets.approval_request import ApprovalRequest + + async def run_test(): + async with App().run_test() as pilot: + results = {} + interrupt_data = { + "action_requests": [ + {"action": {"name": "write_file", "args": {"path": "hello.py"}}} + ] + } + widget = ApprovalRequest(interrupt_data, callback=lambda decision: results.setdefault("decision", decision)) + await pilot.app.mount(widget) + await pilot.pause() + + assert widget.tool_name == "write_file" + assert widget.selected_index == 0 + + widget.action_navigate_down() + assert widget.selected_index == 1 + + widget.action_select_index(2) + widget.action_confirm_selection() + await pilot.pause() + + assert results["decision"] == { + "type": "reject", + "message": "Action rejected by user", + } + assert widget.has_class("resolved") + + asyncio.run(run_test()) + + +def test_chat_log_adds_inline_approval_request(): + """Approval prompts should mount inside the transcript, not as a screen.""" + import asyncio + + from deep_code_agent.tui.widgets.approval_request import ApprovalRequest + from deep_code_agent.tui.widgets.chat_log import ChatLog + + async def run_test(): + async with App().run_test() as pilot: + chat_log = ChatLog() + await pilot.app.mount(chat_log) + await pilot.pause() + + widget = chat_log.add_approval_request( + {"action_requests": [{"action": {"name": "terminal", "args": {"cmd": "pwd"}}}]}, + callback=lambda decision: None, + ) + await pilot.pause() + + assert isinstance(widget, ApprovalRequest) + assert widget in chat_log.children + + asyncio.run(run_test()) + + +def test_chat_log_focuses_inline_approval_request(): + """Keyboard navigation should work without clicking the approval card first.""" + import asyncio + + from deep_code_agent.tui.widgets.chat_log import ChatLog + + async def run_test(): + async with App().run_test() as pilot: + chat_log = ChatLog() + await pilot.app.mount(chat_log) + await pilot.pause() + + widget = chat_log.add_approval_request( + {"action_requests": [{"action": {"name": "terminal", "args": {"cmd": "pwd"}}}]}, + callback=lambda decision: None, + ) + await pilot.pause() + + assert pilot.app.focused is widget + + await pilot.press("down") + await pilot.pause() + assert widget.selected_index == 1 + + asyncio.run(run_test()) + + +def test_approval_request_mouse_click_only_focuses(): + """Clicking an option should not approve or reject the request.""" + import asyncio + + from deep_code_agent.tui.widgets.approval_request import ApprovalChoice, ApprovalRequest + + async def run_test(): + async with App().run_test() as pilot: + decisions = [] + widget = ApprovalRequest( + {"action_requests": [{"action": {"name": "write_file", "args": {"path": "hello.py"}}}]}, + callback=decisions.append, + ) + await pilot.app.mount(widget) + await pilot.pause() + + clicked = await pilot.click(ApprovalChoice) + await pilot.pause() + + assert clicked is True + assert decisions == [] + assert widget.selected_index == 0 + assert pilot.app.focused is widget + + asyncio.run(run_test())