diff --git a/README.md b/README.md index 1cb2247ce6..ccc88f0ecf 100644 --- a/README.md +++ b/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. diff --git a/backend/docs/TUI.md b/backend/docs/TUI.md index 9565b9b671..a8337f6f12 100644 --- a/backend/docs/TUI.md +++ b/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 `/ 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 ` 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 ` to set the active thread goal, `/goal` to show it, and `/goal clear` to clear it. diff --git a/backend/packages/harness/deerflow/tui/AGENTS.md b/backend/packages/harness/deerflow/tui/AGENTS.md index 0ee1f17a8e..e721b9fd2c 100644 --- a/backend/packages/harness/deerflow/tui/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/tui/app.py b/backend/packages/harness/deerflow/tui/app.py index 10e39f446e..325735c0f6 100644 --- a/backend/packages/harness/deerflow/tui/app.py +++ b/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): diff --git a/backend/tests/test_tui_thread_switching.py b/backend/tests/test_tui_thread_switching.py new file mode 100644 index 0000000000..de334b67b2 --- /dev/null +++ b/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)