feat(code): restore interrupted prompt to input on ESC (#4544)

Pressing `ESC` to interrupt the agent now returns the interrupted prompt
to the chat input when the input is empty.

---

When `ESC` interrupts a running agent turn, the prompt that was being
processed is now moved back into the chat input if the input is empty,
so it can be edited and resubmitted. This mirrors the existing
queued-message pop behavior. A non-empty draft in the input is left
untouched so typed text is never clobbered.

Made by [Open
SWE](https://openswe.vercel.app/agents/4708dd80-b2c2-8688-8b1c-ab77cb51228f)

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

libs/code/deepagents_code/app.py

@@ -10268,7 +10268,8 @@ class DeepAgentsApp(App):
             message: The user's message
         """
         # Mount the user message, tracking it so it can be dimmed on interrupt.
-        user_message = UserMessage(message)
+        media_snapshot = self._image_tracker.snapshot()
+        user_message = UserMessage(message, media_snapshot=media_snapshot)
         await self._mount_message(user_message)
         self._active_user_message = user_message
         await self._send_to_agent(message)
@@ -11720,11 +11721,50 @@ class DeepAgentsApp(App):
             self.notify("Queued message discarded", timeout=2)
             return
 
-        if not self._chat_input.value.strip():
-            self._chat_input.set_value_at_end(msg.text)
+        if self._chat_input.value.strip():
+            self.notify("Queued message discarded (input not empty)", timeout=3)
+        elif self._chat_input.set_value_at_end(msg.text):
             self.notify("Queued message moved to input", timeout=2)
         else:
-            self.notify("Queued message discarded (input not empty)", timeout=3)
+            logger.warning(
+                "Text area unavailable during queue pop; "
+                "message text could not be restored: %s",
+                msg.text[:60],
+            )
+            self.notify("Queued message discarded", timeout=2)
+
+    def _restore_interrupted_message_to_input(self, message: UserMessage) -> None:
+        """Return an interrupted prompt to the chat input when it is empty.
+
+        Shares the empty-input guard with `_pop_last_queued_message`: the
+        prompt is moved back into the input (along with any media captured at
+        submission) only when the input holds no draft, so typed text is never
+        clobbered. Unlike the queued-message pop, this path does not consume
+        the message — the interrupted `UserMessage` stays visible in the
+        transcript, dimmed via `set_cancelled()` — so when it does not restore
+        it stays silent rather than reporting a "discarded" outcome.
+        """
+        chat_input = self._chat_input
+        if chat_input is None:
+            logger.debug(
+                "Chat input unavailable during interrupt; "
+                "prompt not restored (message remains visible): %s",
+                message.raw_text[:60],
+            )
+            return
+        if chat_input.value.strip():
+            return
+        if not chat_input.set_value_at_end(message.raw_text):
+            logger.warning(
+                "Text area unavailable during interrupt; "
+                "prompt not restored to input: %s",
+                message.raw_text[:60],
+            )
+            return
+        snapshot = message.media_snapshot
+        if snapshot is not None:
+            self._image_tracker.restore(snapshot)
+        self.notify("Message restored to input", timeout=2)
 
     def _cleanup_external_event_source_sync(self) -> None:
         """Synchronously close the external event listener and unlink its socket.
@@ -11924,7 +11964,12 @@ class DeepAgentsApp(App):
             self._quit_pending = False
             return
 
-        # If agent is running, interrupt it and discard queued messages
+        # If agent is running, interrupt it and discard queued messages.
+        # Unlike the Esc path (`action_interrupt`), Ctrl+C deliberately does
+        # NOT restore the interrupted prompt to the input: Ctrl+C is the
+        # quit/copy flow (double-press quits; branch 7 copies the draft),
+        # whereas prompt-restore belongs to Esc's edit/retract flow. Do not
+        # add a restore call here without reconciling both interrupt paths.
         if self._agent_running and self._agent_worker:
             if self._active_user_message is not None:
                 self._active_user_message.set_cancelled()
@@ -12035,7 +12080,8 @@ class DeepAgentsApp(App):
         5. If approval menu is active, reject it
         6. If ask-user menu is active, cancel it
         7. If queued messages exist, pop the last one (LIFO)
-        8. If agent is running, interrupt it
+        8. If agent is running, interrupt it (restoring the interrupted prompt
+           to the chat input when it is empty)
         9. Otherwise, a second Esc clears the chat input draft (undoable)
         """
         from deepagents_code.tui.widgets.thread_selector import ThreadSelectorScreen
@@ -12111,6 +12157,7 @@ class DeepAgentsApp(App):
         if self._agent_running and self._agent_worker:
             if self._active_user_message is not None:
                 self._active_user_message.set_cancelled()
+                self._restore_interrupted_message_to_input(self._active_user_message)
             self._cancel_worker(self._agent_worker)
             return
 

libs/code/deepagents_code/input.py

@@ -3,7 +3,7 @@
 import logging
 import re
 import shlex
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from difflib import SequenceMatcher
 from pathlib import Path
 from typing import Literal
@@ -202,6 +202,26 @@ class MediaTracker:
         self.next_image_id = 1
         self.next_video_id = 1
 
+    def snapshot(self) -> "MediaTracker":
+        """Return an independent copy of the currently tracked media."""
+        tracker = MediaTracker()
+        tracker.images = [replace(img) for img in self.images]
+        tracker.videos = [replace(vid) for vid in self.videos]
+        tracker.next_image_id = self.next_image_id
+        tracker.next_video_id = self.next_video_id
+        return tracker
+
+    def restore(self, snapshot: "MediaTracker") -> None:
+        """Replace current media state with an independent snapshot copy.
+
+        Args:
+            snapshot: Previously captured media state to restore.
+        """
+        self.images = [replace(img) for img in snapshot.images]
+        self.videos = [replace(vid) for vid in snapshot.videos]
+        self.next_image_id = snapshot.next_image_id
+        self.next_video_id = snapshot.next_video_id
+
     def sync_to_text(
         self,
         text: str,

libs/code/deepagents_code/tui/widgets/chat_input.py

@@ -2900,12 +2900,20 @@ class ChatInput(Vertical):
         if self._text_area:
             self._text_area.text = val
 
-    def set_value_at_end(self, val: str) -> None:
-        """Set the input value and place the cursor at the end of the text."""
+    def set_value_at_end(self, val: str) -> bool:
+        """Set the input value and place the cursor at the end of the text.
+
+        Returns:
+            `True` when the value was written, `False` when the text area is
+            unavailable and the value could not be set. Callers that surface a
+            "restored"/"moved to input" toast should gate it on this so the
+            toast never claims a write that did not happen.
+        """
         if not self._text_area:
-            return
+            return False
         self._text_area.text = val
         self._text_area.move_cursor_to_end()
+        return True
 
     def discard_text(self) -> bool:
         """Clear the draft, keeping it restorable via undo (ctrl+z).

libs/code/deepagents_code/tui/widgets/messages.py

@@ -64,6 +64,8 @@ if TYPE_CHECKING:
     from textual.widgets import Markdown
     from textual.widgets._markdown import MarkdownStream
 
+    from deepagents_code.input import MediaTracker
+
 logger = logging.getLogger(__name__)
 
 
@@ -256,15 +258,38 @@ class UserMessage(Static):
     """
     """`-cancelled` dims a prompt whose turn was interrupted by the user."""
 
-    def __init__(self, content: str, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        content: str,
+        *,
+        media_snapshot: MediaTracker | None = None,
+        **kwargs: Any,
+    ) -> None:
         """Initialize a user message.
 
         Args:
             content: The message content
+            media_snapshot: Optional media tracker state captured at submission.
             **kwargs: Additional arguments passed to parent
         """
         super().__init__(**kwargs)
         self._content = content
+        self._media_snapshot = media_snapshot
+
+    @property
+    def raw_text(self) -> str:
+        """The original, untruncated message text as the user submitted it.
+
+        Named `raw_text` rather than `content` to avoid shadowing Textual's
+        read/write `Static.content` property (backed by a mangled attribute);
+        overriding it getter-only would make `self.content = ...` raise.
+        """
+        return self._content
+
+    @property
+    def media_snapshot(self) -> MediaTracker | None:
+        """Media tracker state captured when the message was submitted."""
+        return self._media_snapshot
 
     def set_cancelled(self) -> None:
         """Dim the message to mark its turn as interrupted by the user."""

libs/code/tests/unit_tests/test_app.py

@@ -58,6 +58,7 @@ from deepagents_code.app import (
     _warn_discarded_goal_channels,
 )
 from deepagents_code.event_bus import ExternalEvent
+from deepagents_code.media_utils import ImageData
 from deepagents_code.tui.widgets.ask_user import AskUserTextArea
 from deepagents_code.tui.widgets.chat_input import ChatInput
 from deepagents_code.tui.widgets.goal_review import GoalReviewMenu, GoalReviewResult
@@ -2377,7 +2378,7 @@ class TestCtrlCCopySelection:
             assert app._quit_pending is False
 
     async def test_ctrl_c_interrupt_marks_active_user_message_cancelled(self) -> None:
-        """Ctrl+C dims the prompt for the interrupted turn."""
+        """Ctrl+C dims the prompt but, unlike Esc, does not restore it."""
         app = DeepAgentsApp()
         async with app.run_test() as pilot:
             await pilot.pause()
@@ -2391,13 +2392,24 @@ class TestCtrlCCopySelection:
             mock_worker = MagicMock()
             app._agent_worker = mock_worker
             app._active_user_message = user_message
+            chat = app._chat_input
+            assert chat is not None
+            chat.value = ""
 
-            app.action_quit_or_interrupt()
+            with patch.object(app, "notify") as mock_notify:
+                app.action_quit_or_interrupt()
             await pilot.pause()
 
             assert user_message.has_class("-cancelled")
             mock_worker.cancel.assert_called_once()
             assert app._quit_pending is False
+            # Ctrl+C is the quit/copy flow: the prompt must NOT be restored to
+            # the input (that behavior is exclusive to the Esc path).
+            assert chat.value == ""
+            assert not any(
+                call.args and call.args[0] == "Message restored to input"
+                for call in mock_notify.call_args_list
+            )
 
     async def test_ctrl_c_non_input_focus_falls_through(self) -> None:
         """Ctrl+C with a non-Input/TextArea widget focused never copies."""
@@ -3485,6 +3497,168 @@ class TestMessageQueue:
                     "Queued message discarded (input not empty)", timeout=3
                 )
 
+    async def test_escape_restores_interrupted_message_to_empty_input(self) -> None:
+        """Interrupting the agent returns the prompt to an empty chat input."""
+        app = DeepAgentsApp()
+        worker = MagicMock()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._agent_running = True
+            app._agent_worker = worker
+            active = UserMessage("do the thing")
+            app._active_user_message = active
+            chat = app._chat_input
+            assert chat is not None
+            chat.value = ""
+
+            with patch.object(app, "notify") as mock_notify:
+                app.action_interrupt()
+
+            assert chat.value == "do the thing"
+            worker.cancel.assert_called_once()
+            assert active.has_class("-cancelled")
+            mock_notify.assert_called_once_with("Message restored to input", timeout=2)
+
+    async def test_escape_restores_interrupted_message_media(self) -> None:
+        """Restored multimodal prompts keep their backing media attachments."""
+        app = DeepAgentsApp()
+        worker = MagicMock()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._agent_running = True
+            app._agent_worker = worker
+            image = ImageData(
+                base64_data="abc123",
+                format="png",
+                placeholder="",
+            )
+            placeholder = app._image_tracker.add_image(image)
+            app._image_tracker.sync_to_text(f"describe {placeholder}")
+            with patch.object(
+                app, "_send_to_agent", new_callable=AsyncMock
+            ) as mock_send:
+                await app._handle_user_message(f"describe {placeholder}")
+
+            mock_send.assert_awaited_once_with("describe [image 1]")
+            active = app._active_user_message
+            assert active is not None
+            app._image_tracker.clear()
+            chat = app._chat_input
+            assert chat is not None
+            chat.value = ""
+
+            with patch.object(app, "notify") as mock_notify:
+                app.action_interrupt()
+
+            images = app._image_tracker.get_images()
+            assert chat.value == "describe [image 1]"
+            assert len(images) == 1
+            assert images[0].base64_data == "abc123"
+            assert images[0].placeholder == "[image 1]"
+            worker.cancel.assert_called_once()
+            assert active.has_class("-cancelled")
+            mock_notify.assert_called_once_with("Message restored to input", timeout=2)
+
+    async def test_escape_interrupt_keeps_existing_input_draft(self) -> None:
+        """A non-empty draft is preserved when the agent is interrupted."""
+        app = DeepAgentsApp()
+        worker = MagicMock()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._agent_running = True
+            app._agent_worker = worker
+            app._active_user_message = UserMessage("in flight")
+            chat = app._chat_input
+            assert chat is not None
+            chat.value = "draft text"
+
+            with patch.object(app, "notify") as mock_notify:
+                app.action_interrupt()
+
+            assert chat.value == "draft text"
+            worker.cancel.assert_called_once()
+            mock_notify.assert_not_called()
+
+    async def test_escape_interrupt_whitespace_only_input_is_restored(self) -> None:
+        """A whitespace-only draft counts as empty, so the prompt is restored."""
+        app = DeepAgentsApp()
+        worker = MagicMock()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._agent_running = True
+            app._agent_worker = worker
+            app._active_user_message = UserMessage("do the thing")
+            chat = app._chat_input
+            assert chat is not None
+            chat.value = "   "
+
+            with patch.object(app, "notify") as mock_notify:
+                app.action_interrupt()
+
+            assert chat.value == "do the thing"
+            worker.cancel.assert_called_once()
+            mock_notify.assert_called_once_with("Message restored to input", timeout=2)
+
+    async def test_escape_interrupt_without_active_message_cancels_only(self) -> None:
+        """Interrupting with no tracked prompt cancels without restoring."""
+        app = DeepAgentsApp()
+        worker = MagicMock()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._agent_running = True
+            app._agent_worker = worker
+            app._active_user_message = None
+            chat = app._chat_input
+            assert chat is not None
+            chat.value = ""
+
+            with patch.object(app, "notify") as mock_notify:
+                app.action_interrupt()
+
+            assert chat.value == ""
+            worker.cancel.assert_called_once()
+            mock_notify.assert_not_called()
+
+    async def test_escape_drains_queue_before_restoring_interrupted_prompt(
+        self,
+    ) -> None:
+        """ESC pops queued messages first; the queued text wins the input.
+
+        Branch 7 (queue pop) runs before branch 8 (interrupt), so the first
+        ESC restores the *queued* text — not the in-flight prompt — and does
+        not cancel. The queued text then occupies the input, so the second
+        ESC interrupts without overwriting it with the active prompt.
+        """
+        app = DeepAgentsApp()
+        worker = MagicMock()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._agent_running = True
+            app._agent_worker = worker
+            active = UserMessage("in flight")
+            app._active_user_message = active
+            app._pending_messages.append(QueuedMessage(text="queued", mode="normal"))
+            app._queued_widgets.append(MagicMock())
+            chat = app._chat_input
+            assert chat is not None
+            chat.value = ""
+
+            # First ESC: pop the queued message into the empty input; the
+            # in-flight prompt is untouched and the worker is not cancelled.
+            app.action_interrupt()
+            assert chat.value == "queued"
+            assert len(app._pending_messages) == 0
+            worker.cancel.assert_not_called()
+            assert not active.has_class("-cancelled")
+
+            # Second ESC: queue drained and the input now holds the queued
+            # text, so the agent is interrupted but the active prompt is NOT
+            # restored over the existing draft.
+            app.action_interrupt()
+            assert chat.value == "queued"
+            worker.cancel.assert_called_once()
+            assert active.has_class("-cancelled")
+
     async def test_escape_pop_single_then_interrupt(self) -> None:
         """Single queued message is popped, then next ESC interrupts agent."""
         app = DeepAgentsApp()