fix(client): count threads, not checkpoints, in list_threads (#6426)

* fix(client): count threads, not checkpoints, in list_threads

checkpointer.list(None, limit=N) returns the newest N checkpoints across
every thread, so one long conversation filled the limit and hid the rest:
the TUI thread picker and --resume <title> only saw the latest one or two
threads. List threads through their single step -1 root checkpoint
instead, then read the latest checkpoint per returned thread.

Add sort_by="updated_at" so --continue keeps resuming the most recently
active thread rather than the most recently created one.

* test(client): pin list_threads scan cost and namespace filtering; docs

* docs(client): state list_threads cost as checkpoints loaded, not rows scanned

* docs(changelog): reference #6426

* fix(client): list threads by checkpoint write order, including branches

Discovering threads by their step -1 input checkpoint dropped Gateway
branches, which are seeded through update_state (step 0), and ordering
by checkpoint ts missed goal writes, which copy the previous ts under a
fresh checkpoint id.

Enumerate threads from the checkpoint primary-key index on SQLite and
Postgres (unwrapping the delta-history cache), falling back to one
list(None) walk for any other saver, and order by checkpoint_id, the
key LangGraph itself uses for a thread's latest checkpoint.

CHANGELOG.md

@@ -804,6 +804,16 @@ This release closes that milestone with **439 merged pull requests**.
 
 ### Fixed
 
+- **client:** `DeerFlowClient.list_threads(limit)` now limits threads rather
+  than checkpoints. It passed `limit` to a checkpoint scan across every thread,
+  and one turn writes several checkpoints, so a single long conversation filled
+  the limit: the TUI thread picker and `--resume <title>` saw only the latest one
+  or two threads, and an older title failed to resolve. SQLite and Postgres now
+  list threads from their checkpoint index, and only each returned thread's
+  first and latest checkpoints are loaded. Gateway branches are listed, and order
+  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])
 - **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
