From 5b2063d8b93b2be870fa36ac1ad2ec8211c85549 Mon Sep 17 00:00:00 2001 From: EastonJiang <3465192960@qq.com> Date: Sat, 17 Oct 2026 19:25:18 +0800 Subject: [PATCH] fix(tui): cancel pending persistence work when the bridge times out (#6927) Co-authored-by: Jaxon1216 <224378083+Jaxon1216@users.noreply.github.com> Co-authored-by: Willem Jiang --- README.md | 3 + .../packages/harness/deerflow/tui/AGENTS.md | 5 + .../harness/deerflow/tui/persistence.py | 10 +- backend/tests/test_tui_persistence.py | 94 +++++++++++++++++++ 4 files changed, 111 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index bdc0f9c9e1..11f770982f 100644 --- a/README.md +++ b/README.md @@ -3296,6 +3296,9 @@ untouched when no input history is available. Recalling identical history text or a saved draft also preserves the cursor and undo history. +Thread visibility and title updates remain best-effort. If a database update times out, +the TUI requests cancellation of the pending update; already committed writes are not rolled back. + See [backend/docs/TUI.md](backend/docs/TUI.md) for the full guide. ## Documentation diff --git a/backend/packages/harness/deerflow/tui/AGENTS.md b/backend/packages/harness/deerflow/tui/AGENTS.md index 566acee4ee..a0e68cd604 100644 --- a/backend/packages/harness/deerflow/tui/AGENTS.md +++ b/backend/packages/harness/deerflow/tui/AGENTS.md @@ -13,6 +13,11 @@ 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). +A persistence bridge timeout requests cancellation of its still-pending coroutine. +This is cooperative cancellation, not rollback of a committed write or ordering +of normally completed title writes. Preserve coroutine-originated errors and +keep the background loop usable for subsequent writes. + 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 diff --git a/backend/packages/harness/deerflow/tui/persistence.py b/backend/packages/harness/deerflow/tui/persistence.py index 122d3eac11..c5870ce7cd 100644 --- a/backend/packages/harness/deerflow/tui/persistence.py +++ b/backend/packages/harness/deerflow/tui/persistence.py @@ -39,7 +39,15 @@ class _LoopThread: def run(self, coro: Awaitable[Any], *, timeout: float = 15.0) -> Any: future = asyncio.run_coroutine_threadsafe(coro, self._loop) - return future.result(timeout) + try: + return future.result(timeout) + except TimeoutError: + # A bridge deadline must not leave pending DB work running after + # the caller has given up. A coroutine's own TimeoutError already + # completed the future and should propagate unchanged. + if not future.done(): + future.cancel() + raise def close(self) -> None: self._loop.call_soon_threadsafe(self._loop.stop) diff --git a/backend/tests/test_tui_persistence.py b/backend/tests/test_tui_persistence.py index 0d4d1fd4a5..34bfdb7176 100644 --- a/backend/tests/test_tui_persistence.py +++ b/backend/tests/test_tui_persistence.py @@ -4,6 +4,9 @@ Uses the in-memory ThreadMetaStore so no SQL engine is required, but exercises the real async store + background-loop wiring used by the TUI. """ +import asyncio +import threading + import pytest from langgraph.store.memory import InMemoryStore @@ -61,3 +64,94 @@ def test_disabled_writer_is_a_silent_noop(): writer.set_title("x", "title") finally: loop.close() + + +def test_loop_timeout_cancels_pending_coroutine(): + loop = _LoopThread() + release = asyncio.Event() + started = threading.Event() + cancelled = threading.Event() + finished = threading.Event() + + async def pending_write(): + started.set() + try: + await release.wait() + except asyncio.CancelledError: + cancelled.set() + raise + finally: + finished.set() + + try: + with pytest.raises(TimeoutError): + loop.run(pending_write(), timeout=0.1) + assert started.wait(2) + assert cancelled.wait(2), "the timed-out coroutine is still running" + finally: + loop._loop.call_soon_threadsafe(release.set) + assert finished.wait(2) + loop.run(asyncio.sleep(0)) + loop.close() + loop._thread.join(timeout=2) + assert not loop._thread.is_alive() + loop._loop.close() + + +def test_timed_out_title_write_cannot_resume_after_newer_title(writer_store_loop, monkeypatch): + writer, store, loop = writer_store_loop + writer.ensure_created("th-timeout") + release = asyncio.Event() + started = threading.Event() + finished = threading.Event() + original_update = store.update_display_name + original_run = loop.run + + async def delayed_update(thread_id, title, **kwargs): + if title == "old title": + started.set() + try: + await release.wait() + return await original_update(thread_id, title, **kwargs) + finally: + finished.set() + return await original_update(thread_id, title, **kwargs) + + def short_run(coro): + return original_run(coro, timeout=0.1) + + monkeypatch.setattr(store, "update_display_name", delayed_update) + monkeypatch.setattr(loop, "run", short_run) + try: + # ThreadMetaWriter intentionally swallows the bridge timeout. + writer.set_title("th-timeout", "old title") + assert started.wait(2) + writer.set_title("th-timeout", "new title") + loop._loop.call_soon_threadsafe(release.set) + assert finished.wait(2) + row = original_run(store.get("th-timeout", user_id="default")) + assert row["display_name"] == "new title" + finally: + loop._loop.call_soon_threadsafe(release.set) + assert finished.wait(2) + original_run(asyncio.sleep(0)) + + +@pytest.mark.parametrize("error_type", [TimeoutError, ValueError]) +def test_loop_preserves_coroutine_errors(error_type): + loop = _LoopThread() + error = error_type("operation failed") + + async def fail(): + raise error + + try: + with pytest.raises(error_type) as exc: + loop.run(fail()) + assert str(exc.value) == "operation failed" + assert loop.run(asyncio.sleep(0, result="still usable")) == "still usable" + finally: + loop.close() + loop._thread.join(timeout=2) + assert not loop._thread.is_alive() + loop._loop.close()