fix(goal): fall back to sync methods on sync-only checkpointers (#6448)

* fix(goal): fall back to sync methods on sync-only checkpointers

The goal helpers called any async checkpointer method that existed. Every
BaseCheckpointSaver defines the async methods, and the sync SqliteSaver and
PostgresSaver used by the TUI and embedded DeerFlowClient raise
NotImplementedError from them, so get_goal/set_goal/clear_goal always failed
there and the TUI printed "Could not set goal.". Fall back to the sync method
only on NotImplementedError; other async errors still propagate and sync
writes stay drained across cancellation.

* docs(changelog): reference #6448

* fix(goal): log when a checkpointer call falls back to the sync method

The fallback was silent, so a sync-only saver wired into the Gateway would
work through the thread-offloaded path with no trace. Emit a debug record
naming the saver and method; debug rather than warning because the TUI and
embedded client take this path on every call.

---------

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

CHANGELOG.md

@@ -850,6 +850,13 @@ This release closes that milestone with **439 merged pull requests**.
   follows checkpoint write order, so a goal-only write counts as activity. A new
   `sort_by="updated_at"` option lets `--continue` keep resuming the most
   recently active thread; the default order stays newest created first. ([#6426])
+- **client:** Goals now work in the TUI and embedded `DeerFlowClient` on the
+  SQLite and Postgres checkpointers. Their synchronous savers define the async
+  checkpoint methods but raise `NotImplementedError` from them, and the goal
+  helpers used any async method that existed, so `/goal` printed "Could not set
+  goal." and `get_goal`/`set_goal`/`clear_goal` raised. The goal helpers now
+  fall back to the synchronous methods for those savers. The web UI was not
+  affected. ([#6448])
 - **frontend:** A failed side-chat send no longer clears the composer. The side
   chat's submit handler showed the error toast and then resolved, which the
   composer treats as success, so the typed text and attachments were lost when
@@ -9159,3 +9166,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6426]: https://github.com/bytedance/deer-flow/pull/6426
 [#6445]: https://github.com/bytedance/deer-flow/pull/6445
 [#6447]: https://github.com/bytedance/deer-flow/pull/6447
+[#6448]: https://github.com/bytedance/deer-flow/pull/6448

CHANGELOG_zh.md

@@ -703,6 +703,10 @@
   最近一两个线程，较早线程的标题无法解析。现在 SQLite 和 Postgres 通过检查点索引列出线程，每个返回的线程只加载首个和最新
   检查点；Gateway 分支也会列出，排序按检查点写入顺序，因此仅写入目标也算作活动。新增的 `sort_by="updated_at"` 选项让
   `--continue` 继续恢复最近活跃的线程；默认顺序仍为按创建时间从新到旧。([#6426])
+- **客户端：** TUI 和内嵌 `DeerFlowClient` 在 SQLite 和 Postgres 检查点后端上现在可以正常使用目标（goal）。这两种
+  同步检查点保存器定义了异步方法，但调用时会抛出 `NotImplementedError`，而目标读写只要异步方法存在就会调用它，因此
+  `/goal` 只会显示 "Could not set goal."，`get_goal`/`set_goal`/`clear_goal` 则直接抛出异常。现在目标读写在这类保存器上
+  会改用同步方法。Web UI 不受影响。([#6448])
 - **前端：** 侧边对话发送失败时不再清空输入框。侧边对话的提交处理在弹出错误提示后仍以成功返回，输入框据此视为成功，
   因此在创建侧边对话或上传附件失败时，已输入的文字和附件都会丢失；发往新侧边对话的第一条消息也会在排队时（实际发送前）
   就被清空。现在提交处理会在提示后抛出错误，排队的首条发送也以其自身结果完成提交，因此草稿会保留以便重试，只有消息
@@ -7502,3 +7506,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6407]: https://github.com/bytedance/deer-flow/pull/6407
 [#6412]: https://github.com/bytedance/deer-flow/pull/6412
 [#6426]: https://github.com/bytedance/deer-flow/pull/6426
+[#6448]: https://github.com/bytedance/deer-flow/pull/6448

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

@@ -173,6 +173,7 @@ a pre-admission miss can become an exact hit.
 - `runtime/context_compaction.py` — compaction via accessor + mutation graph (reference consumer). Runs stamp their effective agent into server-owned checkpoint metadata; manual compaction uses that binding—not request `agent_name`—for memory policy and bucket. Missing/invalid legacy bindings and unreadable agent configs fail closed by skipping the optional flush while compaction may continue with the default model; a missing pre-binding checkpoint emits a warning so the skipped write is observable.
 - `runtime/checkpoint_cache/` + `runtime/checkpointer/cached_saver.py` — delta-mode checkpoint history cache; checkpoint state reads MUST go through `CheckpointStateAccessor`, and the checkpointer may be a `CachedHistorySaver` wrapper — never rely on concrete saver types
 - `runtime/checkpointer/thread_spans.py` — thread enumeration for `DeerFlowClient.list_threads`, ordered by `checkpoint_id` (goal writes keep the old `ts`). SQLite/Postgres savers (unwrapped from `CachedHistorySaver`) use the primary-key index; any other saver walks `list(None)`. Saver type selects speed, never correctness
+- `runtime/goal.py` `_call_checkpointer_method` — tries the async saver method and falls back to the sync one only on `NotImplementedError`: every saver defines the async methods, but the sync TUI/embedded `SqliteSaver`/`PostgresSaver` (also behind `CachedHistorySaver`) raise from them. Sync writes stay in `await_drained`
 - Tests: `tests/test_checkpoint_mode.py` (freeze/detect/gate), `tests/test_checkpoint_state.py` (accessor/mutation graph), `tests/test_delta_channel_checkpointers.py` (saver parity), `tests/test_threads_checkpoint_mode.py`, `tests/test_gateway_checkpoint_mode.py` (dual-mode e2e parity), `tests/test_context_compaction.py` (mutation-graph write, no scheduling), `tests/test_run_worker_rollback.py`, `tests/test_cached_history_saver.py` + `tests/test_cached_history_saver_integration.py` (history cache)
 
 **Checkpoint benchmarks:** Paired full/delta cases and production-shaped runs are documented in `backend/docs/runtime-guidance-details.md`.

backend/packages/harness/deerflow/runtime/goal.py

@@ -620,10 +620,21 @@ def make_goal_continuation_message(goal: GoalState, evaluation: GoalEvaluation)
 
 async def _call_checkpointer_method(checkpointer: Any, async_name: str, sync_name: str, *args: Any, **kwargs: Any) -> Any:
     async_method = getattr(checkpointer, async_name, None)
-    if async_method is not None:
-        result = async_method(*args, **kwargs)
-        return await result if inspect.isawaitable(result) else result
     sync_method = getattr(checkpointer, sync_name, None)
+    if async_method is not None:
+        try:
+            result = async_method(*args, **kwargs)
+            return await result if inspect.isawaitable(result) else result
+        except NotImplementedError:
+            # Every BaseCheckpointSaver defines the async methods, so their
+            # presence proves nothing: the sync SqliteSaver/PostgresSaver used by
+            # the TUI and embedded client (also behind CachedHistorySaver) raise
+            # NotImplementedError from them. Those must use the sync method.
+            if sync_method is None:
+                raise
+            # Debug, not warning: the TUI/embedded path takes this on every
+            # call. It keeps a sync saver wired into the Gateway diagnosable.
+            logger.debug("%s.%s is not implemented; falling back to %s off the event loop", type(checkpointer).__name__, async_name, sync_name)
     if sync_method is None:
         raise AttributeError(f"Missing checkpointer method: {async_name}/{sync_name}")
     # Offload the synchronous checkpointer call so its blocking IO never runs on

backend/tests/test_client.py

@@ -2476,15 +2476,17 @@ class TestThreadQueries:
 
 
 class TestGoalManagement:
-    def test_goal_round_trip_uses_checkpoint(self, client):
-        from langgraph.checkpoint.memory import InMemorySaver
+    # ``sqlite`` and ``cached-sqlite`` are the TUI/embedded defaults: sync-only
+    # savers whose async methods exist but raise NotImplementedError.
+    @pytest.mark.parametrize("saver_kind", SAVER_KINDS)
+    def test_goal_round_trip_uses_checkpoint(self, client, saver_kind):
+        with make_saver(saver_kind) as saver:
+            client._checkpointer = saver
 
-        client._checkpointer = InMemorySaver()
-
-        set_result = client.set_goal("goal-thread", "finish all tests", max_continuations=3)
-        get_result = client.get_goal("goal-thread")
-        clear_result = client.clear_goal("goal-thread")
-        after_clear = client.get_goal("goal-thread")
+            set_result = client.set_goal("goal-thread", "finish all tests", max_continuations=3)
+            get_result = client.get_goal("goal-thread")
+            clear_result = client.clear_goal("goal-thread")
+            after_clear = client.get_goal("goal-thread")
 
         assert set_result["goal"]["objective"] == "finish all tests"
         assert set_result["goal"]["max_continuations"] == 3

backend/tests/test_goal_sync_checkpointer_cancellation.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import threading
 from types import SimpleNamespace
 
 import pytest
 
-from deerflow.runtime.goal import build_goal_state, write_thread_goal
+from deerflow.runtime.goal import build_goal_state, read_thread_goal, write_thread_goal
 
 
 class _BlockingSyncCheckpointer:
@@ -36,9 +37,20 @@ class _BlockingSyncCheckpointer:
             self.put_finished.set()
 
 
+class _AsyncUnsupportedBlockingCheckpointer(_BlockingSyncCheckpointer):
+    """Like langgraph's SqliteSaver: async methods exist but are not implemented."""
+
+    async def aget_tuple(self, _config):
+        raise NotImplementedError
+
+    async def aput(self, _config, _checkpoint, _metadata, _new_versions):
+        raise NotImplementedError
+
+
 @pytest.mark.asyncio
-async def test_sync_goal_checkpoint_write_drains_across_repeated_cancellation() -> None:
-    checkpointer = _BlockingSyncCheckpointer()
+@pytest.mark.parametrize("checkpointer_cls", [_BlockingSyncCheckpointer, _AsyncUnsupportedBlockingCheckpointer])
+async def test_sync_goal_checkpoint_write_drains_across_repeated_cancellation(checkpointer_cls) -> None:
+    checkpointer = checkpointer_cls()
     task = asyncio.create_task(
         write_thread_goal(
             checkpointer,
@@ -67,3 +79,42 @@ async def test_sync_goal_checkpoint_write_drains_across_repeated_cancellation()
         checkpointer.allow_put.set()
         await asyncio.gather(task, return_exceptions=True)
         await asyncio.to_thread(checkpointer.put_finished.wait, 1.0)
+
+
+@pytest.mark.asyncio
+async def test_goal_sync_fallback_is_logged(caplog) -> None:
+    checkpointer = _AsyncUnsupportedBlockingCheckpointer()
+
+    with caplog.at_level(logging.DEBUG, logger="deerflow.runtime.goal"):
+        await read_thread_goal(checkpointer, "thread-1")
+
+    assert "_AsyncUnsupportedBlockingCheckpointer.aget_tuple is not implemented; falling back to get_tuple" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_goal_read_falls_back_to_sync_only_for_not_implemented() -> None:
+    class _BrokenAsyncCheckpointer:
+        def __init__(self) -> None:
+            self.sync_calls = 0
+
+        async def aget_tuple(self, _config):
+            raise ConnectionError("pool closed")
+
+        def get_tuple(self, _config):
+            self.sync_calls += 1
+
+    checkpointer = _BrokenAsyncCheckpointer()
+
+    with pytest.raises(ConnectionError):
+        await read_thread_goal(checkpointer, "thread-1")
+    assert checkpointer.sync_calls == 0
+
+
+@pytest.mark.asyncio
+async def test_goal_read_keeps_not_implemented_without_sync_method() -> None:
+    class _AsyncOnlyUnsupportedCheckpointer:
+        async def aget_tuple(self, _config):
+            raise NotImplementedError
+
+    with pytest.raises(NotImplementedError):
+        await read_thread_goal(_AsyncOnlyUnsupportedCheckpointer(), "thread-1")