mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-22 01:35:28 -04:00
feat(cli): /copy slash command (#3225)
Closes #743 --- Adds a `/copy` slash command that copies the latest completed assistant message to the clipboard. The command reuses the existing clipboard backends, keeps source markdown intact, and reports empty or failed-copy states in chat. Also registers `/copy` in command autocomplete, help, queue policy coverage, and startup tips. --------- Co-authored-by: Mason Daugherty <mason@langchain.dev> Co-authored-by: Mason Daugherty <github@mdrxy.com>
This commit is contained in:
committed by
GitHub
parent
3e9e80597d
commit
646df53642
@@ -4300,7 +4300,7 @@ class DeepAgentsApp(App):
|
||||
await self._mount_message(UserMessage(command))
|
||||
help_body = (
|
||||
"Commands: /quit, /agents, /auth, /clear, /force-clear, "
|
||||
"/offload, /editor, "
|
||||
"/copy, /offload, /editor, "
|
||||
"/mcp, /model [--model-params JSON] [--default], "
|
||||
"/notifications, /reload, /skill:<name>, /remember, "
|
||||
"/skill-creator, /theme, /tokens, /threads, /trace, "
|
||||
@@ -4351,6 +4351,49 @@ class DeepAgentsApp(App):
|
||||
await self._mount_message(
|
||||
AppMessage(f"Started new thread: {new_thread_id}")
|
||||
)
|
||||
elif cmd == "/copy":
|
||||
await self._mount_message(UserMessage(command))
|
||||
# Reverse-scan for the newest assistant message that has finished
|
||||
# streaming and contains visible text. Track whether we passed over
|
||||
# an in-flight stream so we can explain the skip rather than say
|
||||
# "No message to copy yet." misleadingly.
|
||||
content: str | None = None
|
||||
streaming_pending = False
|
||||
for message in reversed(self._message_store.get_all_messages()):
|
||||
if message.type != MessageType.ASSISTANT:
|
||||
continue
|
||||
if not message.content.strip():
|
||||
continue
|
||||
if message.is_streaming:
|
||||
streaming_pending = True
|
||||
continue
|
||||
content = message.content
|
||||
break
|
||||
|
||||
if content is None:
|
||||
empty_msg = (
|
||||
"Latest assistant message is still streaming;"
|
||||
" try again in a moment."
|
||||
if streaming_pending
|
||||
else "No message to copy yet."
|
||||
)
|
||||
await self._mount_message(AppMessage(empty_msg))
|
||||
return
|
||||
|
||||
from deepagents_cli.clipboard import copy_text_to_clipboard
|
||||
|
||||
success, error = copy_text_to_clipboard(self, content)
|
||||
if success:
|
||||
await self._mount_message(
|
||||
AppMessage("Copied latest assistant message to clipboard.")
|
||||
)
|
||||
else:
|
||||
fail_msg = (
|
||||
f"Failed to copy latest assistant message to clipboard: {error}"
|
||||
if error
|
||||
else "Failed to copy latest assistant message to clipboard."
|
||||
)
|
||||
await self._mount_message(AppMessage(fail_msg))
|
||||
elif cmd == "/editor":
|
||||
await self.action_open_editor()
|
||||
elif cmd in {"/offload", "/compact"}:
|
||||
@@ -7338,7 +7381,7 @@ class DeepAgentsApp(App):
|
||||
)
|
||||
progress_modal_visible = not isinstance(self.screen, ModalScreen)
|
||||
if progress_modal_visible:
|
||||
self.push_screen(screen)
|
||||
await self.push_screen(screen)
|
||||
else:
|
||||
self.notify(
|
||||
f"Updating to v{payload.latest}... Logs: {log_path}",
|
||||
|
||||
@@ -13,6 +13,8 @@ from deepagents_cli.config import get_glyphs
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
from textual.app import App
|
||||
|
||||
_PREVIEW_MAX_LENGTH = 40
|
||||
@@ -43,6 +45,54 @@ def _shorten_preview(texts: list[str]) -> str:
|
||||
return dense_text
|
||||
|
||||
|
||||
def copy_text_to_clipboard(app: App, text: str) -> tuple[bool, str | None]:
|
||||
"""Copy text to the system clipboard.
|
||||
|
||||
Args:
|
||||
app: The active Textual app, used for the app clipboard backend.
|
||||
text: Text to copy.
|
||||
|
||||
Returns:
|
||||
Tuple of `(success, error_message)`.
|
||||
|
||||
`success` is `True` when one backend completed without raising.
|
||||
`error_message` is `None` on success and the last backend's error
|
||||
string when every backend failed, suitable for surfacing to the
|
||||
user so they can self-diagnose missing clipboard support.
|
||||
"""
|
||||
# Backend order: pyperclip first (most reliable when installed), then
|
||||
# Textual's app clipboard, then OSC 52 as a last resort for SSH/remote
|
||||
# sessions where no local clipboard is reachable.
|
||||
copy_methods: list[Callable[[str], object]] = [app.copy_to_clipboard]
|
||||
|
||||
try:
|
||||
import pyperclip
|
||||
|
||||
copy_methods.insert(0, pyperclip.copy)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
copy_methods.append(_copy_osc52)
|
||||
|
||||
last_error: str | None = None
|
||||
for copy_fn in copy_methods:
|
||||
try:
|
||||
copy_fn(text)
|
||||
except (OSError, RuntimeError, TypeError) as e:
|
||||
last_error = str(e) or type(e).__name__
|
||||
logger.debug(
|
||||
"Clipboard copy method %s failed: %s",
|
||||
getattr(copy_fn, "__name__", repr(copy_fn)),
|
||||
e,
|
||||
exc_info=True,
|
||||
)
|
||||
continue
|
||||
else:
|
||||
return True, None
|
||||
|
||||
return False, last_error
|
||||
|
||||
|
||||
def copy_selection_to_clipboard(app: App) -> None:
|
||||
"""Copy selected text from app widgets to clipboard.
|
||||
|
||||
@@ -83,46 +133,23 @@ def copy_selection_to_clipboard(app: App) -> None:
|
||||
|
||||
combined_text = "\n".join(selected_texts)
|
||||
|
||||
# Try multiple clipboard methods
|
||||
# Prefer pyperclip/app clipboard first (works reliably on local machines)
|
||||
# OSC 52 is last resort (for SSH/remote where native clipboard unavailable)
|
||||
copy_methods = [app.copy_to_clipboard]
|
||||
success, _ = copy_text_to_clipboard(app, combined_text)
|
||||
if success:
|
||||
# Use markup=False to prevent copied text from being parsed as Rich markup
|
||||
app.notify(
|
||||
f'"{_shorten_preview(selected_texts)}" copied',
|
||||
severity="information",
|
||||
timeout=2,
|
||||
markup=False,
|
||||
)
|
||||
return
|
||||
|
||||
# Try pyperclip if available (preferred - uses pbcopy on macOS)
|
||||
try:
|
||||
import pyperclip
|
||||
|
||||
copy_methods.insert(0, pyperclip.copy)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
# OSC 52 as fallback for remote/SSH sessions
|
||||
copy_methods.append(_copy_osc52)
|
||||
|
||||
for copy_fn in copy_methods:
|
||||
try:
|
||||
copy_fn(combined_text)
|
||||
# Use markup=False to prevent copied text from being parsed as Rich markup
|
||||
app.notify(
|
||||
f'"{_shorten_preview(selected_texts)}" copied',
|
||||
severity="information",
|
||||
timeout=2,
|
||||
markup=False,
|
||||
)
|
||||
except (OSError, RuntimeError, TypeError) as e:
|
||||
logger.debug(
|
||||
"Clipboard copy method %s failed: %s",
|
||||
getattr(copy_fn, "__name__", repr(copy_fn)),
|
||||
e,
|
||||
exc_info=True,
|
||||
)
|
||||
continue
|
||||
else:
|
||||
return
|
||||
|
||||
# If all methods fail, still notify but warn
|
||||
# If all methods fail, still notify but warn. markup=False guards against
|
||||
# this string ever growing dynamic content (e.g., the backend error reason)
|
||||
# that could contain bracket characters.
|
||||
app.notify(
|
||||
"Failed to copy - no clipboard method available",
|
||||
severity="warning",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
@@ -91,6 +91,11 @@ COMMANDS: tuple[SlashCommand, ...] = (
|
||||
bypass_tier=BypassTier.QUEUED,
|
||||
hidden_keywords="reset",
|
||||
),
|
||||
SlashCommand(
|
||||
name="/copy",
|
||||
description="Copy latest assistant message to clipboard",
|
||||
bypass_tier=BypassTier.SIDE_EFFECT_FREE,
|
||||
),
|
||||
SlashCommand(
|
||||
name="/force-clear",
|
||||
description="Interrupt active work, clear chat, and start new thread",
|
||||
|
||||
@@ -37,9 +37,10 @@ from deepagents_cli.config import (
|
||||
from deepagents_cli.widgets._links import open_style_link
|
||||
|
||||
_TIPS: dict[str, int] = {
|
||||
"Use @ to reference files and / for commands": 2,
|
||||
"Use @ to reference files and / for commands": 3,
|
||||
"Try /threads to resume a previous conversation": 2,
|
||||
"Use /offload when your conversation gets long": 2,
|
||||
"Use /copy to copy the latest assistant message": 3,
|
||||
"Use /mcp to see your loaded tools and servers": 1,
|
||||
"Use /remember to save learnings from this conversation": 1,
|
||||
"Use /model to switch models mid-conversation": 2,
|
||||
|
||||
@@ -2456,6 +2456,193 @@ class TestTraceCommand:
|
||||
assert isinstance(app.screen, AuthManagerScreen)
|
||||
|
||||
|
||||
class TestCopyCommand:
|
||||
"""Tests for `/copy` command behavior."""
|
||||
|
||||
async def test_copy_latest_assistant_message_to_clipboard(self) -> None:
|
||||
"""`/copy` copies the latest stored assistant markdown exactly."""
|
||||
from deepagents_cli.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
markdown = "# Result\n\n- keep **markdown** source"
|
||||
app._message_store.append(
|
||||
MessageData(type=MessageType.ASSISTANT, content=markdown)
|
||||
)
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock:
|
||||
await app._handle_command("/copy")
|
||||
await pilot.pause()
|
||||
|
||||
copy_mock.assert_called_once_with(app, markdown)
|
||||
assert any(w._content == "/copy" for w in app.query(UserMessage))
|
||||
assert any(
|
||||
str(w._content) == "Copied latest assistant message to clipboard."
|
||||
for w in app.query(AppMessage)
|
||||
)
|
||||
|
||||
async def test_copy_skips_ineligible_newer_messages(self) -> None:
|
||||
"""`/copy` reverse-scans for the newest completed assistant text."""
|
||||
from deepagents_cli.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
expected = "older completed assistant"
|
||||
app._message_store.append(
|
||||
MessageData(type=MessageType.ASSISTANT, content=expected)
|
||||
)
|
||||
app._message_store.append(MessageData(type=MessageType.USER, content="thanks"))
|
||||
app._message_store.append(MessageData(type=MessageType.ASSISTANT, content=""))
|
||||
app._message_store.append(
|
||||
MessageData(type=MessageType.ASSISTANT, content=" ")
|
||||
)
|
||||
app._message_store.append(MessageData(type=MessageType.APP, content="status"))
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock:
|
||||
await app._handle_command("/copy")
|
||||
await pilot.pause()
|
||||
|
||||
copy_mock.assert_called_once_with(app, expected)
|
||||
|
||||
async def test_copy_falls_back_when_only_streaming_assistant_present(self) -> None:
|
||||
"""`/copy` skips an in-flight stream and copies the prior completed reply."""
|
||||
from deepagents_cli.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
completed = "completed reply"
|
||||
app._message_store.append(
|
||||
MessageData(type=MessageType.ASSISTANT, content=completed)
|
||||
)
|
||||
app._message_store.append(
|
||||
MessageData(
|
||||
type=MessageType.ASSISTANT,
|
||||
content="partial response",
|
||||
is_streaming=True,
|
||||
)
|
||||
)
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock:
|
||||
await app._handle_command("/copy")
|
||||
await pilot.pause()
|
||||
|
||||
copy_mock.assert_called_once_with(app, completed)
|
||||
|
||||
async def test_copy_reports_streaming_pending_when_only_stream_present(
|
||||
self,
|
||||
) -> None:
|
||||
"""`/copy` distinguishes in-flight streams from a truly empty history."""
|
||||
from deepagents_cli.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
app._message_store.append(MessageData(type=MessageType.USER, content="hi"))
|
||||
app._message_store.append(
|
||||
MessageData(
|
||||
type=MessageType.ASSISTANT,
|
||||
content="partial response",
|
||||
is_streaming=True,
|
||||
)
|
||||
)
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock:
|
||||
await app._handle_command("/copy")
|
||||
await pilot.pause()
|
||||
|
||||
copy_mock.assert_not_called()
|
||||
assert any(
|
||||
"still streaming" in str(w._content) for w in app.query(AppMessage)
|
||||
)
|
||||
|
||||
async def test_copy_reports_empty_state_without_clipboard_call(self) -> None:
|
||||
"""`/copy` reports empty state when no assistant text is eligible."""
|
||||
from deepagents_cli.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
app._message_store.append(MessageData(type=MessageType.USER, content="hello"))
|
||||
app._message_store.append(MessageData(type=MessageType.ASSISTANT, content=" "))
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy_mock:
|
||||
await app._handle_command("/copy")
|
||||
await pilot.pause()
|
||||
|
||||
copy_mock.assert_not_called()
|
||||
assert any(
|
||||
str(w._content) == "No message to copy yet."
|
||||
for w in app.query(AppMessage)
|
||||
)
|
||||
|
||||
async def test_copy_reports_clipboard_failure_with_reason(self) -> None:
|
||||
"""`/copy` surfaces the backend error so users can self-diagnose."""
|
||||
from deepagents_cli.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
app._message_store.append(
|
||||
MessageData(type=MessageType.ASSISTANT, content="assistant text")
|
||||
)
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(False, "no clipboard mechanism for your system"),
|
||||
):
|
||||
await app._handle_command("/copy")
|
||||
await pilot.pause()
|
||||
|
||||
assert any(
|
||||
str(w._content)
|
||||
== "Failed to copy latest assistant message to clipboard:"
|
||||
" no clipboard mechanism for your system"
|
||||
for w in app.query(AppMessage)
|
||||
)
|
||||
|
||||
async def test_copy_reports_clipboard_failure_without_reason(self) -> None:
|
||||
"""`/copy` falls back to a generic message when no error string is given."""
|
||||
from deepagents_cli.widgets.message_store import MessageData, MessageType
|
||||
|
||||
app = DeepAgentsApp()
|
||||
app._message_store.append(
|
||||
MessageData(type=MessageType.ASSISTANT, content="assistant text")
|
||||
)
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(False, None),
|
||||
):
|
||||
await app._handle_command("/copy")
|
||||
await pilot.pause()
|
||||
|
||||
assert any(
|
||||
str(w._content)
|
||||
== "Failed to copy latest assistant message to clipboard."
|
||||
for w in app.query(AppMessage)
|
||||
)
|
||||
|
||||
|
||||
class TestRunAgentTaskMediaTracker:
|
||||
"""Tests image tracker wiring from app into textual execution."""
|
||||
|
||||
@@ -4545,7 +4732,7 @@ class TestDeferredActions:
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
for cmd in ("/changelog", "/docs", "/feedback", "/mcp"):
|
||||
for cmd in ("/changelog", "/copy", "/docs", "/feedback", "/mcp"):
|
||||
assert app._can_bypass_queue(cmd) is True
|
||||
|
||||
async def test_queued_commands_do_not_bypass(self) -> None:
|
||||
@@ -5924,6 +6111,43 @@ class TestNotificationCenterIntegration:
|
||||
assert any("Run manually" in m for m in notified)
|
||||
assert any("network unreachable" in m for m in notified)
|
||||
|
||||
async def test_install_immediate_failure_updates_mounted_modal(self) -> None:
|
||||
"""Immediate install failures still render the completed modal state."""
|
||||
from deepagents_cli.config import get_glyphs
|
||||
from deepagents_cli.notifications import ActionId
|
||||
from deepagents_cli.widgets.update_progress import UpdateProgressScreen
|
||||
|
||||
app = DeepAgentsApp(agent=MagicMock(), thread_id="t")
|
||||
entry = _update_entry()
|
||||
app._notice_registry.add(entry)
|
||||
|
||||
def fail_immediately(
|
||||
**kwargs: Any,
|
||||
) -> tuple[bool, str]:
|
||||
assert kwargs["progress"] is not None
|
||||
assert kwargs["log_path"] is not None
|
||||
assert isinstance(app.screen, UpdateProgressScreen)
|
||||
assert app.screen._status_widget is not None
|
||||
assert app.screen._tail_widget is not None
|
||||
return False, "brew: command not found"
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch(
|
||||
"deepagents_cli.update_check.perform_upgrade",
|
||||
new=AsyncMock(side_effect=fail_immediately),
|
||||
):
|
||||
await app._dispatch_notification_action(entry.key, ActionId.INSTALL)
|
||||
await pilot.pause()
|
||||
|
||||
assert isinstance(app.screen, UpdateProgressScreen)
|
||||
status = app.screen.query(Static).filter(".up-status").first()
|
||||
details = app.screen.query(Static).filter(".up-details").first()
|
||||
spinner = app.screen.query(Static).filter(".up-spinner").first()
|
||||
assert "Update failed. Try manually:" in str(status.render())
|
||||
assert details.display is True
|
||||
assert str(spinner.render()) == get_glyphs().checkmark
|
||||
|
||||
async def test_update_skip_once_clears_notified_marker(self) -> None:
|
||||
"""'Remind me next launch' calls clear_update_notified and removes the entry."""
|
||||
from deepagents_cli.notifications import ActionId
|
||||
|
||||
@@ -0,0 +1,274 @@
|
||||
"""Unit tests for `deepagents_cli.clipboard`.
|
||||
|
||||
Covers the clipboard-backend fallback chain (`copy_text_to_clipboard`),
|
||||
selection-driven copy with notification UX (`copy_selection_to_clipboard`),
|
||||
and the OSC 52 escape envelope (`_copy_osc52`).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import io
|
||||
import logging
|
||||
import sys
|
||||
from typing import Self
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from deepagents_cli.clipboard import (
|
||||
_copy_osc52,
|
||||
copy_selection_to_clipboard,
|
||||
copy_text_to_clipboard,
|
||||
logger as clipboard_logger,
|
||||
)
|
||||
|
||||
|
||||
class TestCopyTextToClipboard:
|
||||
"""Test the multi-backend `copy_text_to_clipboard` fallback chain."""
|
||||
|
||||
def test_returns_true_when_pyperclip_succeeds(self) -> None:
|
||||
"""Stops at pyperclip when it succeeds; later backends untouched."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with (
|
||||
patch("pyperclip.copy") as copy,
|
||||
patch("deepagents_cli.clipboard._copy_osc52") as osc52,
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
copy.assert_called_once_with("hello")
|
||||
mock_app.copy_to_clipboard.assert_not_called()
|
||||
osc52.assert_not_called()
|
||||
|
||||
def test_falls_back_to_app_clipboard(self, caplog) -> None:
|
||||
"""Uses Textual `app.copy_to_clipboard` after pyperclip raises."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError("no pyperclip")) as copy,
|
||||
caplog.at_level(logging.DEBUG),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
copy.assert_called_once_with("hello")
|
||||
mock_app.copy_to_clipboard.assert_called_once_with("hello")
|
||||
assert "no pyperclip" in caplog.text
|
||||
|
||||
def test_falls_back_to_osc52(self, caplog) -> None:
|
||||
"""Uses OSC 52 after pyperclip and app clipboard both raise."""
|
||||
mock_app = MagicMock()
|
||||
mock_app.copy_to_clipboard.side_effect = OSError("no app clipboard")
|
||||
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError("no pyperclip")),
|
||||
patch("deepagents_cli.clipboard._copy_osc52") as osc52,
|
||||
caplog.at_level(logging.DEBUG),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
osc52.assert_called_once_with("hello")
|
||||
assert "no pyperclip" in caplog.text
|
||||
assert "no app clipboard" in caplog.text
|
||||
|
||||
def test_returns_last_error_when_all_backends_fail(self, caplog) -> None:
|
||||
"""Returns `(False, last_error)` so callers can surface a reason."""
|
||||
mock_app = MagicMock()
|
||||
mock_app.copy_to_clipboard.side_effect = OSError("no app clipboard")
|
||||
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError("no pyperclip")),
|
||||
patch(
|
||||
"deepagents_cli.clipboard._copy_osc52",
|
||||
side_effect=OSError("no tty"),
|
||||
),
|
||||
caplog.at_level(logging.DEBUG),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
assert success is False
|
||||
assert error == "no tty"
|
||||
assert "no pyperclip" in caplog.text
|
||||
assert "no app clipboard" in caplog.text
|
||||
assert "no tty" in caplog.text
|
||||
|
||||
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()
|
||||
mock_app.copy_to_clipboard.side_effect = OSError()
|
||||
|
||||
with (
|
||||
patch("pyperclip.copy", side_effect=RuntimeError()),
|
||||
patch(
|
||||
"deepagents_cli.clipboard._copy_osc52",
|
||||
side_effect=OSError(),
|
||||
),
|
||||
):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
assert success is False
|
||||
assert error == "OSError"
|
||||
|
||||
def test_pyperclip_import_error_falls_through_to_app_clipboard(self) -> None:
|
||||
"""Missing `pyperclip` module skips that backend cleanly."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with patch.dict(sys.modules, {"pyperclip": None}):
|
||||
success, error = copy_text_to_clipboard(mock_app, "hello")
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
mock_app.copy_to_clipboard.assert_called_once_with("hello")
|
||||
|
||||
def test_pyperclip_receives_markdown_byte_for_byte(self) -> None:
|
||||
"""Markdown source reaches `pyperclip.copy` unmodified.
|
||||
|
||||
Real `copy_text_to_clipboard` runs (no shim); only `pyperclip.copy`
|
||||
is patched, so this catches any future "helpful" stripping or
|
||||
normalization before the backend.
|
||||
"""
|
||||
mock_app = MagicMock()
|
||||
markdown = "# Result\n\n- keep **markdown** source\n\n```py\nx = 1\n```"
|
||||
|
||||
captured: list[str] = []
|
||||
|
||||
with patch("pyperclip.copy", side_effect=captured.append):
|
||||
success, error = copy_text_to_clipboard(mock_app, markdown)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
assert captured == [markdown]
|
||||
mock_app.copy_to_clipboard.assert_not_called()
|
||||
|
||||
|
||||
class TestCopyOsc52:
|
||||
"""Direct coverage of the OSC 52 escape sequence (`_copy_osc52`)."""
|
||||
|
||||
def test_emits_escape_envelope(self, monkeypatch) -> None:
|
||||
r"""Emits ``\x1b]52;c;<base64>\a`` written to ``/dev/tty``."""
|
||||
captured = io.StringIO()
|
||||
|
||||
class _DummyTTY:
|
||||
def __init__(self) -> None:
|
||||
self.buffer = captured
|
||||
|
||||
def __enter__(self) -> Self:
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
pass
|
||||
|
||||
def write(self, s: str) -> int:
|
||||
self.buffer.write(s)
|
||||
return len(s)
|
||||
|
||||
def flush(self) -> None:
|
||||
pass
|
||||
|
||||
monkeypatch.delenv("TMUX", raising=False)
|
||||
text = "hello world"
|
||||
with patch("pathlib.Path.open", return_value=_DummyTTY()):
|
||||
_copy_osc52(text)
|
||||
|
||||
encoded = base64.b64encode(text.encode("utf-8")).decode("ascii")
|
||||
assert captured.getvalue() == f"\033]52;c;{encoded}\a"
|
||||
|
||||
def test_wraps_envelope_for_tmux_passthrough(self, monkeypatch) -> None:
|
||||
"""Inside tmux, the OSC 52 sequence must be wrapped in DCS passthrough."""
|
||||
captured = io.StringIO()
|
||||
|
||||
class _DummyTTY:
|
||||
def __enter__(self) -> Self:
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
pass
|
||||
|
||||
def write(self, s: str) -> int:
|
||||
captured.write(s)
|
||||
return len(s)
|
||||
|
||||
def flush(self) -> None:
|
||||
pass
|
||||
|
||||
monkeypatch.setenv("TMUX", "/tmp/tmux-1000/default,1234,0")
|
||||
text = "hi"
|
||||
with patch("pathlib.Path.open", return_value=_DummyTTY()):
|
||||
_copy_osc52(text)
|
||||
|
||||
encoded = base64.b64encode(text.encode("utf-8")).decode("ascii")
|
||||
inner = f"\033]52;c;{encoded}\a"
|
||||
expected = f"\033Ptmux;\033{inner}\033\\"
|
||||
assert captured.getvalue() == expected
|
||||
|
||||
|
||||
class TestCopySelectionToClipboard:
|
||||
"""Selection-driven copy that delegates to `copy_text_to_clipboard`."""
|
||||
|
||||
def test_delegates_and_notifies_on_success(self) -> None:
|
||||
"""Delegates the side effect and emits an informational toast."""
|
||||
mock_app = MagicMock()
|
||||
selection = MagicMock(end=1)
|
||||
widget = MagicMock()
|
||||
widget.text_selection = selection
|
||||
widget.get_selection.return_value = ("selected text", None)
|
||||
mock_app.query.return_value = [widget]
|
||||
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(True, None),
|
||||
) as copy:
|
||||
copy_selection_to_clipboard(mock_app)
|
||||
|
||||
copy.assert_called_once_with(mock_app, "selected text")
|
||||
mock_app.notify.assert_called_once()
|
||||
assert mock_app.notify.call_args.kwargs["severity"] == "information"
|
||||
assert mock_app.notify.call_args.kwargs["markup"] is False
|
||||
|
||||
def test_warns_with_markup_disabled_when_helper_fails(self) -> None:
|
||||
"""Failure path uses `markup=False` to stay safe under future edits."""
|
||||
mock_app = MagicMock()
|
||||
selection = MagicMock(end=1)
|
||||
widget = MagicMock()
|
||||
widget.text_selection = selection
|
||||
widget.get_selection.return_value = ("selected text", None)
|
||||
mock_app.query.return_value = [widget]
|
||||
|
||||
with patch(
|
||||
"deepagents_cli.clipboard.copy_text_to_clipboard",
|
||||
return_value=(False, "no clipboard mechanism"),
|
||||
):
|
||||
copy_selection_to_clipboard(mock_app)
|
||||
|
||||
mock_app.notify.assert_called_once_with(
|
||||
"Failed to copy - no clipboard method available",
|
||||
severity="warning",
|
||||
timeout=3,
|
||||
markup=False,
|
||||
)
|
||||
|
||||
def test_handles_widget_selection_failures(self, caplog) -> None:
|
||||
"""A failing widget logs and is skipped, not re-raised."""
|
||||
mock_app = MagicMock()
|
||||
mock_widget = MagicMock()
|
||||
mock_widget.text_selection = MagicMock()
|
||||
mock_widget.get_selection.side_effect = AttributeError("No selection")
|
||||
mock_app.query.return_value = [mock_widget]
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
copy_selection_to_clipboard(mock_app)
|
||||
|
||||
assert "Failed to get selection from widget" in caplog.text
|
||||
assert "No selection" in caplog.text
|
||||
|
||||
|
||||
class TestClipboardLogger:
|
||||
"""Sanity check: module exposes a properly named logger."""
|
||||
|
||||
def test_logger_exists_and_is_named(self) -> None:
|
||||
assert clipboard_logger is not None
|
||||
assert clipboard_logger.name == "deepagents_cli.clipboard"
|
||||
@@ -136,6 +136,21 @@ class TestAgentsCommand:
|
||||
assert "/agents" in IMMEDIATE_UI
|
||||
|
||||
|
||||
class TestCopyCommand:
|
||||
"""Validate the `/copy` entry specifically."""
|
||||
|
||||
def test_copy_registered_for_autocomplete(self) -> None:
|
||||
copy_entry = next(entry for entry in SLASH_COMMANDS if entry.name == "/copy")
|
||||
|
||||
assert copy_entry.description == "Copy latest assistant message to clipboard"
|
||||
|
||||
def test_copy_classified_as_side_effect_free(self) -> None:
|
||||
copy_cmd = next(cmd for cmd in COMMANDS if cmd.name == "/copy")
|
||||
|
||||
assert copy_cmd.description == "Copy latest assistant message to clipboard"
|
||||
assert "/copy" in SIDE_EFFECT_FREE
|
||||
|
||||
|
||||
class TestHelpBodyDrift:
|
||||
"""Ensure the /help body in app.py stays in sync with COMMANDS.
|
||||
|
||||
|
||||
@@ -17,10 +17,6 @@ import pytest
|
||||
from tavily import BadRequestError, InvalidAPIKeyError, UsageLimitExceededError
|
||||
from tavily.errors import TimeoutError as TavilyTimeoutError
|
||||
|
||||
from deepagents_cli.clipboard import (
|
||||
copy_selection_to_clipboard,
|
||||
logger as clipboard_logger,
|
||||
)
|
||||
from deepagents_cli.file_ops import FileOpTracker, _safe_read
|
||||
from deepagents_cli.media_utils import (
|
||||
_get_clipboard_via_osascript,
|
||||
@@ -164,33 +160,6 @@ class TestFileOpsExceptionHandling:
|
||||
assert "Failed to read file" in caplog.text
|
||||
|
||||
|
||||
class TestClipboardExceptionHandling:
|
||||
"""Test exception handling in clipboard utilities."""
|
||||
|
||||
def test_copy_handles_widget_selection_failures(self, caplog):
|
||||
"""Test that copy_selection_to_clipboard handles widget failures gracefully."""
|
||||
# Create a mock app with widgets
|
||||
mock_app = MagicMock()
|
||||
mock_widget = MagicMock()
|
||||
mock_widget.text_selection = MagicMock()
|
||||
mock_widget.get_selection.side_effect = AttributeError("No selection")
|
||||
|
||||
mock_app.query.return_value = [mock_widget]
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
# Should not raise
|
||||
copy_selection_to_clipboard(mock_app)
|
||||
|
||||
# Verify the error was logged
|
||||
assert "Failed to get selection from widget" in caplog.text
|
||||
assert "No selection" in caplog.text
|
||||
|
||||
def test_clipboard_logger_exists(self):
|
||||
"""Test that clipboard module has proper logging configured."""
|
||||
assert clipboard_logger is not None
|
||||
assert clipboard_logger.name == "deepagents_cli.clipboard"
|
||||
|
||||
|
||||
class TestMediaUtilsExceptionHandling:
|
||||
"""Test exception handling in media utilities."""
|
||||
|
||||
|
||||
@@ -426,6 +426,10 @@ class TestBuildWelcomeFooter:
|
||||
"""New `--startup-cmd` flag must have a discoverability tip."""
|
||||
assert any("--startup-cmd" in tip for tip in _TIPS)
|
||||
|
||||
def test_copy_command_tip_registered(self) -> None:
|
||||
"""The `/copy` command must have a discoverability tip."""
|
||||
assert "Use /copy to copy the latest assistant message" in _TIPS
|
||||
|
||||
def test_tip_varies_across_calls(self) -> None:
|
||||
"""Tips should rotate (not always the same)."""
|
||||
seen = {build_welcome_footer().plain for _ in range(50)}
|
||||
|
||||
Reference in New Issue
Block a user