mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-21 17:25:26 -04:00
feat(code): collapse large pastes into compact placeholders (#4447)
Large pastes (>800 chars or >2 newlines) are collapsed into a `[Pasted text #N +M lines]` placeholder in the input box. At submission the placeholder is expanded back to full text so the agent receives the complete content. Rendered user messages over 10,000 chars are truncated to head+tail (2,500 each) with an elision marker, matching Claude Code's behaviour. ## Changes - **`paste_collapse.py`** (new) — threshold constants, `PastedContent` dataclass, `should_collapse_paste`, `format_paste_ref`, `parse_paste_refs`, `expand_paste_refs` - **`ChatTextArea`** — new `PastedText` message; `_on_paste` and `_flush_paste_burst` intercept large pastes before Textual inserts them - **`ChatInput`** — stores paste content, inserts placeholder, expands at submit, cleans up orphaned content on delete, expands for clipboard copy - **`UserMessage` / `QueuedUserMessage`** — `_truncate_for_display()` renders head+tail with elision marker for messages over 10k chars; full text preserved in `_content` for copy/select ## Video https://github.com/user-attachments/assets/1d5425ed-d953-41f7-a898-0520610dd711 --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com> Co-authored-by: Mason Daugherty <github@mdrxy.com> Co-authored-by: Mason Daugherty <mason@langchain.dev>
This commit is contained in:
Johannes du Plessis
committed by
GitHub
GitHub
parent
56c5a5e0e6
commit
9ae927d73e
@@ -0,0 +1,103 @@
|
||||
r"""Large paste collapsing for the chat input.
|
||||
|
||||
When the user pastes text exceeding a size or line threshold, the full text
|
||||
is stored off-screen and a compact `[Pasted text #N +M lines]` placeholder
|
||||
is inserted into the input box instead. At submission time the placeholder
|
||||
is expanded back to the original content so the agent receives the full text.
|
||||
|
||||
This mirrors the behavior of Claude Code's paste-collapsing system.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
PASTE_THRESHOLD_CHARS = 800
|
||||
"""Minimum character count for a paste to be collapsed into a placeholder."""
|
||||
|
||||
PASTE_THRESHOLD_LINES = 2
|
||||
"""Minimum line count (newline-separated) for a paste to be collapsed."""
|
||||
|
||||
PASTE_PLACEHOLDER_PATTERN = re.compile(r"\[Pasted text #(\d+)(?: \+(\d+) lines)?\]")
|
||||
"""Regex matching `[Pasted text #N]` or `[Pasted text #N +M lines]`."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PastedContent:
|
||||
"""Stored content for a collapsed paste.
|
||||
|
||||
The paste's numeric identifier is the key under which this record is
|
||||
stored in the input's paste map, so it is not duplicated on the record.
|
||||
|
||||
Attributes:
|
||||
content: The full pasted text.
|
||||
"""
|
||||
|
||||
content: str
|
||||
|
||||
|
||||
def count_lines(text: str) -> int:
|
||||
r"""Return the number of newline characters in `text`.
|
||||
|
||||
Args:
|
||||
text: The text to count newlines in.
|
||||
|
||||
Returns:
|
||||
The number of `\n` occurrences (0 for single-line text).
|
||||
"""
|
||||
return text.count("\n")
|
||||
|
||||
|
||||
def should_collapse_paste(text: str) -> bool:
|
||||
"""Return whether `text` should be collapsed into a placeholder.
|
||||
|
||||
Collapses when the text exceeds the character threshold or the line
|
||||
threshold.
|
||||
|
||||
Args:
|
||||
text: The pasted text to evaluate.
|
||||
|
||||
Returns:
|
||||
`True` if the paste should be collapsed.
|
||||
"""
|
||||
return (
|
||||
len(text) > PASTE_THRESHOLD_CHARS or count_lines(text) > PASTE_THRESHOLD_LINES
|
||||
)
|
||||
|
||||
|
||||
def format_paste_ref(paste_id: int, num_lines: int) -> str:
|
||||
"""Format a paste placeholder reference string.
|
||||
|
||||
Args:
|
||||
paste_id: The numeric paste identifier.
|
||||
num_lines: The number of extra lines (newlines) in the pasted content.
|
||||
|
||||
Returns:
|
||||
`[Pasted text #N]` when `num_lines` is 0, otherwise
|
||||
`[Pasted text #N +M lines]`.
|
||||
"""
|
||||
if num_lines == 0:
|
||||
return f"[Pasted text #{paste_id}]"
|
||||
return f"[Pasted text #{paste_id} +{num_lines} lines]"
|
||||
|
||||
|
||||
def expand_paste_refs(text: str, pasted_contents: dict[int, PastedContent]) -> str:
|
||||
"""Replace all paste placeholders in `text` with their full content.
|
||||
|
||||
Placeholders whose IDs are not in `pasted_contents` are left unchanged.
|
||||
|
||||
Args:
|
||||
text: The text containing placeholders.
|
||||
pasted_contents: Mapping of paste IDs to stored content.
|
||||
|
||||
Returns:
|
||||
The text with all known placeholders expanded.
|
||||
"""
|
||||
|
||||
def _replace(match: re.Match[str]) -> str:
|
||||
content = pasted_contents.get(int(match.group(1)))
|
||||
# Return the stored text literally; unknown IDs keep their placeholder.
|
||||
return content.content if content is not None else match.group(0)
|
||||
|
||||
return PASTE_PLACEHOLDER_PATTERN.sub(_replace, text)
|
||||
@@ -30,6 +30,14 @@ from deepagents_code.config import (
|
||||
is_ascii_mode,
|
||||
)
|
||||
from deepagents_code.input import IMAGE_PLACEHOLDER_PATTERN, VIDEO_PLACEHOLDER_PATTERN
|
||||
from deepagents_code.paste_collapse import (
|
||||
PASTE_PLACEHOLDER_PATTERN,
|
||||
PastedContent,
|
||||
count_lines,
|
||||
expand_paste_refs,
|
||||
format_paste_ref,
|
||||
should_collapse_paste,
|
||||
)
|
||||
from deepagents_code.widgets.autocomplete import (
|
||||
CompletionResult,
|
||||
FuzzyFileController,
|
||||
@@ -138,6 +146,11 @@ if TYPE_CHECKING:
|
||||
from deepagents_code.input import MediaTracker, ParsedPastedPathPayload
|
||||
|
||||
|
||||
def _should_collapse_chat_paste(text: str) -> bool:
|
||||
"""Return whether pasted chat text should be collapsed."""
|
||||
return detect_mode_prefix(text) is None and should_collapse_paste(text)
|
||||
|
||||
|
||||
class CompletionOption(Static):
|
||||
"""A clickable completion option in the autocomplete popup."""
|
||||
|
||||
@@ -521,6 +534,22 @@ class ChatTextArea(TextArea):
|
||||
self.paths = paths
|
||||
super().__init__()
|
||||
|
||||
class PastedText(Message):
|
||||
"""Message sent when a paste is large enough to be collapsed.
|
||||
|
||||
The full text is carried in the message so `ChatInput` can store it
|
||||
and insert a compact placeholder into the text area instead.
|
||||
"""
|
||||
|
||||
def __init__(self, text: str) -> None:
|
||||
"""Initialize with the full pasted text.
|
||||
|
||||
Args:
|
||||
text: The complete pasted text content.
|
||||
"""
|
||||
self.text = text
|
||||
super().__init__()
|
||||
|
||||
class Typing(Message):
|
||||
"""Posted when the user presses a printable key or backspace.
|
||||
|
||||
@@ -993,7 +1022,7 @@ class ChatTextArea(TextArea):
|
||||
return row == 0 and col == 0
|
||||
|
||||
async def _flush_paste_burst(self) -> None:
|
||||
"""Flush buffered burst text through dropped-path parsing.
|
||||
"""Flush buffered burst text through dropped-path and large-paste checks.
|
||||
|
||||
When parsing fails, the buffered text is inserted unchanged so regular
|
||||
typing behavior is preserved.
|
||||
@@ -1009,12 +1038,24 @@ class ChatTextArea(TextArea):
|
||||
|
||||
try:
|
||||
parsed = await asyncio.to_thread(parse_pasted_path_payload, payload)
|
||||
except Exception: # noqa: BLE001 # Treat thread failure as non-path text
|
||||
except Exception:
|
||||
# The parser absorbs OSError/RuntimeError internally, so reaching
|
||||
# here signals an unexpected regression. Leave a breadcrumb (never
|
||||
# the paste content) instead of swallowing it, then fall through to
|
||||
# normal text handling.
|
||||
logger.debug(
|
||||
"Path-payload parsing failed; treating burst as text",
|
||||
exc_info=True,
|
||||
)
|
||||
parsed = None
|
||||
if parsed is not None:
|
||||
self.post_message(self.PastedPaths(payload, parsed.paths))
|
||||
return
|
||||
|
||||
if _should_collapse_chat_paste(payload):
|
||||
self.post_message(self.PastedText(payload))
|
||||
return
|
||||
|
||||
self.insert(payload)
|
||||
|
||||
def _delete_preceding_backslash(self) -> bool:
|
||||
@@ -1172,12 +1213,12 @@ class ChatTextArea(TextArea):
|
||||
self.action_insert_newline()
|
||||
return
|
||||
|
||||
if event.key == "backspace" and self._delete_image_placeholder(backwards=True):
|
||||
if event.key == "backspace" and self._delete_placeholder_token(backwards=True):
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
return
|
||||
|
||||
if event.key == "delete" and self._delete_image_placeholder(backwards=False):
|
||||
if event.key == "delete" and self._delete_placeholder_token(backwards=False):
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
return
|
||||
@@ -1227,8 +1268,8 @@ class ChatTextArea(TextArea):
|
||||
|
||||
await super()._on_key(event)
|
||||
|
||||
def _delete_image_placeholder(self, *, backwards: bool) -> bool:
|
||||
"""Delete a full image placeholder token in one keypress.
|
||||
def _delete_placeholder_token(self, *, backwards: bool) -> bool:
|
||||
"""Delete a full placeholder token (image, video, or paste) in one keypress.
|
||||
|
||||
Args:
|
||||
backwards: Whether the delete action is backwards (`backspace`) or
|
||||
@@ -1241,7 +1282,7 @@ class ChatTextArea(TextArea):
|
||||
return False
|
||||
|
||||
cursor_offset = self.document.get_index_from_location(self.cursor_location) # ty: ignore[unresolved-attribute] # Document has this method; DocumentBase stub is narrower
|
||||
span = self._find_image_placeholder_span(cursor_offset, backwards=backwards)
|
||||
span = self._find_placeholder_span(cursor_offset, backwards=backwards)
|
||||
if span is None:
|
||||
return False
|
||||
|
||||
@@ -1252,19 +1293,32 @@ class ChatTextArea(TextArea):
|
||||
self.move_cursor(start_location)
|
||||
return True
|
||||
|
||||
def _find_image_placeholder_span(
|
||||
def _find_placeholder_span(
|
||||
self, cursor_offset: int, *, backwards: bool
|
||||
) -> tuple[int, int] | None:
|
||||
"""Return placeholder span to delete for current cursor and key direction.
|
||||
|
||||
Covers image, video, and collapsed-paste placeholders so each deletes as
|
||||
a single atomic token. Paste placeholders carry backing content in
|
||||
`ChatInput._pasted_contents`; that map is intentionally left untouched
|
||||
here so an undo can restore the token with its content (it is cleared
|
||||
only at submit).
|
||||
|
||||
Args:
|
||||
cursor_offset: Character offset of the cursor from the start of text.
|
||||
backwards: Whether the delete action is backwards (backspace) or
|
||||
forwards (delete).
|
||||
|
||||
Returns:
|
||||
The `(start, end)` character span of the placeholder to delete, or
|
||||
`None` when the cursor is not adjacent to a placeholder token.
|
||||
"""
|
||||
text = self.text
|
||||
# Check both image and video placeholders
|
||||
for pattern in (IMAGE_PLACEHOLDER_PATTERN, VIDEO_PLACEHOLDER_PATTERN):
|
||||
for pattern in (
|
||||
IMAGE_PLACEHOLDER_PATTERN,
|
||||
VIDEO_PLACEHOLDER_PATTERN,
|
||||
PASTE_PLACEHOLDER_PATTERN,
|
||||
):
|
||||
for match in pattern.finditer(text):
|
||||
start, end = match.span()
|
||||
if backwards:
|
||||
@@ -1284,8 +1338,33 @@ class ChatTextArea(TextArea):
|
||||
return start, end
|
||||
return None
|
||||
|
||||
def replace_placeholder_with_text(self, paste_id: int, content: str) -> bool:
|
||||
"""Replace a `[Pasted text #id]` placeholder with full text in place.
|
||||
|
||||
Used when the same content is pasted again: the compact placeholder is
|
||||
expanded back to the original text where it sits, preserving surrounding
|
||||
input.
|
||||
|
||||
Args:
|
||||
paste_id: The paste id whose placeholder should be expanded.
|
||||
content: The full text to insert where the placeholder was.
|
||||
|
||||
Returns:
|
||||
`True` when a matching placeholder was found and replaced.
|
||||
"""
|
||||
for match in PASTE_PLACEHOLDER_PATTERN.finditer(self.text):
|
||||
if int(match.group(1)) != paste_id:
|
||||
continue
|
||||
start, end = match.span()
|
||||
start_location = self.document.get_location_from_index(start) # ty: ignore[unresolved-attribute] # Document has this method; DocumentBase stub is narrower
|
||||
end_location = self.document.get_location_from_index(end) # ty: ignore[unresolved-attribute]
|
||||
self.delete(start_location, end_location)
|
||||
self.insert(content, start_location)
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _on_paste(self, event: events.Paste) -> None:
|
||||
"""Handle paste events and detect dragged file paths."""
|
||||
"""Handle paste events, detecting file paths and large pastes."""
|
||||
self._backslash_pending_time = None
|
||||
if self._paste_burst_buffer:
|
||||
await self._flush_paste_burst()
|
||||
@@ -1294,17 +1373,32 @@ class ChatTextArea(TextArea):
|
||||
|
||||
try:
|
||||
parsed = await asyncio.to_thread(parse_pasted_path_payload, event.text)
|
||||
except Exception: # noqa: BLE001 # Treat thread failure as non-path text
|
||||
except Exception:
|
||||
# See _flush_paste_burst: swallowing here would silently break the
|
||||
# drag-drop-file path, so log a breadcrumb and fall through to text.
|
||||
logger.debug(
|
||||
"Path-payload parsing failed; treating paste as text",
|
||||
exc_info=True,
|
||||
)
|
||||
parsed = None
|
||||
if parsed is None:
|
||||
# Don't call super() here — Textual's MRO dispatch already calls
|
||||
# TextArea._on_paste after this handler returns. Calling super()
|
||||
# would insert the text a second time, duplicating the paste.
|
||||
if parsed is not None:
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.post_message(self.PastedPaths(event.text, parsed.paths))
|
||||
return
|
||||
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.post_message(self.PastedPaths(event.text, parsed.paths))
|
||||
if _should_collapse_chat_paste(event.text):
|
||||
# Intercept the paste so Textual's default _on_paste doesn't insert
|
||||
# the full text. ChatInput stores the content and inserts a compact
|
||||
# placeholder instead.
|
||||
event.prevent_default()
|
||||
event.stop()
|
||||
self.post_message(self.PastedText(event.text))
|
||||
return
|
||||
|
||||
# Don't call super() here — Textual's MRO dispatch already calls
|
||||
# TextArea._on_paste after this handler returns. Calling super()
|
||||
# would insert the text a second time, duplicating the paste.
|
||||
|
||||
def set_text_from_history(self, text: str, *, cursor_at_end: bool = True) -> None:
|
||||
"""Set text from history navigation.
|
||||
@@ -1557,6 +1651,13 @@ class ChatInput(Vertical):
|
||||
self._completion_view: _CompletionViewAdapter | None = None
|
||||
self._slash_controller: SlashCommandController | None = None
|
||||
|
||||
# Collapsed paste storage: paste_id → full content. When a large paste
|
||||
# arrives, the full text is stored here and a compact
|
||||
# `[Pasted text #N +M lines]` placeholder is inserted into the text
|
||||
# area instead. At submission the placeholder is expanded back.
|
||||
self._pasted_contents: dict[int, PastedContent] = {}
|
||||
self._next_paste_id = 1
|
||||
|
||||
# Guard flag: set True before programmatically stripping the mode
|
||||
# prefix character so the resulting text-change event does not
|
||||
# re-evaluate mode.
|
||||
@@ -2166,23 +2267,38 @@ class ChatInput(Vertical):
|
||||
if self._completion_manager:
|
||||
self._completion_manager.reset()
|
||||
|
||||
# Expand collapsed paste placeholders back to their full content so the
|
||||
# agent receives the original text, not the compact reference.
|
||||
value = expand_paste_refs(value, self._pasted_contents)
|
||||
value = self._replace_submitted_paths_with_images(value)
|
||||
|
||||
mode = self.mode
|
||||
if mode == "normal":
|
||||
detected = detect_mode_prefix(value)
|
||||
if detected is not None:
|
||||
_, mode = detected
|
||||
|
||||
# Prepend mode prefix so the app layer receives the original trigger
|
||||
# form (e.g. "!ls", "/help"). The value may already contain the prefix
|
||||
# when a completion controller wrote it back into the text area before
|
||||
# the strip handler ran.
|
||||
prefix = MODE_PREFIXES.get(self.mode, "")
|
||||
prefix = MODE_PREFIXES.get(mode, "")
|
||||
if prefix and not value.startswith(prefix):
|
||||
value = prefix + value
|
||||
|
||||
self._history.add(value)
|
||||
self.post_message(self.Submitted(value, self.mode))
|
||||
self.post_message(self.Submitted(value, mode))
|
||||
|
||||
if self._text_area:
|
||||
# Preserve submission-time attachments until adapter consumes them.
|
||||
self._skip_media_sync_events += 1
|
||||
self._text_area.clear_text()
|
||||
# Clear only after submit. Ordinary edits are undoable, so removing
|
||||
# backing content earlier can strand a restored placeholder. The input
|
||||
# and its paste map are emptied together here, so IDs can safely restart
|
||||
# at 1 for the next message.
|
||||
self._pasted_contents.clear()
|
||||
self._next_paste_id = 1
|
||||
self.mode = "normal"
|
||||
|
||||
def _sync_media_tracker_to_text(self, text: str) -> None:
|
||||
@@ -2258,11 +2374,26 @@ class ChatInput(Vertical):
|
||||
|
||||
self._insert_pasted_paths(event.raw_text, event.paths)
|
||||
|
||||
def on_chat_text_area_pasted_text(self, event: ChatTextArea.PastedText) -> None:
|
||||
"""Handle large pastes by collapsing into a compact placeholder.
|
||||
|
||||
Stores the full text in `_pasted_contents` and inserts a
|
||||
`[Pasted text #N +M lines]` placeholder into the text area instead
|
||||
of the raw content, keeping the input box compact.
|
||||
|
||||
Args:
|
||||
event: The `PastedText` message carrying the full pasted text.
|
||||
"""
|
||||
if not self._text_area:
|
||||
return
|
||||
self._collapse_and_insert_paste(event.text)
|
||||
|
||||
def handle_external_paste(self, pasted: str) -> bool:
|
||||
"""Handle paste text from app-level routing when input is not focused.
|
||||
|
||||
When the text area is mounted, the paste is always consumed: file paths
|
||||
are attached as images, and plain text is inserted directly.
|
||||
are attached as images, large text is collapsed into a placeholder,
|
||||
and remaining plain text is inserted directly.
|
||||
|
||||
Args:
|
||||
pasted: Raw pasted text payload.
|
||||
@@ -2275,14 +2406,52 @@ class ChatInput(Vertical):
|
||||
return False
|
||||
|
||||
parsed = self._parse_dropped_path_payload(pasted)
|
||||
if parsed is None:
|
||||
self._text_area.insert(pasted)
|
||||
else:
|
||||
if parsed is not None:
|
||||
self._insert_pasted_paths(pasted, parsed.paths)
|
||||
elif _should_collapse_chat_paste(pasted):
|
||||
self._collapse_and_insert_paste(pasted)
|
||||
else:
|
||||
self._text_area.insert(pasted)
|
||||
|
||||
self._text_area.focus()
|
||||
return True
|
||||
|
||||
def _collapse_and_insert_paste(self, text: str) -> None:
|
||||
"""Store full paste content and insert a compact placeholder.
|
||||
|
||||
Pasting content identical to a visible already-collapsed placeholder
|
||||
expands that placeholder back to the full text in place instead of
|
||||
adding a second placeholder — a repeat paste is treated as a request to
|
||||
see the content in full.
|
||||
|
||||
Args:
|
||||
text: The full pasted text to collapse.
|
||||
"""
|
||||
if not self._text_area:
|
||||
logger.debug("Dropping collapsed paste: text area not mounted")
|
||||
return
|
||||
visible_ids = {
|
||||
int(match.group(1))
|
||||
for match in PASTE_PLACEHOLDER_PATTERN.finditer(self._text_area.text)
|
||||
}
|
||||
match_id = next(
|
||||
(
|
||||
pid
|
||||
for pid, stored in self._pasted_contents.items()
|
||||
if pid in visible_ids and stored.content == text
|
||||
),
|
||||
None,
|
||||
)
|
||||
if match_id is not None and self._text_area.replace_placeholder_with_text(
|
||||
match_id, text
|
||||
):
|
||||
return
|
||||
paste_id = self._next_paste_id
|
||||
self._next_paste_id += 1
|
||||
self._pasted_contents[paste_id] = PastedContent(content=text)
|
||||
placeholder = format_paste_ref(paste_id, count_lines(text))
|
||||
self._text_area.insert(placeholder)
|
||||
|
||||
def _apply_inline_dropped_path_replacement(self, text: str) -> bool:
|
||||
"""Replace full dropped-path payload text with image placeholders.
|
||||
|
||||
@@ -2663,7 +2832,7 @@ class ChatInput(Vertical):
|
||||
"""Copy the current draft to the clipboard from the `[ COPY ]` button."""
|
||||
from deepagents_code.clipboard import copy_text_with_feedback
|
||||
|
||||
text = self.value
|
||||
text = expand_paste_refs(self.value, self._pasted_contents)
|
||||
if text:
|
||||
copy_text_with_feedback(
|
||||
self.app,
|
||||
|
||||
@@ -110,6 +110,13 @@ _DEFAULT_TODO_WRAP_WIDTH = 80
|
||||
_TODO_WRAP_GUARD_COLUMNS = 4
|
||||
_MAX_WEB_CONTENT_LEN = 100
|
||||
|
||||
# User message display truncation — when content exceeds this many characters,
|
||||
# only the head and tail are rendered with an elision marker in between.
|
||||
# This keeps very large pastes from flooding the conversation scrollback.
|
||||
_USER_MSG_MAX_DISPLAY_CHARS = 10_000
|
||||
_USER_MSG_TRUNCATE_HEAD_CHARS = 2_500
|
||||
_USER_MSG_TRUNCATE_TAIL_CHARS = 2_500
|
||||
|
||||
# Tools that have their key info already in the header (no need for args line)
|
||||
_TOOLS_WITH_HEADER_INFO: set[str] = {
|
||||
# Filesystem tools
|
||||
@@ -205,6 +212,30 @@ def _select_prompt_body(widget: Static) -> None:
|
||||
}
|
||||
|
||||
|
||||
def _truncate_for_display(text: str) -> str:
|
||||
"""Truncate very long user message text for display in the conversation.
|
||||
|
||||
Keeps the first and last portions and replaces the middle with an elision
|
||||
marker showing the number of newlines in the hidden region. This mirrors
|
||||
Claude Code's `UserPromptMessage` head+tail truncation for rendering
|
||||
performance.
|
||||
|
||||
Args:
|
||||
text: Full message content.
|
||||
|
||||
Returns:
|
||||
Truncated text with an elision marker, or the original text when
|
||||
it does not exceed the display threshold.
|
||||
"""
|
||||
if len(text) <= _USER_MSG_MAX_DISPLAY_CHARS:
|
||||
return text
|
||||
head = text[:_USER_MSG_TRUNCATE_HEAD_CHARS]
|
||||
tail = text[-_USER_MSG_TRUNCATE_TAIL_CHARS:]
|
||||
hidden_text = text[_USER_MSG_TRUNCATE_HEAD_CHARS:-_USER_MSG_TRUNCATE_TAIL_CHARS]
|
||||
hidden_lines = hidden_text.count("\n")
|
||||
return f"{head}\n… +{hidden_lines} lines …\n{tail}"
|
||||
|
||||
|
||||
class UserMessage(Static):
|
||||
"""Widget displaying a user message."""
|
||||
|
||||
@@ -239,7 +270,13 @@ class UserMessage(Static):
|
||||
self.add_class("-cancelled")
|
||||
|
||||
def get_selection(self, selection: Selection) -> tuple[str, str] | None:
|
||||
"""Exclude the prompt prefix glyph from copied text.
|
||||
"""Return selected text, preferring the full content over the render.
|
||||
|
||||
`render()` truncates long messages, so for a full-message selection
|
||||
(select-all / select-to-end, where `selection.end` is `None`) the text
|
||||
is extracted from the untruncated content so copy yields the complete
|
||||
original. A partial selection is extracted from the base (on-screen)
|
||||
render so its offsets stay aligned with what the user highlighted.
|
||||
|
||||
Args:
|
||||
selection: The active selection geometry.
|
||||
@@ -247,7 +284,38 @@ class UserMessage(Static):
|
||||
Returns:
|
||||
The `(text, ending)` selection with the prefix removed, or `None`.
|
||||
"""
|
||||
return _strip_prompt_prefix(super().get_selection(selection), selection)
|
||||
if selection.end is not None:
|
||||
return _strip_prompt_prefix(super().get_selection(selection), selection)
|
||||
text = str(self._build_full_render())
|
||||
return _strip_prompt_prefix((selection.extract(text), "\n"), selection)
|
||||
|
||||
def _prefix_and_body(self) -> tuple[tuple[str, str], str]:
|
||||
"""Compute the styled mode prefix and the body with its trigger stripped.
|
||||
|
||||
Returns:
|
||||
A `(prefix, body)` pair where `prefix` is a `(text, style)` tuple and
|
||||
`body` is the content with any mode-trigger prefix removed.
|
||||
"""
|
||||
colors = theme.get_theme_colors(self)
|
||||
content = self._content
|
||||
mode_match = detect_mode_prefix(content)
|
||||
if mode_match:
|
||||
prefix_text, mode = mode_match
|
||||
glyph = MODE_DISPLAY_GLYPHS.get(mode, prefix_text[0])
|
||||
return (
|
||||
(f"{glyph} ", f"bold {_mode_color(mode, self)}"),
|
||||
content[len(prefix_text) :],
|
||||
)
|
||||
return ("> ", f"bold {colors.primary}"), content
|
||||
|
||||
def _build_full_render(self) -> Content:
|
||||
"""Build a Content from the full content without display truncation.
|
||||
|
||||
Returns:
|
||||
Content with the mode prefix glyph and the full message body.
|
||||
"""
|
||||
prefix, body = self._prefix_and_body()
|
||||
return Content.assemble(prefix, body)
|
||||
|
||||
def text_select_all(self) -> None:
|
||||
"""Select the message body without the prompt prefix glyph."""
|
||||
@@ -269,20 +337,16 @@ class UserMessage(Static):
|
||||
Styled Content with mode prefix and highlighted mentions.
|
||||
"""
|
||||
colors = theme.get_theme_colors(self)
|
||||
parts: list[str | tuple[str, str]] = []
|
||||
content = self._content
|
||||
|
||||
# Use mode-specific prefix indicator when content starts with a
|
||||
# mode trigger character (e.g. "!" for shell, "/" for commands).
|
||||
# The display glyph may differ from the trigger (e.g. "$" for shell).
|
||||
mode_match = detect_mode_prefix(content)
|
||||
if mode_match:
|
||||
prefix_text, mode = mode_match
|
||||
glyph = MODE_DISPLAY_GLYPHS.get(mode, prefix_text[0])
|
||||
parts.append((f"{glyph} ", f"bold {_mode_color(mode, self)}"))
|
||||
content = content[len(prefix_text) :]
|
||||
else:
|
||||
parts.append(("> ", f"bold {colors.primary}"))
|
||||
prefix, content = self._prefix_and_body()
|
||||
parts: list[str | tuple[str, str]] = [prefix]
|
||||
|
||||
# Truncate very long content for display so large pastes don't flood
|
||||
# the conversation. The full text is still available for copy/select.
|
||||
content = _truncate_for_display(content)
|
||||
|
||||
# Highlight @mentions and /commands in the content
|
||||
last_end = 0
|
||||
@@ -351,7 +415,11 @@ class QueuedUserMessage(Static):
|
||||
self.add_class("-ascii")
|
||||
|
||||
def get_selection(self, selection: Selection) -> tuple[str, str] | None:
|
||||
"""Exclude the prompt prefix glyph from copied text.
|
||||
"""Return selected text, preferring the full content over the render.
|
||||
|
||||
See `UserMessage.get_selection`: full-message selections extract from
|
||||
the untruncated content, partial selections defer to the on-screen
|
||||
render so offsets stay aligned.
|
||||
|
||||
Args:
|
||||
selection: The active selection geometry.
|
||||
@@ -359,7 +427,34 @@ class QueuedUserMessage(Static):
|
||||
Returns:
|
||||
The `(text, ending)` selection with the prefix removed, or `None`.
|
||||
"""
|
||||
return _strip_prompt_prefix(super().get_selection(selection), selection)
|
||||
if selection.end is not None:
|
||||
return _strip_prompt_prefix(super().get_selection(selection), selection)
|
||||
text = str(self._build_full_render())
|
||||
return _strip_prompt_prefix((selection.extract(text), "\n"), selection)
|
||||
|
||||
def _prefix_and_body(self) -> tuple[tuple[str, str], str]:
|
||||
"""Compute the muted mode prefix and body with its trigger stripped.
|
||||
|
||||
Returns:
|
||||
A `(prefix, body)` pair where `prefix` is a `(text, style)` tuple.
|
||||
"""
|
||||
colors = theme.get_theme_colors(self)
|
||||
content = self._content
|
||||
mode_match = detect_mode_prefix(content)
|
||||
if mode_match:
|
||||
prefix_text, mode = mode_match
|
||||
glyph = MODE_DISPLAY_GLYPHS.get(mode, prefix_text[0])
|
||||
return (f"{glyph} ", f"bold {colors.muted}"), content[len(prefix_text) :]
|
||||
return ("> ", f"bold {colors.muted}"), content
|
||||
|
||||
def _build_full_render(self) -> Content:
|
||||
"""Build a Content from the full content without display truncation.
|
||||
|
||||
Returns:
|
||||
Content with the mode prefix glyph and the full message body.
|
||||
"""
|
||||
prefix, body = self._prefix_and_body()
|
||||
return Content.assemble(prefix, body)
|
||||
|
||||
def text_select_all(self) -> None:
|
||||
"""Select the message body without the prompt prefix glyph."""
|
||||
@@ -372,15 +467,8 @@ class QueuedUserMessage(Static):
|
||||
Styled Content with dimmed prefix and body.
|
||||
"""
|
||||
colors = theme.get_theme_colors(self)
|
||||
content = self._content
|
||||
mode_match = detect_mode_prefix(content)
|
||||
if mode_match:
|
||||
prefix_text, mode = mode_match
|
||||
glyph = MODE_DISPLAY_GLYPHS.get(mode, prefix_text[0])
|
||||
prefix = (f"{glyph} ", f"bold {colors.muted}")
|
||||
content = content[len(prefix_text) :]
|
||||
else:
|
||||
prefix = ("> ", f"bold {colors.muted}")
|
||||
prefix, content = self._prefix_and_body()
|
||||
content = _truncate_for_display(content)
|
||||
return Content.assemble(prefix, (content, colors.muted))
|
||||
|
||||
|
||||
|
||||
@@ -4259,3 +4259,588 @@ class TestPasteBurstEnterSuppression:
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
|
||||
|
||||
class TestPasteCollapseHelpers:
|
||||
"""Unit tests for the paste_collapse module helpers."""
|
||||
|
||||
def test_should_collapse_short_text(self) -> None:
|
||||
"""Short single-line text should not be collapsed."""
|
||||
from deepagents_code.paste_collapse import should_collapse_paste
|
||||
|
||||
assert should_collapse_paste("hello") is False
|
||||
|
||||
def test_should_collapse_long_text(self) -> None:
|
||||
"""Text exceeding the character threshold should be collapsed."""
|
||||
from deepagents_code.paste_collapse import should_collapse_paste
|
||||
|
||||
assert should_collapse_paste("x" * 801) is True
|
||||
|
||||
def test_should_not_collapse_at_char_boundary(self) -> None:
|
||||
"""Text exactly at the character threshold should not be collapsed."""
|
||||
from deepagents_code.paste_collapse import should_collapse_paste
|
||||
|
||||
assert should_collapse_paste("x" * 800) is False
|
||||
|
||||
def test_should_collapse_multi_line(self) -> None:
|
||||
"""Text with more lines than the threshold should be collapsed."""
|
||||
from deepagents_code.paste_collapse import should_collapse_paste
|
||||
|
||||
assert should_collapse_paste("line1\nline2\nline3\nline4") is True
|
||||
|
||||
def test_should_not_collapse_two_lines(self) -> None:
|
||||
"""Exactly two newlines (three lines) should not be collapsed by line count."""
|
||||
from deepagents_code.paste_collapse import should_collapse_paste
|
||||
|
||||
assert should_collapse_paste("line1\nline2\nline3") is False
|
||||
|
||||
def test_format_paste_ref_single_line(self) -> None:
|
||||
"""Single-line paste ref has no line count suffix."""
|
||||
from deepagents_code.paste_collapse import format_paste_ref
|
||||
|
||||
assert format_paste_ref(1, 0) == "[Pasted text #1]"
|
||||
|
||||
def test_format_paste_ref_multi_line(self) -> None:
|
||||
"""Multi-line paste ref includes the line count."""
|
||||
from deepagents_code.paste_collapse import format_paste_ref
|
||||
|
||||
assert format_paste_ref(3, 5) == "[Pasted text #3 +5 lines]"
|
||||
|
||||
def test_expand_paste_refs(self) -> None:
|
||||
"""expand_paste_refs replaces placeholders with stored content."""
|
||||
from deepagents_code.paste_collapse import PastedContent, expand_paste_refs
|
||||
|
||||
contents = {
|
||||
1: PastedContent(content="FIRST\nSECOND"),
|
||||
2: PastedContent(content="third"),
|
||||
}
|
||||
text = "before [Pasted text #1 +1 lines] after [Pasted text #2] end"
|
||||
expanded = expand_paste_refs(text, contents)
|
||||
assert expanded == "before FIRST\nSECOND after third end"
|
||||
|
||||
def test_expand_paste_refs_out_of_order_and_repeated(self) -> None:
|
||||
"""Placeholders expand by id regardless of order, and repeats reuse content."""
|
||||
from deepagents_code.paste_collapse import PastedContent, expand_paste_refs
|
||||
|
||||
contents = {
|
||||
1: PastedContent(content="ONE"),
|
||||
2: PastedContent(content="TWO"),
|
||||
}
|
||||
text = "[Pasted text #2] [Pasted text #1] [Pasted text #1]"
|
||||
assert expand_paste_refs(text, contents) == "TWO ONE ONE"
|
||||
|
||||
def test_expand_paste_refs_content_with_backslashes(self) -> None:
|
||||
"""Stored content is inserted literally, not as a regex replacement."""
|
||||
from deepagents_code.paste_collapse import PastedContent, expand_paste_refs
|
||||
|
||||
contents = {1: PastedContent(content=r"\1 \g<0> back\slash")}
|
||||
expanded = expand_paste_refs("[Pasted text #1]", contents)
|
||||
assert expanded == r"\1 \g<0> back\slash"
|
||||
|
||||
def test_expand_paste_refs_unknown_id_left_as_is(self) -> None:
|
||||
"""Placeholders with unknown IDs are left unchanged."""
|
||||
from deepagents_code.paste_collapse import expand_paste_refs
|
||||
|
||||
text = "[Pasted text #99 +5 lines]"
|
||||
assert expand_paste_refs(text, {}) == text
|
||||
|
||||
def test_expand_paste_refs_empty_contents(self) -> None:
|
||||
"""Expanding with no stored contents returns text unchanged."""
|
||||
from deepagents_code.paste_collapse import expand_paste_refs
|
||||
|
||||
text = "hello [Pasted text #1] world"
|
||||
assert expand_paste_refs(text, {}) == text
|
||||
|
||||
|
||||
class TestPasteCollapseIntegration:
|
||||
"""Integration tests for paste collapsing in ChatInput."""
|
||||
|
||||
async def test_large_paste_inserts_placeholder(self) -> None:
|
||||
"""Pasting text > 800 chars inserts a placeholder, not the full text."""
|
||||
big_text = "x" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
|
||||
assert "[Pasted text #1]" in chat._text_area.text
|
||||
assert "x" * 900 not in chat._text_area.text
|
||||
assert 1 in chat._pasted_contents
|
||||
assert chat._pasted_contents[1].content == big_text
|
||||
|
||||
async def test_small_paste_inserts_directly(self) -> None:
|
||||
"""Pasting text under the threshold is inserted as-is."""
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste("short text")
|
||||
await pilot.pause()
|
||||
|
||||
assert chat._text_area.text == "short text"
|
||||
assert len(chat._pasted_contents) == 0
|
||||
|
||||
async def test_multi_line_paste_inserts_placeholder(self) -> None:
|
||||
"""Pasting text with > 2 newlines inserts a placeholder."""
|
||||
multi_line = "\n".join(f"line {i}" for i in range(5))
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(multi_line)
|
||||
await pilot.pause()
|
||||
|
||||
assert "[Pasted text #1 +4 lines]" in chat._text_area.text
|
||||
assert "line 0" not in chat._text_area.text
|
||||
|
||||
async def test_submit_expands_placeholder(self) -> None:
|
||||
"""Submitting text with a placeholder sends the full expanded content."""
|
||||
big_text = "A" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
assert app.submitted[0].value == big_text
|
||||
|
||||
async def test_submit_clears_paste_contents(self) -> None:
|
||||
"""Paste contents are cleared after submission."""
|
||||
big_text = "B" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
assert len(chat._pasted_contents) == 1
|
||||
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(chat._pasted_contents) == 0
|
||||
|
||||
async def test_multiple_pastes_get_incrementing_ids(self) -> None:
|
||||
"""Consecutive large pastes get incrementing placeholder IDs."""
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste("X" * 900)
|
||||
await pilot.pause()
|
||||
chat.handle_external_paste("Y" * 900)
|
||||
await pilot.pause()
|
||||
|
||||
assert "[Pasted text #1]" in chat._text_area.text
|
||||
assert "[Pasted text #2]" in chat._text_area.text
|
||||
assert chat._pasted_contents[1].content == "X" * 900
|
||||
assert chat._pasted_contents[2].content == "Y" * 900
|
||||
|
||||
async def test_multiple_pastes_submit_expands_all(self) -> None:
|
||||
"""Submitting with multiple placeholders expands all of them."""
|
||||
text_a = "A" * 900
|
||||
text_b = "B" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(text_a)
|
||||
await pilot.pause()
|
||||
chat.handle_external_paste(text_b)
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
assert app.submitted[0].value == text_a + text_b
|
||||
|
||||
async def test_text_around_placeholder_preserved_on_submit(self) -> None:
|
||||
"""Text typed around a placeholder is preserved on submission."""
|
||||
big_text = "C" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
for char in "fix this: ":
|
||||
await pilot.press(char)
|
||||
await pilot.pause()
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
assert app.submitted[0].value == f"fix this: {big_text}"
|
||||
|
||||
async def test_paste_content_survives_undo_restored_placeholder(self) -> None:
|
||||
"""A restored placeholder still expands after an undoable edit."""
|
||||
big_text = "D" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
placeholder = chat._text_area.text
|
||||
assert 1 in chat._pasted_contents
|
||||
|
||||
chat._text_area.text = "all gone"
|
||||
await pilot.pause()
|
||||
|
||||
chat._text_area.text = f"fix: {placeholder}"
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
assert app.submitted[0].value == f"fix: {big_text}"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("pasted", "mode", "visible"),
|
||||
[
|
||||
("/help " + ("x" * 900), "command", "help " + ("x" * 900)),
|
||||
("!echo " + ("x" * 900), "shell", "echo " + ("x" * 900)),
|
||||
("!!echo " + ("x" * 900), "shell_incognito", "echo " + ("x" * 900)),
|
||||
],
|
||||
)
|
||||
async def test_prefixed_large_paste_renders_mode_instead_of_collapsing(
|
||||
self, pasted: str, mode: str, visible: str
|
||||
) -> None:
|
||||
"""Mode-prefixed large pastes stay visible so the prompt shows mode."""
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(pasted)
|
||||
await pilot.pause()
|
||||
|
||||
assert chat.mode == mode
|
||||
assert chat._text_area.text == visible
|
||||
assert chat._pasted_contents == {}
|
||||
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
assert app.submitted[0].value == pasted
|
||||
assert app.submitted[0].mode == mode
|
||||
|
||||
async def test_identical_second_paste_expands_placeholder(self) -> None:
|
||||
"""Pasting identical content again expands the placeholder to full text."""
|
||||
text = "S" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
assert chat._text_area.text == "[Pasted text #1]"
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
|
||||
# Second identical paste expands the placeholder inline. The stored
|
||||
# copy stays available because this edit can be undone.
|
||||
assert chat._text_area.text == text
|
||||
assert chat._pasted_contents[1].content == text
|
||||
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.submitted[0].value == text
|
||||
|
||||
async def test_repeat_paste_expansion_keeps_content_for_undo(self) -> None:
|
||||
"""Undo-restored placeholders still expand after repeat-paste expansion."""
|
||||
text = "U" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
placeholder = chat._text_area.text
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
assert chat._text_area.text == text
|
||||
|
||||
chat._text_area.text = placeholder
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
|
||||
assert len(app.submitted) == 1
|
||||
assert app.submitted[0].value == text
|
||||
|
||||
async def test_identical_second_paste_expands_preserving_surrounding_text(
|
||||
self,
|
||||
) -> None:
|
||||
"""Expanding a repeated paste keeps text typed around the placeholder."""
|
||||
text = "S" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
for char in "fix: ":
|
||||
await pilot.press(char)
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
assert chat._text_area.text == "fix: [Pasted text #1]"
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
|
||||
assert chat._text_area.text == f"fix: {text}"
|
||||
assert chat._pasted_contents[1].content == text
|
||||
|
||||
async def test_repeat_paste_skips_stale_deleted_placeholder_ids(self) -> None:
|
||||
"""Repeat expansion targets a visible placeholder, not stale paste data."""
|
||||
text = "R" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
assert chat._text_area.text == "[Pasted text #1]"
|
||||
|
||||
await pilot.press("backspace")
|
||||
await pilot.pause()
|
||||
assert chat._text_area.text == ""
|
||||
assert chat._pasted_contents[1].content == text
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
assert chat._text_area.text == "[Pasted text #2]"
|
||||
|
||||
chat.handle_external_paste(text)
|
||||
await pilot.pause()
|
||||
|
||||
assert chat._text_area.text == text
|
||||
assert chat._pasted_contents[1].content == text
|
||||
assert chat._pasted_contents[2].content == text
|
||||
|
||||
async def test_bracketed_paste_event_collapses(self) -> None:
|
||||
"""A real Paste event over the threshold collapses to a placeholder.
|
||||
|
||||
Exercises the production path (`_on_paste` -> `PastedText` message ->
|
||||
`on_chat_text_area_pasted_text`) rather than `handle_external_paste`.
|
||||
"""
|
||||
big_text = "z" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
await chat._text_area._on_paste(events.Paste(big_text))
|
||||
await pilot.pause()
|
||||
|
||||
assert "[Pasted text #1]" in chat._text_area.text
|
||||
assert big_text not in chat._text_area.text
|
||||
assert chat._pasted_contents[1].content == big_text
|
||||
|
||||
async def test_paste_burst_flush_collapses_large_payload(self) -> None:
|
||||
"""A large buffered paste burst collapses to a placeholder on flush."""
|
||||
big_text = "q" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat._text_area._paste_burst_buffer = big_text
|
||||
await chat._text_area._flush_paste_burst()
|
||||
await pilot.pause()
|
||||
|
||||
assert "[Pasted text #1]" in chat._text_area.text
|
||||
assert big_text not in chat._text_area.text
|
||||
assert chat._pasted_contents[1].content == big_text
|
||||
|
||||
async def test_backspace_removes_full_paste_placeholder(self) -> None:
|
||||
"""Backspace deletes a [Pasted text #N] placeholder as a single token."""
|
||||
big_text = "p" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
assert "[Pasted text #1]" in chat._text_area.text
|
||||
|
||||
await pilot.press("backspace")
|
||||
await pilot.pause()
|
||||
|
||||
assert chat._text_area.text == ""
|
||||
# Backing content is preserved so an undo can restore the token;
|
||||
# it is cleared only at submit.
|
||||
assert 1 in chat._pasted_contents
|
||||
|
||||
async def test_forward_delete_removes_full_paste_placeholder(self) -> None:
|
||||
"""Forward delete removes a paste placeholder as a single token."""
|
||||
big_text = "p" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
chat._text_area.move_cursor((0, 0))
|
||||
|
||||
await pilot.press("delete")
|
||||
await pilot.pause()
|
||||
|
||||
assert chat._text_area.text == ""
|
||||
# Like backspace, forward delete leaves the backing content for undo;
|
||||
# it is cleared only at submit.
|
||||
assert 1 in chat._pasted_contents
|
||||
|
||||
async def test_backspace_removes_only_targeted_paste_placeholder(self) -> None:
|
||||
"""Backspace deletes only the placeholder at the cursor, not others."""
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste("A" * 900)
|
||||
await pilot.pause()
|
||||
chat.handle_external_paste("B" * 900)
|
||||
await pilot.pause()
|
||||
assert "[Pasted text #1]" in chat._text_area.text
|
||||
assert "[Pasted text #2]" in chat._text_area.text
|
||||
|
||||
await pilot.press("backspace")
|
||||
await pilot.pause()
|
||||
|
||||
# Exact equality (not a substring check): a non-atomic delete that
|
||||
# removed a single char would leave "[Pasted text #2" and still
|
||||
# satisfy a `"[Pasted text #2]" not in text` assertion.
|
||||
assert chat._text_area.text == "[Pasted text #1]"
|
||||
|
||||
async def test_backspace_removes_multiline_paste_placeholder(self) -> None:
|
||||
"""Backspace atomically deletes the `+M lines` placeholder variant."""
|
||||
multi_line = "\n".join(f"line {i}" for i in range(5))
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(multi_line)
|
||||
await pilot.pause()
|
||||
# The multi-line form carries a "+M lines" suffix, so its span is
|
||||
# longer than the bare "[Pasted text #N]" token.
|
||||
assert chat._text_area.text == "[Pasted text #1 +4 lines]"
|
||||
|
||||
await pilot.press("backspace")
|
||||
await pilot.pause()
|
||||
|
||||
assert chat._text_area.text == ""
|
||||
assert 1 in chat._pasted_contents
|
||||
|
||||
async def test_identical_multiline_repaste_expands_placeholder(self) -> None:
|
||||
"""Repeat-pasting multi-line content expands its `+M lines` placeholder."""
|
||||
multi_line = "\n".join(f"line {i}" for i in range(5))
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(multi_line)
|
||||
await pilot.pause()
|
||||
assert chat._text_area.text == "[Pasted text #1 +4 lines]"
|
||||
|
||||
chat.handle_external_paste(multi_line)
|
||||
await pilot.pause()
|
||||
|
||||
assert chat._text_area.text == multi_line
|
||||
assert chat._pasted_contents[1].content == multi_line
|
||||
|
||||
async def test_copy_button_expands_placeholders(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""The copy button copies expanded text, not the placeholder."""
|
||||
big_text = "E" * 900
|
||||
copied: list[str] = []
|
||||
|
||||
def capture_copy(_app_arg: object, text: str, **_kwargs: object) -> None:
|
||||
copied.append(text)
|
||||
|
||||
# _copy_via_button imports copy_text_with_feedback at call time, so
|
||||
# patching the module attribute is picked up without a manual restore.
|
||||
monkeypatch.setattr(
|
||||
"deepagents_code.clipboard.copy_text_with_feedback", capture_copy
|
||||
)
|
||||
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
|
||||
chat._copy_via_button()
|
||||
await pilot.pause()
|
||||
|
||||
assert len(copied) == 1
|
||||
assert copied[0] == big_text
|
||||
|
||||
async def test_paste_content_survives_undoable_clear(self) -> None:
|
||||
"""Undoable clear (discard_text) must not delete paste contents.
|
||||
|
||||
After clearing and undoing, the restored placeholder must still
|
||||
have its backing content so submission expands correctly.
|
||||
"""
|
||||
big_text = "F" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
assert 1 in chat._pasted_contents
|
||||
|
||||
# Undoable clear empties the text area
|
||||
chat.discard_text()
|
||||
await pilot.pause()
|
||||
|
||||
# Paste contents must survive so undo can restore them
|
||||
assert 1 in chat._pasted_contents
|
||||
|
||||
async def test_orphan_cleanup_skips_empty_text(self) -> None:
|
||||
"""Setting text to empty must not trigger orphan cleanup."""
|
||||
big_text = "G" * 900
|
||||
app = _RecordingApp()
|
||||
async with app.run_test() as pilot:
|
||||
chat = app.query_one(ChatInput)
|
||||
assert chat._text_area is not None
|
||||
|
||||
chat.handle_external_paste(big_text)
|
||||
await pilot.pause()
|
||||
|
||||
chat._text_area.text = ""
|
||||
await pilot.pause()
|
||||
|
||||
assert 1 in chat._pasted_contents
|
||||
|
||||
@@ -3700,3 +3700,139 @@ class TestLiveToolGroupSummary:
|
||||
assert not pilot.app.query(ToolGroupSummary)
|
||||
assert t1.display is True
|
||||
assert t2.display is True
|
||||
|
||||
|
||||
class TestUserMessageTruncation:
|
||||
"""Test head+tail truncation of very long user messages at render time."""
|
||||
|
||||
def test_short_message_not_truncated(self) -> None:
|
||||
"""Messages under the threshold should render in full."""
|
||||
from deepagents_code.widgets.messages import _truncate_for_display
|
||||
|
||||
text = "short message"
|
||||
assert _truncate_for_display(text) == text
|
||||
|
||||
def test_long_message_truncated_with_elision(self) -> None:
|
||||
"""Messages over 10k chars should get head+tail+elision marker."""
|
||||
from deepagents_code.widgets.messages import _truncate_for_display
|
||||
|
||||
text = "A" * 12_000
|
||||
result = _truncate_for_display(text)
|
||||
assert "… +" in result
|
||||
assert " lines …" in result
|
||||
# Head and tail are preserved
|
||||
assert result.startswith("A" * 2500)
|
||||
assert result.endswith("A" * 2500)
|
||||
|
||||
def test_truncation_counts_hidden_lines(self) -> None:
|
||||
"""The elision marker should report the correct hidden line count."""
|
||||
from deepagents_code.widgets.messages import _truncate_for_display
|
||||
|
||||
lines = [f"line {i:04d} " + "x" * 20 for i in range(600)]
|
||||
text = "\n".join(lines)
|
||||
assert len(text) > 10_000
|
||||
result = _truncate_for_display(text)
|
||||
assert "… +" in result
|
||||
assert " lines …" in result
|
||||
|
||||
def test_full_content_preserved_in_widget(self) -> None:
|
||||
"""The widget should store full content even when display is truncated."""
|
||||
big = "B" * 12_000
|
||||
msg = UserMessage(big)
|
||||
assert msg._content == big
|
||||
assert len(msg._content) == 12_000
|
||||
|
||||
def test_message_at_boundary_not_truncated(self) -> None:
|
||||
"""Messages exactly at the threshold should not be truncated."""
|
||||
from deepagents_code.widgets.messages import _truncate_for_display
|
||||
|
||||
text = "C" * 10_000
|
||||
assert _truncate_for_display(text) == text
|
||||
|
||||
async def test_selection_returns_full_content_not_truncated(self) -> None:
|
||||
"""Selecting a truncated message should return the full original text."""
|
||||
from textual.geometry import Offset
|
||||
from textual.selection import Selection
|
||||
|
||||
big = "X" * 12_000
|
||||
msg = UserMessage(big)
|
||||
|
||||
class _TestApp(App[None]):
|
||||
def compose(self) -> ComposeResult:
|
||||
yield msg
|
||||
|
||||
app = _TestApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
# Select the entire message body (skip prefix glyph)
|
||||
selection = Selection(Offset(2, 0), None)
|
||||
result = msg.get_selection(selection)
|
||||
assert result is not None
|
||||
text, _ending = result
|
||||
assert text == big
|
||||
assert "…" not in text
|
||||
|
||||
def test_truncation_reports_exact_hidden_newline_count(self) -> None:
|
||||
"""The elision marker reports the exact number of hidden newlines."""
|
||||
from deepagents_code.widgets.messages import _truncate_for_display
|
||||
|
||||
text = "H" * 6000 + "\n" * 50 + "T" * 6000
|
||||
result = _truncate_for_display(text)
|
||||
assert "… +50 lines …" in result
|
||||
|
||||
async def test_partial_selection_uses_visible_render(self) -> None:
|
||||
"""A partial selection defers to the on-screen (truncated) render.
|
||||
|
||||
Select-all extracts from the full content, but a partial selection must
|
||||
stay aligned with what is visible, so it delegates to the base widget.
|
||||
"""
|
||||
from textual.geometry import Offset
|
||||
from textual.selection import Selection
|
||||
from textual.widgets import Static
|
||||
|
||||
big = "X" * 12_000
|
||||
msg = UserMessage(big)
|
||||
|
||||
class _TestApp(App[None]):
|
||||
def compose(self) -> ComposeResult:
|
||||
yield msg
|
||||
|
||||
app = _TestApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
partial = Selection(Offset(2, 0), Offset(6, 0))
|
||||
expected = _strip_prompt_prefix(Static.get_selection(msg, partial), partial)
|
||||
result = msg.get_selection(partial)
|
||||
# Delegates to the base (truncated) extraction, so it must not
|
||||
# return the full 12k body the way full-content extraction would.
|
||||
assert result == expected
|
||||
assert result is not None
|
||||
assert result[0] != big
|
||||
|
||||
def test_queued_message_render_truncates(self) -> None:
|
||||
"""QueuedUserMessage render truncates long content with an elision marker."""
|
||||
content = _render_content(QueuedUserMessage("Q" * 12_000))
|
||||
assert content.plain.startswith("> ")
|
||||
assert "… +" in content.plain
|
||||
assert len(content.plain) < 12_000
|
||||
|
||||
async def test_queued_selection_returns_full_content(self) -> None:
|
||||
"""Select-all on a truncated QueuedUserMessage returns the full text."""
|
||||
from textual.geometry import Offset
|
||||
from textual.selection import Selection
|
||||
|
||||
big = "Y" * 12_000
|
||||
msg = QueuedUserMessage(big)
|
||||
|
||||
class _TestApp(App[None]):
|
||||
def compose(self) -> ComposeResult:
|
||||
yield msg
|
||||
|
||||
app = _TestApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
result = msg.get_selection(Selection(Offset(2, 0), None))
|
||||
assert result is not None
|
||||
text, _ending = result
|
||||
assert text == big
|
||||
assert "…" not in text
|
||||
|
||||
Reference in New Issue
Block a user