diff --git a/libs/code/deepagents_code/tui/widgets/_inline_prompt.py b/libs/code/deepagents_code/tui/widgets/_inline_prompt.py index 9e89b79863..5f0d32b7b8 100644 --- a/libs/code/deepagents_code/tui/widgets/_inline_prompt.py +++ b/libs/code/deepagents_code/tui/widgets/_inline_prompt.py @@ -163,21 +163,13 @@ async def _on_key(self, event: events.Key) -> None: now = time.monotonic() # Drive the shared paste-burst state machine so a paste replayed as rapid - # key events (no bracketed paste) stays grouped and can be collapsed. + # key events (no bracketed paste) stays grouped without delaying typing. if await self._absorb_key_into_burst(event, now): event.prevent_default() event.stop() return - if self._maybe_start_burst(event, now): - event.prevent_default() - event.stop() - return - - if self._track_burst_run(event, now): - event.prevent_default() - event.stop() - return + self._track_burst_run(event, now) if event.key == "backspace" and self._delete_placeholder_token(backwards=True): event.prevent_default() @@ -207,6 +199,10 @@ async def _on_key(self, event: events.Key) -> None: await super()._on_key(event) + # Must follow `super()._on_key`: promotion verifies the run against the + # document, so the current character has to be in it already. + self._check_burst_run_for_promotion() + async def _on_paste(self, event: events.Paste) -> None: """Reject a dragged media file, else defer to shared paste handling.""" # Flush first, matching the base handler: a rejection returns early, and diff --git a/libs/code/deepagents_code/tui/widgets/_paste_textarea.py b/libs/code/deepagents_code/tui/widgets/_paste_textarea.py index 93fc9eea4a..40eae56fa0 100644 --- a/libs/code/deepagents_code/tui/widgets/_paste_textarea.py +++ b/libs/code/deepagents_code/tui/widgets/_paste_textarea.py @@ -2,10 +2,16 @@ Terminals deliver a paste in one of two shapes: a single bracketed `Paste` event, or — when bracketed paste is unavailable — a rapid stream of individual -key events. Both the primary chat input and the inline free-text prompts need -to (a) keep a multi-line paste grouped instead of submitting on the first -embedded newline, and (b) collapse a large paste into a compact -`[Pasted text #N]` placeholder that expands back to the full text on submit. +key events. Both the primary chat input and the inline free-text prompts keep a +multi-line paste grouped instead of submitting on the first embedded newline. +Bracketed and detected key-event pastes may additionally collapse into a compact +`[Pasted text #N]` placeholder that expands on submit, and dropped-path payloads +(quoted, or bare paths detected by shape) are routed to path/media handling. + +Ordinary typing is never hidden: a rapid run stays in the document until +something confirms it is a paste — an embedded newline, a dropped-path shape, or +a length no human reaches at burst speed — at which point it is promoted into +the hidden buffer. `PasteBurstTextArea` owns the burst detection and Enter-suppression state machine, leaving policy (slash-command context, whether collapsing is enabled, @@ -22,8 +28,10 @@ from textual.binding import Binding from textual.widgets import TextArea +from deepagents_code.input import looks_like_dropped_payload from deepagents_code.paste_collapse import ( PASTE_PLACEHOLDER_PATTERN, + PASTE_THRESHOLD_CHARS, PastedContent, count_lines, expand_paste_refs, @@ -43,15 +51,29 @@ PASTE_BURST_FLUSH_DELAY_SECONDS = 0.08 """Idle timeout before flushing buffered burst text.""" -PASTE_BURST_START_CHARS = {"'", '"'} -"""Characters that can start dropped-path payloads.""" - PASTE_BURST_MIN_CHARS = 3 """Consecutive fast keystrokes before a stream is treated as a paste burst. Terminals that lack bracketed paste replay a paste as individual key events. Counting a short run of rapid chars distinguishes that from human typing, which has much larger inter-key gaps. + +Reaching this count does not by itself hide the run: it arms the Enter +suppression window and makes the run eligible for promotion into +`_paste_burst_buffer`. Promotion itself needs further evidence of a paste — see +`_check_burst_run_for_promotion` and `_consume_enter_as_burst_newline`. +""" + +PASTE_BURST_PROMOTE_CHARS = PASTE_THRESHOLD_CHARS +"""Rapid-run length that on its own confirms a key-event paste. + +A run this long arriving at burst speed (each char within +`PASTE_BURST_CHAR_GAP_SECONDS`) is unreachable by human typing, so it is +promoted into the buffer even without an embedded newline. Derived from the +collapse threshold so a large single-line key-event paste reaches +`[Pasted text #N]` collapsing. The comparison here is `>=` while +`should_collapse_paste` uses `>`, so a run of exactly this length is promoted and +then re-inserted verbatim. """ PASTE_ENTER_SUPPRESS_WINDOW_SECONDS = 0.12 @@ -132,13 +154,13 @@ def __init__(self, **kwargs: Any) -> None: def _init_paste_burst_state(self) -> None: """Reset all paste-burst tracking fields to their initial values.""" - # Buffer high-frequency key bursts from terminals that emulate paste via - # rapid key events instead of dispatching a paste event. + # Holds burst text only after promotion. A rapid run stays in the + # document until something confirms it is a paste. self._paste_burst_buffer = "" self._paste_burst_last_char_time = None self._paste_burst_timer = None - # Counts consecutive rapid keystrokes so a paste-like stream can be - # detected even when it doesn't begin with a quote. + # Counts consecutive rapid keystrokes so a paste replayed as key events + # can be recognized without a bracketed paste event. self._paste_burst_run = 0 self._paste_burst_run_text = "" self._paste_burst_last_key_time = None @@ -198,9 +220,29 @@ def _in_slash_command_context(self) -> bool: # noqa: PLR6301 # overridable hoo return False async def _dispatch_burst_payload(self, payload: str) -> None: - """Handle a flushed burst payload. Base behavior inserts it verbatim.""" + """Handle a flushed burst payload. Base behavior inserts it verbatim. + + Implementations must apply the payload before returning. A payload + applied later — from a posted message or a scheduled callback — is + ordered behind any key event already waiting in this widget's queue, so + the next keystroke would be inserted ahead of the paste. + """ self.insert(payload) + def _burst_run_payload_for_dispatch(self, payload: str) -> str: # noqa: PLR6301 # overridable hook + """Return the payload represented by a visible rapid-key run. + + Most text areas display every character in the run, so the payload is + unchanged. Subclasses with virtual prefixes may restore characters + that were consumed before insertion. + """ + return payload + + def _on_burst_run_promoted( + self, visible_payload: str, dispatch_payload: str + ) -> None: + """React after a visible run has been promoted into the burst buffer.""" + # -- Burst state machine -------------------------------------------------- def _cancel_paste_burst_timer(self) -> None: @@ -234,6 +276,21 @@ def _append_paste_burst(self, text: str, now: float) -> None: self._paste_burst_window_until = now + PASTE_ENTER_SUPPRESS_WINDOW_SECONDS self._schedule_paste_burst_flush() + def _append_recent_paste_burst_text(self, text: str, now: float) -> bool: + """Append text only when it continues an active rapid burst. + + Returns: + `True` when the text was appended, or `False` when there is no + active burst or it has gone idle. + """ + if not self._paste_burst_buffer: + return False + last_time = self._paste_burst_last_char_time + if last_time is None or (now - last_time) > PASTE_BURST_CHAR_GAP_SECONDS: + return False + self._append_paste_burst(text, now) + return True + def _note_paste_burst_keystroke(self, char: str, now: float) -> None: """Track text and timing for consecutive rapid keystrokes.""" last = self._paste_burst_last_key_time @@ -270,10 +327,17 @@ def _enter_inserts_newline_during_burst(self, now: float) -> bool: True when the preceding keystroke was part of a rapid run or the previous `enter` was already suppressed, and the suppression window is - still open. The char-gap check keeps a deliberate `enter` pressed after - a burst settles from being swallowed; the window bounds how long a - replayed paste's newlines stay grouped. Returns `False` immediately in - slash-command context (see `_in_slash_command_context`). + still open. The window bounds how long a replayed paste's newlines stay + grouped. Returns `False` immediately in slash-command context (see + `_in_slash_command_context`), and an active burst buffer short-circuits to + `True` regardless of the window (see below). + + The first suppressed `enter` must remain within the character gap. A + completed single-line burst followed by a deliberate `enter` is + otherwise indistinguishable from a delayed pasted newline, and submit + behavior takes priority once the rapid stream has gone idle. After one + `enter` is suppressed, the wider window keeps consecutive pasted blank + lines grouped. """ if self._in_slash_command_context(): return False @@ -290,27 +354,24 @@ def _enter_inserts_newline_during_burst(self, now: float) -> bool: if last_enter is not None: return True last_key = self._paste_burst_last_key_time - return last_key is not None and (now - last_key) <= PASTE_BURST_CHAR_GAP_SECONDS - - def _should_start_paste_burst(self, char: str) -> bool: - """Return whether a keypress should start paste-burst buffering. - - Quote-prefixed input at an empty cursor is buffered immediately for - dropped-path parsing. Other printable runs are promoted into the same - buffer once they reach `PASTE_BURST_MIN_CHARS` rapid keystrokes. - """ - if char not in PASTE_BURST_START_CHARS: + if last_key is None: return False - if self.text or not self.selection.is_empty: - return False - row, col = self.cursor_location - return row == 0 and col == 0 + return (now - last_key) <= PASTE_BURST_CHAR_GAP_SECONDS async def _flush_paste_burst(self) -> None: """Flush buffered burst text through the payload dispatch hook. When the buffer is empty this is a no-op, so it is safe to call - defensively before handling a bracketed paste. + defensively before handling a bracketed paste. The payload is applied + before this returns, so a caller may go on to handle the current key + knowing it lands after the paste. + + The buffer is cleared before dispatch, and `_promote_paste_burst_run` has + already deleted the run from the document, so a raising dispatch would + leave the text nowhere at all — not on screen, not in the buffer, not in + undo history. Media decoding, attachment tracking and notifications all + run inside that call, so the payload is re-inserted verbatim before the + error propagates. """ payload = self._paste_burst_buffer self._paste_burst_buffer = "" @@ -318,34 +379,69 @@ async def _flush_paste_burst(self) -> None: self._cancel_paste_burst_timer() if not payload: return - await self._dispatch_burst_payload(payload) + try: + await self._dispatch_burst_payload(payload) + except Exception: + logger.warning( + "Burst dispatch failed (%d chars); inserting payload verbatim", + len(payload), + exc_info=True, + ) + self.insert(payload) + raise + + def _promote_paste_burst_run(self, now: float) -> bool: + """Move an already-inserted rapid run out of the document into the buffer. - def _promote_paste_burst_run(self, char: str, now: float) -> bool: - """Move a detected rapid run from the document into the burst buffer. + Deletes the run's characters from the document — they are visible on + screen at this point — and hands them to `_start_paste_burst` so the + eventual flush can apply dropped-path and paste-collapse policy. - The first keys in an unquoted run are inserted normally while the run is - still indistinguishable from typing. Once the threshold is reached, this - removes those keys and buffers the complete run so its eventual flush can - apply dropped-path and paste-collapse policy. + No document mutation happens until every guard has passed, so a `False` + return never leaves a partially-promoted document. A failed verification + does still discard the tracked run, so `False` is not side-effect-free. Args: - char: Current character, which has not yet been inserted. now: Monotonic timestamp for the current key event. Returns: - `True` when the run was promoted and the current key was buffered. + `True` when the run was verified present immediately before the + cursor and moved into the buffer. `False` when promotion is unsafe: + an empty run, a non-empty selection (deleting would clobber the + user's selected range), or a document whose text immediately before + the cursor is no longer the tracked run — which means an intervening + edit desynchronized the tracker. Callers must fall back to handling + the key normally. """ - if not char or not self.selection.is_empty: + payload = self._paste_burst_run_text + if not payload or not self.selection.is_empty: return False - prefix = self._paste_burst_run_text[: -len(char)] cursor = self.cursor_location cursor_offset = self.document.get_index_from_location(cursor) # ty: ignore[unresolved-attribute] # Document has this method; DocumentBase stub is narrower - start_offset = cursor_offset - len(prefix) - if start_offset < 0 or self.text[start_offset:cursor_offset] != prefix: + start_offset = cursor_offset - len(payload) + if start_offset < 0 or self.text[start_offset:cursor_offset] != payload: + # An untracked edit desynchronized the tracker, so drop the stale run + # and let tracking restart on the next keystroke. Logged at warning + # (not debug) so it survives the default INFO level — this is a state + # -machine bug, not a user-reachable condition. The message carries + # only sizes, never the payload, which is user content. + logger.warning( + "Burst run diverged from document (run=%d chars, start=%d, " + "cursor=%d, doc=%d); skipping promotion", + len(payload), + start_offset, + cursor_offset, + len(self.text), + ) + self._reset_paste_burst_run() return False start = self.document.get_location_from_index(start_offset) # ty: ignore[unresolved-attribute] self.delete(start, cursor) - self._start_paste_burst(self._paste_burst_run_text, now) + self._start_paste_burst(payload, now) + # The buffer now owns these characters; clearing the run keeps the + # "run tracks visible text, buffer tracks hidden text" invariant true by + # construction, so a later Enter cannot re-promote the same stale text. + self._reset_paste_burst_run() return True def action_insert_newline(self) -> None: @@ -360,61 +456,125 @@ async def _absorb_key_into_burst(self, event: events.Key, now: float) -> bool: Returns: `True` when the key was buffered and the caller should stop handling it; `False` when there is no active burst (or it was just flushed) - and the caller should continue normal key handling. + and the caller should continue normal key handling. A flush applies + its payload before returning, so handling the key afterwards orders + it after the paste. """ if not self._paste_burst_buffer: return False if event.key == "enter": self._append_paste_burst("\n", now) return True - if event.is_printable and event.character is not None: - last_time = self._paste_burst_last_char_time - if ( - last_time is not None - and (now - last_time) <= PASTE_BURST_CHAR_GAP_SECONDS - ): - self._append_paste_burst(event.character, now) - return True - await self._flush_paste_burst() - return False - - def _maybe_start_burst(self, event: events.Key, now: float) -> bool: - """Start buffering when a keypress looks like the head of a paste. - - Returns: - `True` when a burst was started and the caller should stop handling - the key. - """ if ( event.is_printable and event.character is not None - and self._should_start_paste_burst(event.character) + and self._append_recent_paste_burst_text(event.character, now) ): - self._start_paste_burst(event.character, now) return True + await self._flush_paste_burst() return False - def _track_burst_run(self, event: events.Key, now: float) -> bool: - """Track a rapid run and promote it into the paste buffer once detected. - - Returns: - `True` when the current key was buffered and should not be handled by - the caller. - """ + def _track_burst_run(self, event: events.Key, now: float) -> None: + """Track a rapid run, arming Enter suppression once it looks like a paste.""" if event.is_printable and event.character is not None: - self._paste_burst_last_suppressed_enter_time = None - self._note_paste_burst_keystroke(event.character, now) - if ( - self._paste_burst_run >= PASTE_BURST_MIN_CHARS - and not self._in_slash_command_context() - ): - self._paste_burst_window_until = ( - now + PASTE_ENTER_SUPPRESS_WINDOW_SECONDS - ) - return self._promote_paste_burst_run(event.character, now) + self._note_printable_burst_keystroke(event.character, now) elif event.key != "enter": self._reset_paste_burst_run() - return False + + def _note_printable_burst_keystroke(self, char: str, now: float) -> None: + """Record a printable char in the rapid run and arm Enter suppression. + + Arming is conditional: the run must reach `PASTE_BURST_MIN_CHARS` and + must not be a slash command, where Enter always submits. + + Any printable char also un-latches the suppressed-Enter state, so the next + Enter must re-qualify through the character gap. + + Call this for any character that reaches the document without passing + through `_track_burst_run` — a caller that inserts text itself and + returns early must still keep the tracker in sync, or the run text will + diverge from the document and the run is discarded, losing grouping for + that stretch of the paste. + + Args: + char: The character being inserted into the document. + now: Monotonic timestamp for the current key event. + """ + self._paste_burst_last_suppressed_enter_time = None + self._note_paste_burst_keystroke(char, now) + if ( + self._paste_burst_run >= PASTE_BURST_MIN_CHARS + and not self._in_slash_command_context() + ): + self._paste_burst_window_until = now + PASTE_ENTER_SUPPRESS_WINDOW_SECONDS + + def _check_burst_run_for_promotion(self) -> None: + """Promote a rapid run whose shape or size already confirms a paste. + + Callers must invoke this only once the current character is in the + document, so the run is present and `_promote_paste_burst_run` can find + it. Two confirmations do not need to wait for a newline: + + - A dropped-path shape (`/`, `~`, drive letter, `file://`, UNC). Without + this, an unquoted single-line drop would never reach path parsing or + media rejection. + - A run that reaches `PASTE_BURST_PROMOTE_CHARS`, a length no human + reaches at burst speed. Without this, a large single-line key-event + paste would never collapse into a placeholder. + + Neither case flushes here. The paste is still streaming, so the run is + only a prefix of the payload; once promoted, the remaining characters + are absorbed straight into the buffer by `_absorb_key_into_burst` and + the idle timer flushes the complete payload when the stream stops. + Flushing per keystroke would instead hand each prefix to + `_dispatch_burst_payload` — re-running path parsing, and its filesystem + probes, once per character. + + Ordinary typing reaches neither confirmation: to qualify it would have to + sustain `PASTE_BURST_MIN_CHARS` keystrokes inside + `PASTE_BURST_CHAR_GAP_SECONDS` of each other (~400 WPM) *and* either open + with a path shape or run past the length threshold. + """ + if self._paste_burst_buffer or self._paste_burst_run < PASTE_BURST_MIN_CHARS: + return + payload = self._paste_burst_run_text + dispatch_payload = self._burst_run_payload_for_dispatch(payload) + # In slash-command context Enter always submits and the text is a command, + # not content, so nothing may be hidden — with one exception: a payload the + # dispatch hook rewrote is a path whose leading `/` the mode prefix + # consumed (see `ChatTextArea._burst_run_payload_for_dispatch`), and that + # must reach path handling precisely so it stops being read as a command. + if self._in_slash_command_context() and dispatch_payload == payload: + return + if not ( + looks_like_dropped_payload(dispatch_payload) + or len(payload) >= PASTE_BURST_PROMOTE_CHARS + ): + return + last_key_time = self._paste_burst_last_key_time + if last_key_time is None: + # Unreachable: the run is at least `PASTE_BURST_MIN_CHARS`, and every + # counted keystroke stamps this field. Refuse rather than substitute a + # timestamp, which on a monotonic clock would land decades in the past + # and make the burst flush per character. + logger.warning( + "Qualifying burst run (%d chars) has no key timestamp; " + "skipping promotion", + self._paste_burst_run, + ) + return + if not self._promote_paste_burst_run(last_key_time): + # The shape or length already confirmed a paste, so failing here means + # dropped-path routing, media rejection, and collapsing are all + # silently skipped for this payload. Worth a breadcrumb. + logger.warning( + "Confirmed burst paste (%d chars) could not be promoted; " + "path routing and collapsing are skipped for it", + len(payload), + ) + return + self._paste_burst_buffer = dispatch_payload + self._on_burst_run_promoted(payload, dispatch_payload) def _consume_enter_as_burst_newline(self, now: float) -> bool: """Insert a newline instead of submitting when inside a paste burst. @@ -426,9 +586,31 @@ def _consume_enter_as_burst_newline(self, now: float) -> bool: if not self._enter_inserts_newline_during_burst(now): self._paste_burst_last_suppressed_enter_time = None return False + # This newline confirms a multi-line key-event paste, so pull the + # still-visible run into the buffer and keep the newline with it. Both + # shipped `_on_key`s absorb or flush an active buffer before Enter + # reaches here, so the `_paste_burst_buffer` branch is unreachable. It + # exists because falling through to `_promote_paste_burst_run` would call + # `_start_paste_burst`, which *assigns* the buffer rather than appending + # — silently dropping the text already in it. Loud rather than silently + # wrong if a future caller gets here. + if self._paste_burst_buffer: + logger.warning( + "Enter reached burst-newline handling with a live buffer " + "(%d chars); keeping the newline with the buffer", + len(self._paste_burst_buffer), + ) + self._append_paste_burst("\n", now) + elif self._promote_paste_burst_run(now): + self._append_paste_burst("\n", now) + else: + self.action_insert_newline() + # Set after promotion: `_promote_paste_burst_run` resets run tracking, + # which clears `_paste_burst_last_suppressed_enter_time`. + # `_paste_burst_window_until` is not cleared by that reset; it is + # refreshed here to extend the grouping window. self._paste_burst_window_until = now + PASTE_ENTER_SUPPRESS_WINDOW_SECONDS self._paste_burst_last_suppressed_enter_time = now - self.action_insert_newline() return True # -- Newline affordances (shared by concrete text areas) ------------------ @@ -475,6 +657,7 @@ def _consume_backslash_enter_newline( and enabled and self._backslash_pending_time is not None and (now - self._backslash_pending_time) <= _BACKSLASH_ENTER_GAP_SECONDS + and not self._enter_inserts_newline_during_burst(now) ): self._backslash_pending_time = None if self._delete_preceding_backslash(): diff --git a/libs/code/deepagents_code/tui/widgets/chat_input.py b/libs/code/deepagents_code/tui/widgets/chat_input.py index f0679390a1..2c8d2c9673 100644 --- a/libs/code/deepagents_code/tui/widgets/chat_input.py +++ b/libs/code/deepagents_code/tui/widgets/chat_input.py @@ -31,7 +31,10 @@ get_glyphs, is_ascii_mode, ) -from deepagents_code.input import IMAGE_PLACEHOLDER_PATTERN, VIDEO_PLACEHOLDER_PATTERN +from deepagents_code.input import ( + IMAGE_PLACEHOLDER_PATTERN, + VIDEO_PLACEHOLDER_PATTERN, +) from deepagents_code.paste_collapse import ( PASTE_PLACEHOLDER_PATTERN, PastedContent, @@ -577,31 +580,6 @@ def __init__(self, current_text: str) -> None: class HistoryNext(Message): """Request next history entry.""" - class PastedPaths(Message): - """Message sent when paste payload resolves to file paths.""" - - def __init__(self, raw_text: str, paths: list[Path]) -> None: - """Initialize with raw pasted text and parsed file paths.""" - self.raw_text = raw_text - 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. @@ -620,6 +598,7 @@ def __init__(self, **kwargs: Any) -> None: self._chat_input_owner: ChatInput | None = None self._skip_history_change_events = 0 self._completion_active = False + self._burst_payload_keeps_leading_slash = False # Paste-burst and backslash-pending state is initialized by # PasteBurstTextArea.__init__. # Tracks terminal focus so a click that re-focuses the window only @@ -969,11 +948,26 @@ def _paste_collapse_enabled(self) -> bool: async def _dispatch_burst_payload(self, payload: str) -> None: """Route a flushed burst through dropped-path and large-paste checks. - When parsing fails, the buffered text is inserted unchanged so regular - typing behavior is preserved. + Routed payloads are applied through the owner synchronously rather than + posted to it, so the payload is in the document before this returns. A + posted message lands at the tail of this widget's queue, behind any + keystroke the terminal has already delivered, which would insert that + character ahead of the paste. + + When parsing fails, or there is no owner to route through, the buffered + text is inserted unchanged so regular typing behavior is preserved. """ from deepagents_code.input import parse_pasted_path_payload + keeps_leading_slash = self._burst_payload_keeps_leading_slash + self._burst_payload_keeps_leading_slash = False + owner = self._chat_input_owner + if owner is not None: + # Cleared up front so the verbatim-insert path below cannot leave a + # previous payload's answer standing for + # `_payload_supplied_trailing_space`. + owner._paste_appended_trailing_space = False + try: parsed = await asyncio.to_thread(parse_pasted_path_payload, payload) except Exception: @@ -989,15 +983,104 @@ async def _dispatch_burst_payload(self, payload: str) -> None: exc_info=True, ) parsed = None - if parsed is not None: - self.post_message(self.PastedPaths(payload, parsed.paths)) - return + if owner is not None: + if parsed is not None: + applied = owner.apply_paste_payload(payload, parsed.paths) + elif self._paste_collapse_enabled() and _should_collapse_chat_paste( + payload + ): + applied = owner.apply_paste_payload(payload, None) + else: + applied = False + if applied: + return - if self._paste_collapse_enabled() and _should_collapse_chat_paste(payload): - self.post_message(self.PastedText(payload)) + if keeps_leading_slash and owner is not None: + # Consumed by the change handler this insert triggers, suppressing the + # mode re-detection that would otherwise strip the restored `/`. + owner.suppress_next_prefix_detection() + self.insert(payload) + # A multi-line payload adds rows the same way `action_insert_newline` + # does, and needs the same post-refresh scroll for the same reason: the + # built-in scroll sees stale dimensions and leaves the cursor off screen. + if "\n" in payload: + self.call_after_refresh(self.scroll_cursor_visible) + + def _burst_run_payload_for_dispatch(self, payload: str) -> str: + """Restore a virtual command prefix when a burst is an absolute path. + + A `/` typed at offset 0 switches the input into command mode and is never + inserted, so a dropped absolute path replayed as key events loses its + leading separator. Restoring it lets the run be recognized as a path. + + The restore is deliberately narrow, because a payload rewritten here is + also what takes the input *out* of command mode + (`_on_burst_run_promoted`). Asking `looks_like_dropped_payload` about the + `/`-prefixed candidate cannot decide this — that function is a leading- + token check, so prepending `/` makes it vacuously true for any text. Three + conditions stand in for it instead: + + - The run must start at document offset 0, i.e. it is the text that + directly followed the consumed `/` rather than a later burst. + - The payload must contain its own separator, so `help` stays a command + name while `private/tmp/x` reads as a path tail. + - Nothing before that separator may be whitespace, which keeps a command + with a path argument (`read src/main.py`) from qualifying. + + Returns: + The payload with the consumed leading slash restored, or the payload + unchanged when it does not look like the tail of an absolute path. + """ + owner = self._chat_input_owner + if owner is None or owner.mode != "command": + return payload + cursor_offset = self.document.get_index_from_location(self.cursor_location) # ty: ignore[unresolved-attribute] # Document has this method; DocumentBase stub is narrower + if cursor_offset != len(payload) or not self.text.startswith(payload): + return payload + head, separator, _ = payload.partition("/") + if not separator or any(char.isspace() for char in head): + return payload + # Exactly one `/` was consumed, so exactly one is restored: `lstrip("/")` + # would eat a second leading slash and silently drop a character from a + # `//host/share` payload. + return f"/{payload}" + + def _on_burst_run_promoted( + self, visible_payload: str, dispatch_payload: str + ) -> None: + """Leave command mode when promotion recovered a leading path slash.""" + if visible_payload == dispatch_payload: return + # The payload now leads with the restored `/`, so re-inserting it at + # offset 0 would trip mode-prefix detection a second time and strip the + # slash again — losing the character for good on the paths that do not + # resolve on disk. Flag it so the insert suppresses that detection. + self._burst_payload_keeps_leading_slash = True + owner = self._chat_input_owner + if owner is not None and owner.mode == "command": + owner.mode = "normal" - self.insert(payload) + def _reset_paste_burst_state(self) -> None: + """Reset burst tracking, including the restored-slash flag. + + The flag describes the buffered payload that `super()` is about to + discard, so it must not outlive it: a stale `True` would suppress the next + burst's legitimate mode re-detection. + """ + self._burst_payload_keeps_leading_slash = False + super()._reset_paste_burst_state() + + def _payload_supplied_trailing_space(self) -> bool: + """Return whether the flush appended a trailing space of its own. + + An attached dropped-path payload gets a trailing space from + `_build_path_replacement`; inserting the pending space as well would + double it. The question is what the flush *did*, not what the document + happens to end with — a verbatim payload that merely ends in a space + would otherwise swallow the user's real keystroke. + """ + owner = self._chat_input_owner + return owner is not None and owner._paste_appended_trailing_space async def _on_key(self, event: events.Key) -> None: """Handle key events.""" @@ -1029,8 +1112,32 @@ async def _on_key(self, event: events.Key) -> None: if event.key == "space" and event.character is None: event.prevent_default() event.stop() + # This branch bypasses the burst helpers below, so it has to drive + # them itself: a space inside a replayed paste must reach the buffer + # (if one is live) or the run tracker (if not), otherwise the + # tracker's text diverges from the document and the run is discarded, + # losing grouping for that stretch of the paste. + space_now = time.monotonic() + if self._paste_burst_buffer and self._append_recent_paste_burst_text( + " ", space_now + ): + self.post_message(self.Typing()) + return + # The burst (if any) had gone idle, so this space follows the paste + # rather than belonging to it. Flushing applies the payload before it + # returns, so the space inserted next lands after it. + if self._paste_burst_buffer: + await self._flush_paste_burst() + if self._payload_supplied_trailing_space(): + self.post_message(self.Typing()) + return self.insert(" ") + self._note_printable_burst_keystroke(" ", space_now) self.post_message(self.Typing()) + # The space is in the document, so the run may now qualify — a path + # payload can end on a space, and a long single-line paste can cross + # the length threshold here. + self._check_burst_run_for_promotion() return now = time.monotonic() @@ -1045,17 +1152,9 @@ async def _on_key(self, event: events.Key) -> None: event.stop() return - if self._maybe_start_burst(event, now): - event.prevent_default() - event.stop() - return - - # Promote rapid keystroke runs into the paste buffer so terminals without - # bracketed paste still get newline grouping and large-paste collapsing. - if self._track_burst_run(event, now): - event.prevent_default() - event.stop() - return + # Track rapid keystroke runs so terminals without bracketed paste keep + # embedded newlines grouped without delaying ordinary text insertion. + self._track_burst_run(event, now) # A mode trigger (`!`, `!!`, `/`) typed at the very start of an # unselected input switches modes. Handle it before TextArea inserts the @@ -1071,6 +1170,10 @@ async def _on_key(self, event: events.Key) -> None: ): event.prevent_default() event.stop() + # `_track_burst_run` above already counted this character, but it is + # consumed as a mode switch rather than inserted. Drop the run so the + # tracker does not claim a character the document never received. + self._reset_paste_burst_run() return # Some terminals (e.g. VSCode built-in) send a literal backslash @@ -1109,12 +1212,20 @@ async def _on_key(self, event: events.Key) -> None: # Prevent TextArea's default behavior (e.g., Enter inserting newline) # but let event bubble to ChatInput for completion handling event.prevent_default() + # `space` is the one printable key here, so `_track_burst_run` above + # counted it while this branch inserts nothing. Drop the run so the + # tracker does not claim a character the document never received — + # the same reason as the mode-prefix branch above. + if event.is_printable: + self._reset_paste_burst_run() return # Plain Enter submits, unless a recent keystroke burst suggests this - # newline is part of a paste replayed as key events; then insert a - # newline and keep the window alive so the rest of the paste stays - # grouped instead of submitting mid-stream. + # newline is part of a paste replayed as key events. In that case the + # visible run is pulled off screen into the paste buffer along with this + # newline, and the window is kept alive so the rest of the paste stays + # grouped instead of submitting mid-stream. The text reappears when the + # burst flushes — possibly as a `[Pasted text #N]` placeholder. if event.key == "enter": event.prevent_default() event.stop() @@ -1132,6 +1243,10 @@ async def _on_key(self, event: events.Key) -> None: await super()._on_key(event) + # Must follow `super()._on_key`: promotion verifies the run against the + # document, so the current character has to be in it already. + self._check_burst_run_for_promotion() + def action_delete_right(self) -> None: """Delete a bound placeholder atomically or the next character.""" if not self._delete_placeholder_token(backwards=False): @@ -1300,19 +1415,25 @@ async def _on_paste(self, event: events.Paste) -> None: exc_info=True, ) parsed = None - if parsed is not None: + owner = self._chat_input_owner + if parsed is not None and owner is not None: event.prevent_default() event.stop() - self.post_message(self.PastedPaths(event.text, parsed.paths)) + owner.apply_paste_payload(event.text, parsed.paths) return - if self._paste_collapse_enabled() and _should_collapse_chat_paste(event.text): + if ( + owner is not None + and self._paste_collapse_enabled() + and _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. + # the full text. The owner stores the content and inserts a compact + # placeholder instead — applied here rather than posted, so a + # keystroke queued behind this paste cannot overtake it. event.prevent_default() event.stop() - self.post_message(self.PastedText(event.text)) + owner.apply_paste_payload(event.text, None) return # Don't call super() here — Textual's MRO dispatch already calls @@ -2040,6 +2161,11 @@ def __init__( # immediately recurse into the same replacement path. self._applying_inline_path_replacement = False + # Whether the most recent `apply_paste_payload` appended its own + # trailing space. Read by `ChatTextArea._payload_supplied_trailing_space` + # to decide whether a pending space keystroke would double it. + self._paste_appended_trailing_space = False + # Text area content from the previous Changed event. Used to skip # blocking filesystem path-detection on single-keystroke edits while # still detecting replacement edits that insert a full path payload. @@ -2604,6 +2730,14 @@ def handle_mode_prefix_keystroke(self, char: str) -> bool: True if the keystroke was consumed as a mode selector without inserting the character, otherwise False. """ + # The first slash enters command mode without being inserted. A second + # slash at the same offset can be the leading separator of a UNC-style + # path replayed as key events, so retain it rather than consuming both + # characters as mode triggers. + if char == "/" and self.mode == "command": + self.suppress_next_prefix_detection() + return False + detected_prefix = detect_mode_prefix(char) if detected_prefix is None: return False @@ -2873,26 +3007,42 @@ def on_chat_text_area_history_next( else: self.app.bell() - def on_chat_text_area_pasted_paths(self, event: ChatTextArea.PastedPaths) -> None: - """Handle paste payloads that resolve to dropped file paths.""" - if not self._text_area: - return + def apply_paste_payload(self, text: str, paths: list[Path] | None) -> bool: + """Apply an already-parsed paste payload to the input. - 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. + Callers apply a payload through this method rather than posting it as a + message so it reaches the document synchronously. Textual appends a + posted message to the tail of the receiving widget's FIFO queue, so a + keystroke the terminal already delivered would be handled first and land + ahead of the paste. Args: - event: The `PastedText` message carrying the full pasted text. + text: Raw payload text. + paths: Resolved dropped paths, or `None` to collapse `text` into a + `[Pasted text #N]` placeholder. + + Returns: + `True` when the payload was applied. `False` when there is no text + area to apply it to, in which case the caller still owns the text. """ if not self._text_area: - return - self._collapse_and_insert_paste(event.text) + return False + if paths is not None: + self._paste_appended_trailing_space = self._insert_pasted_paths(text, paths) + else: + self._collapse_and_insert_paste(text) + self._paste_appended_trailing_space = False + return True + + def suppress_next_prefix_detection(self) -> None: + """Skip mode-prefix detection for the next text change. + + Used when inserting text that legitimately starts with a mode trigger, so + the change handler does not consume that character. Shares the guard with + `_strip_mode_prefix`, which reports a guard left uncleared by a missed + change event. + """ + self._stripping_prefix = True def handle_external_paste(self, pasted: str) -> bool: """Handle paste text from app-level routing when input is not focused. @@ -2913,9 +3063,9 @@ def handle_external_paste(self, pasted: str) -> bool: parsed = self._parse_dropped_path_payload(pasted) if parsed is not None: - self._insert_pasted_paths(pasted, parsed.paths) + self.apply_paste_payload(pasted, parsed.paths) elif self._collapse_pastes and _should_collapse_chat_paste(pasted): - self._collapse_and_insert_paste(pasted) + self.apply_paste_payload(pasted, None) else: self._text_area.insert(pasted) @@ -2991,22 +3141,28 @@ def _apply_inline_dropped_path_replacement(self, text: str) -> bool: self._text_area.move_cursor_to_end() return True - def _insert_pasted_paths(self, raw_text: str, paths: list[Path]) -> None: + def _insert_pasted_paths(self, raw_text: str, paths: list[Path]) -> bool: """Insert pasted path payload, attaching images when possible. Args: raw_text: Original paste payload text. paths: Resolved file paths parsed from the payload. + + Returns: + `True` when the inserted text carries a trailing space that + `_build_path_replacement` appended. Unattached payloads are inserted + verbatim, so they never do. """ if not self._text_area: - return + return False replacement, attached = self._build_path_replacement( raw_text, paths, add_trailing_space=True ) if attached: self._text_area.insert(replacement) - return + return replacement.endswith(" ") self._text_area.insert(raw_text) + return False def _build_path_replacement( self, diff --git a/libs/code/tests/unit_tests/test_input_parsing.py b/libs/code/tests/unit_tests/test_input_parsing.py index 35d04a7ac7..e5b3fc17c9 100644 --- a/libs/code/tests/unit_tests/test_input_parsing.py +++ b/libs/code/tests/unit_tests/test_input_parsing.py @@ -435,8 +435,7 @@ def test_dropped_payload_paths_resolves_quoted_payload( """Quoted and bracketed drops resolve, since terminals wrap paths that way. The shape guard strips leading `<`, `'`, and `"` for exactly this reason; - without that strip every quoted drop would look like typed text. A quoted - path is also the designed burst shape — see `PASTE_BURST_START_CHARS`. + without that strip every quoted drop would look like typed text. """ img = tmp_path / "shot.png" img.write_bytes(b"img") diff --git a/libs/code/tests/unit_tests/tui/widgets/test_chat_input.py b/libs/code/tests/unit_tests/tui/widgets/test_chat_input.py index f2600d4af5..f2fd64df9e 100644 --- a/libs/code/tests/unit_tests/tui/widgets/test_chat_input.py +++ b/libs/code/tests/unit_tests/tui/widgets/test_chat_input.py @@ -242,6 +242,10 @@ def compose(self) -> ComposeResult: yield ChatInput(id="chat-input") +class _DispatchError(RuntimeError): + """Stand-in for a burst dispatch that raises (media decode, notify, ...).""" + + class _RecordingApp(App[None]): """App that records ChatInput.Submitted events for assertion.""" @@ -2506,8 +2510,8 @@ async def test_handle_mode_prefix_keystroke_switches_without_text_change( # Non-trigger characters are never consumed. assert chat.handle_mode_prefix_keystroke("a") is False - async def test_redundant_typed_slash_keystroke_stays_command_mode(self) -> None: - """A redundant `/` at the command prompt is consumed as a mode selector.""" + async def test_second_typed_slash_stays_in_command_text(self) -> None: + """A second `/` is retained so key-event path pastes keep both slashes.""" app = _ChatInputTestApp() async with app.run_test() as pilot: chat = app.query_one(ChatInput) @@ -2521,7 +2525,7 @@ async def test_redundant_typed_slash_keystroke_stays_command_mode(self) -> None: await pilot.press("/") await _pause_for_strip(pilot) assert chat.mode == "command" - assert chat._text_area.text == "" + assert chat._text_area.text == "/" async def test_typed_bang_keystroke_skips_strip_round_trip( self, monkeypatch: pytest.MonkeyPatch @@ -3892,6 +3896,36 @@ async def test_key_burst_quoted_path_rewrites_without_showing_raw_path( assert chat._text_area.text == "[image 1] " assert len(app.tracker.get_images()) == 1 + async def test_key_burst_absolute_path_preserves_leading_slash( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A rapid absolute path recovers the slash consumed by command mode.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 1.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + img_path = tmp_path / "absolute-burst.png" + from PIL import Image + + Image.new("RGB", (3, 3), color="navy").save(img_path, format="PNG") + + app = _ImagePasteApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + assert chat._text_area is not None + + for char in str(img_path): + await chat._text_area._on_key(events.Key(char, char)) + + assert chat.mode == "normal" + assert chat._text_area.text == "" + + await pilot.pause(0.35) + + assert chat._text_area.text == "[image 1] " + assert len(app.tracker.get_images()) == 1 + async def test_submit_absolute_path_without_paste_event_attaches_image( self, tmp_path ) -> None: @@ -5210,7 +5244,7 @@ class TestPasteBurstEnterSuppression: async def test_rapid_burst_with_newline_does_not_submit( self, monkeypatch: pytest.MonkeyPatch ) -> None: - """A fast keystroke run followed by enter inserts a newline.""" + """A fast keystroke run stays visible and enter inserts a newline.""" # Widen the burst gap so wall-clock delays between pilot.press calls on # slow CI runners still register as a single rapid burst. monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) @@ -5226,6 +5260,9 @@ async def test_rapid_burst_with_newline_does_not_submit( for char in "hello": await pilot.press(char) + assert ta.text == "hello" + assert ta._paste_burst_buffer == "" + await pilot.press("enter") await pilot.press("w") await pilot.pause(0.15) @@ -5271,6 +5308,8 @@ async def test_single_line_burst_then_manual_enter_submits( ta.text = "abc" now = chat_input_module.time.monotonic() + ta._paste_burst_run = paste_textarea_module.PASTE_BURST_MIN_CHARS + ta._paste_burst_run_text = "abc" ta._paste_burst_last_key_time = ( now - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS - 0.01 ) @@ -5367,6 +5406,949 @@ async def test_slash_command_enter_still_submits_during_burst(self) -> None: assert len(app.submitted) == 1 + async def test_late_enter_after_qualifying_run_still_submits( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A qualifying run does not swallow deliberate enter after going idle.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 0.12 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + ta.text = "abc" + now = chat_input_module.time.monotonic() + # A qualifying run, but the last keystroke landed 50 ms ago — + # outside the 30 ms char gap, still inside the 120 ms window. + ta._paste_burst_run = paste_textarea_module.PASTE_BURST_MIN_CHARS + ta._paste_burst_run_text = "abc" + ta._paste_burst_last_key_time = now - 0.05 + ta._paste_burst_window_until = now + 0.12 + + await ta._on_key(events.Key("enter", None)) + await pilot.pause() + + assert len(app.submitted) == 1 + assert app.submitted[0].value == "abc" + + +class TestPasteBurstPromotion: + """Promotion of a visible rapid run into the hidden paste buffer. + + Rapid typing stays in the document until something confirms a paste: an + embedded newline, a dropped-path shape, or a length no human reaches at + burst speed. These tests drive the real `_on_key` path, since the chat + input's key handling interleaves several branches ahead of the burst + helpers. + """ + + async def test_multiline_key_event_paste_collapses( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A multi-line key-event paste is promoted and collapsed.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + payload = "alpha\n" + "beta gamma delta\n" * 3 + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in payload: + event = ( + events.Key("enter", None) + if char == "\n" + else events.Key(char, char) + ) + await ta._on_key(event) + await pilot.pause(0.35) + + assert "[Pasted text #1" in ta.text + assert chat._pasted_contents[1].content == payload + assert len(app.submitted) == 0 + + async def test_key_event_paste_preserves_backslash_before_newline( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A burst newline takes priority over the backslash+Enter fallback.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in "abc": + await ta._on_key(events.Key(char, char)) + await ta._on_key(events.Key("backslash", "\\")) + await ta._on_key(events.Key("enter", None)) + await pilot.pause(0.35) + + assert ta.text == "abc\\\n" + assert len(app.submitted) == 0 + + async def test_large_single_line_key_event_paste_collapses( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A long single-line key-event paste collapses without any newline. + + `should_collapse_paste` triggers on length as well as line count, so a + newline must not be required to reach collapse handling. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + payload = "y" * 900 + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in payload: + await ta._on_key(events.Key(char, char)) + await pilot.pause(0.35) + + assert "[Pasted text #1]" in ta.text + assert payload not in ta.text + assert chat._pasted_contents[1].content == payload + + async def test_rapid_slash_command_is_not_promoted_as_a_dropped_path( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A key-event `/help` burst must retain command submission semantics.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in "/help": + await pilot.press(char) + + assert chat.mode == "command" + assert ta.text == "help" + assert ta._paste_burst_buffer == "" + + await pilot.press("enter") + await pilot.pause() + + assert len(app.submitted) == 1 + assert app.submitted[0].value == "/help" + assert app.submitted[0].mode == "command" + + @pytest.mark.parametrize("payload", ["hello world", '"hello world"']) + async def test_ordinary_rapid_typing_is_never_promoted( + self, monkeypatch: pytest.MonkeyPatch, payload: str + ) -> None: + """A short rapid run, including quoted text, stays fully visible.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in payload: + await pilot.press(char) + + # Asserted before any pause: the flush timer would restore the text + # and hide the very regression this covers. Typing must be visible + # *while* typing, not once it stops. + assert ta.text == payload + assert ta._paste_burst_buffer == "" + + await pilot.pause(0.15) + + assert ta.text == payload + assert ta._paste_burst_buffer == "" + + async def test_promotion_falls_back_when_selection_is_active( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A selection at Enter blocks promotion; the newline is inserted plainly. + + Promoting would delete the user's selected range rather than the run. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in "abc": + await pilot.press(char) + ta.selection = Selection((0, 0), (0, 3)) + + await ta._on_key(events.Key("enter", None)) + await pilot.pause() + + assert ta._paste_burst_buffer == "" + assert "abc" in ta.text + assert len(app.submitted) == 0 + + async def test_promotion_falls_back_when_run_diverges_from_document( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A run that no longer sits before the cursor is dropped, not deleted. + + Moving the cursor mid-run (e.g. a mouse click, which is not a key + event) desynchronises the tracker. Promoting on a stale run would + delete whatever text now happens to precede the cursor. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + ta.text = "XXXX" + for char in "abc": + await pilot.press(char) + # Relocate the cursor without a key event, so the run survives but + # no longer describes the characters before the cursor. + ta.selection = Selection((0, 2), (0, 2)) + + await ta._on_key(events.Key("enter", None)) + await pilot.pause() + + assert ta._paste_burst_buffer == "" + # Every character survives; only a newline was added at the cursor. + assert ta.text.replace("\n", "") == "abcXXXX" + assert ta.text.count("\n") == 1 + # The diverged run is dropped so it cannot be re-promoted later. + assert ta._paste_burst_run_text == "" + + async def test_vscode_space_workaround_keeps_run_in_sync( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A CSI-u space is tracked, so a burst containing one still promotes. + + VS Code sends space as a key with no character; the workaround inserts + it directly and returns before the burst helpers, so it must feed the + tracker itself or the run text diverges from the document. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in "ab": + await ta._on_key(events.Key(char, char)) + await ta._on_key(events.Key("space", None)) + for char in "cd": + await ta._on_key(events.Key(char, char)) + + assert ta.text == "ab cd" + assert ta._paste_burst_run_text == "ab cd" + + await ta._on_key(events.Key("enter", None)) + await pilot.pause() + + # Promotion succeeded, so the run moved into the buffer rather than + # failing verification and falling back to a plain newline. + assert ta.text == "" + assert ta._paste_burst_buffer == "ab cd\n" + assert len(app.submitted) == 0 + + async def test_vscode_space_workaround_flushes_stale_buffer( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A delayed CSI-u space starts normal input after a completed burst.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + ta._start_paste_burst("abc", stale_time) + + await ta._on_key(events.Key("space", None)) + await pilot.pause() + + assert ta._paste_burst_buffer == "" + assert ta.text == "abc " + assert ta._paste_burst_run_text == " " + + async def test_vscode_space_follows_queued_stale_payload( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A stale burst's queued placeholder is inserted before its space.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + payload = "x" * 900 + ta._start_paste_burst(payload, stale_time) + + await ta._on_key(events.Key("space", None)) + await pilot.pause() + + assert ta.text == "[Pasted text #1] " + assert chat._pasted_contents[1].content == payload + assert ta._paste_burst_run_text == " " + + async def test_vscode_space_stays_ahead_of_already_queued_key( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A key queued behind the space must not overtake it.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + ta._start_paste_burst("abc", stale_time) + + ta.post_message(events.Key("space", None)) + ta.post_message(events.Key("x", "x")) + await pilot.pause() + + assert ta.text == "abc x" + assert ta._paste_burst_run_text == " x" + + async def test_vscode_space_is_absorbed_into_a_live_burst( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A CSI-u space mid-paste joins the buffer instead of flushing it. + + This is the common case for a VS Code key-event paste. Without it, every + space would flush the paste mid-stream into separate fragments. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + # Promote via a newline so the buffer is live and fresh. + for char in "abc": + await ta._on_key(events.Key(char, char)) + await ta._on_key(events.Key("enter", None)) + assert ta._paste_burst_buffer == "abc\n" + + await ta._on_key(events.Key("space", None)) + + # Absorbed into the hidden buffer, not inserted into the document. + assert ta._paste_burst_buffer == "abc\n " + assert ta.text == "" + + await pilot.pause(0.35) + assert ta.text == "abc\n " + + async def test_printable_key_lands_after_a_flushed_payload( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A key that breaks a stale burst is inserted after the payload. + + The payload is applied while handling this key, so it cannot be overtaken. + Applying it by a posted message instead put the character first, splitting + a paste that arrived across a slow terminal read boundary. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + payload = "x" * 900 + ta._start_paste_burst(payload, stale_time) + + await ta._on_key(events.Key("y", "y")) + await pilot.pause() + + assert ta.text == "[Pasted text #1]y" + assert chat._pasted_contents[1].content == payload + + async def test_backspace_after_a_flushed_payload_leaves_earlier_text_alone( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Backspace breaking a stale burst edits the payload, not the text before it. + + The payload lands first, so the deletion applies to it. Applying the + payload later deleted a character the user typed before the paste. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + ta.focus() + ta.insert("ab") + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + ta._start_paste_burst("x" * 900, stale_time) + + # Pressed through the app so the backspace binding actually resolves. + await pilot.press("backspace") + await pilot.pause() + + # The placeholder is deleted as one token, so only the paste is undone. + assert ta.text == "ab" + + async def test_bracketed_paste_stays_ahead_of_a_queued_key(self) -> None: + """A key queued behind a bracketed paste must not overtake it.""" + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + payload = "z" * 900 + ta.post_message(events.Paste(payload)) + ta.post_message(events.Key("q", "q")) + await pilot.pause() + + assert ta.text == "[Pasted text #1]q" + assert chat._pasted_contents[1].content == payload + + async def test_run_of_exactly_the_promote_threshold_is_reinserted_verbatim( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A run of exactly `PASTE_BURST_PROMOTE_CHARS` promotes but does not collapse. + + Promotion uses `>=` while `should_collapse_paste` uses `>`, so this length + is hidden and then restored unchanged. Pinned because the two comparisons + must keep agreeing that no character is lost between them. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.05 + ) + length = paste_textarea_module.PASTE_BURST_PROMOTE_CHARS + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for _ in range(length): + await ta._on_key(events.Key("a", "a")) + + # Confirmed by length alone, so the run is hidden before it flushes. + assert ta._paste_burst_buffer == "a" * length + assert ta.text == "" + + await pilot.pause(0.2) + + assert ta.text == "a" * length + assert chat._pasted_contents == {} + + async def test_burst_state_reset_clears_the_restored_slash_flag(self) -> None: + """A discarded payload must not leave its slash flag set. + + The flag suppresses one mode re-detection. Surviving the payload it + describes would spend that suppression on an unrelated later burst. + """ + app = _RecordingApp() + async with app.run_test(): + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + ta._burst_payload_keeps_leading_slash = True + ta._start_paste_burst("/private/tmp", chat_input_module.time.monotonic()) + + ta.clear_text() + + assert ta._paste_burst_buffer == "" + assert ta._burst_payload_keeps_leading_slash is False + + async def test_keys_queued_behind_a_flushed_payload_keep_their_order( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Printable keys already in the queue land after the payload, in order.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + ta._start_paste_burst("x" * 900, stale_time) + + # The space flushes the stale payload; the rest of the paste is + # already queued behind it. + ta.post_message(events.Key("space", None)) + for char in "world": + ta.post_message(events.Key(char, char)) + await pilot.pause(0.35) + + assert len(app.submitted) == 0 + assert ta.text == "[Pasted text #1] world" + + async def test_dropped_path_replacement_is_not_double_spaced( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The held space is dropped when the payload already ended with one.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + img_path = tmp_path / "spaced.png" + from PIL import Image + + Image.new("RGB", (3, 3), color="teal").save(img_path, format="PNG") + + app = _ImagePasteApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + ta._start_paste_burst(str(img_path), stale_time) + + await ta._on_key(events.Key("space", None)) + await pilot.pause(0.35) + + assert ta.text == "[image 1] " + + async def test_verbatim_payload_ending_in_space_keeps_the_typed_space( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A flushed payload that merely ends in a space must not eat the next one. + + The double-space guard applies only to the trailing space that + `_build_path_replacement` appends. A payload inserted verbatim supplies + no such space, so the user's own keystroke has to survive — checking the + document instead would silently swallow one space of, say, a pasted + indent. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 0.03) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + stale_time = ( + chat_input_module.time.monotonic() + - paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + - 0.01 + ) + ta._start_paste_burst("hello ", stale_time) + + await ta._on_key(events.Key("space", None)) + await pilot.pause(0.35) + + assert ta.text == "hello " + + async def test_failed_dispatch_reinserts_the_payload_verbatim( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A raising dispatch must not destroy the buffered paste. + + Promotion has already deleted the run from the document and the flush + clears the buffer before dispatching, so without the guard the text + exists nowhere — not on screen, not in the buffer, not in undo. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + async def _boom(payload: str) -> None: # noqa: ARG001, RUF029 + raise _DispatchError + + monkeypatch.setattr(ta, "_dispatch_burst_payload", _boom) + + ta._start_paste_burst("important paste", chat_input_module.time.monotonic()) + + with pytest.raises(_DispatchError): + await ta._flush_paste_burst() + await pilot.pause() + + assert ta.text == "important paste" + assert ta._paste_burst_buffer == "" + + async def test_flushed_run_is_not_re_promoted_by_a_later_enter( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A second Enter after a flush must not re-promote the same run. + + Promotion hands the run's characters to the buffer, which flushes them + back into the document. If the run tracker still claimed them, a second + Enter inside the window would find them sitting before the cursor, + delete them, and re-dispatch text the user had already committed. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in "abc": + await ta._on_key(events.Key(char, char)) + await ta._on_key(events.Key("enter", None)) + for char in "abc": + await ta._on_key(events.Key(char, char)) + # Let the burst flush its payload back into the document. + await pilot.pause(0.35) + assert ta.text == "abc\nabc" + assert ta._paste_burst_buffer == "" + + await ta._on_key(events.Key("enter", None)) + await pilot.pause() + + # A newline was added; the trailing "abc" was not swallowed. + assert ta.text == "abc\nabc\n" + + async def test_consumed_mode_prefix_resets_the_run(self) -> None: + """A mode trigger is counted but never inserted, so it clears the run.""" + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + await ta._on_key(events.Key("!", "!")) + await pilot.pause() + + assert ta._paste_burst_run_text == "" + + async def test_rapid_typing_in_command_mode_stays_visible( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Fast typing after a `/` is not mistaken for a dropped path. + + The slash-recovery hook prepends `/` before asking whether the payload + looks like a path, so a guard phrased as a question about the `/`-prefixed + candidate is vacuously true for any text. If that is the only guard, every + rapid run in command mode is hidden and the input silently leaves command + mode mid-command. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in "/git": + await pilot.press(char) + for char in "add": + await pilot.press(char) + await pilot.pause(0.35) + + assert ta.text == "gitadd" + assert chat.mode == "command" + assert ta._paste_burst_buffer == "" + + async def test_rapid_slash_command_with_path_argument_is_not_promoted( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A command whose argument is a path keeps its command semantics. + + The payload contains a separator, so a separator-only test would treat + `read src/main.py` as the tail of an absolute path — injecting a `/` and + dropping out of command mode. Whitespace before the separator rules it out. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + # Driven through `_on_key` rather than `pilot.press` so slash-command + # completion cannot rewrite the text out from under the assertion. + for char in "/read src/main.py": + key = "space" if char == " " else char + await ta._on_key(events.Key(key, char)) + await pilot.pause(0.35) + + # Still a command, and nothing was hidden or slash-prefixed. (The + # space itself is swallowed by the open completion popup, so the exact + # text is not asserted here.) + assert chat.mode == "command" + assert ta._paste_burst_buffer == "" + assert not ta.text.startswith("/") + assert ta.text.endswith("src/main.py") + + async def test_rapid_absolute_path_that_does_not_exist_keeps_its_slash( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A recovered slash survives the insert that follows a failed parse. + + Only an existing path takes the dropped-path branch. Everything else + falls through to a plain insert at offset 0, which trips mode-prefix + detection a second time — stripping the recovered slash again and losing + the character for good. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + missing = tmp_path / "no-such-file.txt" + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in str(missing): + await ta._on_key(events.Key(char, char)) + await pilot.pause(0.35) + + assert ta.text == str(missing) + assert chat.mode == "normal" + assert chat.value == str(missing) + + async def test_rapid_double_slash_path_keeps_both_leading_slashes( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A key-event UNC-style path does not lose its second slash to mode handling. + + The second slash is text rather than another mode trigger. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + payload = "//host/share" + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in payload: + await ta._on_key(events.Key(char, char)) + await pilot.pause(0.35) + + assert ta.text == payload + assert chat.value == payload + + async def test_run_just_below_the_promote_threshold_stays_visible( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Length-based promotion brackets exactly at `PASTE_BURST_PROMOTE_CHARS`.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + payload = "z" * (paste_textarea_module.PASTE_BURST_PROMOTE_CHARS - 1) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in payload: + await ta._on_key(events.Key(char, char)) + + # Before the pause: waiting for the flush timer would restore the + # text and mask a run that had wrongly been promoted. + assert ta.text == payload + assert ta._paste_burst_buffer == "" + + await pilot.pause(0.35) + + assert ta.text == payload + assert ta._paste_burst_buffer == "" + + async def test_run_resets_at_human_typing_speed(self) -> None: + """Real inter-key gaps keep the run at one, so nothing ever qualifies. + + Every other test here widens the char gap so each keystroke counts as + burst speed. That is the right worst case for visibility, but it never + exercises the reset — and if the reset regressed, a slowly typed long + paragraph would vanish into a placeholder. + """ + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + typed = "" + for char in "abcde": + await ta._on_key(events.Key(char, char)) + typed += char + # Checked after every keystroke rather than only at the end: the + # inter-key sleep is shorter than the flush delay, so a promoted + # run would be hidden right here and restored before the final + # assertion could see it. + assert ta.text == typed + await asyncio.sleep( + paste_textarea_module.PASTE_BURST_CHAR_GAP_SECONDS + 0.01 + ) + await pilot.pause() + + assert ta.text == "abcde" + assert ta._paste_burst_run == 1 + + async def test_completion_space_resets_the_run( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A space swallowed for completion is counted but never inserted. + + `space` is the one printable key the completion-navigation branch + intercepts, so without a reset the tracker claims a character the + document never received and later promotions fail verification. + """ + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + + app = _RecordingApp() + async with app.run_test() as pilot: + chat = app.query_one(ChatInput) + ta = chat._text_area + assert ta is not None + + for char in "ab": + await ta._on_key(events.Key(char, char)) + ta._completion_active = True + await ta._on_key(events.Key("space", " ")) + await pilot.pause() + + assert ta.text == "ab" + assert ta._paste_burst_run_text == "" + class TestPasteCollapseHelpers: """Unit tests for the paste_collapse module helpers.""" @@ -5831,9 +6813,9 @@ async def test_bracketed_paste_event_collapses( ) -> 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`, - and asserts the collapse toast fires on that path too. + Exercises the production path (`_on_paste` -> `apply_paste_payload`) + rather than `handle_external_paste`, and asserts the collapse toast fires + on that path too. """ big_text = "z" * 900 app = _RecordingApp() diff --git a/libs/code/tests/unit_tests/tui/widgets/test_inline_prompt.py b/libs/code/tests/unit_tests/tui/widgets/test_inline_prompt.py index 99e344fb3c..2c11db00c7 100644 --- a/libs/code/tests/unit_tests/tui/widgets/test_inline_prompt.py +++ b/libs/code/tests/unit_tests/tui/widgets/test_inline_prompt.py @@ -101,6 +101,13 @@ async def test_unquoted_key_event_paste_collapses_and_expands( monkeypatch.setattr( paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 ) + # Promotion now happens on the first suppressed Enter, so the + # suppression window is load-bearing here: without widening it, a slow + # runner can spend more than the default 0.12s between the third + # character and the Enter and the paste would submit instead. + monkeypatch.setattr( + paste_textarea_module, "PASTE_ENTER_SUPPRESS_WINDOW_SECONDS", 60.0 + ) payload = "alpha\nbeta\ngamma\ndelta" app = _PromptApp() async with app.run_test() as pilot: @@ -190,7 +197,7 @@ async def test_modified_backspace_after_tab_deletes_placeholder_atomically( async def test_key_burst_with_newline_does_not_submit( self, monkeypatch: pytest.MonkeyPatch ) -> None: - """A multi-line paste replayed as key events inserts a newline, no submit.""" + """Rapid key-event text stays visible and inserts a newline, no submit.""" # Widen the burst gap/window so wall-clock delays on slow runners still # register as one rapid burst. monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) @@ -205,6 +212,9 @@ async def test_key_burst_with_newline_does_not_submit( for char in "hello": await pilot.press(char) + assert ta.text == "hello" + assert ta._paste_burst_buffer == "" + await pilot.press("enter") await pilot.press("w") await pilot.pause(0.15) @@ -212,6 +222,26 @@ async def test_key_burst_with_newline_does_not_submit( assert app.submissions == [] assert "\n" in ta.text + @pytest.mark.parametrize("payload", ["hello", '"hello"']) + async def test_rapid_typing_stays_visible( + self, monkeypatch: pytest.MonkeyPatch, payload: str + ) -> None: + """Rapid ordinary typing, including quoted text, stays visible.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + app = _PromptApp() + async with app.run_test() as pilot: + ta = app.query_one(InlinePromptTextArea) + ta.focus() + await pilot.pause() + + for char in payload: + await pilot.press(char) + await pilot.pause() + + assert ta.text == payload + assert ta._paste_burst_buffer == "" + assert not list(app._notifications) + async def test_deliberate_enter_submits( self, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -524,6 +554,33 @@ async def test_key_burst_of_video_is_rejected(self, tmp_path: Path) -> None: assert latest.message.startswith(MEDIA_UNSUPPORTED_TOAST_PREFIX) assert "clip.mp4" in latest.message + async def test_unquoted_media_key_burst_is_rejected( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """An unquoted media path replayed as rapid keys is rejected, not typed.""" + monkeypatch.setattr(paste_textarea_module, "PASTE_BURST_CHAR_GAP_SECONDS", 60.0) + monkeypatch.setattr( + paste_textarea_module, "PASTE_BURST_FLUSH_DELAY_SECONDS", 0.25 + ) + clip = tmp_path / "clip.mp4" + clip.write_bytes(b"vid") + app = _PromptApp() + async with app.run_test() as pilot: + ta = app.query_one(InlinePromptTextArea) + ta.focus() + await pilot.pause() + + for char in str(clip): + event = Key(char, char) + await ta._on_key(event) + + await pilot.pause(0.35) + + assert ta.text == "" + latest = list(app._notifications)[-1] + assert latest.message.startswith(MEDIA_UNSUPPORTED_TOAST_PREFIX) + assert "clip.mp4" in latest.message + async def test_non_media_path_burst_is_inserted( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -664,8 +721,8 @@ async def test_quoted_image_drop_via_real_key_events_is_rejected( Drives the production route end to end — `_on_key` -> burst promotion -> flush timer -> `_dispatch_burst_payload` — rather than calling the - dispatcher directly. A leading quote is in `PASTE_BURST_START_CHARS` - precisely so a dropped path buffers, so this is the designed shape. + dispatcher directly. The leading quote stays visible until the rapid + run contains enough of the absolute path to confirm a dropped payload. """ monkeypatch.setattr( paste_textarea_module, "_collapse_pastes_enabled", lambda: False @@ -717,13 +774,14 @@ async def test_pending_burst_is_flushed_before_a_media_paste_is_refused( ta.focus() await pilot.pause() - for char in "'abc": + pending = f"'{tmp_path / 'draft.txt'}'" + for char in pending: await ta._on_key(Key(char, char)) assert ta._paste_burst_buffer, "expected a pending burst to flush" await _paste(pilot, str(img)) - assert ta.text == "'abc" + assert ta.text == pending assert not ta._paste_burst_buffer latest = list(app._notifications)[-1] assert latest.message.startswith(MEDIA_UNSUPPORTED_TOAST_PREFIX)