fix(tui): let quit skip an interrupted run's unfinished step (#7326)

* fix(tui): let quit skip an interrupted run's unfinished step

After Ctrl+C the synchronous agent stream keeps running until its next
event. The run was a Textual thread worker, i.e. a thread on asyncio's
default executor, which App.run() joins on shutdown (up to 300 s on
3.12) and interpreter exit joins again with no timeout. Quitting while
a long step was still running therefore hung silently after the
terminal had been restored, and further Ctrl+C presses only printed a
traceback.

Run workers on a daemon thread instead. Quit now waits visibly for
running workers (an interrupted step still checkpoints, a finished run
still saves its title) and refuses sends meanwhile; a second quit
request exits at once. Report unexpected worker exceptions as an error
row and end the run, since Textual no longer catches them.

* docs(changelog): reference #7326 for the TUI quit-after-interrupt fix

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -1397,6 +1397,13 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**.
 
 ### Fixed
 
+- **tui:** Quitting after `Ctrl+C` interrupted a run no longer hangs on the
+  step that run is still finishing. The run thread was joined on exit, so the
+  TUI closed but the shell prompt stayed blocked until a long tool call or model
+  request returned, and further `Ctrl+C` only printed a traceback. Quitting now
+  says it is waiting for that step and closes when it finishes; press `Ctrl+C`
+  again to quit at once. ([#7326])
+  
 - **tui:** The streamed transcript no longer drops tokens that repeat or extend
   the text already shown. Deltas were merged by prefix, so `ha` + `ha` rendered
   as `ha`, `1` + `10` as `10`, and an answer that opened with a Markdown marker
@@ -13027,3 +13034,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7258]: https://github.com/bytedance/deer-flow/pull/7258
 [#7277]: https://github.com/bytedance/deer-flow/pull/7277
 [#7325]: https://github.com/bytedance/deer-flow/pull/7325
+[#7326]: https://github.com/bytedance/deer-flow/pull/7326

CHANGELOG_zh.md

@@ -1112,6 +1112,11 @@
 
 ### 修复
 
+- **TUI：** 按 `Ctrl+C` 中断运行后再退出，不再卡在该运行仍在完成的步骤上。此前运行
+  线程会在退出时被等待，TUI 已关闭但 shell 提示符一直阻塞，直到较长的工具调用或模型请求
+  返回，再按 `Ctrl+C` 也只会打印 traceback。现在退出时会提示正在等待该步骤，并在其完成后
+  关闭；再按一次 `Ctrl+C` 即可立即退出。([#7326])
+  
 - **TUI：** 流式对话记录不再丢弃与已显示文本重复或在其基础上延伸的 token。此前增量
   按前缀合并，`ha` + `ha` 显示为 `ha`，`1` + `10` 显示为 `10`，以 Markdown 标记开头的
   回答还会丢失之后单独出现的同一标记（闭合的 `**`、列表符号、表格竖线、闭合代码块）。
@@ -10399,3 +10404,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7258]: https://github.com/bytedance/deer-flow/pull/7258
 [#7277]: https://github.com/bytedance/deer-flow/pull/7277
 [#7325]: https://github.com/bytedance/deer-flow/pull/7325
+[#7326]: https://github.com/bytedance/deer-flow/pull/7326

README.md

@@ -3610,7 +3610,8 @@ After an interrupt, late stream actions from the previous run cannot change the
 next run's display or status, even when both prompts use the same conversation.
 A tool call already in progress still finishes, so until it does a new prompt in
 that conversation is held back with a notice instead of racing it; `/new` and
-`/resume` stay available.
+`/resume` stay available. Quitting waits for that step as well and closes the TUI
+once it finishes; press `Ctrl+C` again to quit at once without saving it.
 If a run cannot start, the TUI reports an error and returns to idle so you can retry.
 A message that is held back or cannot start stays in the composer, so `Enter` resends it.
 

README_ko.md

@@ -1927,7 +1927,7 @@ deerflow --recursion-limit 250 --print "task" # override the headless agent-loop
 
 스트리밍 대화록(Markdown으로 렌더링된 답변), 간결한 도구 활동 카드, `/` 슬래시 명령 팔레트, 화면 전용 `/clear`, `/goal` 목표 관리, `/model` 및 `/threads` 선택기, 입력 기록, PageUp/PageDown 대화록 탐색, `Esc` / `Ctrl+C` 중단을 갖춘 키보드 중심의 채팅 화면입니다. 입력창(composer)은 붙여 넣은 코드, 스택 트레이스, 여러 문단으로 된 프롬프트의 줄바꿈과 들여쓰기를 보존하며, `Enter`를 누르면 전체 문서를 보냅니다. 대화록이 갱신될 때 위로 스크롤한 상태라면 읽던 위치를 유지하고, 맨 아래로 돌아오면 다시 새 출력을 따라갑니다. `/clear`는 스레드나 저장된 대화를 삭제하지 않고 현재 터미널 화면에서 행만 제거합니다. `/new`와 `/clear`는 실행 중에는 진행 중인 화면 상태를 재설정하는 대신 기다려 달라고 안내합니다. TUI에서 연 세션은 웹 UI 사이드바에도 표시됩니다. TUI는 로컬 기본 사용자로 공유 스레드 저장소에 기록하므로, **Gateway를 실행하지 않고도** 터미널과 웹이 동기화된 상태를 유지합니다.
 
-실행 중에는 `/resume`, `/threads`, `/switch`가 대화를 전환하기 전에 기다려 달라고 안내합니다. 잘못된 `/resume` 참조는 TUI를 닫거나 현재 대화를 바꾸지 않고 오류만 표시합니다. 중단 후에는 두 프롬프트가 같은 대화를 사용하더라도 이전 실행에서 늦게 도착한 스트림 동작이 다음 실행의 화면이나 상태를 바꿀 수 없습니다. 이미 진행 중인 도구 호출은 끝까지 완료되므로, 완료될 때까지 그 대화의 새 프롬프트는 경쟁 상태를 피하도록 알림과 함께 보류되며, `/new`와 `/resume`은 계속 사용할 수 있습니다. 실행을 시작할 수 없으면 TUI가 오류를 보고하고 유휴 상태로 돌아가므로 재시도할 수 있습니다. 보류되었거나 시작할 수 없는 메시지는 입력창에 남아 있으므로 `Enter`로 다시 보낼 수 있습니다.
+실행 중에는 `/resume`, `/threads`, `/switch`가 대화를 전환하기 전에 기다려 달라고 안내합니다. 잘못된 `/resume` 참조는 TUI를 닫거나 현재 대화를 바꾸지 않고 오류만 표시합니다. 중단 후에는 두 프롬프트가 같은 대화를 사용하더라도 이전 실행에서 늦게 도착한 스트림 동작이 다음 실행의 화면이나 상태를 바꿀 수 없습니다. 이미 진행 중인 도구 호출은 끝까지 완료되므로, 완료될 때까지 그 대화의 새 프롬프트는 경쟁 상태를 피하도록 알림과 함께 보류되며, `/new`와 `/resume`은 계속 사용할 수 있습니다. 종료할 때도 그 단계가 끝나기를 기다렸다가 TUI를 닫으며, `Ctrl+C`를 한 번 더 누르면 그 단계를 저장하지 않고 바로 종료합니다. 실행을 시작할 수 없으면 TUI가 오류를 보고하고 유휴 상태로 돌아가므로 재시도할 수 있습니다. 보류되었거나 시작할 수 없는 메시지는 입력창에 남아 있으므로 `Enter`로 다시 보낼 수 있습니다.
 
 입력창의 마지막 행에서 `Down`은 입력 기록을 탐색하는 중이 아니면 보내지 않은 초안을 건드리지 않습니다. 기록을 불러온 뒤에는 앞으로 이동해 저장해 둔 초안을 복원합니다.
 

backend/docs/TUI.md

@@ -88,7 +88,7 @@ for trusted embedded CLI runs.
 | `PageUp` / `PageDown` | Scroll the transcript without moving focus from the composer |
 | `Tab` | Complete the highlighted command (adds a trailing space) |
 | `Esc` | Close the palette / overlay |
-| `Ctrl+C` | Interrupt the active run, or quit when idle |
+| `Ctrl+C` | Interrupt the active run, or quit when idle (again to skip waiting for an interrupted step) |
 | `Ctrl+L` | Redraw · `Ctrl+U` clear composer |
 
 At the last input row, `↓` preserves the current draft unless you are browsing
@@ -125,6 +125,9 @@ An interrupted run cannot be stopped mid-step, so a tool call already in progres
 same conversation shows a notice instead of starting a second run there — two runs
 on one thread would overwrite each other's history. Send again once it finishes,
 or use `/new` or `/resume` to continue in another conversation meanwhile.
+Quitting (`Ctrl+C` or `/quit`) also waits for that step, or for a finished run
+still saving its title, with a notice; the TUI closes as soon as it is done.
+Press `Ctrl+C` again to quit at once; the unfinished step is then not saved.
 If a run cannot start, the TUI reports an error and returns to idle so you can retry.
 
 Use `/goal <condition>` to set the active thread goal, `/goal` to show it, and

backend/packages/harness/deerflow/tui/AGENTS.md

@@ -48,9 +48,13 @@ order completed title writes, roll back a write already in progress, or force
 stop synchronous backend work.
 An interrupted worker keeps streaming until its next event, so a long tool step
 still checkpoints. Until that worker returns (`_Run.started`/`stopped`, set by the
-worker itself; Textual reports a cancelled thread worker finished at once), reject
-sends on its thread: whichever run checkpoints last would drop the other turn.
-Other threads stay usable; a worker cancelled before it started holds nothing.
+worker itself), reject sends on its thread: whichever run checkpoints last would
+drop the other turn. Other threads stay usable; a worker cancelled before it
+started holds nothing.
+Run workers are daemon threads, not Textual thread workers: asyncio's default
+executor is joined by `App.run()` and again at interpreter exit, so quitting hung
+until the step ended. Quit waits for running workers with a notice and refuses
+sends; a second quit request exits at once and drops the unfinished step.
 
 `InputHistory.down()` returns `None` when history navigation is inactive; the
 app must then leave the composer untouched, including its cursor and undo state.

backend/packages/harness/deerflow/tui/app.py

@@ -1,7 +1,7 @@
 """The Textual application — a terminal workbench over the embedded harness.
 
 The app keeps a single immutable :class:`ViewState` and re-renders it through the
-pure renderers. Agent runs execute on a worker *thread* (``DeerFlowClient.stream``
+pure renderers. Agent runs execute on a daemon thread (``DeerFlowClient.stream``
 is a synchronous generator); each yielded action is marshalled back onto the UI
 thread via ``call_from_thread`` and folded into the reducer.
 """
@@ -10,8 +10,7 @@ from __future__ import annotations
 
 import uuid
 from dataclasses import dataclass, field
-from functools import partial
-from threading import Event
+from threading import Event, Thread
 
 from textual.app import App, ComposeResult
 from textual.binding import Binding
@@ -28,6 +27,7 @@ from .render import render_header, render_status, render_transcript
 from .runtime import stream_actions
 from .theme import SYMBOLS, THEME
 from .view_state import (
+    AssistantError,
     ClearRows,
     RunEnded,
     RunStarted,
@@ -41,6 +41,7 @@ from .widgets.composer import ComposerInput
 
 _HELP_KEYS = "Keys:  Enter send · PgUp/PgDn scroll transcript · Ctrl+C interrupt or quit · Ctrl+L redraw · / commands · Esc close overlay"
 _HELP_TEXT = f"{format_command_help()}\n{_HELP_KEYS}"
+_QUIT_PENDING_TEXT = "Quitting once the previous run stops; its current step finishes first. Press Ctrl+C again to quit now."
 
 
 @dataclass(eq=False)
@@ -50,8 +51,8 @@ class _Run:
     thread_id: str
     cancelled: Event = field(default_factory=Event)
     finished: bool = False
-    # Set by the worker thread itself. Textual marks a cancelled thread worker
-    # finished at once, while the synchronous agent stream keeps running.
+    # Set by the worker thread itself: the synchronous agent stream cannot be
+    # stopped, so it keeps running after an interrupt until its next event.
     started: Event = field(default_factory=Event)
     stopped: Event = field(default_factory=Event)
 
@@ -211,6 +212,7 @@ class DeerFlowTUI(App):
         self._streaming = False
         self._run: _Run | None = None
         self._interrupted_runs: list[_Run] = []
+        self._quit_pending = False
         self._skills_meta: list[dict] = []
         self._model_override: str | None = None
         self._palette_open = False
@@ -466,7 +468,7 @@ class DeerFlowTUI(App):
             # post-run persistence, e.g. thread title) is quietly abandoned.
             if self._streaming:
                 self._interrupt_run()
-            self.exit()
+            self._request_quit()
         elif name == "help":
             self._dispatch(SystemMessage(_HELP_TEXT))
         elif name == "new":
@@ -680,6 +682,9 @@ class DeerFlowTUI(App):
         if self._streaming:
             self._dispatch_still_working()
             return False
+        if self._quit_pending:
+            self._dispatch(SystemMessage(_QUIT_PENDING_TEXT, tone="info"))
+            return False
         # An interrupted worker cannot be killed: it stops at its next stream
         # event, so a long tool call keeps running and then checkpoints. A new
         # run on the same thread would race it, and whichever checkpoint lands
@@ -695,12 +700,7 @@ class DeerFlowTUI(App):
         self._streaming = True
         self._dispatch(UserSubmitted(text))
         try:
-            self.run_worker(
-                partial(self._stream_worker, text, run),
-                thread=True,
-                exclusive=True,
-                group="agent",
-            )
+            self._start_run_thread(text, run)
         except Exception:  # noqa: BLE001 - worker creation must release the busy reservation
             run.cancelled.set()
             self._run = None
@@ -709,12 +709,24 @@ class DeerFlowTUI(App):
             return False
         return True
 
+    def _start_run_thread(self, text: str, run: _Run) -> None:
+        # Not a Textual thread worker: those run on asyncio's default executor,
+        # which App.run() and interpreter exit both join, so quitting would hang
+        # until an interrupted step ended. Quit waits for running workers itself.
+        Thread(target=self._stream_worker, args=(text, run), name="deerflow-tui-run", daemon=True).start()
+
     def _stream_worker(self, text: str, run: _Run) -> None:
         # Mark the start before checking cancellation: an interrupt either
         # stops this worker here or sees it started and waits for it.
         run.started.set()
         try:
             self._stream_run(text, run)
+        except Exception as exc:  # noqa: BLE001 - no Textual worker reports a failure of this thread
+            try:
+                if self.call_from_thread(self._on_stream_action, run, AssistantError(str(exc) or exc.__class__.__name__)):
+                    self.call_from_thread(self._on_stream_action, run, RunEnded())
+            except Exception:  # noqa: BLE001 - the app may already have exited
+                pass
         finally:
             run.stopped.set()
 
@@ -783,7 +795,7 @@ class DeerFlowTUI(App):
         if self._streaming:
             self._interrupt_run()
         else:
-            self.exit()
+            self._request_quit()
 
     def action_escape(self) -> None:
         if self._palette_open:
@@ -795,11 +807,29 @@ class DeerFlowTUI(App):
         if self._run is not None:
             self._run.cancelled.set()
             self._interrupted_runs.append(self._run)
-        self.workers.cancel_group(self, "agent")
         self._streaming = False
         self.state = reduce(self.state, RunEnded())
         self._dispatch(SystemMessage("Interrupted.", tone="info"))
 
+    def _running_workers(self) -> list[_Run]:
+        runs = [*self._interrupted_runs, *([self._run] if self._run is not None else [])]
+        return [run for run in runs if run.worker_running()]
+
+    def _request_quit(self) -> None:
+        # An interrupted worker finishes its current step (and checkpoints it);
+        # a completed one may still be writing its title. Exit once they return,
+        # unless a second quit request asks to leave now and drop that work.
+        if self._quit_pending or not self._running_workers():
+            self.exit()
+            return
+        self._quit_pending = True
+        self._dispatch(SystemMessage(_QUIT_PENDING_TEXT, tone="info"))
+        self.set_interval(0.1, self._exit_once_workers_stop)
+
+    def _exit_once_workers_stop(self) -> None:
+        if not self._running_workers():
+            self.exit()
+
     def action_redraw(self) -> None:
         self.refresh(layout=True)
         self._refresh_all()

backend/tests/test_tui_app.py

@@ -6,6 +6,7 @@ loop: keypress -> submit -> worker thread -> stream_actions -> reducer -> state.
 
 import asyncio
 import threading
+import time
 
 import pytest
 from textual import events
@@ -363,29 +364,32 @@ async def test_quit_interrupts_an_active_stream_before_exiting():
     which checks ``self._streaming`` and interrupts before any teardown.
     Left unfixed, the worker thread survives the app's exit; its next
     ``call_from_thread`` call then fails silently and the in-flight turn is
-    abandoned without a trace.
+    abandoned without a trace. The interrupted worker is still inside its
+    step, so the app says it is waiting and exits once that step returns.
     """
     session = _BlockedSession()
     app = DeerFlowTUI(session, LaunchPlan(mode="tui"))
     async with app.run_test() as pilot:
-        await pilot.pause()
-        await pilot.press("h", "i")
-        await pilot.press("enter")
-        await _wait_until(lambda: app._streaming, pilot)
+        try:
+            await pilot.pause()
+            await pilot.press("h", "i")
+            await pilot.press("enter")
+            await _wait_until(lambda: app._streaming, pilot)
 
-        for ch in "/quit":
-            await pilot.press(ch)
-        await pilot.press("enter")
-        await pilot.pause()
+            for ch in "/quit":
+                await pilot.press(ch)
+            await pilot.press("enter")
+            await pilot.pause()
 
-        # The run must be interrupted (state cleaned up, message surfaced)
-        # before the app is allowed to exit — /quit still quits, but safely.
-        assert app._streaming is False
-        assert any(r.kind == "system" and "Interrupt" in r.text for r in app.state.rows)
-        assert app._exit is True
+            assert app._streaming is False
+            assert any(r.kind == "system" and "Interrupt" in r.text for r in app.state.rows)
+            assert any(r.kind == "system" and "Press Ctrl+C again to quit now" in r.text for r in app.state.rows)
+            assert app._exit is False
 
-    # Unblock the worker thread so it doesn't leak past the test.
-    session.client.release.set()
+            session.client.release.set()
+            assert await _wait_until(lambda: app._exit, pilot)
+        finally:
+            session.client.release.set()
 
 
 @pytest.mark.asyncio
@@ -409,6 +413,173 @@ async def test_ctrl_c_interrupts_an_active_stream_without_exiting():
     session.client.release.set()
 
 
+@pytest.mark.asyncio
+async def test_ctrl_c_quit_waits_for_the_interrupted_run_and_refuses_new_sends():
+    session = _BlockedSession()
+    app = DeerFlowTUI(session, LaunchPlan(mode="tui"))
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            await pilot.press("h", "i")
+            await pilot.press("enter")
+            await _wait_until(lambda: app._streaming, pilot)
+
+            await pilot.press("ctrl+c")  # interrupt
+            # Another thread is free to send to, but not while quitting: the
+            # exit would cut that new run off as soon as the old one stops.
+            app._handle_submit("/new")
+            await pilot.press("ctrl+c")  # quit
+            await pilot.pause()
+            assert app._exit is False
+            assert any(r.kind == "system" and "Press Ctrl+C again to quit now" in r.text for r in app.state.rows)
+
+            app.query_one("#composer").value = "one more"
+            await pilot.press("enter")
+            await pilot.pause()
+            assert not app._streaming
+            assert app.query_one("#composer").value == "one more"
+            assert not any(r.kind == "user" and r.text == "one more" for r in app.state.rows)
+
+            session.client.release.set()
+            assert await _wait_until(lambda: app._exit, pilot)
+        finally:
+            session.client.release.set()
+
+
+@pytest.mark.asyncio
+async def test_second_quit_request_exits_without_waiting_for_the_interrupted_run():
+    session = _BlockedSession()
+    app = DeerFlowTUI(session, LaunchPlan(mode="tui"))
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            await pilot.press("h", "i")
+            await pilot.press("enter")
+            await _wait_until(lambda: app._streaming, pilot)
+
+            await pilot.press("ctrl+c")  # interrupt
+            await pilot.press("ctrl+c")  # quit: waits
+            await pilot.press("ctrl+c")  # quit now
+            await pilot.pause()
+
+            assert app._exit is True
+            assert app._run.worker_running()
+        finally:
+            session.client.release.set()
+
+
+@pytest.mark.asyncio
+async def test_quit_waits_for_a_completed_run_still_saving_its_title():
+    class _BlockingTitleWriter:
+        def __init__(self):
+            self.writing = threading.Event()
+            self.release = threading.Event()
+            self.titles = []
+
+        def ensure_created(self, *args, **kwargs):
+            pass
+
+        def set_title(self, thread_id, title):
+            self.writing.set()
+            self.release.wait(5)
+            self.titles.append(title)
+
+    class _TitledClient(_FakeClient):
+        def stream(self, message, *, thread_id=None, **kwargs):
+            yield StreamEvent(type="values", data={"title": "Saved title"})
+            yield StreamEvent(type="end", data={"usage": {}})
+
+    session = _FakeSession()
+    session.client = _TitledClient()
+    session.writer = _BlockingTitleWriter()
+    app = DeerFlowTUI(session, LaunchPlan(mode="tui"))
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            await pilot.press("h", "i")
+            await pilot.press("enter")
+            assert await _wait_until(lambda: session.writer.writing.is_set() and not app._streaming, pilot)
+
+            await pilot.press("ctrl+c")
+            await pilot.pause()
+            assert app._exit is False
+
+            session.writer.release.set()
+            assert await _wait_until(lambda: app._exit, pilot)
+            assert session.writer.titles == ["Saved title"]
+        finally:
+            session.writer.release.set()
+
+
+@pytest.mark.asyncio
+async def test_unexpected_worker_failure_ends_the_run_with_an_error_row():
+    class _FailingWriter:
+        def ensure_created(self, *args, **kwargs):
+            raise RuntimeError("thread store exploded")
+
+    session = _FakeSession()
+    session.writer = _FailingWriter()
+    app = DeerFlowTUI(session, LaunchPlan(mode="tui"))
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        await pilot.press("h", "i")
+        await pilot.press("enter")
+        assert await _wait_until(lambda: not app._streaming, pilot)
+
+        assert any(r.kind == "assistant" and r.error and "thread store exploded" in r.text for r in app.state.rows)
+        assert app._run.stopped.is_set()
+        assert session.client.stream_calls == []
+
+
+_FORCE_QUIT_SCRIPT = """
+import threading, time
+from deerflow.client import StreamEvent
+from deerflow.tui.app import DeerFlowTUI
+from deerflow.tui.cli import LaunchPlan
+from deerflow.tui.session import Session
+
+in_step = threading.Event()
+
+class Client:
+    def list_models(self): return {"models": []}
+    def list_skills(self, **kwargs): return {"skills": []}
+    def stream(self, message, *, thread_id=None, **kwargs):
+        yield StreamEvent(type="messages-tuple", data={"type": "ai", "content": "working", "id": "m1"})
+        in_step.set()
+        time.sleep(60)  # a long tool step that emits no event
+        yield StreamEvent(type="end", data={"usage": {}})
+
+class App(DeerFlowTUI):
+    def on_mount(self):
+        super().on_mount()
+        self._send_to_agent("q")
+        self.set_interval(0.05, self._press_ctrl_c_once_in_step)
+
+    def _press_ctrl_c_once_in_step(self):
+        if in_step.is_set() and not self._exit:
+            for _ in range(3):  # interrupt, quit, quit now
+                self.action_interrupt()
+
+App(Session(client=Client(), writer=None), LaunchPlan(mode="tui", thread_id="t-1")).run(headless=True)
+print("exited")
+"""
+
+
+def test_quit_now_exits_the_process_while_the_interrupted_step_is_still_running():
+    """Drive the real ``App.run()``: a Textual thread worker would be joined by
+    asyncio's default-executor shutdown and again at interpreter exit, holding
+    the process until the 60 s step ended."""
+    import subprocess
+    import sys
+
+    started = time.monotonic()
+    result = subprocess.run([sys.executable, "-c", _FORCE_QUIT_SCRIPT], capture_output=True, text=True, timeout=45)
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip().endswith("exited")
+    assert time.monotonic() - started < 30
+
+
 @pytest.mark.asyncio
 async def test_tab_keeps_focus_on_composer_when_palette_closed():
     app = DeerFlowTUI(_FakeSession(), LaunchPlan(mode="tui"))

backend/tests/test_tui_interrupted_run_drain.py

@@ -163,17 +163,17 @@ async def test_a_stopping_run_does_not_block_another_thread():
 
 @pytest.mark.asyncio
 async def test_an_interrupt_before_the_worker_starts_does_not_block_the_thread(monkeypatch):
-    # Textual may cancel a queued thread worker before its function runs; that
-    # run never touches the backend and must not hold its thread.
+    # A run interrupted before its worker thread runs never touches the
+    # backend and must not hold its thread.
     client = _GraphClient()
     app = DeerFlowTUI(Session(client=client, writer=None), LaunchPlan(mode="tui", thread_id="thread-a"))
     async with app.run_test() as pilot:
         await pilot.pause()
-        original_run_worker = app.run_worker
-        monkeypatch.setattr(app, "run_worker", lambda *args, **kwargs: None)
+        original_start = app._start_run_thread
+        monkeypatch.setattr(app, "_start_run_thread", lambda *args, **kwargs: None)
         await _submit(app, pilot, "never started")
         await pilot.press("ctrl+c")
-        monkeypatch.setattr(app, "run_worker", original_run_worker)
+        monkeypatch.setattr(app, "_start_run_thread", original_start)
 
         await _submit(app, pilot, "new question")
         await _settle(pilot, lambda: "new question" in client.finished and not app._streaming)

backend/tests/test_tui_run_isolation.py

@@ -231,15 +231,15 @@ async def test_worker_start_failure_restores_idle_state_and_allows_retry(monkeyp
     client.release_new.set()
     app = DeerFlowTUI(Session(client=client, writer=writer), LaunchPlan(mode="tui", thread_id="thread-a"))
     failed_runs = []
-    original_start = app.run_worker
+    original_start = app._start_run_thread
 
     def fail_first_agent_worker(*args, **kwargs):
-        if kwargs.get("group") == "agent" and not failed_runs:
+        if not failed_runs:
             failed_runs.append(app._run)
             raise RuntimeError("private worker startup detail")
         return original_start(*args, **kwargs)
 
-    monkeypatch.setattr(app, "run_worker", fail_first_agent_worker)
+    monkeypatch.setattr(app, "_start_run_thread", fail_first_agent_worker)
     async with app.run_test() as pilot:
         app.query_one("#composer").value = "old question"
         await pilot.press("enter")
@@ -273,14 +273,11 @@ async def test_worker_start_failure_restores_idle_state_and_allows_retry(monkeyp
 async def test_launch_message_that_fails_to_start_stays_in_the_composer(monkeypatch):
     client = _Client()
     app = DeerFlowTUI(Session(client=client), LaunchPlan(mode="tui", thread_id="thread-a", message="launch question"))
-    original_start = app.run_worker
 
     def fail_agent_worker(*args, **kwargs):
-        if kwargs.get("group") == "agent":
-            raise RuntimeError("Cannot start the launch worker")
-        return original_start(*args, **kwargs)
+        raise RuntimeError("Cannot start the launch worker")
 
-    monkeypatch.setattr(app, "run_worker", fail_agent_worker)
+    monkeypatch.setattr(app, "_start_run_thread", fail_agent_worker)
     async with app.run_test() as pilot:
         await pilot.pause()
         assert not app._streaming
@@ -360,7 +357,7 @@ async def test_completed_run_keeps_title_write_after_ui_moves_on(monkeypatch, ne
                 def fail_to_start(*args, **kwargs):
                     raise RuntimeError("Cannot start the next worker")
 
-                monkeypatch.setattr(app, "run_worker", fail_to_start)
+                monkeypatch.setattr(app, "_start_run_thread", fail_to_start)
             app.query_one("#composer").value = "/resume thread-b" if next_action == "switch" else "new question"
             await pilot.press("enter")
             if next_action == "failed-send":