fix: generate fallback title for interrupted first-turn runs (#3874)

* fix: generate title for interrupted first turn

* test(title): cover partial-exchange + dict-form messages

Harden the interrupted-run fallback path added in 19fc34fd:

- TitleMiddleware._should_generate_title now accepts a lone first-turn
  user message when allow_partial_exchange=True, so the worker can still
  derive a title if cancellation lands before any AI chunk is checkpointed.
- runtime/runs/worker._ensure_interrupted_title computes the next
  checkpoint step defensively (treat missing/non-int step as 0) and
  renames a shadowed ckpt_config local for readability.
- Add four unit tests in tests/test_title_middleware_core_logic.py:
  partial-exchange allows user-only, partial-exchange still respects an
  existing title, dict-form messages are recognized, and the sync
  fallback path derives a title from dict-form messages — matching what
  channel_values stores in the checkpoint.

Refs #3859.

* fix: persist interrupted-title via channel_versions bump

Address PR #3874 review feedback: ``_ensure_interrupted_title`` previously
called ``aput(..., new_versions={})``. LangGraph's DB-backed savers
(``PostgresSaver`` and the v4 ``SqliteSaver`` blob layout) strip inline
``channel_values`` from ``put`` and only persist blobs for channels named
in ``new_versions`` — so the fallback ``title`` channel was dropped on
read-back and ``threads_meta.display_name`` stayed ``"Untitled"`` after
refresh on those backends. The original in-memory e2e passed because
``InMemorySaver`` keeps the inline snapshot verbatim.

Fix mirrors ``_rollback_to_pre_run_checkpoint`` in the same file: bump
``channel_versions["title"]`` (via ``checkpointer.get_next_version`` when
available, else int/string fallbacks), persist the new version on the
checkpoint, and declare it in ``new_versions`` so the DB savers actually
write the blob.

Regression coverage in ``tests/test_run_worker_rollback.py``:

- ``test_ensure_interrupted_title_bumps_channel_version_and_declares_it_in_new_versions``
  — exact ``aput`` invariants: ``new_versions == {"title": 1}``, the
  written checkpoint's ``channel_versions["title"]`` is bumped, and the
  pre-existing ``messages`` version is preserved.
- ``test_ensure_interrupted_title_bumps_existing_string_version`` —
  string-shaped prior version (some savers use UUID-style versions);
  bumped value must differ from the prior, no overwrite-in-place.
- ``test_ensure_interrupted_title_skips_when_title_already_set`` — title
  short-circuit; no extra ``aput``.
- ``test_ensure_interrupted_title_returns_none_when_no_checkpoint`` —
  no checkpoint yet; returns ``None`` without writing.
- ``test_ensure_interrupted_title_round_trip_with_real_sqlite_checkpointer``
  — full round-trip against a real ``AsyncSqliteSaver`` on a disk-backed
  DB, then closes and re-opens the saver to simulate a fresh
  connection. The fallback title must still be present on the second
  ``aget_tuple``. This is the exact scenario the review flagged.

Validated locally with the full backend suite: 5195 passed, 18 skipped.

Refs #3859. Addresses review on #3874.

* test(worker, title): harden interrupted-title fallback for every saver

Defensive coverage on top of the channel_versions fix (commit 05253957),
addressing edge cases surfaced during a second-pass review of #3874.

Worker:
- Extract version bump into ``_bump_channel_version(checkpointer, current)``
  with explicit fallbacks for int / float / numeric-string / UUID-shaped
  string / None / bool, AND a wrap-around defense when the saver's
  ``get_next_version`` raises or returns an unchanged value. The
  invariant is: returned version MUST differ from the prior. Without
  this, a saver bug (or a custom backend) could leave
  ``new_versions={"title": v}`` no-op on DB savers — the very class of
  bug the original review pointed out.

Title middleware:
- Coerce ``state.get("messages")`` from ``None`` to ``[]`` on both
  ``_should_generate_title`` and ``_build_title_prompt``. A
  partially-initialized checkpoint can carry ``messages=None`` on the
  channel_values channel (the worker reads raw channel_values, not
  BaseMessages), and the default kwarg only protects against a missing
  key. Repro: ``TypeError: 'NoneType' object is not iterable`` from
  the next() generator — confirmed by reverting the fix and watching
  ``test_*_handles_none_messages_channel`` go red.

Tests (TDD-verified red→green for the new asserts):
- ``test_run_worker_rollback.py``:
  * ``_bump_channel_version`` — 8 tests covering every version type
    (int, float, numeric string, UUID-style string, None, bool) and
    every saver-side fault mode (no ``get_next_version`` / raising /
    stuck on identity).
  * ``test_ensure_interrupted_title_*`` — 5 additional helper
    boundary tests: title.enabled=false short-circuit; empty
    messages list; messages=None; aput-error propagation (helper
    contract: caller swallows, not the helper); idempotency on a
    real InMemorySaver across two invocations.
  * ``test_ensure_interrupted_title_preserves_non_title_channel_versions``
    — pins that ``new_versions`` only contains ``"title"`` and that
    other channels' versions are untouched (regression anchor for a
    sloppier draft that bumped every channel).
  * ``test_worker_finally_block_swallows_helper_exceptions`` — pins
    the integration contract: even if the helper raises, the worker's
    threads_meta status sync still runs and ``publish_end`` is still
    awaited so the SSE stream closes cleanly.

- ``test_title_middleware_core_logic.py``:
  * 4 additional tests: ``messages=None`` on both
    ``_should_generate_title`` and ``_build_title_prompt``; the
    ``role: user`` / ``role: assistant`` (OpenAI-style) dict
    normalization; partial-exchange path with a dict-form message.

Verification:
- ``PYTHONPATH=. uv run pytest tests/ -x --ignore=tests/blocking_io -q``
  → 5215 passed, 18 skipped.
- ``ruff check`` + ``ruff format --check`` clean on every touched file.
- Red/green TDD verification: temporarily reverted the
  ``new_versions={}`` fix → 4 new tests went red as expected; restored
  and the suite is green again. Same red/green dance for the
  ``messages=None`` coercion.

Refs #3859. Addresses second-pass review on #3874.

* fix(title): ignore dict context reminders in fallback

* fix(worker): link interrupted-title checkpoint to its parent

The title-bump checkpoint written by ``_ensure_interrupted_title`` was
landing without a ``parent_checkpoint_id`` — a real orphan in the
LangGraph history graph. Reproduction (disk-backed AsyncSqliteSaver):

  [seed] checkpoint_id = 1f173dbc...
  [helper] wrote title = "Why is the sky blue?"
  [issue 1] new checkpoint = 1f173dbc..., parent = None
  [issue 1] is new checkpoint orphaned? True

Root cause: ``_ensure_interrupted_title`` built ``write_config`` as
``{"thread_id": ..., "checkpoint_ns": ...}`` only. ``BaseCheckpointSaver``
implementations read ``configurable.checkpoint_id`` from that config as
the *parent* id when inserting (see ``langgraph/checkpoint/sqlite/aio.py``
``aput``: ``config["configurable"].get("checkpoint_id")`` becomes the
``parent_checkpoint_id`` column). With no value, the saver writes NULL —
the new checkpoint is a tree root.