@@ -9109,3 +9119,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6400]: https://github.com/bytedance/deer-flow/pull/6400
 [#6401]: https://github.com/bytedance/deer-flow/pull/6401
 [#6407]: https://github.com/bytedance/deer-flow/pull/6407
+[#6426]: https://github.com/bytedance/deer-flow/pull/6426

CHANGELOG_zh.md

@@ -694,6 +694,11 @@
 
 ### 修复
 
+- **客户端：** `DeerFlowClient.list_threads(limit)` 现在限制的是线程数而不是检查点数。此前 `limit` 被传给跨所有线程的
+  检查点扫描，而一轮对话会写入多个检查点，因此一个较长的对话就会占满上限：TUI 线程选择器和 `--resume <标题>` 只能看到
+  最近一两个线程，较早线程的标题无法解析。现在 SQLite 和 Postgres 通过检查点索引列出线程，每个返回的线程只加载首个和最新
+  检查点；Gateway 分支也会列出，排序按检查点写入顺序，因此仅写入目标也算作活动。新增的 `sort_by="updated_at"` 选项让
+  `--continue` 继续恢复最近活跃的线程；默认顺序仍为按创建时间从新到旧。([#6426])
 - **前端：** 侧边对话发送失败时不再清空输入框。侧边对话的提交处理在弹出错误提示后仍以成功返回，输入框据此视为成功，
   因此在创建侧边对话或上传附件失败时，已输入的文字和附件都会丢失；发往新侧边对话的第一条消息也会在排队时（实际发送前）
   就被清空。现在提交处理会在提示后抛出错误，排队的首条发送也以其自身结果完成提交，因此草稿会保留以便重试，只有消息
@@ -7491,3 +7496,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6400]: https://github.com/bytedance/deer-flow/pull/6400
 [#6401]: https://github.com/bytedance/deer-flow/pull/6401
 [#6407]: https://github.com/bytedance/deer-flow/pull/6407
+[#6426]: https://github.com/bytedance/deer-flow/pull/6426

backend/packages/harness/deerflow/client.py

@@ -757,57 +757,51 @@ class DeerFlowClient:
             pass
         return {"goal": None}
 
-    def list_threads(self, limit: int = 10) -> dict:
+    def list_threads(self, limit: int = 10, *, sort_by: Literal["created_at", "updated_at"] = "created_at") -> dict:
         """List the recent N threads.
 
+        ``limit`` counts threads, not checkpoints, and only each returned
+        thread's first and latest checkpoints are loaded (see
+        :mod:`deerflow.runtime.checkpointer.thread_spans`).
+
         Args:
             limit: Maximum number of threads to return. Default is 10.
+            sort_by: ``"created_at"`` (default) orders by each thread's first
+                checkpoint, ``"updated_at"`` by its latest. Both follow
+                checkpoint write order, which the reported ``ts`` values can
+                lag: a goal write keeps the previous checkpoint's ``ts``.
 
         Returns:
             Dict with "thread_list" key containing list of thread info dicts,
-            sorted by thread creation time descending.
+            newest first by ``sort_by``.
         """
+        from deerflow.runtime.checkpointer.thread_spans import list_thread_spans
+
+        if sort_by not in ("created_at", "updated_at"):
+            raise ValueError(f"sort_by must be 'created_at' or 'updated_at', got {sort_by!r}")
         checkpointer = self._get_thread_checkpointer()
+        spans = list_thread_spans(checkpointer, order_by="first" if sort_by == "created_at" else "latest", limit=limit)
 
-        thread_info_map = {}
+        def _checkpoint(thread_id: str, checkpoint_id: str):
+            return checkpointer.get_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": "", "checkpoint_id": checkpoint_id}})
 
-        for cp in checkpointer.list(config=None, limit=limit):
-            cfg = cp.config.get("configurable", {})
-            thread_id = cfg.get("thread_id")
-            if not thread_id:
+        threads = []
+        for span in spans:
+            latest = _checkpoint(span.thread_id, span.latest_checkpoint_id)
+            if latest is None:
                 continue
-
-            ts = cp.checkpoint.get("ts")
-            checkpoint_id = cfg.get("checkpoint_id")
-
-            if thread_id not in thread_info_map:
-                channel_values = cp.checkpoint.get("channel_values", {})
-                thread_info_map[thread_id] = {
-                    "thread_id": thread_id,
-                    "created_at": ts,
-                    "updated_at": ts,
-                    "latest_checkpoint_id": checkpoint_id,
-                    "title": channel_values.get("title"),
+            first = latest if span.first_checkpoint_id == span.latest_checkpoint_id else _checkpoint(span.thread_id, span.first_checkpoint_id)
+            threads.append(
+                {
+                    "thread_id": span.thread_id,
+                    "created_at": first.checkpoint.get("ts") if first is not None else None,
+                    "updated_at": latest.checkpoint.get("ts"),
+                    "latest_checkpoint_id": span.latest_checkpoint_id,
+                    "title": latest.checkpoint.get("channel_values", {}).get("title"),
                 }
-            else:
-                # Explicitly compare timestamps to ensure accuracy when iterating over unordered namespaces.
-                # Treat None as "missing" and only compare when existing values are non-None.
-                if ts is not None:
-                    current_created = thread_info_map[thread_id]["created_at"]
-                    if current_created is None or ts < current_created:
-                        thread_info_map[thread_id]["created_at"] = ts
+            )
 
-                    current_updated = thread_info_map[thread_id]["updated_at"]
-                    if current_updated is None or ts > current_updated:
-                        thread_info_map[thread_id]["updated_at"] = ts
-                        thread_info_map[thread_id]["latest_checkpoint_id"] = checkpoint_id
-                        channel_values = cp.checkpoint.get("channel_values", {})
-                        thread_info_map[thread_id]["title"] = channel_values.get("title")
-
-        threads = list(thread_info_map.values())
-        threads.sort(key=lambda x: x.get("created_at") or "", reverse=True)
-
-        return {"thread_list": threads[:limit]}
+        return {"thread_list": threads}
 
     def get_thread(self, thread_id: str) -> dict:
         """Get the complete materialized checkpoint history for a thread."""

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

@@ -172,6 +172,7 @@ a pre-admission miss can become an exact hit.
 - `agents/thread_state.py` — `ThreadState`/`DeltaThreadState`, `delta_messages_field` / `DELTA_MESSAGES_FIELD` (`DeltaChannel` at the configured `snapshot_frequency`, default 10), schema adaptation helpers
 - `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
 - 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/checkpointer/thread_spans.py

@@ -0,0 +1,97 @@
+"""Enumerate the threads a checkpointer holds, in checkpoint write order.
+
+LangGraph has no thread-listing API, and ``BaseCheckpointSaver.list(None)``
+walks every checkpoint of every thread: a listing built on it either loads whole
+histories or, given ``limit``, counts checkpoints instead of threads.
+
+Order follows ``checkpoint_id``. Savers mint time-ordered UUID6 ids, and
+LangGraph itself defines a thread's latest checkpoint as its largest id.
+``checkpoint["ts"]`` is not a recency signal: a goal write copies the previous
+checkpoint's ``ts`` under a fresh id. Discovery does not assume any particular
+first checkpoint either — a Gateway branch is seeded by ``update_state``
+(``step`` 0), not by an input checkpoint (``step`` -1).
+
+The SQLite and Postgres savers answer from their primary-key index
+``(thread_id, checkpoint_ns, checkpoint_id)``, one row per thread. Any other
+saver falls back to a single ``list(None)`` walk: equally correct, only slower.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Literal
+
+from langgraph.checkpoint.sqlite import SqliteSaver
+
+from deerflow.runtime.checkpointer.cached_saver import CachedHistorySaver
+
+try:  # optional extra: deerflow-harness[postgres]
+    from langgraph.checkpoint.postgres import PostgresSaver
+except ImportError:  # pragma: no cover - exercised only without the extra
+    PostgresSaver = None
+
+ThreadSpanOrder = Literal["first", "latest"]
+
+_ORDER_COLUMNS: dict[str, str] = {"first": "first_checkpoint_id", "latest": "latest_checkpoint_id"}
+
+_SPANS_SQL = """
+    SELECT thread_id, MIN(checkpoint_id) AS first_checkpoint_id, MAX(checkpoint_id) AS latest_checkpoint_id
+    FROM checkpoints
+    WHERE checkpoint_ns = ''
+    GROUP BY thread_id
+    ORDER BY {order_column} DESC, thread_id
+"""
+
+
+@dataclass(frozen=True)
+class ThreadSpan:
+    """A thread's first and latest root-namespace checkpoint ids."""
+
+    thread_id: str
+    first_checkpoint_id: str
+    latest_checkpoint_id: str
+
+
+def list_thread_spans(checkpointer: Any, *, order_by: ThreadSpanOrder, limit: int | None = None) -> list[ThreadSpan]:
+    """Return up to *limit* threads, newest first by their first or latest checkpoint."""
+    if order_by not in _ORDER_COLUMNS:
+        raise ValueError(f"order_by must be 'first' or 'latest', got {order_by!r}")
+    if limit is not None and limit <= 0:
+        return []
+    # The delta-history cache wraps the saver but never owns which threads exist.
+    saver = checkpointer._inner if isinstance(checkpointer, CachedHistorySaver) else checkpointer
+    sql = _SPANS_SQL.format(order_column=_ORDER_COLUMNS[order_by])
+    if isinstance(saver, SqliteSaver):
+        params: tuple = ()
+        if limit is not None:
+            sql += " LIMIT ?"
+            params = (limit,)
+        with saver.cursor(transaction=False) as cur:
+            cur.execute(sql, params)
+            return [ThreadSpan(*row) for row in cur.fetchall()]
+    if PostgresSaver is not None and isinstance(saver, PostgresSaver):
+        params = ()
+        if limit is not None:
+            sql += " LIMIT %s"
+            params = (limit,)
+        with saver._cursor() as cur:
+            cur.execute(sql, params)
+            return [ThreadSpan(row["thread_id"], row["first_checkpoint_id"], row["latest_checkpoint_id"]) for row in cur.fetchall()]
+    return _scan_thread_spans(checkpointer, order_by=order_by, limit=limit)
+
+
+def _scan_thread_spans(checkpointer: Any, *, order_by: ThreadSpanOrder, limit: int | None) -> list[ThreadSpan]:
+    bounds: dict[str, tuple[str, str]] = {}
+    for cp in checkpointer.list(None):
+        cfg = cp.config.get("configurable", {})
+        thread_id = cfg.get("thread_id")
+        checkpoint_id = cfg.get("checkpoint_id")
+        # Subgraph namespaces keep their own checkpoints; only the root namespace describes the thread.
+        if not thread_id or not checkpoint_id or cfg.get("checkpoint_ns"):
+            continue
+        first, latest = bounds.get(thread_id, (checkpoint_id, checkpoint_id))
+        bounds[thread_id] = (min(first, checkpoint_id), max(latest, checkpoint_id))
+    spans = [ThreadSpan(thread_id, first, latest) for thread_id, (first, latest) in bounds.items()]
+    spans.sort(key=lambda span: span.thread_id)
+    spans.sort(key=lambda span: getattr(span, _ORDER_COLUMNS[order_by]), reverse=True)
+    return spans if limit is None else spans[:limit]

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

@@ -9,7 +9,7 @@ A terminal-native UI over the embedded harness, exposed as the `deerflow` consol
 - `message_format.py` / `command_registry.py` / `input_history.py` / `render.py` / `theme.py` — pure helpers (tool summaries, slash registry + `resolve()`, ↑/↓ history, Rich renderers). The command registry must exclude every shared `RESERVED_SLASH_SKILL_NAMES` entry that the agent runtime rejects from both its picker and resolver; the `context` skill is the exception for ordinary task text, while the exact composer-only `/context compact` alias remains unavailable as a skill activation.
 - `app.py` — Textual `App`. Runs `DeerFlowClient.stream()` (sync) on a worker thread and marshals actions to the UI thread via `call_from_thread`. Slash palette with `/goal` management + model/thread modal pickers; routes idle display-only `/clear` through `ClearRows` without replacing the active thread, and blocks state-resetting local commands like `/new` and `/clear` with the standard "Still working" message during an active run; priority key bindings gated by `check_action` so they never steal keys from overlays or the composer. Application-level PageUp/PageDown bindings scroll the transcript while preserving composer focus; streaming follows output only while the transcript remains at the bottom.
 - `widgets/composer.py` — the composer, a Textual `TextArea`: bracketed multiline pastes retain their line breaks and indentation, `Enter` submits the whole document unchanged (only an all-whitespace document counts as empty), and Up/Down move inside it before falling back to input history at the first/last visual row (soft-wrapped rows count, so a long wrapped prompt is navigable).
-- `session.py` / `persistence.py` — builds the client + checkpointer and the `ThreadMetaWriter`.
+- `session.py` / `persistence.py` — builds the client + checkpointer and the `ThreadMetaWriter`. The thread picker and `--resume <title>` read `DeerFlowClient.list_threads()`, where `limit` counts threads (newest created first); `--continue` passes `sort_by="updated_at"` to resume the most recently active thread.
 
 **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).
 

backend/packages/harness/deerflow/tui/session.py

@@ -29,7 +29,7 @@ class Session:
         if plan.thread_id:
             return self.resolve_ref(plan.thread_id)
         if plan.continue_recent:
-            threads = self.client.list_threads(limit=1).get("thread_list", [])
+            threads = self.client.list_threads(limit=1, sort_by="updated_at").get("thread_list", [])
             if threads:
                 return threads[0].get("thread_id")
         return None

backend/tests/_thread_checkpoint_helpers.py

@@ -0,0 +1,115 @@
+"""Real checkpointers and run-shaped checkpoint writers for thread-listing tests.
+
+Each saver kind takes a different path through
+``deerflow.runtime.checkpointer.thread_spans``: ``memory`` walks ``list(None)``
+(one thread at a time), ``sqlite`` and ``postgres`` query the primary-key
+index, and ``cached-sqlite`` must unwrap the delta-history cache first.
+``postgres`` runs only with ``DEERFLOW_TEST_POSTGRES_URL`` (or
+``TEST_POSTGRES_URI``) and the ``postgres`` extra installed, in a throwaway
+schema.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+import pytest
+from langgraph.checkpoint.base import empty_checkpoint, uuid6
+
+SAVER_KINDS = ("memory", "sqlite", "cached-sqlite", "postgres")
+INDEXED_SAVER_KINDS = frozenset({"sqlite", "cached-sqlite", "postgres"})
+
+
+@contextmanager
+def make_saver(kind: str) -> Iterator[Any]:
+    if kind == "memory":
+        from langgraph.checkpoint.memory import InMemorySaver
+
+        yield InMemorySaver()
+    elif kind in ("sqlite", "cached-sqlite"):
+        from langgraph.checkpoint.sqlite import SqliteSaver
+
+        conn = sqlite3.connect(":memory:", check_same_thread=False)
+        try:
+            saver = SqliteSaver(conn)
+            if kind == "cached-sqlite":
+                from deerflow.runtime.checkpoint_cache.memory import MemoryCheckpointHistoryCache
+                from deerflow.runtime.checkpointer.cached_saver import CachedHistorySaver
+
+                saver = CachedHistorySaver(saver, MemoryCheckpointHistoryCache(128), key_prefix="thread-list-test")
+            yield saver
+        finally:
+            conn.close()
+    elif kind == "postgres":
+        url = os.getenv("DEERFLOW_TEST_POSTGRES_URL") or os.getenv("TEST_POSTGRES_URI")
+        if not url:
+            pytest.skip("set TEST_POSTGRES_URI or DEERFLOW_TEST_POSTGRES_URL to run live PostgreSQL tests")
+        postgres = pytest.importorskip("langgraph.checkpoint.postgres")
+        import psycopg
+
+        url = url.replace("postgresql+asyncpg://", "postgresql://")
+        schema = f"thread_list_{uuid.uuid4().hex[:12]}"
+        with psycopg.connect(url, autocommit=True) as admin:
+            admin.execute(f'CREATE SCHEMA "{schema}"')
+            try:
+                with psycopg.connect(url, autocommit=True, options=f"-c search_path={schema}") as conn:
+                    saver = postgres.PostgresSaver(conn)
+                    saver.setup()
+                    yield saver
+            finally:
+                admin.execute(f'DROP SCHEMA "{schema}" CASCADE')
+    else:
+        raise ValueError(kind)
+
+
+def _put(saver: Any, config: dict, checkpoint: dict, metadata: dict, title: str | None, version: int) -> dict:
+    new_versions: dict[str, Any] = {}
+    if title is not None:
+        # Channel values persist only for channels named in ``new_versions``,
+        # and InMemorySaver keys their blobs by version.
+        new_versions = {"title": version}
+        checkpoint["channel_values"] = {"title": title}
+        checkpoint["channel_versions"] = {"title": version}
+    return saver.put(config, checkpoint, metadata, new_versions)
+
+
+def put_thread(
+    saver: Any,
+    thread_id: str,
+    timestamps: list[str],
+    *,
+    title: str | None = None,
+    checkpoint_ns: str = "",
+    first_step: int = -1,
+    first_source: str = "input",
+) -> list[str]:
+    """Write one checkpoint per timestamp, chained like a run, and return their ids.
+
+    The default first checkpoint is a run's input checkpoint (``step`` -1). A
+    Gateway branch is seeded by ``update_state`` instead: pass
+    ``first_step=0, first_source="branch"``.
+    """
+    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns}}
+    checkpoint_ids = []
+    for offset, ts in enumerate(timestamps):
+        checkpoint = empty_checkpoint()
+        checkpoint["ts"] = ts
+        metadata = {"source": first_source if offset == 0 else "loop", "step": first_step + offset, "parents": {}}
+        config = _put(saver, config, checkpoint, metadata, f"{title} #{offset}" if title else None, offset + 1)
+        checkpoint_ids.append(config["configurable"]["checkpoint_id"])
+    return checkpoint_ids
+
+
+def put_goal_write(saver: Any, thread_id: str) -> str:
+    """Mirror ``runtime.goal.write_thread_goal``: copy the latest checkpoint under a fresh id, keeping its ``ts``."""
+    latest = saver.get_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}})
+    checkpoint = dict(latest.checkpoint)
+    checkpoint["id"] = str(uuid6())
+    metadata = {**latest.metadata, "source": "update", "step": latest.metadata.get("step", 0) + 1}
+    config = saver.put(latest.config, checkpoint, metadata, {})
+    return config["configurable"]["checkpoint_id"]

backend/tests/test_client.py

@@ -14,6 +14,7 @@ from types import SimpleNamespace
 from unittest.mock import ANY, MagicMock, patch
 
 import pytest
+from _thread_checkpoint_helpers import INDEXED_SAVER_KINDS, SAVER_KINDS, make_saver, put_goal_write, put_thread
 from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage, ToolMessage  # noqa: F401
 from langchain_core.tools import StructuredTool
 
@@ -2218,44 +2219,133 @@ class TestThreadQueries:
         snapshot.created_at = checkpoint_tuple.checkpoint["ts"]
         return snapshot
 
-    def test_list_threads_empty(self, client):
-        mock_checkpointer = MagicMock()
-        mock_checkpointer.list.return_value = []
-        client._checkpointer = mock_checkpointer
+    @pytest.fixture(params=SAVER_KINDS)
+    def saver(self, request):
+        with make_saver(request.param) as saver:
+            saver.kind = request.param
+            yield saver
 
-        result = client.list_threads()
-        assert result == {"thread_list": []}
-        mock_checkpointer.list.assert_called_once_with(config=None, limit=10)
+    def test_list_threads_empty(self, client, saver):
+        client._checkpointer = saver
 
-    def test_list_threads_basic(self, client):
-        mock_checkpointer = MagicMock()
-        client._checkpointer = mock_checkpointer
+        assert client.list_threads() == {"thread_list": []}
 
-        cp1 = self._make_mock_checkpoint_tuple("t1", "c1", "2023-01-01T10:00:00Z", title="Thread 1")
-        cp2 = self._make_mock_checkpoint_tuple("t1", "c2", "2023-01-01T10:05:00Z", title="Thread 1 Updated")
-        cp3 = self._make_mock_checkpoint_tuple("t2", "c3", "2023-01-02T10:00:00Z", title="Thread 2")
-        cp_empty = self._make_mock_checkpoint_tuple("", "c4", "2023-01-03T10:00:00Z", title="Thread Empty")
+    def test_list_threads_limit_counts_threads_not_checkpoints(self, client, saver):
+        client._checkpointer = saver
+        put_thread(saver, "t-old", ["2023-01-01T10:00:00Z", "2023-01-01T10:01:00Z"])
+        # A long conversation writes far more checkpoints than the limit.
+        put_thread(saver, "t-busy", [f"2023-01-02T10:{minute:02d}:00Z" for minute in range(30)])
 
-        # Mock list returns out of order to test the timestamp sorting/comparison
-        # Also includes a checkpoint with an empty thread_id which should be skipped
-        mock_checkpointer.list.return_value = [cp2, cp1, cp_empty, cp3]
+        threads = client.list_threads(limit=2)["thread_list"]
 
-        result = client.list_threads(limit=5)
-        mock_checkpointer.list.assert_called_once_with(config=None, limit=5)
+        assert [t["thread_id"] for t in threads] == ["t-busy", "t-old"]
 
-        threads = result["thread_list"]
-        assert len(threads) == 2
+    def test_list_threads_reports_first_and_latest_checkpoint(self, client, saver):
+        client._checkpointer = saver
+        checkpoint_ids = put_thread(saver, "t1", ["2023-01-01T10:00:00Z", "2023-01-01T10:05:00Z", "2023-01-01T10:09:00Z"], title="Thread 1")
 
-        # t2 should be first because its created_at (2023-01-02) is newer than t1 (2023-01-01)
-        assert threads[0]["thread_id"] == "t2"
-        assert threads[0]["created_at"] == "2023-01-02T10:00:00Z"
-        assert threads[0]["title"] == "Thread 2"
+        (thread,) = client.list_threads()["thread_list"]
 
-        assert threads[1]["thread_id"] == "t1"
-        assert threads[1]["created_at"] == "2023-01-01T10:00:00Z"
-        assert threads[1]["updated_at"] == "2023-01-01T10:05:00Z"
-        assert threads[1]["latest_checkpoint_id"] == "c2"
-        assert threads[1]["title"] == "Thread 1 Updated"
+        assert thread == {
+            "thread_id": "t1",
+            "created_at": "2023-01-01T10:00:00Z",
+            "updated_at": "2023-01-01T10:09:00Z",
+            "latest_checkpoint_id": checkpoint_ids[-1],
+            "title": "Thread 1 #2",
+        }
+
+    def test_list_threads_sorts_by_creation_then_slices(self, client, saver):
+        client._checkpointer = saver
+        put_thread(saver, "t-first", ["2023-01-01T10:00:00Z"])
+        put_thread(saver, "t-second", ["2023-01-02T10:00:00Z"])
+        put_thread(saver, "t-third", ["2023-01-03T10:00:00Z"])
+        put_thread(saver, "t-first", ["2023-01-05T10:00:00Z"])  # later activity does not change creation order
+
+        threads = client.list_threads(limit=2)["thread_list"]
+
+        assert [t["thread_id"] for t in threads] == ["t-third", "t-second"]
+
+    def test_list_threads_sort_by_updated_at_returns_most_recently_written(self, client, saver):
+        client._checkpointer = saver
+        put_thread(saver, "t-first", ["2023-01-01T10:00:00Z"])
+        put_thread(saver, "t-second", ["2023-01-02T10:00:00Z"])
+        put_thread(saver, "t-third", ["2023-01-03T10:00:00Z"])
+        put_thread(saver, "t-first", ["2023-01-05T10:00:00Z"])
+
+        threads = client.list_threads(limit=2, sort_by="updated_at")["thread_list"]
+
+        assert [t["thread_id"] for t in threads] == ["t-first", "t-third"]
+
+    def test_list_threads_includes_threads_seeded_without_an_input_checkpoint(self, client, saver):
+        client._checkpointer = saver
+        put_thread(saver, "t-run", ["2023-01-01T10:00:00Z", "2023-01-01T10:01:00Z"])
+        # The Gateway seeds a branch through update_state, so its first checkpoint is step 0.
+        put_thread(saver, "t-branch", ["2023-01-02T10:00:00Z", "2023-01-02T10:00:01Z"], title="My branch", first_step=0, first_source="branch")
+
+        threads = client.list_threads()["thread_list"]
+
+        assert [(t["thread_id"], t["created_at"]) for t in threads] == [("t-branch", "2023-01-02T10:00:00Z"), ("t-run", "2023-01-01T10:00:00Z")]
+
+    def test_list_threads_orders_goal_writes_by_checkpoint_not_ts(self, client, saver):
+        client._checkpointer = saver
+        put_thread(saver, "t-a", ["2023-01-01T10:00:00Z"])
+        put_thread(saver, "t-b", ["2023-01-02T10:00:00Z"])
+        # A goal write keeps the copied checkpoint's ts but is the newest write.
+        goal_checkpoint_id = put_goal_write(saver, "t-a")
+
+        (thread,) = client.list_threads(limit=1, sort_by="updated_at")["thread_list"]
+
+        assert thread["thread_id"] == "t-a"
+        assert thread["latest_checkpoint_id"] == goal_checkpoint_id
+        assert thread["updated_at"] == "2023-01-01T10:00:00Z"
+
+    def test_list_threads_ignores_subgraph_namespaces(self, client, saver):
+        client._checkpointer = saver
+        # Subgraph checkpoints are written both before and after the thread's own.
+        put_thread(saver, "t1", ["2023-01-01T09:00:00Z"], checkpoint_ns="subgraph-a")
+        checkpoint_ids = put_thread(saver, "t1", ["2023-01-01T10:00:00Z", "2023-01-01T10:05:00Z"])
+        put_thread(saver, "t1", ["2023-01-01T10:09:00Z"], checkpoint_ns="subgraph-b")
+
+        (thread,) = client.list_threads()["thread_list"]
+
+        assert thread["created_at"] == "2023-01-01T10:00:00Z"
+        assert thread["updated_at"] == "2023-01-01T10:05:00Z"
+        assert thread["latest_checkpoint_id"] == checkpoint_ids[-1]
+
+    def test_list_threads_reads_sql_savers_from_the_index(self, client, saver):
+        if saver.kind not in INDEXED_SAVER_KINDS:
+            pytest.skip("only SQL savers have a thread index")
+        client._checkpointer = saver
+        put_thread(saver, "t1", ["2023-01-01T10:00:00Z", "2023-01-01T10:01:00Z"])
+
+        with patch.object(type(saver), "list", side_effect=AssertionError("walked every checkpoint")):
+            threads = client.list_threads()["thread_list"]
+
+        assert [t["thread_id"] for t in threads] == ["t1"]
+
+    def test_list_threads_rejects_unknown_sort_key(self, client, saver):
+        client._checkpointer = saver
+
+        with pytest.raises(ValueError, match="sort_by"):
+            client.list_threads(sort_by="title")
+
+    def test_tui_session_resolves_past_a_busy_thread(self, client, saver):
+        from deerflow.tui.cli import LaunchPlan
+        from deerflow.tui.session import Session
+
+        client._checkpointer = saver
+        put_thread(saver, "t-old", ["2023-01-01T10:00:00Z"], title="Old chat")
+        put_thread(saver, "t-branch", ["2023-01-01T11:00:00Z"], title="My branch", first_step=0, first_source="branch")
+        put_thread(saver, "t-busy", [f"2023-01-02T10:{minute:02d}:00Z" for minute in range(120)], title="Busy chat")
+        put_goal_write(saver, "t-old")
+        session = Session(client=client)
+
+        # --resume <title> finds threads whose checkpoints fall outside the newest 100,
+        # including a branch that never had an input checkpoint.
+        assert session.resolve_ref("Old chat #0") == "t-old"
+        assert session.resolve_ref("My branch #0") == "t-branch"
+        # --continue resumes the thread written most recently, even by a goal write.
+        assert session.resolve_thread(LaunchPlan(mode="tui", continue_recent=True)) == "t-old"
 
     def test_list_threads_fallback_checkpointer(self, client):
         mock_checkpointer = MagicMock()

backend/tests/test_tui_session.py

@@ -8,7 +8,7 @@ class _Client:
     def __init__(self, threads):
         self._threads = threads
 
-    def list_threads(self, limit=10):
+    def list_threads(self, limit=10, *, sort_by="created_at"):
         return {"thread_list": self._threads}