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.

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

@@ -579,6 +579,27 @@ async def _ensure_interrupted_title(*, checkpointer: Any, thread_id: str, app_co
     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/PostgresSaver) actually persist the new
+    # blob — those savers strip inline ``channel_values`` and only write blobs
+    # for channels listed in ``new_versions``. Mirrors ``_rollback_to_pre_run_checkpoint``.
+    channel_versions = dict(checkpoint.get("channel_versions", {}) or {})
+    get_next_version = getattr(checkpointer, "get_next_version", None)
+    current_title_version = channel_versions.get("title")
+    if callable(get_next_version):
+        next_title_version = get_next_version(current_title_version, None)
+    elif isinstance(current_title_version, int):
+        next_title_version = current_title_version + 1
+    elif isinstance(current_title_version, str):
+        try:
+            next_title_version = str(int(current_title_version) + 1)
+        except ValueError:
+            next_title_version = current_title_version + ".1"
+    else:
+        next_title_version = 1
+    channel_versions["title"] = next_title_version
+    checkpoint["channel_versions"] = channel_versions
+
     metadata = dict(getattr(ckpt_tuple, "metadata", {}) or {})
     metadata["source"] = "update"
     prev_step = metadata.get("step")
@@ -589,7 +610,15 @@ async def _ensure_interrupted_title(*, checkpointer: Any, thread_id: str, app_co
     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 ""
     write_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns}}
-    await _call_checkpointer_method(checkpointer, "aput", "put", write_config, checkpoint, metadata, {})
+    await _call_checkpointer_method(
+        checkpointer,
+        "aput",
+        "put",
+        write_config,
+        checkpoint,
+        metadata,
+        {"title": next_title_version},
+    )
     return title
 
 

backend/tests/test_run_worker_rollback.py

@@ -13,6 +13,7 @@ from deerflow.runtime.runs.worker import (
     RunContext,
     _agent_factory_supports_app_config,
     _build_runtime_context,
+    _ensure_interrupted_title,
     _extract_llm_error_fallback_message,
     _install_runtime_context,
     _rollback_to_pre_run_checkpoint,
@@ -672,3 +673,205 @@ def test_extract_llm_error_fallback_message_updates_mode_no_fallback():
         ]
     }
     assert _extract_llm_error_fallback_message(update_chunk) is None
