fix: generate title for interrupted first turn

README.md

@@ -598,6 +598,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/CLAUDE.md

@@ -214,7 +214,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
 10. **SummarizationMiddleware** - Context reduction when approaching token limits (optional, if enabled)
 11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
 12. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional); subagent usage is cached by `tool_call_id` only while token usage is enabled and merged back into the dispatching AIMessage by message position rather than message id
-13. **TitleMiddleware** - Auto-generates thread title after first complete exchange and normalizes structured message content before prompting the title model
+13. **TitleMiddleware** - Auto-generates thread title after 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` persists a local fallback title from the checkpoint during interrupted-run cleanup and then syncs it to `threads_meta.display_name`.
 14. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses)
 15. **ViewImageMiddleware** - Injects base64 image data before LLM call (conditional on vision support)
 16. **DeferredToolFilterMiddleware** - Hides deferred (MCP) tool schemas from the bound model using a build-time deferred-name set + catalog hash, reading per-thread promotions from `ThreadState.promoted` (hash-scoped, no ContextVar); a tool becomes bound on subsequent turns after `tool_search` returns its schema (optional, if `tool_search.enabled`)

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

@@ -63,10 +63,27 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
         return ""
 
     @staticmethod
-    def _is_user_message_for_title(message: object) -> bool:
-        return getattr(message, "type", None) == "human" and not is_dynamic_context_reminder(message)
+    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
 
-    def _should_generate_title(self, state: TitleMiddlewareState) -> bool:
+    @staticmethod
+    def _message_content(message: object) -> object:
+        if isinstance(message, dict):
+            return message.get("content", "")
+        return getattr(message, "content", "")
+
+    @staticmethod
+    def _is_user_message_for_title(message: object) -> bool:
+        return TitleMiddleware._message_type(message) == "human" and not is_dynamic_context_reminder(message)
+
+    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:
@@ -83,10 +100,10 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
 
         # 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
+        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.
@@ -96,8 +113,8 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
         config = self._get_title_config()
         messages = state.get("messages", [])
 
-        user_msg_content = next((m.content for m in messages if self._is_user_message_for_title(m)), "")
-        assistant_msg_content = next((m.content for m in messages if m.type == "ai"), "")
+        user_msg_content = next((self._message_content(m) for m in messages if self._is_user_message_for_title(m)), "")
+        assistant_msg_content = next((self._message_content(m) for m in messages if self._message_type(m) == "ai"), "")
 
         user_msg = self._normalize_content(user_msg_content)
         assistant_msg = self._strip_think_tags(self._normalize_content(assistant_msg_content))
@@ -143,9 +160,9 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
         config["tags"] = [*(config.get("tags") or []), "middleware:title"]
         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._build_title_prompt(state)

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

@@ -412,6 +412,12 @@ 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 _ensure_interrupted_title(checkpointer=checkpointer, thread_id=thread_id, app_config=ctx.app_config)
+            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:
@@ -551,6 +557,44 @@ def _new_checkpoint_marker() -> dict[str, str]:
     return {"id": marker["id"], "ts": marker["ts"]}
 
 
+async def _ensure_interrupted_title(*, checkpointer: Any, thread_id: str, app_config: AppConfig | None) -> str | None:
+    """Persist a local fallback title for interrupted first-turn runs."""
+    ckpt_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+    ckpt_tuple = await _call_checkpointer_method(checkpointer, "aget_tuple", "get_tuple", ckpt_config)
+    if ckpt_tuple is None:
+        return None
+
+    checkpoint = copy.deepcopy(getattr(ckpt_tuple, "checkpoint", {}) or {})
+    channel_values = dict(checkpoint.get("channel_values", {}) or {})
+    existing_title = channel_values.get("title")
+    if existing_title:
+        return existing_title
+
+    from deerflow.agents.middlewares.title_middleware import TitleMiddleware
+
+    middleware = TitleMiddleware(app_config=app_config) if app_config is not None else TitleMiddleware()
+    result = middleware._generate_title_result(channel_values, allow_partial_exchange=True)
+    title = result.get("title") if isinstance(result, dict) else None
+    if not title:
+        return None
+
+    channel_values["title"] = title
+    marker = _new_checkpoint_marker()
+    checkpoint.update({"id": marker["id"], "ts": marker["ts"], "channel_values": channel_values})
+
+    metadata = dict(getattr(ckpt_tuple, "metadata", {}) or {})
+    metadata["source"] = "update"
+    metadata["step"] = metadata.get("step", 0) + 1
+    metadata["writes"] = {"runtime_interrupt_title": {"title": title}}
+
+    ckpt_config = getattr(ckpt_tuple, "config", {}) or {}
+    ckpt_configurable = ckpt_config.get("configurable", {}) if isinstance(ckpt_config, dict) else {}
+    checkpoint_ns = ckpt_configurable.get("checkpoint_ns", "") if isinstance(ckpt_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, {})
+    return title
+
+
 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,13 @@ class _ScriptedAgent:
         title: str,
         answer: str,
         block_after_first_chunk: 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.write_title = write_title
         self.checkpointer: Any | None = None
         self.store: Any | None = None
         self.metadata = {"model_name": "fake-test-model"}
@@ -101,7 +103,9 @@ class _ScriptedAgent:
             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 +269,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",
@@ -581,6 +604,51 @@ 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"
+
+
 @pytest.mark.anyio
 async def test_sse_consumer_disconnect_cancels_inflight_run():
     """A disconnected SSE request should cancel an in-flight run when configured."""