Consequences:
- Any future LangGraph ``runs.resume_from`` / time-travel feature has no
  backward edge to walk past the title-bump.
- History-visualization UIs built on ``alist()`` render the title-bump as
  a sibling of the prior checkpoint, not its descendant.

Fix: read ``checkpoint_id`` off the tuple's own config and thread it into
``write_config["configurable"]["checkpoint_id"]`` before calling ``aput``,
the same pattern every middleware-driven write uses.

Three new regression tests against real ``AsyncSqliteSaver`` (disk-backed,
fresh connections so we exercise the on-disk read path):

- ``test_ensure_interrupted_title_links_new_checkpoint_to_its_parent`` —
  asserts ``latest.parent_config["configurable"]["checkpoint_id"]`` equals
  the seeded checkpoint id. TDD red-green verified: reverting the fix
  flips this test red with ``AssertionError: title-bump checkpoint must
  have a parent_config``.
- ``test_ensure_interrupted_title_appears_in_history_with_audit_marker``
  — pins the audit contract: the title-bump entry in ``alist()`` carries
  ``metadata.source == "update"`` and ``metadata.writes`` contains
  ``runtime_interrupt_title``. This is a deliberate design choice — we
  do NOT hide the entry from history (audit trail belongs in the saver),
  but its source and writes marker MUST be unambiguous so UIs/tools can
  identify it.
- ``test_ensure_interrupted_title_survives_immediate_next_turn`` —
  cancel → immediate user follow-up scenario. Simulates the agent's next
  turn appending a (user, ai) pair without touching the title channel,
  then opens a fresh saver and verifies the title is still present after
  the next-turn checkpoint write. Pins the channel-version-blob
  invariant established by commit 05253957 — without the
  ``new_versions={"title": v}`` declaration there, the title blob would
  vanish from the DB and this test would read back ``None``.

Verification:
- ``PYTHONPATH=. uv run pytest tests/ -x --ignore=tests/blocking_io -q``
  → 5222 passed, 15 skipped.
- ``ruff check`` + ``ruff format --check`` clean on every touched file.
- Reproduction script confirms ``parent_checkpoint_id`` is now non-null
  and the next-turn read-back preserves the fallback title.

Refs #3859.

* Revert "fix(worker): link interrupted-title checkpoint to its parent"

This reverts commit c763ed9334781db1acdce0f5f33d663d8d5f80ff.

* test: trim over-engineered test coverage

Reduce review surface area on PR #3874 by dropping defensive tests that
don't pin a real invariant. After self-review:

- ``_bump_channel_version``: 8 tests → 2 (happy path + saver-error
  fallback). Dropped float / bool / numeric-string / UUID-string /
  missing-get-next-version / stuck-get-next-version branches — those
  are speculative scaffolding for savers we don't ship.
- ``_ensure_interrupted_title``: dropped ``returns_none_when_title_disabled``,
  ``returns_none_with_no_user_message``, ``returns_none_when_no_checkpoint``
  — boundary guards already exercised by the e2e test and the
  ``handles_none_messages_channel`` regression anchor.

Net: -107 lines of test code. Remaining coverage still pins every
red-green-verified invariant (channel_versions bump, string-version
bump, idempotency, sqlite round-trip, non-title channel preservation,
aput-error contract, worker finally swallowing, partial-exchange).

Verification: 5209 passed, 15 skipped.

* fix: harden interrupted title finalization

* fix: serialize interrupted title finalization

* fix: preserve interrupt semantics during title finalization

* fix: preserve delayed interrupted title recovery

---------

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

README.md

@@ -606,6 +606,8 @@ Tools follow the same philosophy. DeerFlow comes with a core toolset — web sea
 
 Gateway-generated follow-up suggestions now normalize both plain-string model output and block/list-style rich content before parsing the JSON array response, so provider-specific content wrappers do not silently drop suggestions.
 
