fix(tui): guard conversation switching and invalid resume references (#6249)

* fix(tui): guard conversation switching and invalid resume references

* fix(tui): discard late stream actions after conversation switches

---------

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

README.md

@@ -2483,6 +2483,12 @@ Headless `--print` and `--json` exit with status `1` when the run fails, includi
 
 A keyboard-driven chat surface with a streaming transcript (Markdown-rendered answers), compact tool-activity cards, a `/` slash-command palette, display-only `/clear`, `/goal` goal management, `/model` and `/threads` pickers, input history, PageUp/PageDown transcript navigation, and `Esc` / `Ctrl+C` interrupt. The composer preserves line breaks and indentation in pasted code, stack traces, and multi-paragraph prompts; `Enter` sends the complete document. Transcript refreshes preserve your reading position after you scroll upward and resume following new output when you return to the bottom. `/clear` removes rows from the current terminal display without deleting the thread or its persisted conversation; `/new` and `/clear` ask you to wait during an active run instead of resetting in-flight display state. Sessions opened in the TUI also appear in the Web UI sidebar — it writes the shared thread store under the local default user, so terminal and web stay in sync **without running the Gateway**.
 
+During an active run, `/resume`, `/threads`, and `/switch` ask you to wait before
+switching conversations. An invalid `/resume` reference displays an error without
+closing the TUI or changing the current conversation.
+After an interrupt and conversation switch, late stream actions from the previous
+thread cannot change the new conversation's display or run state.
+
 At the last composer row, `Down` leaves an unsent draft untouched unless you are
 browsing input history; after recalling history, it moves forward to restore your
 saved draft.

backend/docs/TUI.md

@@ -100,12 +100,20 @@ return to the bottom with `PageDown`.
 
 ### Slash commands
 
-`/help` `/new` `/clear` `/goal` `/threads` (`/switch`) `/model` `/skills` `/tools`
+`/help` `/new` `/clear` `/goal` `/resume` `/threads` (`/switch`) `/model` `/skills` `/tools`
 `/mcp` `/memory` `/uploads` `/usage` `/config` `/quit`, plus
 `/<skill-name> task` to activate any enabled skill for the current turn (same
 semantics as elsewhere in DeerFlow). `/model` and `/threads` open modal pickers.
 
 `/clear` removes the current transcript rows from the terminal display only; it keeps the active thread and persisted conversation intact. During an active run, `/new` and `/clear` ask you to wait for the run to finish instead of resetting in-flight display state.
+
+`/resume <id-or-title>` switches conversations, while `/resume` without an argument
+opens the thread picker. During an active run, `/resume`, `/threads`, and `/switch`
+ask you to wait instead of switching away from in-flight output. An invalid
+`/resume` reference displays an error and leaves the current conversation intact.
+After an interrupt and conversation switch, late stream actions from the previous
+thread are discarded when they reach the UI.
+
 Use `/goal <condition>` to set the active thread goal, `/goal` to show it, and
 `/goal clear` to clear it.
 

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

@@ -13,6 +13,18 @@ A terminal-native UI over the embedded harness, exposed as the `deerflow` consol
 
 **Web UI visibility**: the Web UI lists threads from the `threads_meta` SQL table (user-scoped), not the checkpointer. `persistence.py` writes a `threads_meta` row under the default user (`"default"`) into the same DB the Gateway reads — via the harness-only `deerflow.persistence.engine.init_engine_from_config()` — so TUI sessions appear in the Web UI sidebar **without** running the Gateway. Best-effort: a no-op on the `memory` backend. All DB work runs on one long-lived background event loop (a SQLAlchemy async engine is bound to its creating loop).
 
+Thread switching must reject active runs both before resolving/opening a picker
+and when applying the picker callback, which may outlive the idle state in which
+it opened. Keep the current thread and transcript unchanged when rejecting a
+switch. Report invalid resume references as error rows without relaxing the
+canonical thread-id validation or terminating the app.
+Worker actions must carry their originating thread id into the UI callback.
+Check it against the displayed thread on delivery, not just before scheduling:
+an interrupt/switch can race with the worker's cancellation check. Drop actions
+for another thread before they reach the reducer or change streaming state.
+
+`InputHistory.down()` returns `None` when history navigation is inactive; the
+app must then leave the composer untouched, including its cursor and undo state.
 `InputHistory.up()` returns `None` when history is empty, and `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

@@ -495,6 +495,9 @@ class DeerFlowTUI(App):
         self.push_screen(SelectScreen("Select model", options), on_choice)
 
     def _open_thread_switcher(self) -> None:
+        if self._streaming:
+            self._dispatch_still_working()
+            return
         try:
             threads = self.session.recent_threads(limit=20)
         except Exception:  # noqa: BLE001
@@ -518,13 +521,24 @@ class DeerFlowTUI(App):
 
     def _resume_thread(self, ref: str) -> None:
         """/resume [id-or-title]: switch to a thread, or open the picker if blank."""
+        if self._streaming:
+            self._dispatch_still_working()
+            return
         ref = ref.strip()
         if not ref:
             self._open_thread_switcher()
             return
-        self._switch_to_thread(self.session.resolve_ref(ref))
+        try:
+            thread_id = self.session.resolve_ref(ref)
+        except ValueError as exc:
+            self._dispatch(SystemMessage(str(exc), tone="error"))
+            return
+        self._switch_to_thread(thread_id)
 
     def _switch_to_thread(self, thread_id: str) -> None:
+        if self._streaming:
+            self._dispatch_still_working()
+            return
         self._conv_thread_id = thread_id
         self.state = initial_state()
         self._dispatch(SystemMessage(f"Resumed thread {thread_id[:8]}."))
@@ -657,13 +671,20 @@ class DeerFlowTUI(App):
                 break
             if isinstance(action, ThreadTitle):
                 latest_title = action.title
-            self.call_from_thread(self._on_action, action)
+            self.call_from_thread(self._on_stream_action, thread_id, action)
 
         # Only persist a title for a run that completed normally — an interrupted
         # run may only have emitted the title middleware's first, truncated guess.
         if writer is not None and latest_title and not self._cancelled:
             writer.set_title(thread_id, latest_title)
 
+    def _on_stream_action(self, thread_id: str, action) -> None:
+        # Interrupt/switch may happen after the worker's cancellation check.
+        # Validate the destination when Textual delivers the action to the UI.
+        if thread_id != self._conv_thread_id:
+            return
+        self._on_action(action)
+
     def _on_action(self, action) -> None:
         self.state = reduce(self.state, action)
         if isinstance(action, RunStarted):

backend/tests/test_tui_thread_switching.py

@@ -0,0 +1,178 @@
+"""Conversation-switch boundaries through real Textual events and Session resolution."""
+
+import asyncio
+import threading
+
+import pytest
+
+from deerflow.client import StreamEvent
+from deerflow.tui.app import DeerFlowTUI, SelectScreen
+from deerflow.tui.cli import LaunchPlan
+from deerflow.tui.session import Session
+from deerflow.tui.view_state import AssistantDelta, RunEnded, RunStarted
+
+
+class _ControlledClient:
+    def __init__(self):
+        self.release = threading.Event()
+        self.calls = []
+
+    def list_models(self):
+        return {"models": []}
+
+    def list_skills(self, **kwargs):
+        return {"skills": []}
+
+    def list_threads(self, **kwargs):
+        return {"thread_list": [{"thread_id": "thread-b", "title": "Other thread"}, {"thread_id": "thread-a", "title": "Original thread"}]}
+
+    def stream(self, message, *, thread_id=None, **kwargs):
+        self.calls.append((message, thread_id))
+        yield StreamEvent(type="messages-tuple", data={"type": "ai", "content": "before ", "id": "answer"})
+        if not self.release.wait(10):
+            raise TimeoutError("Test worker was not released")
+        yield StreamEvent(type="messages-tuple", data={"type": "ai", "content": "after", "id": "answer"})
+        yield StreamEvent(type="end", data={})
+
+
+async def _settle(pilot, predicate):
+    for _ in range(100):
+        await pilot.pause()
+        if predicate():
+            return
+        await asyncio.sleep(0.01)
+    raise AssertionError("Expected TUI event did not arrive")
+
+
+@pytest.mark.asyncio
+async def test_interrupt_then_resume_discards_old_thread_delivery(monkeypatch):
+    client = _ControlledClient()
+    app = DeerFlowTUI(Session(client=client), LaunchPlan(mode="tui", thread_id="thread-a"))
+    ready = threading.Event()
+    deliver = threading.Event()
+    delivered = threading.Event()
+    original = app.call_from_thread
+
+    def paused_delivery(callback, *args):
+        action = args[-1]
+        if isinstance(action, AssistantDelta) and action.text == "after":
+            ready.set()
+            if not deliver.wait(10):
+                raise TimeoutError("UI delivery was not released")
+            result = original(callback, *args)
+            delivered.set()
+            return result
+        return original(callback, *args)
+
+    # Pause a real stream action after the worker's cancellation check but
+    # before Textual executes its callback on the UI thread.
+    monkeypatch.setattr(app, "call_from_thread", paused_delivery)
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            app.query_one("#composer").value = "original question"
+            await pilot.press("enter")
+            await _settle(pilot, lambda: app._streaming and any(row.kind == "assistant" for row in app.state.rows))
+            client.release.set()
+            await _settle(pilot, ready.is_set)
+
+            app.action_interrupt()
+            app.query_one("#composer").value = "/resume thread-b"
+            await pilot.press("enter")
+            await _settle(pilot, lambda: app._conv_thread_id == "thread-b")
+            deliver.set()
+            await _settle(pilot, delivered.is_set)
+            assert not any(row.kind == "assistant" for row in app.state.rows)
+        finally:
+            client.release.set()
+            deliver.set()
+            await pilot.pause()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("command", ["/resume thread-b", "/resume", "/threads", "/switch"])
+async def test_active_run_rejects_thread_switch_commands(command):
+    client = _ControlledClient()
+    app = DeerFlowTUI(Session(client=client), LaunchPlan(mode="tui", thread_id="thread-a"))
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            app.query_one("#composer").value = "original question"
+            await pilot.press("enter")
+            await _settle(pilot, lambda: app._streaming and any(row.kind == "assistant" for row in app.state.rows))
+
+            app.query_one("#composer").value = command
+            await pilot.press("enter")
+            await pilot.pause()
+            assert not isinstance(app.screen, SelectScreen)
+            assert app._conv_thread_id == "thread-a"
+            assert any(row.kind == "system" and "Still working" in row.text for row in app.state.rows)
+            assert any(row.kind == "user" and row.text == "original question" for row in app.state.rows)
+
+            client.release.set()
+            await _settle(pilot, lambda: not app._streaming)
+            assert client.calls == [("original question", "thread-a")]
+            assert [row.text for row in app.state.rows if row.kind == "assistant"] == ["before after"]
+
+            # The same command must remain available once the run has ended.
+            app.query_one("#composer").value = command
+            await pilot.press("enter")
+            await pilot.pause()
+            if isinstance(app.screen, SelectScreen):
+                await pilot.press("enter")
+            await _settle(pilot, lambda: app._conv_thread_id == "thread-b")
+        finally:
+            client.release.set()
+            await _settle(pilot, lambda: not app._streaming)
+
+
+@pytest.mark.asyncio
+async def test_picker_callback_rechecks_run_state_before_switching():
+    client = _ControlledClient()
+    app = DeerFlowTUI(Session(client=client), LaunchPlan(mode="tui", thread_id="thread-a"))
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app._open_thread_switcher()
+        await pilot.pause()
+        assert isinstance(app.screen, SelectScreen)
+        # A pending picker may have been opened before RunStarted reaches the UI.
+        app._on_action(RunStarted())
+        await pilot.press("enter")
+        await pilot.pause()
+        assert app._conv_thread_id == "thread-a"
+        assert any(row.kind == "system" and "Still working" in row.text for row in app.state.rows)
+        app._on_action(RunEnded())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("ref", ["nonexistent title with spaces", "../outside", "x" * 65])
+async def test_invalid_resume_shows_error_and_keeps_app_usable(ref):
+    client = _ControlledClient()
+    client.release.set()
+    app = DeerFlowTUI(Session(client=client), LaunchPlan(mode="tui", thread_id="thread-a"))
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.query_one("#composer").value = f"/resume {ref}"
+        await pilot.press("enter")
+        await pilot.pause()
+        assert app._conv_thread_id == "thread-a"
+        assert any(row.kind == "system" and row.tone == "error" and "not a valid thread id" in row.text for row in app.state.rows)
+
+        app.query_one("#composer").value = "still usable"
+        await pilot.press("enter")
+        await _settle(pilot, lambda: client.calls and not app._streaming)
+        assert client.calls == [("still usable", "thread-a")]
+        assert any(row.kind == "assistant" and row.text == "before after" for row in app.state.rows)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("ref", ["Other thread", "thread-b", "new-valid-id"])
+async def test_idle_resume_preserves_title_id_and_new_namespace_resolution(ref):
+    app = DeerFlowTUI(Session(client=_ControlledClient()), LaunchPlan(mode="tui", thread_id="thread-a"))
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.query_one("#composer").value = f"/resume {ref}"
+        await pilot.press("enter")
+        expected = "new-valid-id" if ref == "new-valid-id" else "thread-b"
+        await _settle(pilot, lambda: app._conv_thread_id == expected)
+        assert any(row.kind == "system" and "Resumed thread" in row.text for row in app.state.rows)