+
+
+class _FakeCheckpointTuple:
+    """Minimal stand-in for ``CheckpointTuple`` used by ``_ensure_interrupted_title``."""
+
+    def __init__(self, *, checkpoint: dict, metadata: dict, config: dict | None = None):
+        self.checkpoint = checkpoint
+        self.metadata = metadata
+        self.config = config or {"configurable": {"thread_id": "thread-1", "checkpoint_ns": ""}}
+
+
+class _TitleCheckpointer:
+    """Captures ``aput`` arguments and exposes ``get_next_version`` like DB savers."""
+
+    def __init__(self, *, tuple_value: _FakeCheckpointTuple | None, put_result: dict | None = None):
+        self.aget_tuple = AsyncMock(return_value=tuple_value)
+        self.aput = AsyncMock(return_value=put_result or {})
+
+    def get_next_version(self, current, _channel):
+        if current is None:
+            return 1
+        if isinstance(current, int):
+            return current + 1
+        if isinstance(current, str):
+            try:
+                return str(int(current) + 1)
+            except ValueError:
+                return f"{current}.1"
+        return 1
+
+
+@pytest.mark.anyio
+async def test_ensure_interrupted_title_bumps_channel_version_and_declares_it_in_new_versions(monkeypatch):
+    """Regression for #3859 review: DB-backed savers (Sqlite/Postgres) strip inline
+    ``channel_values`` from ``put`` and only persist blobs for channels listed in
+    ``new_versions``. The helper must therefore bump ``channel_versions["title"]``
+    and pass ``{"title": next_version}`` so the fallback title actually survives
+    a fresh ``aget_tuple`` after the worker's finally hook.
+    """
+    from deerflow.agents.middlewares.title_middleware import TitleMiddleware
+
+    monkeypatch.setattr(
+        TitleMiddleware,
+        "_generate_title_result",
+        lambda self, state, allow_partial_exchange=False: {"title": "Generated Title"},
+    )
+
+    initial_checkpoint = {
+        "id": "ckpt-1",
+        "ts": "2026-06-29T00:00:00Z",
+        "channel_values": {"messages": [{"type": "human", "content": "hi"}]},
+        "channel_versions": {"messages": 5},
+    }
+    checkpointer = _TitleCheckpointer(
+        tuple_value=_FakeCheckpointTuple(
+            checkpoint=initial_checkpoint,
+            metadata={"source": "loop", "step": 7},
+        ),
+    )
+
+    title = await _ensure_interrupted_title(checkpointer=checkpointer, thread_id="thread-1", app_config=None)
+
+    assert title == "Generated Title"
+    checkpointer.aput.assert_awaited_once()
+    write_config, written_checkpoint, written_metadata, new_versions = checkpointer.aput.await_args.args
+
+    # The title channel must be declared in new_versions — without this, DB
+    # savers drop the inline channel_values["title"] from the persisted blob.
+    assert new_versions == {"title": 1}
+    # Channel versions on the checkpoint itself must also reflect the bump,
+    # so a subsequent aget_tuple reconstructs channel_values with the title.
+    assert written_checkpoint["channel_versions"]["title"] == 1
+    # Pre-existing channel versions must be preserved.
+    assert written_checkpoint["channel_versions"]["messages"] == 5
+    # The fallback title rides into channel_values for the (legacy / single-table)
+    # savers that inline the snapshot.
+    assert written_checkpoint["channel_values"]["title"] == "Generated Title"
+    assert written_metadata["source"] == "update"
+    assert written_metadata["step"] == 8
+    assert written_metadata["writes"] == {"runtime_interrupt_title": {"title": "Generated Title"}}
+    assert write_config == {"configurable": {"thread_id": "thread-1", "checkpoint_ns": ""}}
+
+
+@pytest.mark.anyio
+async def test_ensure_interrupted_title_bumps_existing_string_version(monkeypatch):
+    """When the checkpointer lacks ``get_next_version`` and the prior title
+    version is a string (some savers use UUID-shaped versions), the helper must
+    still produce a strictly different value rather than overwriting in place.
+    """
+    from deerflow.agents.middlewares.title_middleware import TitleMiddleware
+
+    monkeypatch.setattr(
+        TitleMiddleware,
+        "_generate_title_result",
+        lambda self, state, allow_partial_exchange=False: {"title": "T"},
+    )
+
+    initial_checkpoint = {
+        "id": "ckpt-1",
+        "ts": "2026-06-29T00:00:00Z",
+        "channel_values": {"messages": [{"type": "human", "content": "hi"}]},
+        "channel_versions": {"title": "v3"},
+    }
+
+    class _NoGetNextVersion:
+        def __init__(self):
+            self.aget_tuple = AsyncMock(
+                return_value=_FakeCheckpointTuple(
+                    checkpoint=initial_checkpoint,
+                    metadata={},
+                ),
+            )
+            self.aput = AsyncMock(return_value={})
+
+    checkpointer = _NoGetNextVersion()
+    await _ensure_interrupted_title(checkpointer=checkpointer, thread_id="thread-1", app_config=None)
+
+    _, written_checkpoint, _, new_versions = checkpointer.aput.await_args.args
+    bumped = written_checkpoint["channel_versions"]["title"]
+    assert bumped != "v3", "title version must change so DB savers persist the update"
+    assert new_versions == {"title": bumped}
+
+
+@pytest.mark.anyio
+async def test_ensure_interrupted_title_skips_when_title_already_set():
+    """If the checkpoint already carries a title, no new checkpoint is written."""
+    checkpointer = _TitleCheckpointer(
+        tuple_value=_FakeCheckpointTuple(
+            checkpoint={
+                "id": "ckpt-1",
+                "channel_values": {"messages": [], "title": "Already there"},
+                "channel_versions": {"title": 1},
+            },
+            metadata={},
+        ),
+    )
+
+    title = await _ensure_interrupted_title(checkpointer=checkpointer, thread_id="thread-1", app_config=None)
+
+    assert title == "Already there"
+    checkpointer.aput.assert_not_awaited()
+
+
+@pytest.mark.anyio
+async def test_ensure_interrupted_title_returns_none_when_no_checkpoint():
+    """No checkpoint exists yet → nothing to update."""
+    checkpointer = _TitleCheckpointer(tuple_value=None)
+    assert await _ensure_interrupted_title(checkpointer=checkpointer, thread_id="thread-1", app_config=None) is None
+    checkpointer.aput.assert_not_awaited()
+
+
+@pytest.mark.anyio
+async def test_ensure_interrupted_title_round_trip_with_real_sqlite_checkpointer(tmp_path):
+    """Full round-trip against a real ``AsyncSqliteSaver`` on a disk-backed DB.
+
+    Mirrors what Gateway constructs in production via ``make_checkpointer`` when
+    ``database.backend == "sqlite"``, then closes and re-opens the saver to
+    simulate a fresh connection. The fallback title must survive that boundary —
+    this is the scenario the #3874 review flagged as broken before the
+    ``new_versions={"title": ...}`` fix.
+    """
+    from langchain_core.messages import HumanMessage
+    from langgraph.checkpoint.base import empty_checkpoint
+    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+    from deerflow.config.title_config import TitleConfig
+
+    db_path = str(tmp_path / "ckpt.db")
+    thread_cfg = {"configurable": {"thread_id": "thread-1", "checkpoint_ns": ""}}
+
+    # 1. Seed a first-turn checkpoint that has a human message and NO title —
+    #    the same shape the agent leaves behind when interrupted mid-stream.
+    async with AsyncSqliteSaver.from_conn_string(db_path) as writer:
+        await writer.setup()
+        ck = empty_checkpoint()
+        ck["channel_values"] = {
+            "messages": [HumanMessage(content="Why is the sky blue?").model_dump()],
+        }
+        ck["channel_versions"] = {"messages": 1}
+        await writer.aput(thread_cfg, ck, {"source": "loop", "step": 1, "writes": {}}, {"messages": 1})
+
+    # 2. Run the worker helper through a *fresh* saver instance — this is what
+    #    the lifespan-owned checkpointer pool does for each request.
+    title_config = TitleConfig(enabled=True, max_chars=40, max_words=20)
+    app_config = SimpleNamespace(title=title_config)
+    async with AsyncSqliteSaver.from_conn_string(db_path) as worker_saver:
+        title = await _ensure_interrupted_title(
+            checkpointer=worker_saver,
+            thread_id="thread-1",
+            app_config=app_config,
+        )
+    assert title, "fallback title must be generated from the seeded user message"
+
+    # 3. Open ANOTHER fresh saver and confirm the title survives — this is the
+    #    invariant the #3874 review was guarding: ``new_versions={}`` would
+    #    cause DB savers to drop the title blob, so a fresh aget_tuple would
+    #    read back without it.
+    async with AsyncSqliteSaver.from_conn_string(db_path) as reader:
+        tup = await reader.aget_tuple(thread_cfg)
+    assert tup is not None
+    persisted = tup.checkpoint.get("channel_values", {}).get("title")
+    assert persisted == title