+Interrupted first-turn runs still persist a fallback conversation title, so stopping a streaming response does not leave the thread as "Untitled" after refresh.
+
 ```
 # Paths inside the sandbox container
 /mnt/skills/public

backend/AGENTS.md

@@ -224,7 +224,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the
 14. **SummarizationMiddleware** - *(optional, if enabled)* Context reduction when approaching token limits
 15. **TodoListMiddleware** - *(optional, if `is_plan_mode`)* Task tracking with the `write_todos` tool
 16. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is merged back into the dispatching AIMessage by message position
-17. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model
+17. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint.
 18. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses)
 19. **ViewImageMiddleware** - *(optional, if the model supports vision)* Injects base64 image data before the LLM call
 20. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)

backend/docs/TITLE_GENERATION_IMPLEMENTATION.md

@@ -186,6 +186,12 @@ pytest
 2. LangGraph Platform 会自动持久化
 3. 检查数据库确认 checkpointer 工作正常
 
+### 中断首轮后仍显示默认标题？
+
+1. `runtime/runs/worker.py` 会在 interrupted-run cleanup 中保持 run 处于 finalizing 状态，避免同线程新 run 在 fallback title 写入期间覆盖 checkpoint
+2. 如果取消发生在可用 checkpoint 写入前，worker 会使用本次 `graph_input` 中的首条用户消息生成本地 fallback title
+3. fallback title 写入前会重新读取 latest checkpoint；如果同线程状态已经前进，只对最新 snapshot 做 title-only 更新，避免旧消息重新成为 latest
+
 ---
 
 ## 📊 性能影响

backend/packages/harness/deerflow/agents/middlewares/title_middleware.py

@@ -63,16 +63,42 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
 
         return ""
 
+    @staticmethod
+    def _message_type(message: object) -> str | None:
+        message_type = getattr(message, "type", None)
+        if message_type is None and isinstance(message, dict):
+            message_type = message.get("type") or message.get("role")
+        if message_type == "user":
+            return "human"
+        if message_type == "assistant":
+            return "ai"
+        return message_type if isinstance(message_type, str) else None
+
+    @staticmethod
+    def _message_content(message: object) -> object:
+        if isinstance(message, dict):
+            return message.get("content", "")
+        return getattr(message, "content", "")
+
+    @staticmethod
+    def _is_dynamic_context_reminder_message(message: object) -> bool:
+        if is_dynamic_context_reminder(message):
+            return True
+        if isinstance(message, dict):
+            additional_kwargs = message.get("additional_kwargs")
+            return isinstance(additional_kwargs, dict) and bool(additional_kwargs.get("dynamic_context_reminder"))
+        return False
+
     @staticmethod
     def _is_user_message_for_title(message: object) -> bool:
-        return getattr(message, "type", None) == "human" and not is_dynamic_context_reminder(message)
+        return TitleMiddleware._message_type(message) == "human" and not TitleMiddleware._is_dynamic_context_reminder_message(message)
 
     def _get_title_user_message(self, state: TitleMiddlewareState) -> str:
-        messages = state.get("messages", [])
-        user_msg_content = next((m.content for m in messages if self._is_user_message_for_title(m)), "")
+        messages = state.get("messages") or []
+        user_msg_content = next((self._message_content(m) for m in messages if self._is_user_message_for_title(m)), "")
         return self._normalize_content(user_msg_content)
 
-    def _should_generate_title(self, state: TitleMiddlewareState) -> bool:
+    def _should_generate_title(self, state: TitleMiddlewareState, *, allow_partial_exchange: bool = False) -> bool:
         """Check if we should generate a title for this thread."""
         config = self._get_title_config()
         if not config.enabled:
@@ -82,17 +108,23 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
         if state.get("title"):
             return False
 
-        # Check if this is the first turn (has at least one user message and one assistant response)
-        messages = state.get("messages", [])
-        if len(messages) < 2:
+        # Check if this is the first turn (has at least one user message and one assistant response).
+        # Defensively coerce a None ``messages`` channel (possible when reading a
+        # partially-initialized checkpoint) into an empty list so ``len()`` is safe.
+        messages = state.get("messages") or []
+        min_messages = 1 if allow_partial_exchange else 2
+        if len(messages) < min_messages:
             return False
 
         # Count user and assistant messages
         user_messages = [m for m in messages if self._is_user_message_for_title(m)]
-        assistant_messages = [m for m in messages if m.type == "ai"]
+        assistant_messages = [m for m in messages if self._message_type(m) == "ai"]
 
-        # Generate title after first complete exchange
-        return len(user_messages) == 1 and len(assistant_messages) >= 1
+        # Normal path: title only after first complete exchange. Interrupted path
+        # (``allow_partial_exchange=True``) accepts a lone first-turn user message
+        # so a fallback title can still be persisted when the run is cancelled
+        # before any AI chunk reaches the checkpoint.
+        return len(user_messages) == 1 and (len(assistant_messages) >= 1 or allow_partial_exchange)
 
     def _build_title_prompt(self, state: TitleMiddlewareState) -> tuple[str, str]:
         """Extract user/assistant messages and build the title prompt.
@@ -100,9 +132,9 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
         Returns (prompt_string, user_msg) so callers can use user_msg as fallback.
         """
         config = self._get_title_config()
-        messages = state.get("messages", [])
+        messages = state.get("messages") or []
 
-        assistant_msg_content = next((m.content for m in messages if m.type == "ai"), "")
+        assistant_msg_content = next((self._message_content(m) for m in messages if self._message_type(m) == "ai"), "")
 
         user_msg = self._get_title_user_message(state)
         assistant_msg = self._strip_think_tags(self._normalize_content(assistant_msg_content))
@@ -152,9 +184,9 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
         ]
         return config
 
-    def _generate_title_result(self, state: TitleMiddlewareState) -> dict | None:
+    def _generate_title_result(self, state: TitleMiddlewareState, *, allow_partial_exchange: bool = False) -> dict | None:
         """Generate a local fallback title without blocking on an LLM call."""
-        if not self._should_generate_title(state):
+        if not self._should_generate_title(state, allow_partial_exchange=allow_partial_exchange):
             return None
 
         user_msg = self._get_title_user_message(state)

backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -104,6 +104,7 @@ class RunRecord:
     message_count: int = 0
     last_ai_message: str | None = None
     first_human_message: str | None = None
+    finalizing: bool = False
 
 
 class RunManager:
@@ -484,6 +485,64 @@ class RunManager:
         await self._persist_status(record, status, error=error)
         logger.info("Run %s -> %s", run_id, status.value)
 
+    async def set_finalizing(self, run_id: str, finalizing: bool) -> None:
+        """Mark whether a run is performing post-cancel cleanup."""
+        async with self._lock:
+            record = self._runs.get(run_id)
+            if record is None:
+                logger.warning("set_finalizing called for unknown run %s", run_id)
+                return
+            record.finalizing = finalizing
+            record.updated_at = _now_iso()
+
+    async def wait_for_prior_finalizing(
+        self,
+        thread_id: str,
+        run_id: str,
+        *,
+        poll_interval: float = 0.01,
+    ) -> None:
+        """Wait until older same-thread runs have finished post-cancel cleanup."""
+        while True:
+            async with self._lock:
+                found_current = False
+                prior_finalizing = False
+                for record in self._thread_records_locked(thread_id):
+                    if record.run_id == run_id:
+                        found_current = True
+                        break
+                    if record.finalizing:
+                        prior_finalizing = True
+
+                if not found_current or not prior_finalizing:
+                    return
+
+            await asyncio.sleep(poll_interval)
+
+    async def has_later_run(self, thread_id: str, run_id: str) -> bool:
+        """Return whether a newer in-memory run has been admitted for the thread."""
+        async with self._lock:
+            seen_current = False
+            for record in self._thread_records_locked(thread_id):
+                if record.run_id == run_id:
+                    seen_current = True
+                    continue
+                if seen_current:
+                    return True
+        return False
+
+    async def has_later_started_run(self, thread_id: str, run_id: str) -> bool:
+        """Return whether a newer same-thread run may have already advanced state."""
+        async with self._lock:
+            seen_current = False
+            for record in self._thread_records_locked(thread_id):
+                if record.run_id == run_id:
+                    seen_current = True
+                    continue
+                if seen_current and (record.status != RunStatus.pending or record.finalizing):
+                    return True
+        return False
+
     async def _persist_model_name(self, run_id: str, model_name: str | None) -> None:
         """Best-effort persist model_name update to the backing store."""
         if self._store is None:
@@ -532,7 +591,9 @@ class RunManager:
                 return False
             record.abort_action = action
             record.abort_event.set()
-            if record.task is not None and not record.task.done():
+            task_active = record.task is not None and not record.task.done()
+            record.finalizing = task_active
+            if task_active:
                 record.task.cancel()
             record.status = RunStatus.interrupted
             record.updated_at = _now_iso()
@@ -571,7 +632,7 @@ class RunManager:
             if multitask_strategy not in _supported_strategies:
                 raise UnsupportedStrategyError(f"Multitask strategy '{multitask_strategy}' is not yet supported. Supported strategies: {', '.join(_supported_strategies)}")
 
-            inflight = [r for r in self._thread_records_locked(thread_id) if r.status in (RunStatus.pending, RunStatus.running)]
+            inflight = [r for r in self._thread_records_locked(thread_id) if r.status in (RunStatus.pending, RunStatus.running) or r.finalizing]
 
             if multitask_strategy == "reject" and inflight:
                 raise ConflictError(f"Thread {thread_id} already has an active run")
@@ -615,9 +676,13 @@ class RunManager:
 
             if multitask_strategy in ("interrupt", "rollback") and inflight:
                 for r in inflight:
+                    if r.finalizing:
+                        continue
                     r.abort_action = multitask_strategy
                     r.abort_event.set()
-                    if r.task is not None and not r.task.done():
+                    task_active = r.task is not None and not r.task.done()
+                    r.finalizing = task_active
+                    if task_active:
                         r.task.cancel()
                     r.status = RunStatus.interrupted
                     r.updated_at = now
@@ -685,7 +750,7 @@ class RunManager:
     async def has_inflight(self, thread_id: str) -> bool:
         """Return ``True`` if *thread_id* has a pending or running run."""
         async with self._lock:
-            return any(r.status in (RunStatus.pending, RunStatus.running) for r in self._thread_records_locked(thread_id))
+            return any(r.status in (RunStatus.pending, RunStatus.running) or r.finalizing for r in self._thread_records_locked(thread_id))
 
     async def cleanup(self, run_id: str, *, delay: float = 300) -> None:
         """Remove a run record after an optional delay."""

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -159,6 +159,8 @@ async def run_agent(
         )
 
     try:
+        await run_manager.wait_for_prior_finalizing(thread_id, run_id)
+
         # Initialize RunJournal + write human_message event.
         # These are inside the try block so any exception (e.g. a DB
         # error writing the event) flows through the except/finally
@@ -335,6 +337,7 @@ async def run_agent(
 
         # 8. Final status
         if record.abort_event.is_set():
+            await run_manager.set_finalizing(run_id, True)
             action = record.abort_action
             if action == "rollback":
                 await run_manager.set_status(run_id, RunStatus.error, error="Rolled back by user")
@@ -362,6 +365,7 @@ async def run_agent(
             await run_manager.set_status(run_id, RunStatus.success)
 
     except asyncio.CancelledError:
+        await run_manager.set_finalizing(run_id, True)
         action = record.abort_action
         if action == "rollback":
             await run_manager.set_status(run_id, RunStatus.error, error="Rolled back by user")
@@ -409,6 +413,14 @@ async def run_agent(
             except Exception:
                 logger.warning("Failed to persist run completion for %s (non-fatal)", run_id, exc_info=True)
 
+        if checkpointer is not None and record.status == RunStatus.interrupted:
+            try:
+                await run_manager.wait_for_prior_finalizing(thread_id, run_id)
+                if not await run_manager.has_later_started_run(thread_id, run_id):
+                    await _ensure_interrupted_title(checkpointer=checkpointer, thread_id=thread_id, app_config=ctx.app_config, graph_input=graph_input)
+            except Exception:
+                logger.debug("Failed to generate interrupted title for thread %s (non-fatal)", thread_id)
+
         # Sync title from checkpoint to threads_meta.display_name
         if checkpointer is not None and thread_store is not None:
             try:
@@ -430,6 +442,9 @@ async def run_agent(
             except Exception:
                 logger.debug("Failed to update thread_meta status for %s (non-fatal)", thread_id)
 
+        if record.finalizing:
+            await run_manager.set_finalizing(run_id, False)
+
         await bridge.publish_end(run_id)
         asyncio.create_task(bridge.cleanup(run_id, delay=60))
 
@@ -548,6 +563,167 @@ def _new_checkpoint_marker() -> dict[str, str]:
     return {"id": marker["id"], "ts": marker["ts"]}
 
 
+def _bump_channel_version(checkpointer: Any, current_version: Any) -> Any:
+    """Return a strictly-different next version for a checkpoint channel.
+
+    DB-backed LangGraph savers (PostgresSaver / v4 SqliteSaver blob layout)
+    persist channel blobs keyed by ``channel_versions[<channel>]``, so the
+    new value MUST differ from the prior value. We delegate to the
+    checkpointer's ``get_next_version`` when available — that is the canonical
+    versioning scheme each saver picks (int, monotonic float, or
+    UUID-shaped string). When the checkpointer doesn't expose it (or it
+    returns ``None``/an unchanged value), fall back to a defensive bump that
+    still guarantees inequality.
+    """
+    get_next_version = getattr(checkpointer, "get_next_version", None)
+    if callable(get_next_version):
+        try:
+            next_version = get_next_version(current_version, None)
+        except Exception:
+            next_version = None
+        if next_version is not None and next_version != current_version:
+            return next_version
+        # fall through to defensive bump
+
+    if isinstance(current_version, bool):
+        # ``bool`` is a subclass of ``int``; treat True/False as 1/0 instead of
+        # adding to the boolean itself, which would produce an int anyway but
+        # via a path that surprises readers.
+        return int(current_version) + 1
+    if isinstance(current_version, int):
+        return current_version + 1
+    if isinstance(current_version, float):
+        # Match LangGraph's default float versioning (monotonic increment).
+        return current_version + 1.0
+    if isinstance(current_version, str):
+        try:
+            return str(int(current_version) + 1)
+        except ValueError:
+            return f"{current_version}.1"
+    return 1
+
+
+def _checkpoint_identity(ckpt_tuple: Any | None, checkpoint: dict[str, Any]) -> str | None:
+    tuple_config = getattr(ckpt_tuple, "config", {}) or {}
+    tuple_configurable = tuple_config.get("configurable", {}) if isinstance(tuple_config, dict) else {}
+    if isinstance(tuple_configurable, dict):
+        checkpoint_id = tuple_configurable.get("checkpoint_id")
+        if isinstance(checkpoint_id, str) and checkpoint_id:
+            return checkpoint_id
+    checkpoint_id = checkpoint.get("id")
+    return checkpoint_id if isinstance(checkpoint_id, str) and checkpoint_id else None
+
+
+def _checkpoint_namespace(ckpt_tuple: Any | None) -> str:
+    tuple_config = getattr(ckpt_tuple, "config", {}) or {}
+    tuple_configurable = tuple_config.get("configurable", {}) if isinstance(tuple_config, dict) else {}
+    checkpoint_ns = tuple_configurable.get("checkpoint_ns", "") if isinstance(tuple_configurable, dict) else ""
+    return checkpoint_ns if isinstance(checkpoint_ns, str) else ""
+
+
+def _graph_input_messages(graph_input: Any | None) -> list[Any]:
+    if not isinstance(graph_input, dict):
+        return []
+    messages = graph_input.get("messages")
+    if isinstance(messages, list):
+        return messages
+    if isinstance(messages, tuple):
+        return list(messages)
+    return []
+
+
+def _title_generation_state(channel_values: dict[str, Any], graph_input: Any | None) -> dict[str, Any]:
+    state = dict(channel_values)
+    messages = state.get("messages")
+    if not messages:
+        fallback_messages = _graph_input_messages(graph_input)
+        if fallback_messages:
+            state["messages"] = fallback_messages
+    return state
+
+
+async def _ensure_interrupted_title(*, checkpointer: Any, thread_id: str, app_config: AppConfig | None, graph_input: Any | None = None) -> str | None:
+    """Persist a local fallback title for interrupted first-turn runs.
+
+    Returns the title that is now persisted (existing or newly written), or
+    ``None`` when no checkpoint is available or no title text can be derived.
+    Idempotent: re-invoking against a checkpoint that already carries a title
+    short-circuits without writing a new checkpoint.
+    """
+    from deerflow.agents.middlewares.title_middleware import TitleMiddleware
+
+    middleware = TitleMiddleware(app_config=app_config) if app_config is not None else TitleMiddleware()
+    ckpt_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+
+    for _attempt in range(3):
+        ckpt_tuple = await _call_checkpointer_method(checkpointer, "aget_tuple", "get_tuple", ckpt_config)
+        checkpoint = copy.deepcopy(getattr(ckpt_tuple, "checkpoint", {}) or {}) if ckpt_tuple is not None else empty_checkpoint()
+        channel_values = dict(checkpoint.get("channel_values", {}) or {})
+        existing_title = channel_values.get("title")
+        if existing_title:
+            return existing_title
+
+        result = middleware._generate_title_result(_title_generation_state(channel_values, graph_input), allow_partial_exchange=True)
+        title = result.get("title") if isinstance(result, dict) else None
+        if not title:
+            return None
+
+        # ``empty_checkpoint()`` creates a fresh id every time; only real tuples
+        # carry an identity stable enough for the stale-snapshot comparison.
+        base_identity = _checkpoint_identity(ckpt_tuple, checkpoint) if ckpt_tuple is not None else None
+        latest_tuple = await _call_checkpointer_method(checkpointer, "aget_tuple", "get_tuple", ckpt_config)
+        latest_checkpoint = copy.deepcopy(getattr(latest_tuple, "checkpoint", {}) or {}) if latest_tuple is not None else empty_checkpoint()
+        latest_identity = _checkpoint_identity(latest_tuple, latest_checkpoint) if latest_tuple is not None else None
+        if base_identity is None:
+            if latest_identity is not None:
+                continue
+        elif latest_identity != base_identity:
+            continue
+
+        checkpoint = latest_checkpoint
+        channel_values = dict(checkpoint.get("channel_values", {}) or {})
+        existing_title = channel_values.get("title")
+        if existing_title:
+            return existing_title
+
+        channel_values["title"] = title
+        marker = _new_checkpoint_marker()
+        checkpoint.update({"id": marker["id"], "ts": marker["ts"], "channel_values": channel_values})
+
+        # Bump ``channel_versions["title"]`` and declare the bump in ``new_versions``
+        # so DB-backed savers (SqliteSaver v4 / PostgresSaver) actually persist the
+        # new blob — those savers strip inline ``channel_values`` from ``put`` and
+        # only write blobs for channels listed in ``new_versions``. The legacy
+        # single-table sqlite saver ignores ``new_versions`` and inlines the
+        # snapshot, so this path is correct for both layouts. Mirrors
+        # ``_rollback_to_pre_run_checkpoint`` in the same file.
+        channel_versions = dict(checkpoint.get("channel_versions", {}) or {})
+        next_title_version = _bump_channel_version(checkpointer, channel_versions.get("title"))
+        channel_versions["title"] = next_title_version
+        checkpoint["channel_versions"] = channel_versions
+
+        metadata = dict(getattr(latest_tuple, "metadata", {}) or {})
+        metadata["source"] = "update"
+        prev_step = metadata.get("step")
+        metadata["step"] = (prev_step + 1) if isinstance(prev_step, int) else 1
+        metadata["writes"] = {"runtime_interrupt_title": {"title": title}}
+
+        checkpoint_ns = _checkpoint_namespace(latest_tuple)
+        write_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns}}
+        await _call_checkpointer_method(
+            checkpointer,
+            "aput",
+            "put",
+            write_config,
+            checkpoint,
+            metadata,
+            {"title": next_title_version},
+        )
+        return title
+
+    return None
+
+
 def _lg_mode_to_sse_event(mode: str) -> str:
     """Map LangGraph internal stream_mode name to SSE event name.
 

backend/tests/test_runtime_lifecycle_e2e.py

@@ -80,11 +80,15 @@ class _ScriptedAgent:
         title: str,
         answer: str,
         block_after_first_chunk: bool = False,
+        block_before_checkpoint: bool = False,
+        write_title: bool = True,
     ) -> None:
         self.controller = controller
         self.title = title
         self.answer = answer
         self.block_after_first_chunk = block_after_first_chunk
+        self.block_before_checkpoint = block_before_checkpoint
+        self.write_title = write_title
         self.checkpointer: Any | None = None
         self.store: Any | None = None
         self.metadata = {"model_name": "fake-test-model"}
@@ -98,10 +102,15 @@ class _ScriptedAgent:
 
         try:
             thread_id = _thread_id_from_config(config)
+            if self.block_before_checkpoint:
+                while not self.controller.release.is_set():
+                    await asyncio.sleep(0.05)
             human_text = _last_human_text(graph_input)
             human = HumanMessage(content=human_text)
             ai = await self.model.ainvoke([human], config=config)
-            state = {"messages": [human.model_dump(), ai.model_dump()], "title": self.title}
+            state = {"messages": [human.model_dump(), ai.model_dump()]}
+            if self.write_title:
+                state["title"] = self.title
 
             if self.checkpointer is not None:
                 await _write_checkpoint(self.checkpointer, thread_id=thread_id, state=state)
@@ -265,6 +274,25 @@ def isolated_app(isolated_deer_flow_home: Path, monkeypatch: pytest.MonkeyPatch)
     return create_app()
 
 
+@pytest.fixture
+def isolated_app_with_title(isolated_deer_flow_home: Path, monkeypatch: pytest.MonkeyPatch):
+    config_path = isolated_deer_flow_home.parent / "config-title-enabled.yaml"
+    config_path.write_text(_MINIMAL_CONFIG_YAML.replace("title:\n  enabled: false", "title:\n  enabled: true"), encoding="utf-8")
+    monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
+
+    _preserve_process_config_singletons(monkeypatch)
+    _reset_process_singletons(monkeypatch)
+
+    from deerflow.config import app_config as app_config_module
+
+    cfg = app_config_module.get_app_config()
+    cfg.database.sqlite_dir = str(isolated_deer_flow_home / "db")
+
+    from app.gateway.app import create_app
+
+    return create_app()
+
+
 def _register_user(client, *, email: str = "runtime-e2e@example.com") -> str:
     response = client.post(
         "/api/v1/auth/register",
@@ -379,6 +407,34 @@ def _wait_for_status(client, thread_id: str, run_id: str, status: str, *, timeou
     raise AssertionError(f"Run {run_id} did not reach {status!r}; last={last!r}")
 
 
+def _wait_for_thread_title(client, thread_id: str, expected_title: str, *, timeout: float = 5.0) -> dict:
+    deadline = time.monotonic() + timeout
+    last: dict | None = None
+    while time.monotonic() < deadline:
+        response = client.get(f"/api/threads/{thread_id}")
+        assert response.status_code == 200, response.text
+        last = response.json()
+        if last.get("values", {}).get("title") == expected_title:
+            return last
+        time.sleep(0.05)
+    raise AssertionError(f"Thread {thread_id} did not reach title {expected_title!r}; last={last!r}")
+
+
+def _wait_for_search_title(client, csrf_token: str, thread_id: str, expected_title: str, *, timeout: float = 5.0) -> dict:
+    deadline = time.monotonic() + timeout
+    last_match: dict | None = None
+    while time.monotonic() < deadline:
+        response = client.post("/api/threads/search", json={"limit": 20}, headers={"X-CSRF-Token": csrf_token})
+        assert response.status_code == 200, response.text
+        matching = [item for item in response.json() if item["thread_id"] == thread_id]
+        if matching:
+            last_match = matching[0]
+            if last_match.get("values", {}).get("title") == expected_title:
+                return last_match
+        time.sleep(0.05)
+    raise AssertionError(f"Search result for {thread_id} did not reach title {expected_title!r}; last={last_match!r}")
+
+
 def _thread_id_from_config(config: dict | None) -> str:
     config = config or {}
     context = config.get("context") if isinstance(config.get("context"), dict) else {}
@@ -581,6 +637,99 @@ def test_cancel_interrupt_stops_running_background_run(isolated_app):
         assert thread.json()["status"] == "idle"
 
 
+def test_cancel_interrupt_generates_missing_title_from_checkpoint(isolated_app_with_title):
+    """Interrupted first-turn runs should still persist an automatic thread title."""
+    from starlette.testclient import TestClient
+
+    controller = _RunController()
+    factory = _make_agent_factory(
+        controller,
+        title="",
+        answer="This run should be interrupted before a title is written.",
+        block_after_first_chunk=True,
+        write_title=False,
+    )
+
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=factory),
+        TestClient(isolated_app_with_title) as client,
+    ):
+        csrf_token = _register_user(client, email="interrupt-title-e2e@example.com")
+        thread_id = _create_thread(client, csrf_token)
+
+        created = client.post(
+            f"/api/threads/{thread_id}/runs",
+            json=_run_body(),
+            headers={"X-CSRF-Token": csrf_token},
+        )
+        assert created.status_code == 200, created.text
+        run_id = created.json()["run_id"]
+        assert controller.checkpoint_written.wait(5), "fake agent never wrote checkpoint"
+
+        cancelled = client.post(
+            f"/api/threads/{thread_id}/runs/{run_id}/cancel?wait=true&action=interrupt",
+            headers={"X-CSRF-Token": csrf_token},
+        )
+        assert cancelled.status_code == 204, cancelled.text
+
+        thread = client.get(f"/api/threads/{thread_id}")
+        assert thread.status_code == 200, thread.text
+        assert thread.json()["values"]["title"] == "Run lifecycle E2E prompt"
+
+        search = client.post("/api/threads/search", json={"limit": 20}, headers={"X-CSRF-Token": csrf_token})
+        assert search.status_code == 200, search.text
+        matching = [item for item in search.json() if item["thread_id"] == thread_id]
+        assert matching[0]["values"]["title"] == "Run lifecycle E2E prompt"
+
+
+def test_cancel_wait_false_generates_title_from_graph_input_before_checkpoint(isolated_app_with_title):
+    """Fire-and-forget cancel should title early interruptions before checkpoint."""
+    from starlette.testclient import TestClient
+
+    controller = _RunController()
+    factory = _make_agent_factory(
+        controller,
+        title="",
+        answer="This answer should never be checkpointed.",
+        block_before_checkpoint=True,
+        write_title=False,
+    )
+
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=factory),
+        TestClient(isolated_app_with_title) as client,
+    ):
+        csrf_token = _register_user(client, email="interrupt-title-early-e2e@example.com")
+        thread_id = _create_thread(client, csrf_token)
+
+        created = client.post(
+            f"/api/threads/{thread_id}/runs",
+            json=_run_body(),
+            headers={"X-CSRF-Token": csrf_token},
+        )
+        assert created.status_code == 200, created.text
+        run_id = created.json()["run_id"]
+        assert controller.started.wait(5), "fake agent never started"
+        assert not controller.checkpoint_written.is_set()
+
+        cancelled = client.post(
+            f"/api/threads/{thread_id}/runs/{run_id}/cancel?wait=false&action=interrupt",
+            headers={"X-CSRF-Token": csrf_token},
+        )
+        assert cancelled.status_code == 202, cancelled.text
+        assert controller.cancelled.wait(5), "fake agent task was not cancelled"
+        assert not controller.checkpoint_written.is_set()
+
+        run = _wait_for_status(client, thread_id, run_id, "interrupted")
+        assert run["status"] == "interrupted"
+
+        thread = _wait_for_thread_title(client, thread_id, "Run lifecycle E2E prompt")
+        assert thread["values"]["title"] == "Run lifecycle E2E prompt"
+
+        matching = _wait_for_search_title(client, csrf_token, thread_id, "Run lifecycle E2E prompt")
+        assert matching["values"]["title"] == "Run lifecycle E2E prompt"
+
+
 @pytest.mark.anyio
 async def test_sse_consumer_disconnect_cancels_inflight_run():
     """A disconnected SSE request should cancel an in-flight run when configured."""

backend/tests/test_title_middleware_core_logic.py

@@ -367,6 +367,113 @@ class TestTitleMiddlewareCoreLogic:
         assert "<system-reminder>" not in prompt
         assert "User prefers Python" not in prompt
 
+    def test_should_generate_title_partial_exchange_allows_user_only(self):
+        """Interrupted-run path can produce a fallback from a lone human message."""
+        _set_test_title_config(enabled=True)
+        middleware = TitleMiddleware()
+        state = {"messages": [HumanMessage(content="只有人类消息，AI 还没回复")]}
+
+        assert middleware._should_generate_title(state) is False
+        assert middleware._should_generate_title(state, allow_partial_exchange=True) is True
+
+    def test_should_generate_title_partial_exchange_skips_when_titled(self):
+        """Existing title still wins, even on the interrupted-run path."""
+        _set_test_title_config(enabled=True)
+        middleware = TitleMiddleware()
+        state = {
+            "messages": [HumanMessage(content="问题")],
+            "title": "Already set",
+        }
+        assert middleware._should_generate_title(state, allow_partial_exchange=True) is False
+
+    def test_should_generate_title_handles_dict_messages(self):
+        """Checkpoint channel_values store messages as dicts; the middleware must accept them."""
+        _set_test_title_config(enabled=True)
+        middleware = TitleMiddleware()
+        state = {
+            "messages": [
+                {"type": "human", "content": "问"},
+                {"type": "ai", "content": "答"},
+            ]
+        }
+        assert middleware._should_generate_title(state) is True
+
+    def test_sync_generate_title_from_dict_messages(self):
+        """Sync fallback path can derive title text from dict-form messages."""
+        _set_test_title_config(max_chars=20)
+        middleware = TitleMiddleware()
+        state = {
+            "messages": [
+                {"role": "user", "content": "请帮我写测试"},
+                {"role": "assistant", "content": "好的"},
+            ]
+        }
+        assert middleware._generate_title_result(state) == {"title": "请帮我写测试"}
+
+    def test_should_generate_title_handles_none_messages_channel(self):
+        """A checkpoint with ``messages=None`` (partially-initialized state) must not crash."""
+        _set_test_title_config(enabled=True)
+        middleware = TitleMiddleware()
+        # ``messages`` key exists but is None — ``state.get("messages", [])`` would
+        # have returned ``None`` (default only applies on missing key), so this
+        # exercises the ``or []`` coercion the helper relies on.
+        state = {"messages": None}
+
+        assert middleware._should_generate_title(state) is False
+        assert middleware._should_generate_title(state, allow_partial_exchange=True) is False
+
+    def test_build_title_prompt_handles_none_messages_channel(self):
+        """``_build_title_prompt`` must also tolerate a None messages channel."""
+        _set_test_title_config(enabled=True)
+        middleware = TitleMiddleware()
+        state = {"messages": None}
+
+        prompt, user_msg = middleware._build_title_prompt(state)
+        assert user_msg == ""
+        # Prompt is still well-formed — just empty user/assistant slots.
+        assert "{user_msg}" not in prompt  # the template was formatted, not left raw
+
+    def test_should_generate_title_dict_messages_role_normalization(self):
+        """Dict-form messages may use either ``type`` or ``role``; both must map correctly."""
+        _set_test_title_config(enabled=True)
+        middleware = TitleMiddleware()
+        state = {
+            "messages": [
+                # ``role: user`` should be normalized to ``human``
+                {"role": "user", "content": "Q"},
+                # ``role: assistant`` should be normalized to ``ai``
+                {"role": "assistant", "content": "A"},
+            ]
+        }
+        assert middleware._should_generate_title(state) is True
+
+    def test_partial_exchange_with_dict_human_message(self):
+        """Interrupted-run path must accept a lone dict-form first-turn user message."""
+        _set_test_title_config(enabled=True, max_chars=20)
+        middleware = TitleMiddleware()
+        state = {"messages": [{"role": "user", "content": "请帮我写测试"}]}
+
+        result = middleware._generate_title_result(state, allow_partial_exchange=True)
+        assert result == {"title": "请帮我写测试"}
+
+    def test_partial_exchange_ignores_dict_dynamic_context_reminder(self):
+        """Checkpoint dicts can include hidden memory reminders that should not count as real user turns."""
+        _set_test_title_config(enabled=True, max_chars=20)
+        middleware = TitleMiddleware()
+        state = {
+            "messages": [
+                {
+                    "type": "human",
+                    "content": "<memory>User prefers concise titles.</memory>",
+                    "additional_kwargs": {"hide_from_ui": True, _DYNAMIC_CONTEXT_REMINDER_KEY: True},
+                },
+                {"type": "human", "content": "请帮我写测试", "additional_kwargs": {}},
+            ]
+        }
+
+        assert middleware._should_generate_title(state, allow_partial_exchange=True) is True
+        assert middleware._generate_title_result(state, allow_partial_exchange=True) == {"title": "请帮我写测试"}
+
     def test_generate_title_async_strips_think_tags_in_response(self, monkeypatch):
         """Async title generation strips <think> blocks from the model response."""
         _set_test_title_config(max_chars=50, model_name="title-model")

backend/tests/test_worker_langfuse_metadata.py

@@ -36,6 +36,15 @@ class _FakeAgent:
 
 
 class _FakeRunManager:
+    async def wait_for_prior_finalizing(self, *_args, **_kwargs) -> None:
+        return None
+
+    async def has_later_run(self, *_args, **_kwargs) -> bool:
+        return False
+
+    async def has_later_started_run(self, *_args, **_kwargs) -> bool:
+        return False
+
     async def set_status(self, *_args, **_kwargs) -> None:
         return None
 

frontend/AGENTS.md

@@ -65,8 +65,9 @@ The frontend is a stateful chat application. Users create **threads** (conversat
 
 1. User input → thread hooks (`core/threads/hooks.ts`) → LangGraph SDK streaming
 2. Stream events update thread state (messages, artifacts, todos)
-3. TanStack Query manages server state; localStorage stores user settings
-4. Components subscribe to thread state and render updates
+3. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits
+4. TanStack Query manages server state; localStorage stores user settings
+5. Components subscribe to thread state and render updates
 
 ### Key Patterns
 

frontend/src/core/threads/hooks.ts

@@ -606,6 +606,58 @@ export function upsertThreadInInfiniteCache(
   );
 }
 
+export function invalidateStoppedThreadCaches(
+  queryClient: QueryClient,
+  threadId: string | null | undefined,
+  isMock = false,
+) {
+  void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
+  void queryClient.invalidateQueries({
+    queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
+  });
+
+  if (!threadId || isMock) {
+    return;
+  }
+
+  void queryClient.invalidateQueries({ queryKey: ["thread", threadId] });
+  void queryClient.invalidateQueries({
+    queryKey: ["thread", "metadata", threadId, isMock],
+  });
+  void queryClient.invalidateQueries({
+    queryKey: threadTokenUsageQueryKey(threadId),
+  });
+}
+
+export const STOP_THREAD_FINALIZATION_REFETCH_DELAY_MS = 1500;
+
+function scheduleStoppedThreadFinalizationRefetch(
+  queryClient: QueryClient,
+  threadId: string | null | undefined,
+  isMock = false,
+) {
+  if (isMock) {
+    return;
+  }
+  globalThis.setTimeout(() => {
+    invalidateStoppedThreadCaches(queryClient, threadId, isMock);
+  }, STOP_THREAD_FINALIZATION_REFETCH_DELAY_MS);
+}
+
+export async function stopThreadAndInvalidateCaches(
+  queryClient: QueryClient,
+  stop: () => Promise<void> | void,
+  threadId: string | null | undefined,
+  isMock = false,
+) {
+  try {
+    await stop();
+  } finally {
+    invalidateStoppedThreadCaches(queryClient, threadId, isMock);
+    scheduleStoppedThreadFinalizationRefetch(queryClient, threadId, isMock);
+  }
+}
+
 function getStreamErrorMessage(error: unknown): string {
   if (typeof error === "string" && error.trim()) {
     return error;
@@ -958,21 +1010,21 @@ export function useThreadStream({
           .map(messageIdentity)
           .filter((id): id is string => Boolean(id)),
       );
-      void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
-      void queryClient.invalidateQueries({
-        queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
-      });
-      if (threadIdRef.current && !isMock) {
-        void queryClient.invalidateQueries({
-          queryKey: ["thread", threadIdRef.current],
-        });
-        void queryClient.invalidateQueries({
-          queryKey: threadTokenUsageQueryKey(threadIdRef.current),
-        });
-      }
+      invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock);
     },
   });
 
+  const stopThread = useCallback(async () => {
+    const stoppedThreadId =
+      threadIdRef.current ?? displayThreadId ?? threadId ?? null;
+    await stopThreadAndInvalidateCaches(
+      queryClient,
+      () => thread.stop(),
+      stoppedThreadId,
+      isMock,
+    );
+  }, [displayThreadId, isMock, queryClient, thread, threadId]);
+
   const hasVisibleStreamState =
     Boolean(threadId) || liveMessagesThreadId === currentViewThreadId;
   const persistedMessages = useMemo(
@@ -1443,6 +1495,7 @@ export function useThreadStream({
   // History messages may overlap with thread.messages; thread.messages take precedence
   const mergedThread = {
     ...thread,
+    stop: stopThread,
     values: hasVisibleStreamState ? thread.values : EMPTY_THREAD_VALUES,
     messages: mergedMessages,
   } as typeof thread;

frontend/tests/unit/core/threads/infinite.test.ts

@@ -1,12 +1,19 @@
-import { describe, expect, test } from "@rstest/core";
-import { QueryClient, type InfiniteData } from "@tanstack/react-query";
+import { describe, expect, rs, test } from "@rstest/core";
+import {
+  QueryClient,
+  QueryObserver,
+  type InfiniteData,
+} from "@tanstack/react-query";
 
 import {
   filterInfiniteThreadsCache,
   getInfiniteThreadsNextPageParam,
   INFINITE_THREADS_PAGE_SIZE,
   INFINITE_THREADS_QUERY_KEY_PREFIX,
+  invalidateStoppedThreadCaches,
   mapInfiniteThreadsCache,
+  STOP_THREAD_FINALIZATION_REFETCH_DELAY_MS,
+  stopThreadAndInvalidateCaches,
   upsertThreadInInfiniteCache,
 } from "@/core/threads/hooks";
 import type { AgentThread } from "@/core/threads/types";
@@ -226,3 +233,183 @@ describe("upsertThreadInInfiniteCache", () => {
     expect(cache?.pages[0]?.[0]?.values.title).toBe("Old title");
   });
 });
+
+describe("invalidateStoppedThreadCaches", () => {
+  function invalidatedQueryKeys(client: QueryClient) {
+    const invalidate = rs.spyOn(client, "invalidateQueries");
+    return {
+      invalidate,
+      queryKeys: () =>
+        invalidate.mock.calls.map(([filters]) => filters?.queryKey),
+    };
+  }
+
+  test("refreshes current thread and sidebar caches after fire-and-forget stop", () => {
+    const client = new QueryClient();
+    const { queryKeys } = invalidatedQueryKeys(client);
+
+    invalidateStoppedThreadCaches(client, "thread-1", false);
+
+    expect(queryKeys()).toContainEqual(["threads", "search"]);
+    expect(queryKeys()).toContainEqual(INFINITE_THREADS_QUERY_KEY_PREFIX);
+    expect(queryKeys()).toContainEqual(["thread", "thread-1"]);
+    expect(queryKeys()).toContainEqual([
+      "thread",
+      "metadata",
+      "thread-1",
+      false,
+    ]);
+    expect(queryKeys()).toContainEqual(["thread-token-usage", "thread-1"]);
+  });
+
+  test("does not refresh per-thread API caches for mock threads", () => {
+    const client = new QueryClient();
+    const { queryKeys } = invalidatedQueryKeys(client);
+
+    invalidateStoppedThreadCaches(client, "thread-1", true);
+
+    expect(queryKeys()).toContainEqual(["threads", "search"]);
+    expect(queryKeys()).toContainEqual(INFINITE_THREADS_QUERY_KEY_PREFIX);
+    expect(queryKeys()).not.toContainEqual(["thread", "thread-1"]);
+    expect(queryKeys()).not.toContainEqual([
+      "thread",
+      "metadata",
+      "thread-1",
+      true,
+    ]);
+    expect(queryKeys()).not.toContainEqual(["thread-token-usage", "thread-1"]);
+  });
+
+  test("wraps SDK stop and refreshes caches after it resolves", async () => {
+    const client = new QueryClient();
+    const stop = rs.fn(() => Promise.resolve());
+    const { queryKeys } = invalidatedQueryKeys(client);
+
+    await stopThreadAndInvalidateCaches(client, stop, "thread-1", false);
+
+    expect(stop).toHaveBeenCalledTimes(1);
+    expect(queryKeys()).toContainEqual([
+      "thread",
+      "metadata",
+      "thread-1",
+      false,
+    ]);
+  });
+
+  test("still refreshes caches when SDK stop rejects", async () => {
+    const client = new QueryClient();
+    const stop = rs.fn(async () => {
+      throw new Error("cancel failed");
+    });
+    const { queryKeys } = invalidatedQueryKeys(client);
+
+    await expect(
+      stopThreadAndInvalidateCaches(client, stop, "thread-1", false),
+    ).rejects.toThrow("cancel failed");
+
+    expect(queryKeys()).toContainEqual(["threads", "search"]);
+    expect(queryKeys()).toContainEqual([
+      "thread",
+      "metadata",
+      "thread-1",
+      false,
+    ]);
+  });
+
+  test("schedules sidebar refetch even if stopped thread id is not known", async () => {
+    rs.useFakeTimers();
+
+    const client = new QueryClient();
+    const { queryKeys } = invalidatedQueryKeys(client);
+
+    try {
+      await stopThreadAndInvalidateCaches(
+        client,
+        () => Promise.resolve(),
+        null,
+        false,
+      );
+
+      const countSearchInvalidations = () =>
+        queryKeys().filter(
+          (queryKey) =>
+            queryKey?.length === 2 &&
+            queryKey[0] === "threads" &&
+            queryKey[1] === "search",
+        ).length;
+
+      expect(countSearchInvalidations()).toBe(1);
+
+      await rs.advanceTimersByTimeAsync(
+        STOP_THREAD_FINALIZATION_REFETCH_DELAY_MS,
+      );
+
+      expect(countSearchInvalidations()).toBe(2);
+      expect(queryKeys()).not.toContainEqual(["thread", null]);
+    } finally {
+      client.clear();
+      rs.useRealTimers();
+    }
+  });
+
+  test("scheduled refetch lets sidebar receive delayed backend title finalization", async () => {
+    rs.useFakeTimers();
+
+    const client = new QueryClient({
+      defaultOptions: { queries: { retry: false } },
+    });
+    let finalized = false;
+    let fetchCount = 0;
+    const observer = new QueryObserver<AgentThread[]>(client, {
+      queryKey: ["threads", "search"],
+      queryFn: async () => {
+        fetchCount += 1;
+        return [
+          makeThread(
+            "thread-1",
+            finalized ? "Generated Title" : "New Conversation",
+          ),
+        ];
+      },
+    });
+    const unsubscribe = observer.subscribe((result) => {
+      void result.status;
+    });
+
+    try {
+      await observer.refetch();
+      expect(
+        client.getQueryData<AgentThread[]>(["threads", "search"])?.[0]?.values
+          ?.title,
+      ).toBe("New Conversation");
+
+      await stopThreadAndInvalidateCaches(
+        client,
+        () => Promise.resolve(),
+        "thread-1",
+        false,
+      );
+      await Promise.resolve();
+
+      expect(
+        client.getQueryData<AgentThread[]>(["threads", "search"])?.[0]?.values
+          ?.title,
+      ).toBe("New Conversation");
+
+      finalized = true;
+      await rs.advanceTimersByTimeAsync(
+        STOP_THREAD_FINALIZATION_REFETCH_DELAY_MS,
+      );
+
+      expect(
+        client.getQueryData<AgentThread[]>(["threads", "search"])?.[0]?.values
+          ?.title,
+      ).toBe("Generated Title");
+      expect(fetchCount).toBeGreaterThanOrEqual(3);
+    } finally {
+      unsubscribe();
+      client.clear();
+      rs.useRealTimers();
+    }
+  });
+});