mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-22 01:35:28 -04:00
feat(code): clear chat input via esc+esc, add [ X ]/[ COPY ] buttons (#4000)
Clear the chat input with <kbd>esc</kbd>+<kbd>esc</kbd> (undoable via <kbd>ctrl+z</kbd>) or the new on-screen `[ X ]`/`[ COPY ]` buttons. --- Adds quick ways to discard or copy a chat-input draft when you change your mind, instead of selecting all and deleting. - <kbd>esc + esc</kbd> clears the draft: the first <kbd>Esc</kbd> (when there is nothing else to interrupt) arms a pending clear and shows a hint; a second <kbd>Esc</kbd> within the window clears it. - <kbd>ctrl+z</kbd> to undo: the clear is an undoable `TextArea` edit (via `discard_text`), so a mistaken clear can be restored. - **`[ X ]` and `[ COPY ]` buttons** on the right of the input row provide discoverable mouse alternatives — `[ X ]` clears (undoable), `[ COPY ]` copies the draft to the clipboard. This addresses the request for buttons so these operations don't require memorizing shortcuts. The buttons use fixed widths because `width: auto` collapsed to 0 next to the `1fr` text area. Made by [Open SWE](https://openswe.vercel.app) --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
@@ -1008,6 +1008,20 @@ _DEFERRED_APPROVAL_TIMEOUT_SECONDS: float = 30.0
|
||||
"""Maximum seconds the deferred-approval worker will wait for the user to stop
|
||||
typing before showing the approval widget regardless."""
|
||||
|
||||
_RAPID_QUIT_CTRL_C_PRESSES: int = 2
|
||||
"""Consecutive rapid `Ctrl+C` presses that force the quit sequence.
|
||||
|
||||
When a draft is present, a single `Ctrl+C` copies it (matching terminal copy
|
||||
semantics), which otherwise leaves no way to reach the quit arm by pressing
|
||||
`Ctrl+C`. Mashing `Ctrl+C` is the universal "get me out" reflex, so the second
|
||||
rapid press bypasses the copy branches and arms quit instead.
|
||||
"""
|
||||
|
||||
_RAPID_QUIT_CTRL_C_WINDOW_SECONDS: float = 1.0
|
||||
"""Window within which repeated `Ctrl+C` presses count toward the rapid-quit
|
||||
escape hatch. Tight enough that a deliberate copy-then-interrupt sequence,
|
||||
which has much larger gaps, never trips it."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class QueuedMessage:
|
||||
@@ -2033,6 +2047,16 @@ class DeepAgentsApp(App):
|
||||
self._quit_pending = False
|
||||
"""True after a first `Ctrl+C` so a second press within the window quits."""
|
||||
|
||||
self._ctrl_c_times: list[float] = []
|
||||
"""Monotonic timestamps of recent `Ctrl+C` presses, pruned to
|
||||
`_RAPID_QUIT_CTRL_C_WINDOW_SECONDS`. Drives the rapid-quit escape hatch
|
||||
so mashing `Ctrl+C` bypasses the clipboard-copy branches and reaches the
|
||||
quit arm even when a draft is present."""
|
||||
|
||||
self._clear_input_pending = False
|
||||
"""True after a first `Esc` (with nothing else to interrupt) so a second
|
||||
press within the window clears the chat input draft."""
|
||||
|
||||
self._thread_switching = False
|
||||
"""Re-entry guard for `/threads` switches; blocks message handling
|
||||
until the new thread's history finishes loading."""
|
||||
@@ -8361,11 +8385,26 @@ class DeepAgentsApp(App):
|
||||
4. If ask_user menu is active, cancel it
|
||||
5. If agent is running, interrupt it (preserve input)
|
||||
6. If double press (quit_pending), quit
|
||||
7. Otherwise show quit hint
|
||||
7. If a focused input has text, copy the whole draft (no selection)
|
||||
8. Otherwise show quit hint
|
||||
|
||||
Rapid escape hatch: the clipboard-copy branches (1 and 7) are skipped
|
||||
once `Ctrl+C` is pressed `_RAPID_QUIT_CTRL_C_PRESSES` times within
|
||||
`_RAPID_QUIT_CTRL_C_WINDOW_SECONDS`. Without this, a non-empty draft
|
||||
makes branch 7 copy on every press, so the quit arm is unreachable by
|
||||
`Ctrl+C` alone. Mashing `Ctrl+C` then falls through to arm quit (and a
|
||||
further press exits). The interrupt branches (2-5) stay unconditional so
|
||||
a repeated press still cancels in-flight work rather than quitting.
|
||||
"""
|
||||
now = _monotonic()
|
||||
window = _RAPID_QUIT_CTRL_C_WINDOW_SECONDS
|
||||
self._ctrl_c_times = [t for t in self._ctrl_c_times if now - t <= window]
|
||||
self._ctrl_c_times.append(now)
|
||||
rapid = len(self._ctrl_c_times) >= _RAPID_QUIT_CTRL_C_PRESSES
|
||||
|
||||
# If a focused input widget has selected text, copy it instead of
|
||||
# quitting/interrupting so Ctrl+C matches standard terminal behavior.
|
||||
if self._copy_focused_selection():
|
||||
if not rapid and self._copy_focused_selection():
|
||||
self._quit_pending = False
|
||||
return
|
||||
|
||||
@@ -8399,11 +8438,21 @@ class DeepAgentsApp(App):
|
||||
self._quit_pending = False
|
||||
return
|
||||
|
||||
# Double Ctrl+C to quit
|
||||
# Double Ctrl+C to quit. Once the quit hint is visible, preserve the
|
||||
# armed quit path before draft-copy handling gets another chance to
|
||||
# consume Ctrl+C and clear `_quit_pending`.
|
||||
if self._quit_pending:
|
||||
self.exit()
|
||||
else:
|
||||
self._arm_quit_pending("Ctrl+C")
|
||||
return
|
||||
|
||||
# No selection and nothing to interrupt: copy the whole input draft so
|
||||
# Ctrl+C copies what was typed instead of arming quit. Skipped on a rapid
|
||||
# repeat press so mashing Ctrl+C escapes the copy loop and reaches quit.
|
||||
if not rapid and self._copy_focused_input_text():
|
||||
self._quit_pending = False
|
||||
return
|
||||
|
||||
self._arm_quit_pending("Ctrl+C")
|
||||
|
||||
def _copy_focused_selection(self) -> bool:
|
||||
"""Copy the focused input's selection to the clipboard, if any.
|
||||
@@ -8428,19 +8477,43 @@ class DeepAgentsApp(App):
|
||||
if not selected_text:
|
||||
return False
|
||||
|
||||
from deepagents_code.clipboard import copy_text_to_clipboard
|
||||
from deepagents_code.clipboard import copy_text_with_feedback
|
||||
|
||||
success, error = copy_text_to_clipboard(self, selected_text)
|
||||
if not success:
|
||||
self.notify(
|
||||
f"Failed to copy selection: {error}"
|
||||
if error
|
||||
else "Failed to copy selection - no clipboard method available",
|
||||
severity="warning",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
return success
|
||||
return copy_text_with_feedback(self, selected_text, failure_noun="selection")
|
||||
|
||||
def _copy_focused_input_text(self) -> bool:
|
||||
"""Copy the focused input's full text to the clipboard, if non-empty.
|
||||
|
||||
Ctrl+C fallback used when there is no active selection, so the whole
|
||||
draft is copied instead of arming quit.
|
||||
|
||||
Returns:
|
||||
`True` when non-empty text was handled by a clipboard attempt.
|
||||
"""
|
||||
from textual.widgets import Input, TextArea
|
||||
|
||||
widget = self.focused
|
||||
if not isinstance(widget, (TextArea, Input)):
|
||||
return False
|
||||
if isinstance(widget, Input) and widget.password:
|
||||
return False
|
||||
|
||||
text = widget.text if isinstance(widget, TextArea) else widget.value
|
||||
if not text:
|
||||
return False
|
||||
|
||||
from deepagents_code.clipboard import copy_text_with_feedback
|
||||
|
||||
# Return True regardless of copy success: the keypress is consumed
|
||||
# either way (a failed copy already warned), so it never falls through
|
||||
# to arming quit.
|
||||
copy_text_with_feedback(
|
||||
self,
|
||||
text,
|
||||
failure_noun="input",
|
||||
success_message="Input copied to clipboard",
|
||||
)
|
||||
return True
|
||||
|
||||
def _arm_quit_pending(self, shortcut: str) -> None:
|
||||
"""Set the pending-quit flag and show a matching hint.
|
||||
@@ -8469,9 +8542,19 @@ class DeepAgentsApp(App):
|
||||
6. If ask-user menu is active, cancel it
|
||||
7. If queued messages exist, pop the last one (LIFO)
|
||||
8. If agent is running, interrupt it
|
||||
9. Otherwise, a second Esc clears the chat input draft (undoable)
|
||||
"""
|
||||
from deepagents_code.widgets.thread_selector import ThreadSelectorScreen
|
||||
|
||||
# Any higher-priority Esc breaks the double-Esc clear sequence: only two
|
||||
# consecutive Escs with nothing else to handle should clear the draft.
|
||||
# Disarm up front and restore only at the terminal clear branch, so an
|
||||
# intervening interrupt (agent cancel, popup dismiss, queued-message pop,
|
||||
# ...) can't leave a stale flag that clears a later draft on a single
|
||||
# press.
|
||||
clear_was_pending = self._clear_input_pending
|
||||
self._clear_input_pending = False
|
||||
|
||||
if (
|
||||
isinstance(self.screen, ThreadSelectorScreen)
|
||||
and self.screen.is_delete_confirmation_open
|
||||
@@ -8530,6 +8613,52 @@ class DeepAgentsApp(App):
|
||||
self._cancel_worker(self._agent_worker)
|
||||
return
|
||||
|
||||
# Nothing left to interrupt: a double Esc clears the chat input draft.
|
||||
# Restore the armed state captured above so a genuine consecutive Esc
|
||||
# still confirms the clear.
|
||||
self._clear_input_pending = clear_was_pending
|
||||
self._handle_clear_input_escape()
|
||||
|
||||
def _handle_clear_input_escape(self) -> None:
|
||||
"""Clear the chat input draft on a double `Esc` press.
|
||||
|
||||
With nothing else to interrupt, the first `Esc` arms a pending flag and
|
||||
shows a hint; a second `Esc` within the window clears the draft. The
|
||||
clear is undoable via ctrl+z so a mistaken clear can be restored.
|
||||
|
||||
When the draft is empty there is nothing to clear, so no hint is shown
|
||||
and any pending flag is reset.
|
||||
"""
|
||||
chat_input = self._chat_input
|
||||
if chat_input is None or not chat_input.value:
|
||||
self._clear_input_pending = False
|
||||
return
|
||||
if self._clear_input_pending:
|
||||
self._clear_input_pending = False
|
||||
# The non-empty `value` guard above already implies a clear, so this
|
||||
# is defensive: it only suppresses the toast if `discard_text` ever
|
||||
# reports nothing cleared (e.g. a future `value` that diverges from
|
||||
# the text area), keeping the confirmation honest.
|
||||
if chat_input.discard_text():
|
||||
self.notify(
|
||||
"Input cleared (ctrl+z to undo)",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
return
|
||||
self._arm_clear_input_pending()
|
||||
|
||||
def _arm_clear_input_pending(self) -> None:
|
||||
"""Set the clear-input flag and show a matching hint."""
|
||||
self._clear_input_pending = True
|
||||
timeout = 3
|
||||
self.notify(
|
||||
"Press Esc again to clear input",
|
||||
timeout=timeout,
|
||||
markup=False,
|
||||
)
|
||||
self.set_timer(timeout, lambda: setattr(self, "_clear_input_pending", False))
|
||||
|
||||
def action_quit_app(self) -> None:
|
||||
"""Handle quit action (Ctrl+D)."""
|
||||
from deepagents_code.widgets.auth import (
|
||||
|
||||
@@ -6,7 +6,7 @@ import base64
|
||||
import logging
|
||||
import os
|
||||
import pathlib
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Literal
|
||||
|
||||
from textual.dom import NoScreen
|
||||
|
||||
@@ -80,7 +80,12 @@ def copy_text_to_clipboard(app: App, text: str) -> tuple[bool, str | None]:
|
||||
for copy_fn in copy_methods:
|
||||
try:
|
||||
copy_fn(text)
|
||||
except (OSError, RuntimeError, TypeError) as e:
|
||||
# ValueError covers UnicodeEncodeError from the `text.encode("utf-8")`
|
||||
# done by both `_copy_osc52` and `App.copy_to_clipboard` on draft text
|
||||
# containing lone surrogates, so a bad code point degrades to the next
|
||||
# backend instead of escaping the (success, error) contract and crashing
|
||||
# the caller.
|
||||
except (OSError, RuntimeError, TypeError, ValueError) as e:
|
||||
last_error = str(e) or type(e).__name__
|
||||
logger.debug(
|
||||
"Clipboard copy method %s failed: %s",
|
||||
@@ -95,6 +100,47 @@ def copy_text_to_clipboard(app: App, text: str) -> tuple[bool, str | None]:
|
||||
return False, last_error
|
||||
|
||||
|
||||
def copy_text_with_feedback(
|
||||
app: App,
|
||||
text: str,
|
||||
*,
|
||||
failure_noun: Literal["input", "selection"],
|
||||
success_message: str | None = None,
|
||||
) -> bool:
|
||||
"""Copy text to the clipboard and surface the outcome as a toast.
|
||||
|
||||
Centralizes the copy-then-notify pattern shared by the Ctrl+C and
|
||||
`[ COPY ]` paths so the success/failure messaging stays consistent.
|
||||
|
||||
Args:
|
||||
app: The active Textual app, used for the clipboard backend and toasts.
|
||||
text: Text to copy.
|
||||
failure_noun: Noun used in the warning toast (e.g. `"input"`,
|
||||
`"selection"`).
|
||||
success_message: Toast shown on success. When `None`, success is silent
|
||||
(used by the selection-copy path, which has no visible draft to
|
||||
confirm).
|
||||
|
||||
Returns:
|
||||
`True` when the copy succeeded.
|
||||
"""
|
||||
success, error = copy_text_to_clipboard(app, text)
|
||||
if success:
|
||||
if success_message is not None:
|
||||
app.notify(success_message, timeout=3, markup=False)
|
||||
else:
|
||||
# markup=False so the dynamic error string is never parsed as markup.
|
||||
app.notify(
|
||||
f"Failed to copy {failure_noun}: {error}"
|
||||
if error
|
||||
else f"Failed to copy {failure_noun} - no clipboard method available",
|
||||
severity="warning",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
return success
|
||||
|
||||
|
||||
def copy_selection_to_clipboard(app: App) -> None:
|
||||
"""Copy selected text from app widgets to clipboard.
|
||||
|
||||
|
||||
@@ -7,7 +7,7 @@ import contextlib
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, ClassVar
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, Literal, assert_never
|
||||
|
||||
from rich.cells import cell_len
|
||||
from rich.segment import Segment
|
||||
@@ -227,6 +227,69 @@ class CompletionOption(Static):
|
||||
self.post_message(self.Clicked(self._index))
|
||||
|
||||
|
||||
InputAction = Literal["clear", "copy"]
|
||||
"""Closed set of actions an `InputActionButton` can dispatch."""
|
||||
|
||||
|
||||
class InputActionButton(Static):
|
||||
"""Small clickable button shown at the right edge of the chat input row.
|
||||
|
||||
Provides discoverable mouse alternatives to keyboard shortcuts for
|
||||
clearing (`[ X ]`) and copying (`[ COPY ]`) the current draft.
|
||||
"""
|
||||
|
||||
DEFAULT_CSS = """
|
||||
InputActionButton {
|
||||
height: 1;
|
||||
margin: 0 0 0 1;
|
||||
text-style: bold;
|
||||
}
|
||||
|
||||
InputActionButton.input-action-clear {
|
||||
width: 5;
|
||||
color: $error;
|
||||
}
|
||||
|
||||
InputActionButton.input-action-copy {
|
||||
width: 8;
|
||||
color: $primary;
|
||||
}
|
||||
|
||||
InputActionButton.input-action-clear:hover {
|
||||
background: $error;
|
||||
color: auto;
|
||||
}
|
||||
|
||||
InputActionButton.input-action-copy:hover {
|
||||
background: $primary;
|
||||
color: auto;
|
||||
}
|
||||
"""
|
||||
|
||||
class Clicked(Message):
|
||||
"""Message sent when an input action button is clicked."""
|
||||
|
||||
def __init__(self, action: InputAction) -> None:
|
||||
"""Initialize with the action identifier (`clear` or `copy`)."""
|
||||
super().__init__()
|
||||
self.action = action
|
||||
|
||||
@property
|
||||
def allow_select(self) -> bool:
|
||||
"""Disable terminal text selection for the action label."""
|
||||
return False
|
||||
|
||||
def __init__(self, label: str, action: InputAction, **kwargs: Any) -> None:
|
||||
"""Initialize the button with a label and an action identifier."""
|
||||
super().__init__(label, markup=False, **kwargs)
|
||||
self._action = action
|
||||
|
||||
def on_click(self, event: Click) -> None:
|
||||
"""Relay the click as a typed `Clicked` message."""
|
||||
event.stop()
|
||||
self.post_message(self.Clicked(self._action))
|
||||
|
||||
|
||||
class CompletionPopup(VerticalScroll):
|
||||
"""Popup widget that displays completion suggestions as clickable options."""
|
||||
|
||||
@@ -330,6 +393,14 @@ class CompletionPopup(VerticalScroll):
|
||||
self.hide()
|
||||
return
|
||||
|
||||
# The DOM mutations above can await, during which a hide() (or a newer
|
||||
# rebuild) bumps the generation to cancel this one. The top-of-function
|
||||
# guard ran before that await, so re-check here: without it a stale
|
||||
# rebuild would re-show a popup that was dismissed mid-flight (e.g. when
|
||||
# a completion is applied and the popup hidden in the same key press).
|
||||
if generation != self._rebuild_generation:
|
||||
return
|
||||
|
||||
self.show()
|
||||
|
||||
if 0 <= selected_index < len(self._options):
|
||||
@@ -784,6 +855,20 @@ class ChatTextArea(TextArea):
|
||||
self._paste_burst_last_key_time = None
|
||||
self._paste_burst_last_suppressed_enter_time = None
|
||||
|
||||
def _reset_paste_burst_state(self) -> None:
|
||||
"""Reset all paste-burst and backslash tracking to a clean slate.
|
||||
|
||||
Shared by the text-replacing entry points (`set_text_from_history`,
|
||||
`clear_text`, `discard_text`) so a wholesale text swap never leaves
|
||||
stale burst/backslash timing that would misclassify the next keystroke.
|
||||
"""
|
||||
self._paste_burst_buffer = ""
|
||||
self._paste_burst_last_char_time = None
|
||||
self._paste_burst_window_until = None
|
||||
self._reset_paste_burst_run()
|
||||
self._cancel_paste_burst_timer()
|
||||
self._backslash_pending_time = None
|
||||
|
||||
def _enter_inserts_newline_during_burst(self, now: float) -> bool:
|
||||
"""Return whether `enter` should insert a newline rather than submit.
|
||||
|
||||
@@ -1144,12 +1229,7 @@ class ChatTextArea(TextArea):
|
||||
to preserve historical cursor-at-end behavior for callers
|
||||
that don't specify a direction.
|
||||
"""
|
||||
self._paste_burst_buffer = ""
|
||||
self._paste_burst_last_char_time = None
|
||||
self._paste_burst_window_until = None
|
||||
self._reset_paste_burst_run()
|
||||
self._cancel_paste_burst_timer()
|
||||
self._backslash_pending_time = None
|
||||
self._reset_paste_burst_state()
|
||||
self._skip_history_change_events += 1
|
||||
self.text = text
|
||||
if cursor_at_end:
|
||||
@@ -1169,15 +1249,26 @@ class ChatTextArea(TextArea):
|
||||
# set_text_from_history is still suppressed, plus one for the
|
||||
# self.text = "" assignment below.
|
||||
self._skip_history_change_events += 1
|
||||
self._paste_burst_buffer = ""
|
||||
self._paste_burst_last_char_time = None
|
||||
self._paste_burst_window_until = None
|
||||
self._reset_paste_burst_run()
|
||||
self._cancel_paste_burst_timer()
|
||||
self._backslash_pending_time = None
|
||||
self._reset_paste_burst_state()
|
||||
self.text = ""
|
||||
self.move_cursor((0, 0))
|
||||
|
||||
def discard_text(self) -> bool:
|
||||
"""Clear the draft via an undoable edit (restorable with ctrl+z).
|
||||
|
||||
Unlike `clear_text`, the deletion is recorded in the undo history and
|
||||
the resulting `Changed` event is allowed to propagate, so completion
|
||||
and argument-hint state stay in sync.
|
||||
|
||||
Returns:
|
||||
`True` when there was text to clear.
|
||||
"""
|
||||
if not self.text:
|
||||
return False
|
||||
self._reset_paste_burst_state()
|
||||
self.clear()
|
||||
return True
|
||||
|
||||
|
||||
class _CompletionViewAdapter:
|
||||
"""Translate completion-space replacements to text-area coordinates."""
|
||||
@@ -1224,6 +1315,11 @@ class ChatInput(Vertical):
|
||||
|
||||
DEFAULT_CSS = """
|
||||
ChatInput {
|
||||
height: auto;
|
||||
layers: base actions;
|
||||
}
|
||||
|
||||
ChatInput #input-box {
|
||||
height: auto;
|
||||
min-height: 3;
|
||||
max-height: 25;
|
||||
@@ -1232,20 +1328,33 @@ class ChatInput(Vertical):
|
||||
border: solid $primary;
|
||||
}
|
||||
|
||||
ChatInput.mode-shell {
|
||||
ChatInput.mode-shell #input-box {
|
||||
border: solid $mode-bash;
|
||||
}
|
||||
|
||||
ChatInput.mode-command {
|
||||
ChatInput.mode-command #input-box {
|
||||
border: solid $mode-command;
|
||||
}
|
||||
|
||||
ChatInput.mode-shell-incognito {
|
||||
ChatInput.mode-shell-incognito #input-box {
|
||||
border: solid $mode-incognito;
|
||||
border-title-color: $mode-incognito;
|
||||
border-title-style: bold;
|
||||
}
|
||||
|
||||
/* Action buttons float on their own z-layer over the top border line, so
|
||||
they cost no content row and never overlap the draft text. The row docks
|
||||
to the right edge and sizes to its buttons (`width: auto`), overlaying
|
||||
only the right portion of the border line and leaving the rest clear. */
|
||||
ChatInput #input-actions {
|
||||
layer: actions;
|
||||
dock: right;
|
||||
width: auto;
|
||||
height: 1;
|
||||
margin-right: 1;
|
||||
display: none;
|
||||
}
|
||||
|
||||
ChatInput .input-row {
|
||||
height: auto;
|
||||
width: 100%;
|
||||
@@ -1333,6 +1442,8 @@ class ChatInput(Vertical):
|
||||
super().__init__(**kwargs)
|
||||
self._cwd = Path(cwd) if cwd else Path.cwd()
|
||||
self._image_tracker = image_tracker
|
||||
self._input_box: Vertical | None = None
|
||||
self._action_buttons: Horizontal | None = None
|
||||
self._text_area: ChatTextArea | None = None
|
||||
self._popup: CompletionPopup | None = None
|
||||
self._completion_manager: MultiCompletionManager | None = None
|
||||
@@ -1384,17 +1495,38 @@ class ChatInput(Vertical):
|
||||
Yields:
|
||||
Widgets for the input row and completion popup.
|
||||
"""
|
||||
with Horizontal(classes="input-row"):
|
||||
yield Static(">", classes="input-prompt", id="prompt")
|
||||
yield ChatTextArea(id="chat-input")
|
||||
# The bordered box owns the prompt, text area, and completion popup so
|
||||
# the action buttons (a sibling) can float on its top border line; a
|
||||
# widget can only render on its sibling's border, not its parent's.
|
||||
with Vertical(id="input-box"):
|
||||
with Horizontal(classes="input-row"):
|
||||
yield Static(">", classes="input-prompt", id="prompt")
|
||||
yield ChatTextArea(id="chat-input")
|
||||
yield CompletionPopup(id="completion-popup")
|
||||
|
||||
yield CompletionPopup(id="completion-popup")
|
||||
# Action buttons float on their own z-layer over the top border line so
|
||||
# they cost no content row and never overlap the draft text.
|
||||
with Horizontal(id="input-actions"):
|
||||
yield InputActionButton(
|
||||
"[ X ]",
|
||||
"clear",
|
||||
id="clear-button",
|
||||
classes="input-action input-action-clear",
|
||||
)
|
||||
yield InputActionButton(
|
||||
"[ COPY ]",
|
||||
"copy",
|
||||
id="copy-button",
|
||||
classes="input-action input-action-copy",
|
||||
)
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Initialize components after mount."""
|
||||
self._input_box = self.query_one("#input-box", Vertical)
|
||||
self._action_buttons = self.query_one("#input-actions", Horizontal)
|
||||
if is_ascii_mode():
|
||||
colors = theme.get_theme_colors(self)
|
||||
self.styles.border = ("ascii", colors.primary)
|
||||
self._input_box.styles.border = ("ascii", colors.primary)
|
||||
|
||||
self._text_area = self.query_one("#chat-input", ChatTextArea)
|
||||
self._popup = self.query_one("#completion-popup", CompletionPopup)
|
||||
@@ -1492,9 +1624,28 @@ class ChatInput(Vertical):
|
||||
|
||||
self._text_area.argument_hint = ""
|
||||
|
||||
def _set_action_buttons_visible(self, *, visible: bool) -> None:
|
||||
"""Show or hide the clear/copy action buttons on the input border.
|
||||
|
||||
Only writes `display` when it actually changes. Mutating it on every
|
||||
keystroke would trigger a layout reflow each time, which perturbs the
|
||||
completion popup's deferred (`call_after_refresh`) show/hide ordering.
|
||||
"""
|
||||
if self._action_buttons is not None and self._action_buttons.display != visible:
|
||||
self._action_buttons.display = visible
|
||||
|
||||
def on_text_area_changed(self, event: TextArea.Changed) -> None:
|
||||
"""Detect input mode and update completions."""
|
||||
text = event.text_area.text
|
||||
# Reveal the clear/copy buttons only when there is a meaningful draft to
|
||||
# act on, so an empty input keeps a clean, uncluttered border.
|
||||
# Whitespace-only input (e.g. stray spaces or newlines) has nothing
|
||||
# worth clearing or copying, so it stays hidden too. Done before the
|
||||
# early returns below so recalled-history text shows them as well.
|
||||
# NOTE: this `strip()` gate is deliberately stricter than the keyboard
|
||||
# paths (esc+esc clear, Ctrl+C copy), which act on the raw value so a
|
||||
# whitespace-only draft is still clearable/copyable without the buttons.
|
||||
self._set_action_buttons_visible(visible=bool(text.strip()))
|
||||
# Drag-drop / bracketed paste arrive as one Changed event with a
|
||||
# multi-character inserted span. Normal typing arrives one character at
|
||||
# a time. Checking the changed span (rather than net length delta)
|
||||
@@ -2266,10 +2417,10 @@ class ChatInput(Vertical):
|
||||
self.mode = "normal"
|
||||
return
|
||||
prompt.update(glyph or ">")
|
||||
if mode == "shell_incognito":
|
||||
self.border_title = "incognito"
|
||||
else:
|
||||
self.border_title = None
|
||||
if self._input_box is not None:
|
||||
self._input_box.border_title = (
|
||||
"incognito" if mode == "shell_incognito" else None
|
||||
)
|
||||
|
||||
self.call_after_refresh(_apply)
|
||||
self.post_message(self.ModeChanged(mode))
|
||||
@@ -2303,6 +2454,57 @@ class ChatInput(Vertical):
|
||||
self._text_area.text = val
|
||||
self._text_area.move_cursor_to_end()
|
||||
|
||||
def discard_text(self) -> bool:
|
||||
"""Clear the draft, keeping it restorable via undo (ctrl+z).
|
||||
|
||||
Returns:
|
||||
`True` when there was text to clear.
|
||||
"""
|
||||
if self._text_area is None:
|
||||
return False
|
||||
if self._text_area.text:
|
||||
self._skip_media_sync_events += 1
|
||||
return self._text_area.discard_text()
|
||||
|
||||
def on_input_action_button_clicked(self, event: InputActionButton.Clicked) -> None:
|
||||
"""Handle clicks on the `[ X ]` / `[ COPY ]` input buttons."""
|
||||
event.stop()
|
||||
if event.action == "clear":
|
||||
self._clear_via_button()
|
||||
elif event.action == "copy":
|
||||
self._copy_via_button()
|
||||
else:
|
||||
assert_never(event.action)
|
||||
|
||||
def _clear_via_button(self) -> None:
|
||||
"""Clear the draft from the `[ X ]` button (undoable with ctrl+z).
|
||||
|
||||
Also exits any active slash/shell mode, unlike the Esc-driven clear.
|
||||
"""
|
||||
cleared = self.discard_text()
|
||||
self.exit_mode()
|
||||
if cleared:
|
||||
self.app.notify("Input cleared (ctrl+z to undo)", timeout=3, markup=False)
|
||||
if self._text_area is not None:
|
||||
self._text_area.focus()
|
||||
|
||||
def _copy_via_button(self) -> None:
|
||||
"""Copy the current draft to the clipboard from the `[ COPY ]` button."""
|
||||
from deepagents_code.clipboard import copy_text_with_feedback
|
||||
|
||||
text = self.value
|
||||
if text:
|
||||
copy_text_with_feedback(
|
||||
self.app,
|
||||
text,
|
||||
failure_noun="input",
|
||||
success_message="Input copied to clipboard",
|
||||
)
|
||||
# Refocus the input so clicking the button never strands focus on the
|
||||
# (non-focusable) button.
|
||||
if self._text_area is not None:
|
||||
self._text_area.focus()
|
||||
|
||||
@property
|
||||
def input_widget(self) -> ChatTextArea | None:
|
||||
"""Get the underlying TextArea widget.
|
||||
|
||||
@@ -13,7 +13,7 @@ import time
|
||||
import webbrowser
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING, Any, ClassVar
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, cast
|
||||
from unittest.mock import AsyncMock, MagicMock, call, patch
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -1447,8 +1447,8 @@ class TestCtrlCCopySelection:
|
||||
exit_mock.assert_not_called()
|
||||
assert app._quit_pending is False
|
||||
|
||||
async def test_ctrl_c_without_selection_still_quits(self) -> None:
|
||||
"""Ctrl+C with no selection in ChatTextArea falls through to quit hint."""
|
||||
async def test_ctrl_c_without_selection_copies_full_input(self) -> None:
|
||||
"""Ctrl+C with no selection in ChatTextArea copies the full input."""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
@@ -1473,18 +1473,215 @@ class TestCtrlCCopySelection:
|
||||
):
|
||||
app.action_quit_or_interrupt()
|
||||
|
||||
copy_mock.assert_not_called()
|
||||
copy_mock.assert_called_once_with(app, "hello world")
|
||||
exit_mock.assert_not_called()
|
||||
assert app._quit_pending is True
|
||||
assert app._quit_pending is False
|
||||
notify_mock.assert_called_once_with(
|
||||
"Press Ctrl+C again to quit",
|
||||
"Input copied to clipboard",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
async def test_ctrl_c_rapid_presses_force_quit_over_draft_copy(self) -> None:
|
||||
"""Mashing Ctrl+C with a draft skips copy, arms quit, then exits.
|
||||
|
||||
A non-empty draft makes the copy branch absorb every single press, so
|
||||
the rapid escape hatch is the only way to reach quit via Ctrl+C alone.
|
||||
"""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.text = "draft text"
|
||||
await pilot.pause()
|
||||
text_area.focus()
|
||||
await pilot.pause()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock,
|
||||
patch.object(app, "notify"),
|
||||
patch.object(app, "exit") as exit_mock,
|
||||
patch(
|
||||
"deepagents_code.app._monotonic",
|
||||
side_effect=[0.0, 0.2, 0.4, 0.6],
|
||||
),
|
||||
):
|
||||
# First press copies the draft (standard terminal copy).
|
||||
app.action_quit_or_interrupt()
|
||||
assert app._quit_pending is False
|
||||
copy_mock.assert_called_once()
|
||||
exit_mock.assert_not_called()
|
||||
|
||||
# Second rapid press skips copy and arms quit instead.
|
||||
app.action_quit_or_interrupt()
|
||||
assert app._quit_pending is True
|
||||
copy_mock.assert_called_once() # copy was skipped this time
|
||||
exit_mock.assert_not_called()
|
||||
|
||||
# Third rapid press exits.
|
||||
app.action_quit_or_interrupt()
|
||||
exit_mock.assert_called_once()
|
||||
|
||||
async def test_ctrl_c_quit_pending_beats_draft_copy(self) -> None:
|
||||
"""Once the quit prompt is shown, the next Ctrl+C exits before copying."""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.text = "draft text"
|
||||
await pilot.pause()
|
||||
text_area.focus()
|
||||
await pilot.pause()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock,
|
||||
patch.object(app, "notify"),
|
||||
patch.object(app, "exit") as exit_mock,
|
||||
patch(
|
||||
"deepagents_code.app._monotonic",
|
||||
side_effect=[0.0, 0.2, 1.4],
|
||||
),
|
||||
):
|
||||
app.action_quit_or_interrupt()
|
||||
assert app._quit_pending is False
|
||||
copy_mock.assert_called_once()
|
||||
exit_mock.assert_not_called()
|
||||
|
||||
app.action_quit_or_interrupt()
|
||||
assert app._quit_pending is True
|
||||
copy_mock.assert_called_once()
|
||||
exit_mock.assert_not_called()
|
||||
|
||||
app.action_quit_or_interrupt()
|
||||
exit_mock.assert_called_once()
|
||||
copy_mock.assert_called_once()
|
||||
|
||||
async def test_ctrl_c_slow_presses_keep_copying_draft(self) -> None:
|
||||
"""Ctrl+C presses spaced beyond the rapid window keep copying, never quit."""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.text = "draft text"
|
||||
await pilot.pause()
|
||||
text_area.focus()
|
||||
await pilot.pause()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock,
|
||||
patch.object(app, "notify"),
|
||||
patch.object(app, "exit") as exit_mock,
|
||||
patch(
|
||||
"deepagents_code.app._monotonic",
|
||||
side_effect=[0.0, 2.0, 4.0],
|
||||
),
|
||||
):
|
||||
for _ in range(3):
|
||||
app.action_quit_or_interrupt()
|
||||
|
||||
assert copy_mock.call_count == 3
|
||||
assert app._quit_pending is False
|
||||
exit_mock.assert_not_called()
|
||||
|
||||
async def test_ctrl_c_window_boundary_is_inclusive(self) -> None:
|
||||
"""Two presses exactly one window apart count as rapid (`<=` boundary).
|
||||
|
||||
Pins the inclusive edge of `now - t <= _RAPID_QUIT_CTRL_C_WINDOW_SECONDS`
|
||||
so a future flip to `<` (or a window tweak) is caught.
|
||||
"""
|
||||
from deepagents_code.app import _RAPID_QUIT_CTRL_C_WINDOW_SECONDS
|
||||
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.text = "draft text"
|
||||
await pilot.pause()
|
||||
text_area.focus()
|
||||
await pilot.pause()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock,
|
||||
patch.object(app, "notify"),
|
||||
patch.object(app, "exit") as exit_mock,
|
||||
patch(
|
||||
"deepagents_code.app._monotonic",
|
||||
side_effect=[0.0, _RAPID_QUIT_CTRL_C_WINDOW_SECONDS],
|
||||
),
|
||||
):
|
||||
# Press 1 copies the draft.
|
||||
app.action_quit_or_interrupt()
|
||||
copy_mock.assert_called_once()
|
||||
assert app._quit_pending is False
|
||||
|
||||
# Press 2 sits exactly on the window edge: still rapid, so it
|
||||
# skips the copy and arms quit instead.
|
||||
app.action_quit_or_interrupt()
|
||||
copy_mock.assert_called_once()
|
||||
assert app._quit_pending is True
|
||||
exit_mock.assert_not_called()
|
||||
|
||||
async def test_ctrl_c_just_past_window_keeps_copying(self) -> None:
|
||||
"""A press just past the window is not rapid: it copies again, never quits."""
|
||||
from deepagents_code.app import _RAPID_QUIT_CTRL_C_WINDOW_SECONDS
|
||||
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.text = "draft text"
|
||||
await pilot.pause()
|
||||
text_area.focus()
|
||||
await pilot.pause()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock,
|
||||
patch.object(app, "notify"),
|
||||
patch.object(app, "exit") as exit_mock,
|
||||
patch(
|
||||
"deepagents_code.app._monotonic",
|
||||
side_effect=[0.0, _RAPID_QUIT_CTRL_C_WINDOW_SECONDS + 0.001],
|
||||
),
|
||||
):
|
||||
# Press 1 copies; press 2 falls just outside the window, so the
|
||||
# first timestamp is pruned and it copies again rather than
|
||||
# arming quit.
|
||||
app.action_quit_or_interrupt()
|
||||
app.action_quit_or_interrupt()
|
||||
assert copy_mock.call_count == 2
|
||||
assert app._quit_pending is False
|
||||
exit_mock.assert_not_called()
|
||||
|
||||
async def test_ctrl_c_copies_input_selection(self) -> None:
|
||||
"""Ctrl+C with a selection in a focused Input copies it, no quit."""
|
||||
app = DeepAgentsApp()
|
||||
@@ -15283,3 +15480,355 @@ class TestNotifyOrphanedTracingDisabled:
|
||||
app._notify_orphaned_tracing_disabled()
|
||||
|
||||
log_mock.assert_called_once()
|
||||
|
||||
|
||||
class TestClearInputEscape:
|
||||
"""Tests for the double-Esc chat-input clear fallback."""
|
||||
|
||||
@staticmethod
|
||||
def _make_app() -> DeepAgentsApp:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="t")
|
||||
app.notify = MagicMock() # ty: ignore[invalid-assignment]
|
||||
app.set_timer = MagicMock() # ty: ignore[invalid-assignment]
|
||||
return app
|
||||
|
||||
def test_double_escape_clears_input(self) -> None:
|
||||
"""First Esc arms the clear; the second clears the draft."""
|
||||
app = self._make_app()
|
||||
chat_input = MagicMock()
|
||||
chat_input.value = "draft text"
|
||||
chat_input.discard_text.return_value = True
|
||||
app._chat_input = chat_input
|
||||
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is True
|
||||
chat_input.discard_text.assert_not_called()
|
||||
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is False
|
||||
chat_input.discard_text.assert_called_once_with()
|
||||
|
||||
def test_arm_and_clear_emit_separate_toasts(self) -> None:
|
||||
"""The arm hint and the clear confirmation are distinct toasts.
|
||||
|
||||
The first Esc hints to press again (no premature undo hint); the second
|
||||
Esc confirms the clear and surfaces the ctrl+z undo hint at the moment
|
||||
it becomes actionable.
|
||||
"""
|
||||
app = self._make_app()
|
||||
chat_input = MagicMock()
|
||||
chat_input.value = "draft text"
|
||||
chat_input.discard_text.return_value = True
|
||||
app._chat_input = chat_input
|
||||
|
||||
notify = cast("MagicMock", app.notify)
|
||||
|
||||
app._handle_clear_input_escape()
|
||||
notify.assert_called_once_with(
|
||||
"Press Esc again to clear input",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
notify.reset_mock()
|
||||
app._handle_clear_input_escape()
|
||||
notify.assert_called_once_with(
|
||||
"Input cleared (ctrl+z to undo)",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
def test_clear_toast_suppressed_when_nothing_discarded(self) -> None:
|
||||
"""No confirmation toast fires if `discard_text` reports nothing cleared."""
|
||||
app = self._make_app()
|
||||
chat_input = MagicMock()
|
||||
chat_input.value = "draft text"
|
||||
chat_input.discard_text.return_value = False
|
||||
app._chat_input = chat_input
|
||||
|
||||
notify = cast("MagicMock", app.notify)
|
||||
|
||||
app._handle_clear_input_escape() # arm
|
||||
notify.reset_mock()
|
||||
app._handle_clear_input_escape() # clear attempt
|
||||
|
||||
chat_input.discard_text.assert_called_once_with()
|
||||
notify.assert_not_called()
|
||||
|
||||
def test_escape_no_op_when_input_empty(self) -> None:
|
||||
"""Esc never arms a clear when the draft is empty."""
|
||||
app = self._make_app()
|
||||
chat_input = MagicMock()
|
||||
chat_input.value = ""
|
||||
app._chat_input = chat_input
|
||||
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is False
|
||||
chat_input.discard_text.assert_not_called()
|
||||
|
||||
def test_whitespace_only_draft_is_clearable(self) -> None:
|
||||
"""esc+esc acts on a whitespace-only draft via raw `value`.
|
||||
|
||||
Deliberately broader than the `[ X ]`/`[ COPY ]` buttons, which gate on
|
||||
`text.strip()` and stay hidden for whitespace-only input.
|
||||
"""
|
||||
app = self._make_app()
|
||||
chat_input = MagicMock()
|
||||
chat_input.value = " \n "
|
||||
chat_input.discard_text.return_value = True
|
||||
app._chat_input = chat_input
|
||||
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is True
|
||||
chat_input.discard_text.assert_not_called()
|
||||
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is False
|
||||
chat_input.discard_text.assert_called_once_with()
|
||||
|
||||
def test_pending_resets_when_input_emptied_between_presses(self) -> None:
|
||||
"""A pending clear is disarmed if the draft becomes empty before press 2."""
|
||||
app = self._make_app()
|
||||
chat_input = MagicMock()
|
||||
chat_input.value = "x"
|
||||
app._chat_input = chat_input
|
||||
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is True
|
||||
|
||||
chat_input.value = ""
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is False
|
||||
chat_input.discard_text.assert_not_called()
|
||||
|
||||
def test_pending_resets_after_timer_expiry(self) -> None:
|
||||
"""When the arm window elapses, the timer disarms so a later Esc re-arms.
|
||||
|
||||
Without the reset, a stale pending flag would let a much-later Esc clear
|
||||
a draft the user typed long after the hint vanished.
|
||||
"""
|
||||
app = self._make_app()
|
||||
chat_input = MagicMock()
|
||||
chat_input.value = "draft text"
|
||||
chat_input.discard_text.return_value = True
|
||||
app._chat_input = chat_input
|
||||
|
||||
set_timer = cast("MagicMock", app.set_timer)
|
||||
|
||||
app._handle_clear_input_escape() # arm
|
||||
assert app._clear_input_pending is True
|
||||
|
||||
# Fire the scheduled reset callback as Textual's timer would.
|
||||
_delay, reset_callback = set_timer.call_args.args
|
||||
reset_callback()
|
||||
assert app._clear_input_pending is False
|
||||
|
||||
# The next Esc re-arms instead of clearing the untouched draft.
|
||||
app._handle_clear_input_escape()
|
||||
assert app._clear_input_pending is True
|
||||
chat_input.discard_text.assert_not_called()
|
||||
|
||||
async def test_double_escape_clears_via_full_escape_chain(self) -> None:
|
||||
"""Through the real Esc handler, esc+esc clears when nothing else pends.
|
||||
|
||||
Exercises the precedence chain: the clear only fires as the last resort
|
||||
of `action_interrupt`, after every higher-priority interrupt has passed.
|
||||
"""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("a draft I regret")
|
||||
await pilot.pause()
|
||||
|
||||
app.action_interrupt()
|
||||
await pilot.pause()
|
||||
assert app._clear_input_pending is True
|
||||
assert chat_input.value == "a draft I regret"
|
||||
|
||||
app.action_interrupt()
|
||||
await pilot.pause()
|
||||
assert app._clear_input_pending is False
|
||||
assert chat_input.value == ""
|
||||
|
||||
# The clear is undoable.
|
||||
text_area.undo()
|
||||
await pilot.pause()
|
||||
assert chat_input.value == "a draft I regret"
|
||||
|
||||
async def test_second_escape_clears_edited_draft_not_a_snapshot(self) -> None:
|
||||
"""The second Esc clears whatever is in the draft now, not what was armed.
|
||||
|
||||
Arming does not snapshot the draft: editing between the two presses means
|
||||
the second Esc clears the edited content (and undo restores that).
|
||||
"""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("draft A")
|
||||
await pilot.pause()
|
||||
|
||||
app.action_interrupt() # arm with "draft A"
|
||||
await pilot.pause()
|
||||
assert app._clear_input_pending is True
|
||||
|
||||
text_area.insert(" plus B") # edit within the arm window
|
||||
await pilot.pause()
|
||||
assert chat_input.value == "draft A plus B"
|
||||
|
||||
app.action_interrupt() # second Esc clears the edited draft
|
||||
await pilot.pause()
|
||||
assert app._clear_input_pending is False
|
||||
assert chat_input.value == ""
|
||||
|
||||
text_area.undo()
|
||||
await pilot.pause()
|
||||
assert chat_input.value == "draft A plus B"
|
||||
|
||||
async def test_higher_priority_interrupt_disarms_pending_clear(self) -> None:
|
||||
"""An intervening interrupt breaks the sequence so the next Esc re-arms.
|
||||
|
||||
Without disarming, an Esc that (e.g.) cancels the agent would leave the
|
||||
clear armed, and the very next Esc would wipe the draft on a single
|
||||
press — surprising the user who only pressed clear-Esc once.
|
||||
"""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("keep me")
|
||||
await pilot.pause()
|
||||
|
||||
# First Esc arms the clear (nothing else to interrupt).
|
||||
app.action_interrupt()
|
||||
await pilot.pause()
|
||||
assert app._clear_input_pending is True
|
||||
|
||||
# A higher-priority interrupt fires: the running agent is cancelled,
|
||||
# which must also disarm the pending clear and leave the draft intact.
|
||||
app._agent_running = True
|
||||
app._agent_worker = MagicMock()
|
||||
with patch.object(app, "_cancel_worker"):
|
||||
app.action_interrupt()
|
||||
await pilot.pause()
|
||||
assert app._clear_input_pending is False
|
||||
assert chat_input.value == "keep me"
|
||||
|
||||
# With the agent idle again, the next Esc re-arms instead of clearing
|
||||
# on a single press.
|
||||
app._agent_running = False
|
||||
app._agent_worker = None
|
||||
app.action_interrupt()
|
||||
await pilot.pause()
|
||||
assert app._clear_input_pending is True
|
||||
assert chat_input.value == "keep me"
|
||||
|
||||
|
||||
class TestCopyFocusedInputText:
|
||||
"""Tests for the Ctrl+C copy-whole-input fallback (no active selection)."""
|
||||
|
||||
@staticmethod
|
||||
def _make_app() -> DeepAgentsApp:
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="t")
|
||||
app.notify = MagicMock() # ty: ignore[invalid-assignment]
|
||||
return app
|
||||
|
||||
def test_copies_whole_input_when_no_selection(self, monkeypatch) -> None:
|
||||
"""A focused, non-empty input with no selection is copied in full."""
|
||||
from textual.widgets import TextArea
|
||||
|
||||
import deepagents_code.clipboard as clipboard_module
|
||||
|
||||
copied: list[str] = []
|
||||
|
||||
def fake_copy(_app: object, text: str) -> tuple[bool, str | None]:
|
||||
copied.append(text)
|
||||
return True, None
|
||||
|
||||
monkeypatch.setattr(clipboard_module, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
app = self._make_app()
|
||||
text_area = TextArea()
|
||||
text_area.text = "whole input draft"
|
||||
monkeypatch.setattr(type(app), "focused", property(lambda _self: text_area))
|
||||
|
||||
assert app._copy_focused_input_text() is True
|
||||
assert copied == ["whole input draft"]
|
||||
|
||||
def test_failed_input_copy_is_handled(self, monkeypatch) -> None:
|
||||
"""A focused input copy failure must not fall through to quit handling."""
|
||||
from textual.widgets import TextArea
|
||||
|
||||
import deepagents_code.clipboard as clipboard_module
|
||||
|
||||
copied: list[str] = []
|
||||
|
||||
def fake_copy(_app: object, text: str) -> tuple[bool, str | None]:
|
||||
copied.append(text)
|
||||
return False, "no clipboard backend"
|
||||
|
||||
monkeypatch.setattr(clipboard_module, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
app = self._make_app()
|
||||
notify = MagicMock()
|
||||
app.notify = notify # ty: ignore[invalid-assignment]
|
||||
text_area = TextArea()
|
||||
text_area.text = "whole input draft"
|
||||
monkeypatch.setattr(type(app), "focused", property(lambda _self: text_area))
|
||||
|
||||
assert app._copy_focused_input_text() is True
|
||||
assert copied == ["whole input draft"]
|
||||
notify.assert_called_once()
|
||||
|
||||
def test_no_copy_when_input_empty(self, monkeypatch) -> None:
|
||||
"""An empty focused input is not copied."""
|
||||
from textual.widgets import TextArea
|
||||
|
||||
import deepagents_code.clipboard as clipboard_module
|
||||
|
||||
copied: list[str] = []
|
||||
|
||||
def fake_copy(_app: object, text: str) -> tuple[bool, str | None]:
|
||||
copied.append(text)
|
||||
return True, None
|
||||
|
||||
monkeypatch.setattr(clipboard_module, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
app = self._make_app()
|
||||
text_area = TextArea()
|
||||
monkeypatch.setattr(type(app), "focused", property(lambda _self: text_area))
|
||||
|
||||
assert app._copy_focused_input_text() is False
|
||||
assert copied == []
|
||||
|
||||
def test_no_copy_when_nothing_focused(self, monkeypatch) -> None:
|
||||
"""When the focused widget is not an input, nothing is copied."""
|
||||
app = self._make_app()
|
||||
monkeypatch.setattr(type(app), "focused", property(lambda _self: None))
|
||||
assert app._copy_focused_input_text() is False
|
||||
|
||||
async def test_no_copy_from_password_input(self) -> None:
|
||||
"""A focused masked password Input is never copied by the whole-input path."""
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
|
||||
password_input = Input(value="secret-api-key", password=True)
|
||||
await app.mount(password_input)
|
||||
password_input.focus()
|
||||
await pilot.pause()
|
||||
|
||||
with patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock:
|
||||
assert app._copy_focused_input_text() is False
|
||||
|
||||
copy_mock.assert_not_called()
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import html
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
@@ -11,8 +12,10 @@ from textual.app import App, ComposeResult
|
||||
from textual.containers import Container
|
||||
from textual.widgets import Static
|
||||
|
||||
from deepagents_code import _textual_patches as _textual_patches
|
||||
from deepagents_code.command_registry import SLASH_COMMANDS
|
||||
from deepagents_code.input import MediaTracker
|
||||
from deepagents_code.media_utils import ImageData
|
||||
from deepagents_code.widgets import chat_input as chat_input_module
|
||||
from deepagents_code.widgets.autocomplete import MAX_SUGGESTIONS
|
||||
from deepagents_code.widgets.chat_input import (
|
||||
@@ -252,6 +255,340 @@ class TestChatTextAreaKeybindings:
|
||||
assert "alt+backspace" in word_delete_keys
|
||||
|
||||
|
||||
class TestDiscardText:
|
||||
"""Tests for the undoable draft clear behind esc+esc and the `[ X ]` button."""
|
||||
|
||||
async def test_discard_text_clears_and_reports_cleared(self) -> None:
|
||||
"""`discard_text` empties the draft and returns True when text existed."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("a draft I changed my mind about")
|
||||
await pilot.pause()
|
||||
|
||||
assert chat_input.discard_text() is True
|
||||
await pilot.pause()
|
||||
assert chat_input.value == ""
|
||||
|
||||
async def test_discard_text_no_op_when_empty(self) -> None:
|
||||
"""`discard_text` returns False and leaves the media-skip counter alone."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
await pilot.pause()
|
||||
before = chat_input._skip_media_sync_events
|
||||
assert chat_input.discard_text() is False
|
||||
# An empty no-op must not bump the skip counter: a stray increment
|
||||
# would later swallow a legitimate media sync, desyncing placeholders.
|
||||
assert chat_input._skip_media_sync_events == before
|
||||
|
||||
async def test_discard_text_is_undoable(self) -> None:
|
||||
"""The cleared draft is restorable via the TextArea undo (ctrl+z)."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("restore me")
|
||||
await pilot.pause()
|
||||
|
||||
assert chat_input.discard_text() is True
|
||||
await pilot.pause()
|
||||
assert chat_input.value == ""
|
||||
|
||||
text_area.undo()
|
||||
await pilot.pause()
|
||||
assert chat_input.value == "restore me"
|
||||
|
||||
async def test_discard_text_preserves_media_for_undo(self) -> None:
|
||||
"""Undoing a cleared media draft keeps placeholder media attached."""
|
||||
app = _ImagePasteApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
placeholder = app.tracker.add_image(
|
||||
ImageData(base64_data="abc", format="png", placeholder="")
|
||||
)
|
||||
text_area.insert(placeholder)
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.tracker.get_images()) == 1
|
||||
assert chat_input.discard_text() is True
|
||||
await pilot.pause()
|
||||
assert chat_input.value == ""
|
||||
assert len(app.tracker.get_images()) == 1
|
||||
|
||||
text_area.undo()
|
||||
await pilot.pause()
|
||||
assert chat_input.value == placeholder
|
||||
assert len(app.tracker.get_images()) == 1
|
||||
|
||||
|
||||
class TestInputActionButtons:
|
||||
"""Tests for the `[ X ]` clear and `[ COPY ]` buttons in the chat input."""
|
||||
|
||||
async def test_buttons_render_labels(self) -> None:
|
||||
"""The action button labels render as text, not Rich markup tags."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
# Buttons only appear once a draft exists.
|
||||
text_area.insert("draft")
|
||||
await pilot.pause()
|
||||
rendered = html.unescape(app.export_screenshot()).replace("\xa0", " ")
|
||||
|
||||
assert "[ X ]" in rendered
|
||||
assert "[ COPY ]" in rendered
|
||||
|
||||
async def test_buttons_render_on_input_border(self) -> None:
|
||||
"""Buttons sit on the box's top border line, above full-width text.
|
||||
|
||||
They render on the border row (not a content row), so the text area
|
||||
keeps the full width and the draft is never overlapped. The top-right
|
||||
corner stays visible, a first-row text click still reaches the text
|
||||
area, and a button click hits the button.
|
||||
"""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test(size=(60, 24)) as pilot:
|
||||
await pilot.pause()
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
# A long single line would wrap to the text area's full width.
|
||||
text_area.insert("Z" * 200)
|
||||
await pilot.pause()
|
||||
|
||||
box = chat_input.query_one("#input-box")
|
||||
clear = chat_input.query_one("#clear-button", Static)
|
||||
copy = chat_input.query_one("#copy-button", Static)
|
||||
|
||||
# Text area spans the full width inside the border.
|
||||
assert text_area.region.right == box.content_region.right
|
||||
|
||||
# Buttons render on the top border row, above the first text row.
|
||||
assert clear.region.y == box.region.y
|
||||
assert copy.region.y == box.region.y
|
||||
assert text_area.region.y > box.region.y
|
||||
|
||||
# The top-right corner stays visible (buttons stop short of the edge).
|
||||
assert copy.region.right < box.region.right
|
||||
|
||||
# No overlap: a first-row click reaches the text area, and a click on
|
||||
# a button hits the button.
|
||||
left_widget, _ = app.screen.get_widget_at(
|
||||
text_area.region.x + 1, text_area.region.y
|
||||
)
|
||||
assert left_widget is text_area
|
||||
button_widget, _ = app.screen.get_widget_at(
|
||||
copy.region.x + 1, copy.region.y
|
||||
)
|
||||
assert button_widget is copy
|
||||
|
||||
async def test_buttons_hidden_until_draft_entered(self) -> None:
|
||||
"""The buttons appear only while the draft has non-whitespace content."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
actions = chat_input.query_one("#input-actions")
|
||||
|
||||
# Empty draft: nothing to clear or copy, so the buttons stay hidden.
|
||||
assert actions.display is False
|
||||
|
||||
# Whitespace-only input has nothing worth acting on: still hidden.
|
||||
text_area.insert(" \n\n ")
|
||||
await pilot.pause()
|
||||
assert actions.display is False
|
||||
|
||||
# Real content reveals them.
|
||||
text_area.insert("draft")
|
||||
await pilot.pause()
|
||||
assert actions.display is True
|
||||
|
||||
# Clearing the draft hides them again.
|
||||
chat_input.discard_text()
|
||||
await pilot.pause()
|
||||
assert actions.display is False
|
||||
|
||||
async def test_copy_button_double_click_does_not_select_label(self) -> None:
|
||||
"""Double-clicking `[ COPY ]` should not trigger Textual word selection."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("draft") # buttons only render with a draft
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.double_click("#copy-button", offset=(3, 0))
|
||||
await pilot.pause()
|
||||
|
||||
assert app.screen.get_selected_text() is None
|
||||
|
||||
async def test_clear_button_clears_input(self) -> None:
|
||||
"""Clicking `[ X ]` empties the draft."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("clear me")
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.click("#clear-button")
|
||||
await pilot.pause()
|
||||
assert chat_input.value == ""
|
||||
|
||||
async def test_clear_button_is_undoable(self) -> None:
|
||||
"""A draft cleared via `[ X ]` can be restored with undo."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("undo me")
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.click("#clear-button")
|
||||
await pilot.pause()
|
||||
assert chat_input.value == ""
|
||||
|
||||
text_area.undo()
|
||||
await pilot.pause()
|
||||
assert chat_input.value == "undo me"
|
||||
|
||||
async def test_clear_button_exits_command_mode(self) -> None:
|
||||
"""Clicking `[ X ]` should not leave a stale slash-command mode active."""
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
|
||||
text_area.insert("/")
|
||||
await _pause_for_strip(pilot)
|
||||
assert chat_input.mode == "command"
|
||||
assert chat_input._current_suggestions
|
||||
|
||||
text_area.insert("help")
|
||||
await pilot.pause()
|
||||
await pilot.click("#clear-button")
|
||||
await pilot.pause()
|
||||
|
||||
assert chat_input.mode == "normal"
|
||||
assert chat_input.value == ""
|
||||
assert chat_input._current_suggestions == []
|
||||
|
||||
text_area.insert("hello")
|
||||
await pilot.pause()
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
assert app.submitted[0].value == "hello"
|
||||
assert app.submitted[0].mode == "normal"
|
||||
|
||||
async def test_copy_button_copies_input(self, monkeypatch) -> None:
|
||||
"""Clicking `[ COPY ]` sends the draft to the clipboard helper."""
|
||||
import deepagents_code.clipboard as clipboard_module
|
||||
|
||||
copied: list[str] = []
|
||||
|
||||
def fake_copy(_app: object, text: str) -> tuple[bool, str | None]:
|
||||
copied.append(text)
|
||||
return True, None
|
||||
|
||||
monkeypatch.setattr(clipboard_module, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("copy me")
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.click("#copy-button")
|
||||
await pilot.pause()
|
||||
|
||||
assert copied == ["copy me"]
|
||||
|
||||
async def test_copy_button_failure_warns(self, monkeypatch) -> None:
|
||||
"""A failed `[ COPY ]` surfaces a warning toast instead of failing silently."""
|
||||
import deepagents_code.clipboard as clipboard_module
|
||||
|
||||
def fake_copy(_app: object, _text: str) -> tuple[bool, str | None]:
|
||||
return False, "boom"
|
||||
|
||||
monkeypatch.setattr(clipboard_module, "copy_text_to_clipboard", fake_copy)
|
||||
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("copy me")
|
||||
await pilot.pause()
|
||||
|
||||
notifications: list[tuple[str, object]] = []
|
||||
monkeypatch.setattr(
|
||||
app,
|
||||
"notify",
|
||||
lambda message, **kwargs: notifications.append(
|
||||
(message, kwargs.get("severity"))
|
||||
),
|
||||
)
|
||||
|
||||
await pilot.click("#copy-button")
|
||||
await pilot.pause()
|
||||
|
||||
assert notifications == [("Failed to copy input: boom", "warning")]
|
||||
|
||||
async def test_clear_button_refocuses_input(self) -> None:
|
||||
"""Clicking `[ X ]` returns focus to the text area so typing can continue."""
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("clear me")
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.click("#clear-button")
|
||||
await pilot.pause()
|
||||
assert text_area.has_focus
|
||||
|
||||
async def test_copy_button_refocuses_input(self, monkeypatch) -> None:
|
||||
"""`[ COPY ]` returns focus to the input (not the non-focusable button)."""
|
||||
import deepagents_code.clipboard as clipboard_module
|
||||
|
||||
monkeypatch.setattr(
|
||||
clipboard_module,
|
||||
"copy_text_to_clipboard",
|
||||
lambda _app, _text: (True, None),
|
||||
)
|
||||
|
||||
app = _ChatInputTestApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
text_area = chat_input.input_widget
|
||||
assert text_area is not None
|
||||
text_area.insert("copy me") # buttons only render with a draft
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.click("#copy-button")
|
||||
await pilot.pause()
|
||||
assert text_area.has_focus
|
||||
|
||||
|
||||
class _ImagePasteApp(App[None]):
|
||||
"""App that wires a shared tracker into ChatInput for paste tests."""
|
||||
|
||||
@@ -322,8 +659,9 @@ class TestPromptIndicator:
|
||||
chat_input.mode = "shell_incognito"
|
||||
await pilot.pause()
|
||||
|
||||
input_box = chat_input.query_one("#input-box")
|
||||
assert _prompt_text(prompt) == "$"
|
||||
assert chat_input.border_title == "incognito"
|
||||
assert input_box.border_title == "incognito"
|
||||
assert chat_input.has_class("mode-shell-incognito")
|
||||
|
||||
async def test_incognito_shell_to_shell_clears_incognito_styling(self) -> None:
|
||||
@@ -336,14 +674,15 @@ class TestPromptIndicator:
|
||||
async with app.run_test() as pilot:
|
||||
chat_input = app.query_one(ChatInput)
|
||||
|
||||
input_box = chat_input.query_one("#input-box")
|
||||
chat_input.mode = "shell_incognito"
|
||||
await pilot.pause()
|
||||
assert chat_input.border_title == "incognito"
|
||||
assert input_box.border_title == "incognito"
|
||||
assert chat_input.has_class("mode-shell-incognito")
|
||||
|
||||
chat_input.mode = "shell"
|
||||
await pilot.pause()
|
||||
assert chat_input.border_title is None
|
||||
assert input_box.border_title is None
|
||||
assert not chat_input.has_class("mode-shell-incognito")
|
||||
assert chat_input.has_class("mode-shell")
|
||||
|
||||
|
||||
@@ -18,6 +18,7 @@ from deepagents_code.clipboard import (
|
||||
_copy_osc52,
|
||||
copy_selection_to_clipboard,
|
||||
copy_text_to_clipboard,
|
||||
copy_text_with_feedback,
|
||||
logger as clipboard_logger,
|
||||
)
|
||||
|
||||
@@ -96,6 +97,28 @@ class TestCopyTextToClipboard:
|
||||
assert "no app clipboard" in caplog.text
|
||||
assert "no tty" in caplog.text
|
||||
|
||||
def test_unicode_encode_error_is_caught_not_propagated(self, caplog) -> None:
|
||||
"""A backend raising `UnicodeEncodeError` is caught, not propagated.
|
||||
|
||||
`_copy_osc52` does `text.encode("utf-8")`, which raises
|
||||
`UnicodeEncodeError` (a `ValueError` subclass) on lone surrogate code
|
||||
points. The loop must treat it like any other backend failure so the
|
||||
`(success, error)` contract holds instead of crashing the caller.
|
||||
"""
|
||||
mock_app = MagicMock()
|
||||
mock_app.copy_to_clipboard.side_effect = OSError("no app clipboard")
|
||||
boom = UnicodeEncodeError("utf-8", "\ud800", 0, 1, "surrogates not allowed")
|
||||
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError("no pyperclip")),
|
||||
patch("deepagents_code.clipboard._copy_osc52", side_effect=boom),
|
||||
caplog.at_level(logging.DEBUG),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "\ud800")
|
||||
|
||||
assert success is False
|
||||
assert "surrogates not allowed" in (error or "")
|
||||
|
||||
def test_returns_exception_class_name_when_message_empty(self) -> None:
|
||||
"""An exception with empty `str()` still produces a non-empty reason."""
|
||||
mock_app = MagicMock()
|
||||
@@ -145,6 +168,90 @@ class TestCopyTextToClipboard:
|
||||
mock_app.copy_to_clipboard.assert_not_called()
|
||||
|
||||
|
||||
class TestCopyTextWithFeedback:
|
||||
"""The copy-then-notify helper shared by Ctrl+C and the `[ COPY ]` button."""
|
||||
|
||||
def test_success_with_message_notifies(self) -> None:
|
||||
"""A successful copy with a message emits a confirmation toast."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy:
|
||||
result = copy_text_with_feedback(
|
||||
mock_app,
|
||||
"draft",
|
||||
failure_noun="input",
|
||||
success_message="Input copied to clipboard",
|
||||
)
|
||||
|
||||
assert result is True
|
||||
copy.assert_called_once_with(mock_app, "draft")
|
||||
mock_app.notify.assert_called_once_with(
|
||||
"Input copied to clipboard",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
def test_success_without_message_is_silent(self) -> None:
|
||||
"""With no success message, a successful copy emits no toast."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
):
|
||||
result = copy_text_with_feedback(
|
||||
mock_app,
|
||||
"selection",
|
||||
failure_noun="selection",
|
||||
)
|
||||
|
||||
assert result is True
|
||||
mock_app.notify.assert_not_called()
|
||||
|
||||
def test_failure_warns_with_noun_and_markup_disabled(self) -> None:
|
||||
"""A failed copy warns using the failure noun, with `markup=False`."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(False, "boom"),
|
||||
):
|
||||
result = copy_text_with_feedback(
|
||||
mock_app,
|
||||
"draft",
|
||||
failure_noun="input",
|
||||
success_message="Input copied to clipboard",
|
||||
)
|
||||
|
||||
assert result is False
|
||||
mock_app.notify.assert_called_once_with(
|
||||
"Failed to copy input: boom",
|
||||
severity="warning",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
def test_failure_without_error_uses_fallback_message(self) -> None:
|
||||
"""A failure with no error detail falls back to the generic warning."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with patch(
|
||||
"deepagents_code.clipboard.copy_text_to_clipboard",
|
||||
return_value=(False, None),
|
||||
):
|
||||
copy_text_with_feedback(mock_app, "draft", failure_noun="input")
|
||||
|
||||
mock_app.notify.assert_called_once_with(
|
||||
"Failed to copy input - no clipboard method available",
|
||||
severity="warning",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
|
||||
class TestCopyOsc52:
|
||||
"""Direct coverage of the OSC 52 escape sequence (`_copy_osc52`)."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user