diff --git a/CHANGELOG.md b/CHANGELOG.md index d69007c87d..7a36c59860 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 20e122a09a..067a400c5e 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/README.md b/README.md index becba0d30a..1ddda0ebc7 100644 --- a/README.md +++ b/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. diff --git a/README_ko.md b/README_ko.md index c5cdf4bd67..e7d8a586e5 100644 --- a/README_ko.md +++ b/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`은 입력 기록을 탐색하는 중이 아니면 보내지 않은 초안을 건드리지 않습니다. 기록을 불러온 뒤에는 앞으로 이동해 저장해 둔 초안을 복원합니다. diff --git a/backend/docs/TUI.md b/backend/docs/TUI.md index 86bbbdf686..a572263fe4 100644 --- a/backend/docs/TUI.md +++ b/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 ` to set the active thread goal, `/goal` to show it, and diff --git a/backend/packages/harness/deerflow/tui/AGENTS.md b/backend/packages/harness/deerflow/tui/AGENTS.md index 4cb37674a8..3c8f44d5ba 100644 --- a/backend/packages/harness/deerflow/tui/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/tui/app.py b/backend/packages/harness/deerflow/tui/app.py index 6904051d8d..394e594f5a 100644 --- a/backend/packages/harness/deerflow/tui/app.py +++ b/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() diff --git a/backend/tests/test_tui_app.py b/backend/tests/test_tui_app.py index 1d152f5ee0..abfaf71ce9 100644 --- a/backend/tests/test_tui_app.py +++ b/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")) diff --git a/backend/tests/test_tui_interrupted_run_drain.py b/backend/tests/test_tui_interrupted_run_drain.py index a2827f342e..051c9f6420 100644 --- a/backend/tests/test_tui_interrupted_run_drain.py +++ b/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) diff --git a/backend/tests/test_tui_run_isolation.py b/backend/tests/test_tui_run_isolation.py index 77d1452d9c..d0984d5219 100644 --- a/backend/tests/test_tui_run_isolation.py +++ b/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":