fix(code): skip Esc prompt-restore once output generation begins (#4582)

Pressing `Esc` to interrupt an agent turn now only returns the prompt to
the chat input if the model has not yet started generating output; once
a response has begun streaming, the prompt is left as-is.

---

Follow-up to #4544. That change returns an interrupted prompt to the
chat input whenever the input is empty, but it does so even after the
model has already begun producing a response. Returning the prompt at
that point is confusing: the model already started (or finished)
answering, so re-offering the text invites an accidental re-submission
of an already-answered request.

This gates the restore on whether the turn has produced any output yet.
`execute_task_textual` now notifies the app (via a new
`on_output_started` adapter callback) the first time a turn emits
streamed text or a tool call. `DeepAgentsApp` tracks this per turn in
`_active_turn_output_started` (reset at the start of each send) and
`_restore_interrupted_message_to_input` skips the restore once it is
set. The empty-input guard and the Ctrl+C behavior are unchanged.

Made by [Open
SWE](https://openswe.vercel.app/agents/6a02202c-6d98-6070-b0e9-c7a658b2aa05)

---------

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

libs/code/deepagents_code/app.py

@@ -2523,6 +2523,12 @@ class DeepAgentsApp(App):
         """The `UserMessage` widget that started the in-flight turn, tracked so
         it can be dimmed if the turn is interrupted."""
 
+        self._active_turn_visible_output_started = False
+        """True once the current turn has displayed model text or a tool call.
+
+        Gates Esc prompt restore without counting hidden agent activity.
+        """
+
         self._active_tool_group: ToolGroupSummary | None = None
         """Open tool-group summary for the current step. Tools are folded into
         it as they stream and it is closed at the next step boundary."""
@@ -3304,6 +3310,7 @@ class DeepAgentsApp(App):
             on_auto_approve_enabled=self._on_auto_approve_enabled,
             set_spinner=self._set_spinner,
             set_active_message=self._set_active_message,
+            on_user_visible_output_started=self._on_user_visible_output_started,
             sync_message_content=self._sync_message_content,
             sync_tool_message=self._sync_tool_message_state,
             request_ask_user=self._request_ask_user,
@@ -10566,6 +10573,9 @@ class DeepAgentsApp(App):
         # Check if agent is available
         if self._agent and self._ui_adapter and self._session_state:
             self._agent_running = True
+            # Fresh turn: no model text or tool call is visible yet, so an Esc
+            # interrupt may still return this prompt to the input.
+            self._active_turn_visible_output_started = False
 
             # Flush any buffered non-incognito `!` shell output into thread
             # state so this turn's model sees commands run since the last turn.
@@ -11026,6 +11036,10 @@ class DeepAgentsApp(App):
         self._agent_running = False
         self._agent_worker = None
         self._active_user_message = None
+        # Clear the output-started gate alongside its lifecycle siblings so the
+        # "False at turn start" invariant holds locally, not just via the
+        # start-of-turn reset in `_send_to_agent`.
+        self._active_turn_visible_output_started = False
 
         # Remove spinner if present
         await self._set_spinner(None)
@@ -11961,6 +11975,15 @@ class DeepAgentsApp(App):
             for widget in collapsible:
                 widget.remove_class("-grouped")
 
+    def _on_user_visible_output_started(self) -> None:
+        """Record that the current turn has rendered model text or a tool call.
+
+        Hidden model and subagent activity does not call this. Once set, an Esc
+        interrupt no longer returns the prompt to the input because the user has
+        seen work produced from it.
+        """
+        self._active_turn_visible_output_started = True
+
     def _set_active_message(self, message_id: str | None) -> None:
         """Set the active streaming message (won't be pruned).
 
@@ -12095,7 +12118,14 @@ class DeepAgentsApp(App):
         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.
+
+        Restore is also skipped once model text or a tool call is visible for
+        the turn (`_active_turn_visible_output_started`). Returning the prompt
+        then would invite a confusing re-submission of a request that has already
+        produced user-visible work.
         """
+        if self._active_turn_visible_output_started:
+            return
         chat_input = self._chat_input
         if chat_input is None:
             logger.debug(
@@ -12438,7 +12468,8 @@ class DeepAgentsApp(App):
         6. If ask-user menu is active, cancel it
         7. If queued messages exist, pop the last one (LIFO)
         8. If agent is running, interrupt it (restoring the interrupted prompt
-           to the chat input when it is empty)
+           to the chat input when it is empty and no user-visible model output
+           — text or a tool call — has appeared yet for the turn)
         9. Otherwise, a second Esc clears the chat input draft (undoable)
         """
         from deepagents_code.tui.widgets.thread_selector import ThreadSelectorScreen

libs/code/deepagents_code/tui/textual_adapter.py

@@ -298,6 +298,7 @@ class TextualUIAdapter:
         on_auto_approve_enabled: Callable[[], Awaitable[None] | None] | None = None,
         set_spinner: Callable[[SpinnerStatus], Awaitable[None]] | None = None,
         set_active_message: Callable[[str | None], None] | None = None,
+        on_user_visible_output_started: Callable[[], None] | None = None,
         sync_message_content: Callable[[str, str], None] | None = None,
         sync_tool_message: Callable[[ToolCallMessage], None] | None = None,
         request_ask_user: (
@@ -334,6 +335,13 @@ class TextualUIAdapter:
         self._set_active_message = set_active_message
         """Callback to set the active streaming message ID (pass `None` to clear)."""
 
+        self._on_user_visible_output_started = on_user_visible_output_started
+        """Callback fired after the first model text or tool-call widget renders.
+
+        Hidden model and subagent output does not trigger it. A turn interrupted
+        before any user-visible model output produces zero firings.
+        """
+
         self._sync_message_content = sync_message_content
         """Callback to sync final message content back to the store after streaming."""
 
@@ -654,6 +662,33 @@ async def execute_task_textual(
         adapter._on_tokens_pending()
 
     file_op_tracker = FileOpTracker(assistant_id=assistant_id, backend=backend)
+    # Fires at most once per turn, after the first main-agent text or tool-call
+    # widget becomes visible, so hidden model activity cannot block prompt restore.
+    user_visible_output_started = False
+
+    def _notify_user_visible_output_started() -> None:
+        """Fire the output-started callback once, on the first visible output.
+
+        Call only from main-agent, post-filter paths: the "hidden output does
+        not count" guarantee lives in the placement of the call sites (all sit
+        after the subagent and summarization `continue`s), not in any check
+        here — this helper only dedupes.
+        """
+        nonlocal user_visible_output_started
+        if user_visible_output_started:
+            return
+        user_visible_output_started = True
+        if adapter._on_user_visible_output_started:
+            try:
+                adapter._on_user_visible_output_started()
+            except Exception:
+                # A prompt-restore gate update must never abort agent
+                # streaming — log and keep going (mirrors `_on_tool_complete`).
+                logger.warning(
+                    "on_user_visible_output_started callback failed",
+                    exc_info=True,
+                )
+
     displayed_tool_ids: set[str] = set()
     tool_call_buffers: dict[ToolCallBufferKey, ToolCallBuffer] = {}
     # Tool-call ids that already received terminal hooks before a resumed
@@ -853,6 +888,7 @@ async def execute_task_textual(
                                                     tool_id,
                                                 )
                                             else:
+                                                _notify_user_visible_output_started()
                                                 # Fire tool.use and latch the id
                                                 # together, only once the widget
                                                 # is mounted, so the "every
@@ -1192,6 +1228,7 @@ async def execute_task_textual(
                                 # streaming (uses MarkdownStream internally for
                                 # better performance)
                                 await current_msg.append_content(text)
+                                _notify_user_visible_output_started()
 
                         elif block_type in {"tool_call_chunk", "tool_call"}:
                             chunk_name = block.get("name")
@@ -1298,6 +1335,7 @@ async def execute_task_textual(
                                         buffer_id,
                                     )
                                 else:
+                                    _notify_user_visible_output_started()
                                     # Mark running so the group row reflects live
                                     # progress; the row itself is hidden inside
                                     # the group, so this drives state, not a

libs/code/tests/unit_tests/test_app.py

@@ -3512,6 +3512,8 @@ class TestMessageQueue:
             chat = app._chat_input
             assert chat is not None
             chat.value = ""
+            # No visible output yet, so the gate must not suppress restore.
+            assert app._active_turn_visible_output_started is False
 
             with patch.object(app, "notify") as mock_notify:
                 app.action_interrupt()
@@ -3621,6 +3623,122 @@ class TestMessageQueue:
             worker.cancel.assert_called_once()
             mock_notify.assert_not_called()
 
+    async def test_escape_after_visible_output_started_does_not_restore_prompt(
+        self,
+    ) -> None:
+        """Once output is visible, Esc interrupts without restoring the 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("do the thing")
+            app._active_user_message = active
+            # Simulate the adapter reporting the first streamed output.
+            app._on_user_visible_output_started()
+            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()
+            assert active.has_class("-cancelled")
+            mock_notify.assert_not_called()
+
+    async def test_ui_adapter_wires_visible_output_started_callback(self) -> None:
+        """The constructed adapter forwards output-started to the app handler.
+
+        Both halves of the gate are unit-tested in isolation; this pins the
+        seam at `_post_paint_init` so a dropped `on_user_visible_output_started`
+        kwarg cannot silently disable the feature while every other test stays
+        green.
+        """
+        app = DeepAgentsApp()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            adapter = app._ui_adapter
+            assert adapter is not None
+            assert (
+                adapter._on_user_visible_output_started
+                == app._on_user_visible_output_started
+            )
+
+    async def test_interrupt_restores_before_cancelling_worker(self) -> None:
+        """Restore reads the gate before the worker's cleanup can reset it.
+
+        `_cleanup_agent_task` resets `_active_turn_visible_output_started` on the
+        worker's teardown, so `_restore_interrupted_message_to_input` must run
+        before `_cancel_worker`. Pin that order in `action_interrupt` rather than
+        leave it holding only by construction (a mock worker never triggers
+        cleanup, so an ordering regression would otherwise pass unnoticed).
+        """
+        app = DeepAgentsApp()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._agent_running = True
+            app._agent_worker = MagicMock()
+            app._active_user_message = UserMessage("do the thing")
+
+            calls = MagicMock()
+            with (
+                patch.object(
+                    app, "_restore_interrupted_message_to_input", calls.restore
+                ),
+                patch.object(app, "_cancel_worker", calls.cancel),
+            ):
+                app.action_interrupt()
+
+            assert [call[0] for call in calls.mock_calls] == ["restore", "cancel"]
+
+    async def test_send_to_agent_resets_visible_output_started_flag(self) -> None:
+        """A fresh turn clears the output-started flag so Esc can restore again.
+
+        Without this reset the gate would be sticky: once any turn produced
+        output, every later turn's Esc-interrupt would stop restoring the
+        prompt. Closing the worker coroutine leaves the flag as `_send_to_agent`
+        set it, without running the turn.
+        """
+        app = DeepAgentsApp()
+        app._agent = MagicMock()
+        app._agent.aupdate_state = AsyncMock()
+        app._ui_adapter = MagicMock()
+        app._session_state = MagicMock()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            # A prior turn produced output.
+            app._on_user_visible_output_started()
+            assert app._active_turn_visible_output_started is True
+
+            with patch.object(app, "run_worker") as mock_rw:
+                mock_rw.return_value = MagicMock()
+                await app._send_to_agent("next question")
+                coro = mock_rw.call_args[0][0]
+                coro.close()
+
+            assert app._active_turn_visible_output_started is False
+
+    async def test_cleanup_agent_task_resets_visible_output_started_flag(self) -> None:
+        """Turn cleanup clears the output-started flag alongside its siblings.
+
+        Keeps the "False at turn start" invariant local rather than relying on
+        the start-of-turn reset being reached on every entry path.
+        """
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
+        app._process_next_from_queue = AsyncMock()  # ty: ignore
+        app._maybe_drain_deferred = AsyncMock()  # ty: ignore
+        app._set_spinner = AsyncMock()  # ty: ignore
+        app._schedule_git_branch_refresh = MagicMock()  # ty: ignore
+        app._on_user_visible_output_started()
+        assert app._active_turn_visible_output_started is True
+
+        await app._cleanup_agent_task()
+
+        assert app._active_turn_visible_output_started is False
+
     async def test_escape_drains_queue_before_restoring_interrupted_prompt(
         self,
     ) -> None:

libs/code/tests/unit_tests/tui/test_textual_adapter.py

@@ -127,6 +127,26 @@ class TestTextualUIAdapterInit:
         )
         assert adapter._on_tool_complete is callback
 
