fix(tui): wait for an interrupted run before sending to its thread (#6350)

* fix(tui): wait for an interrupted run before sending to its thread

Textual cannot stop a thread worker, so after an interrupt the agent
stream keeps running until its next event. A tool step already in
progress still completes and checkpoints while a new prompt runs on the
same thread; whichever run checkpoints last becomes the thread's state,
so the new turn could drop out of the conversation history.

Track when each worker actually starts and returns (Textual reports a
cancelled thread worker finished at once) and reject sends to a thread
whose interrupted worker is still running. Other threads stay usable,
and a worker cancelled before it started holds nothing.

* docs(changelog): reference #6350 for the interrupted TUI run fix

---------

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

CHANGELOG.md

@@ -623,6 +623,14 @@ This release closes that milestone with **439 merged pull requests**.
 
 ### Fixed
 
+- **tui:** A new prompt no longer races an interrupted run in the same
+  conversation. After `Ctrl+C` the run's worker keeps going until its current
+  step returns, so a tool call already in progress (a long shell command, for
+  example) still finished and checkpointed while the next prompt ran on the same
+  thread; whichever run checkpointed last became the conversation's history, so
+  the new turn could silently disappear from it. Until the interrupted worker
+  returns, a prompt in that conversation now shows a notice instead of starting a
+  second run; other conversations stay available through `/new` and `/resume`. ([#6350])
 - **sandbox:** With host bash enabled, the local sandbox no longer keeps a
   thread on the skill view of the last restricted Agent that ran there. That
   view is only maintained while host bash is off, but `LocalSandboxProvider`
@@ -8809,3 +8817,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6343]: https://github.com/bytedance/deer-flow/pull/6343
 [#6344]: https://github.com/bytedance/deer-flow/pull/6344
 [#6347]: https://github.com/bytedance/deer-flow/pull/6347
+[#6350]: https://github.com/bytedance/deer-flow/pull/6350

CHANGELOG_zh.md

@@ -558,6 +558,11 @@
 
 ### 修复
 
+- **TUI：** 新的提示不再与同一会话中被中断的运行发生竞争。按下 `Ctrl+C` 后，该运行的工作线程会
+  一直执行到当前步骤返回，因此正在进行的工具调用（例如耗时较长的 shell 命令）仍会完成并写入检查点，
+  而下一条提示已在同一线程上运行；最后写入检查点的运行会成为会话历史，新的一轮可能因此悄然消失。
+  现在在被中断的工作线程返回之前，在该会话中发送提示会显示提示信息而不会启动第二个运行；
+  其他会话仍可通过 `/new` 和 `/resume` 使用。([#6350])
 - **沙箱：** 启用 host bash 时，本地沙箱不再把线程固定在上一个受限 Agent 的技能视图上。
   该视图只在 host bash 关闭时维护，但 `LocalSandboxProvider` 只要它存在就会挂载，导致该线程之后
   不受限的运行仍沿用旧的 allowlist，读不到 `/mnt/skills` 下其他已启用或新增的技能。现在这些运行
@@ -7297,3 +7302,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6338]: https://github.com/bytedance/deer-flow/pull/6338
 [#6343]: https://github.com/bytedance/deer-flow/pull/6343
 [#6344]: https://github.com/bytedance/deer-flow/pull/6344
+[#6350]: https://github.com/bytedance/deer-flow/pull/6350

README.md

@@ -2570,6 +2570,9 @@ switching conversations. An invalid `/resume` reference displays an error withou
 closing the TUI or changing the current conversation.
 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.
 If a run cannot start, the TUI reports an error and returns to idle so you can retry.
 
 At the last composer row, `Down` leaves an unsent draft untouched unless you are

backend/docs/TUI.md

@@ -114,6 +114,11 @@ ask you to wait instead of switching away from in-flight output. An invalid
 After an interrupt, late stream actions from the previous run are discarded when
 they reach the UI, even if the next prompt uses the same conversation. Its final
 usage, title, and completion event cannot replace those of the new run.
+An interrupted run cannot be stopped mid-step, so a tool call already in progress
+(a long shell command, for example) finishes first. Until it does, sending to the
+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.
 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

@@ -29,6 +29,11 @@ uncancelled, normally completed run still writes its title to its original
 thread after the UI moves on; keep that write off the UI thread. This does not
 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.
 
 `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

@@ -50,6 +50,14 @@ 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.
+    started: Event = field(default_factory=Event)
+    stopped: Event = field(default_factory=Event)
+
+    def worker_running(self) -> bool:
+        """Whether this run's worker may still write to its thread."""
+        return self.started.is_set() and not self.stopped.is_set()
 
 
 _TRANSPARENT_CSS = """
@@ -202,6 +210,7 @@ class DeerFlowTUI(App):
         self._spinner_idx = 0
         self._streaming = False
         self._run: _Run | None = None
+        self._interrupted_runs: list[_Run] = []
         self._skills_meta: list[dict] = []
         self._model_override: str | None = None
         self._palette_open = False
@@ -653,6 +662,14 @@ class DeerFlowTUI(App):
         if self._streaming:
             self._dispatch_still_working()
             return
+        # 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
+        # last becomes the thread's state, dropping the other turn.
+        self._interrupted_runs = [run for run in self._interrupted_runs if run.worker_running()]
+        if any(run.thread_id == self._conv_thread_id for run in self._interrupted_runs):
+            self._dispatch(SystemMessage("The interrupted run is still stopping on this thread. Send again once it finishes, or use /new or /resume to continue elsewhere.", tone="info"))
+            return
         if self._conv_thread_id is None:
             self._conv_thread_id = str(uuid.uuid4())
         run = _Run(self._conv_thread_id)
@@ -673,6 +690,15 @@ class DeerFlowTUI(App):
             self._dispatch(SystemMessage("Could not start the run. Please try again.", tone="error"))
 
     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)
+        finally:
+            run.stopped.set()
+
+    def _stream_run(self, text: str, run: _Run) -> None:
         if run.cancelled.is_set():
             return
         thread_id = run.thread_id
@@ -748,6 +774,7 @@ class DeerFlowTUI(App):
     def _interrupt_run(self) -> None:
         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())

backend/tests/test_tui_interrupted_run_drain.py

@@ -0,0 +1,219 @@
+"""A new send must not race an interrupted run that is still writing its thread.
+
+Textual cannot kill a thread worker, so after Ctrl+C the synchronous agent
+stream keeps running until its next event. When it is inside a long tool call,
+that step still completes and checkpoints. A new run on the same thread started
+meanwhile forks from the older checkpoint, and whichever run checkpoints last
+becomes the thread's state, so the other turn silently drops out of it.
+
+These tests drive the real TUI through Textual's pilot with a client that
+streams a real compiled LangGraph graph backed by a checkpointer.
+"""
+
+import asyncio
+import operator
+import threading
+from typing import Annotated, TypedDict
+
+import pytest
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, StateGraph
+
+from deerflow.client import StreamEvent
+from deerflow.tui.app import DeerFlowTUI
+from deerflow.tui.cli import LaunchPlan
+from deerflow.tui.session import Session
+
+STILL_STOPPING = "The interrupted run is still stopping on this thread"
+
+
+class _State(TypedDict):
+    messages: Annotated[list, operator.add]
+
+
+class _GraphClient:
+    """Streams a real checkpointed graph; "old question" blocks in a tool step."""
+
+    def __init__(self):
+        self.old_in_tool = threading.Event()
+        self.release_old_tool = threading.Event()
+        self.calls: list[tuple[str, str | None]] = []
+        self.finished: set[str] = set()
+        self.active: dict[str | None, int] = {}
+        self.max_active_per_thread = 0
+        self._lock = threading.Lock()
+
+        def tool_step(state):
+            last = state["messages"][-1]
+            if last == "old question":
+                self.old_in_tool.set()
+                if not self.release_old_tool.wait(10):
+                    raise TimeoutError("old tool step was not released")
+                return {"messages": ["old tool result"]}
+            return {"messages": [f"answer to {last}"]}
+
+        graph = StateGraph(_State)
+        graph.add_node("tool_step", tool_step)
+        graph.add_edge(START, "tool_step")
+        graph.add_edge("tool_step", END)
+        self.graph = graph.compile(checkpointer=InMemorySaver())
+
+    def list_models(self):
+        return {"models": []}
+
+    def list_skills(self, **kwargs):
+        return {"skills": []}
+
+    def list_threads(self, **kwargs):
+        return {"thread_list": []}
+
+    def messages(self, thread_id: str) -> list:
+        return self.graph.get_state({"configurable": {"thread_id": thread_id}}).values.get("messages", [])
+
+    def stream(self, message, *, thread_id=None, **kwargs):
+        self.calls.append((message, thread_id))
+        with self._lock:
+            self.active[thread_id] = self.active.get(thread_id, 0) + 1
+            self.max_active_per_thread = max(self.max_active_per_thread, self.active[thread_id])
+        try:
+            for _ in self.graph.stream({"messages": [message]}, {"configurable": {"thread_id": thread_id}}, stream_mode="values"):
+                yield StreamEvent(type="values", data={})
+            yield StreamEvent(type="end", data={"usage": {}})
+        finally:
+            with self._lock:
+                self.active[thread_id] -= 1
+            self.finished.add(message)
+
+
+async def _settle(pilot, predicate):
+    for _ in range(500):
+        await pilot.pause()
+        if predicate():
+            return
+        await asyncio.sleep(0.01)
+    raise AssertionError("Expected controlled UI event did not arrive")
+
+
+async def _submit(app, pilot, text: str) -> None:
+    app.query_one("#composer").value = text
+    await pilot.press("enter")
+
+
+def _system_texts(app) -> list[str]:
+    return [row.text for row in app.state.rows if row.kind == "system"]
+
+
+async def _interrupt_old_run_in_its_tool_step(app, pilot, client) -> None:
+    await _submit(app, pilot, "old question")
+    await _settle(pilot, client.old_in_tool.is_set)
+    await pilot.press("ctrl+c")
+
+
+@pytest.mark.asyncio
+async def test_send_waits_for_the_interrupted_run_before_reusing_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:
+        try:
+            await pilot.pause()
+            await _interrupt_old_run_in_its_tool_step(app, pilot, client)
+
+            await _submit(app, pilot, "new question")
+            await pilot.pause()
+            assert any(STILL_STOPPING in text for text in _system_texts(app))
+            assert [message for message, _ in client.calls] == ["old question"]
+            assert not app._streaming
+
+            client.release_old_tool.set()
+            await _settle(pilot, lambda: "old question" in client.finished and app._run.stopped.is_set())
+            await _submit(app, pilot, "new question")
+            await _settle(pilot, lambda: "new question" in client.finished and not app._streaming)
+        finally:
+            client.release_old_tool.set()
+
+    assert client.max_active_per_thread == 1
+    assert client.messages(app._conv_thread_id) == ["old question", "old tool result", "new question", "answer to new question"]
+
+
+@pytest.mark.asyncio
+async def test_a_stopping_run_does_not_block_another_thread():
+    client = _GraphClient()
+    app = DeerFlowTUI(Session(client=client, writer=None), LaunchPlan(mode="tui", thread_id="thread-a"))
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            await _interrupt_old_run_in_its_tool_step(app, pilot, client)
+            old_thread = app._conv_thread_id
+
+            await _submit(app, pilot, "/new")
+            await _submit(app, pilot, "new question")
+            await _settle(pilot, lambda: "new question" in client.finished and not app._streaming)
+            new_thread = app._conv_thread_id
+
+            assert not any(STILL_STOPPING in text for text in _system_texts(app))
+            assert not client.release_old_tool.is_set()
+        finally:
+            client.release_old_tool.set()
+        await _settle(pilot, lambda: "old question" in client.finished)
+
+    assert new_thread != old_thread
+    assert client.messages(new_thread) == ["new question", "answer to new question"]
+
+
+@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.
+    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)
+        await _submit(app, pilot, "never started")
+        await pilot.press("ctrl+c")
+        monkeypatch.setattr(app, "run_worker", original_run_worker)
+
+        await _submit(app, pilot, "new question")
+        await _settle(pilot, lambda: "new question" in client.finished and not app._streaming)
+
+    assert not any(STILL_STOPPING in text for text in _system_texts(app))
+    assert [message for message, _ in client.calls] == ["new question"]
+
+
+@pytest.mark.asyncio
+async def test_a_completed_run_still_saving_its_title_does_not_block_the_next_send():
+    # Only interrupted runs are waited for: after a normal completion the agent
+    # stream is done and the worker only writes the thread title.
+    class _BlockingTitleWriter:
+        def __init__(self):
+            self.writing = threading.Event()
+            self.release = threading.Event()
+
+        def ensure_created(self, *args, **kwargs):
+            pass
+
+        def set_title(self, thread_id, title):
+            self.writing.set()
+            self.release.wait(10)
+
+    class _TitledClient(_GraphClient):
+        def stream(self, message, *, thread_id=None, **kwargs):
+            yield StreamEvent(type="values", data={"title": f"title for {message}"})
+            yield from super().stream(message, thread_id=thread_id, **kwargs)
+
+    client, writer = _TitledClient(), _BlockingTitleWriter()
+    app = DeerFlowTUI(Session(client=client, writer=writer), LaunchPlan(mode="tui", thread_id="thread-a"))
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            await _submit(app, pilot, "first question")
+            await _settle(pilot, writer.writing.is_set)
+            await _settle(pilot, lambda: not app._streaming)
+
+            await _submit(app, pilot, "second question")
+            await _settle(pilot, lambda: "second question" in client.finished)
+        finally:
+            writer.release.set()
+
+    assert not any(STILL_STOPPING in text for text in _system_texts(app))

backend/tests/test_tui_run_isolation.py

@@ -113,6 +113,18 @@ async def _late_delivery(monkeypatch, gate, switch_thread=False, restart=True):
                 if switch_thread:
                     app.query_one("#composer").value = "/resume thread-b"
                     await pilot.press("enter")
+                else:
+                    # The old worker is still running on this thread, so a
+                    # send there waits for it (see
+                    # test_tui_interrupted_run_drain.py); its late action is
+                    # still discarded once delivered.
+                    app.query_one("#composer").value = "new question"
+                    await pilot.press("enter")
+                    await pilot.pause()
+                    assert client.calls == [("old question", "thread-a")]
+                    assert any("still stopping" in row.text for row in app.state.rows if row.kind == "system")
+                    deliver.set()
+                    await _settle(pilot, old_done.is_set)
                 app.query_one("#composer").value = "new question"
                 await pilot.press("enter")
                 await _settle(pilot, lambda: client.new_started.is_set() and app._streaming)