+    def test_on_user_visible_output_started_defaults_to_none_and_accepts_callback(
+        self,
+    ) -> None:
+        """Verify the user-visible-output callback can be assigned."""
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+        )
+        assert adapter._on_user_visible_output_started is None
+
+        callback = MagicMock()
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=callback,
+        )
+        assert adapter._on_user_visible_output_started is callback
+
     def test_set_token_callbacks(self) -> None:
         """Verify token callbacks can be assigned."""
         adapter = TextualUIAdapter(
@@ -2149,6 +2169,239 @@ def _text_message(text: str) -> SimpleNamespace:
     return SimpleNamespace(content_blocks=[{"type": "text", "text": text}])
 
 
+class TestExecuteTaskTextualUserVisibleOutputStarted:
+    """The callback fires once on the first output rendered for the user."""
+
+    async def test_fires_once_on_first_streamed_text(self) -> None:
+        """Streaming text triggers `on_user_visible_output_started` a single time."""
+        user_visible_output_started = MagicMock()
+        chunks = [
+            ((), "messages", (_text_message("hello"), {})),
+            ((), "messages", (_text_message(" world"), {})),
+        ]
+
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=_FakeAgent(chunks),
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_called_once_with()
+
+    async def test_fires_once_across_text_then_tool_call(self) -> None:
+        """Text followed by a tool call in one turn still fires exactly once."""
+        user_visible_output_started = MagicMock()
+        chunks = [
+            ((), "messages", (_text_message("thinking"), {})),
+            ((), "messages", (_tool_call_message("task", {"task": "a"}, "t-a"), {})),
+        ]
+
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=_FakeAgent(chunks),
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_called_once_with()
+
+    async def test_fires_on_first_tool_call_without_text(self) -> None:
+        """A turn that opens with a tool call still reports output started."""
+        user_visible_output_started = MagicMock()
+        chunks = [
+            ((), "messages", (_tool_call_message("task", {"task": "a"}, "t-a"), {})),
+        ]
+
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=_FakeAgent(chunks),
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_called_once_with()
+
+    async def test_fires_on_synthesized_ask_user_tool_call(self) -> None:
+        """An updates-only `ask_user` row reports visible output after mounting."""
+        user_visible_output_started = MagicMock()
+        future: asyncio.Future[AskUserWidgetResult] = asyncio.Future()
+        future.set_result({"type": "answered", "answers": ["Alice"]})
+
+        async def request_ask_user(
+            _questions: list[Question],
+        ) -> asyncio.Future[AskUserWidgetResult] | None:
+            await asyncio.sleep(0)
+            return future
+
+        questions: list[Question] = [{"question": "Name?", "type": "text"}]
+        agent = _SequencedAgent(
+            streams_by_call=[
+                [
+                    _ask_user_interrupt_chunk(
+                        {
+                            "type": "ask_user",
+                            "questions": questions,
+                            "tool_call_id": "ask-1",
+                        }
+                    )
+                ],
+                [],
+            ]
+        )
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            request_ask_user=request_ask_user,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=agent,
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=False),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_called_once_with()
+
+    async def test_not_fired_when_no_output_is_produced(self) -> None:
+        """A turn that streams no text or tool call never reports output."""
+        user_visible_output_started = MagicMock()
+
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=_FakeAgent([]),
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_not_called()
+
+    async def test_not_fired_for_subagent_output(self) -> None:
+        """Text and tool calls hidden in a subagent namespace do not count."""
+        user_visible_output_started = MagicMock()
+        chunks = [
+            (("subagent",), "messages", (_text_message("hidden"), {})),
+            (
+                ("subagent",),
+                "messages",
+                (_tool_call_message("read_file", {"path": "x"}, "t-a"), {}),
+            ),
+        ]
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=_FakeAgent(chunks),
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_not_called()
+
+    async def test_not_fired_for_hidden_summarization_output(self) -> None:
+        """Hidden main-namespace summarization text does not count."""
+        user_visible_output_started = MagicMock()
+        chunks = [
+            (
+                (),
+                "messages",
+                (_text_message("hidden summary"), {"lc_source": "summarization"}),
+            ),
+        ]
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=_FakeAgent(chunks),
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_not_called()
+
+    async def test_not_fired_when_tool_widget_does_not_mount(self) -> None:
+        """A tool call that never reaches the transcript does not count."""
+        user_visible_output_started = MagicMock()
+
+        async def fail_mount(_widget: object) -> None:
+            await asyncio.sleep(0)
+            msg = "mount failed"
+            raise RuntimeError(msg)
+
+        adapter = TextualUIAdapter(
+            mount_message=fail_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            on_user_visible_output_started=user_visible_output_started,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=_FakeAgent(
+                [
+                    (
+                        (),
+                        "messages",
+                        (_tool_call_message("read_file", {"path": "x"}, "t-a"), {}),
+                    )
+                ]
+            ),
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        user_visible_output_started.assert_not_called()
+
+
 class TestExecuteTaskTextualParallelToolSpinner:
     """Regression tests for #1796: premature spinner with parallel tools."""