Merge pull request #4145 from nesquena/stage-3998v2

Release NK (v0.51.398): auto-generate titles for imported CLI sessions (#3987)

CHANGELOG.md

@@ -3,6 +3,12 @@
 
 ## [Unreleased]
 
+## [v0.51.398] — 2026-06-13 — Release NK (auto-generate titles for imported CLI sessions, #3987)
+
+### Fixed
+
+- **Imported CLI/external sessions now get a generated title instead of a raw id or placeholder (#3987).** When a session is imported without a WebUI title, the backend runs the existing `generate_session_title` path (reusing `_looks_like_default_cli_title` detection). The async title persist re-resolves the **latest** canonical session under `_get_session_agent_lock(sid)` immediately before saving (preferring `SESSIONS[sid]`, else `Session.load(sid)`) and re-checks the default-title guard on that latest object — so a WebUI reply that lands while title generation is in flight is never clobbered by a stale pre-generation snapshot. Manual regenerate still works. (#3987)
+
 ## [v0.51.397] — 2026-06-13 — Release NJ (wire /credits through WebUI command dispatch, #4071)
 
 ### Fixed

api/routes.py

@@ -28,6 +28,7 @@ from contextlib import closing
 from urllib.parse import parse_qs, urlsplit
 from api.agent_sessions import (
     MESSAGING_SOURCES,
+    _looks_like_default_cli_title,
     is_cli_session_row,
     is_cli_session_row_visible,
     read_session_lineage_report,
@@ -56,6 +57,132 @@ def _publish_session_list_changed(reason: str, *, profile: str | None = None) ->
         publish_session_list_changed(reason)
 
 
+def _sync_session_title_to_insights(session) -> None:
+    """Write title-only session metadata updates through to state.db when enabled."""
+    try:
+        if not load_settings().get("sync_to_insights"):
+            return
+        from api.state_sync import sync_session_usage
+
+        messages = getattr(session, "messages", None) or []
+        sync_session_usage(
+            session_id=session.session_id,
+            input_tokens=getattr(session, "input_tokens", None) or 0,
+            output_tokens=getattr(session, "output_tokens", None) or 0,
+            estimated_cost=getattr(session, "estimated_cost", 0.0),
+            model=getattr(session, "model", ""),
+            title=session.title,
+            message_count=len(messages),
+            profile=getattr(session, "profile", None),
+        )
+    except Exception:
+        logger.debug("Failed to update session title in state.db", exc_info=True)
+
+
+def _persist_generated_session_title(
+    session,
+    next_title: str,
+    *,
+    event_reason: str,
+    require_default_title: bool = False,
+) -> str:
+    normalized_title = str(next_title or "").strip()[:80] or "Untitled"
+    sid = str(getattr(session, "session_id", "") or "")
+    original_session = session
+    with _get_session_agent_lock(sid):
+        with LOCK:
+            latest = SESSIONS.get(sid)
+            if latest is not None and str(getattr(latest, "session_id", "") or "") != sid:
+                SESSIONS.pop(sid, None)
+                latest = None
+            elif latest is not None:
+                SESSIONS.move_to_end(sid)
+        if latest is None:
+            latest = Session.load(sid)
+            if latest is None:
+                raise KeyError(sid)
+        session = _ensure_full_session_before_mutation(sid, latest)
+        if getattr(session, "read_only", False):
+            raise PermissionError(f"Session {sid} is read-only")
+        if require_default_title:
+            latest_meta = {
+                "title": getattr(session, "title", None),
+                "source_tag": getattr(session, "source_tag", None),
+                "raw_source": getattr(session, "raw_source", None),
+                "session_source": getattr(session, "session_source", None),
+                "source_label": getattr(session, "source_label", None),
+            }
+            if not _looks_like_default_cli_title(latest_meta):
+                return session.title
+        session.title = normalized_title
+        from api.session_ops import mark_session_title_generated
+
+        # mark_session_title_generated sets s.llm_title_generated = True and clears manual_title.
+        mark_session_title_generated(session)
+        session.save(touch_updated_at=False)
+        with LOCK:
+            SESSIONS[sid] = session
+            SESSIONS.move_to_end(sid)
+            while len(SESSIONS) > SESSIONS_MAX:
+                SESSIONS.popitem(last=False)
+    _sync_session_title_to_insights(session)
+    _publish_session_list_changed(event_reason, profile=getattr(session, "profile", None))
+    if original_session is not session:
+        original_session.title = session.title
+        original_session.llm_title_generated = session.llm_title_generated
+        original_session.manual_title = session.manual_title
+    return session.title
+
+
+def _queue_generated_title_for_imported_session(session, cli_meta: dict | None) -> None:
+    try:
+        cli_meta = dict(cli_meta or {})
+        if not session or cli_meta.get("read_only") or not _looks_like_default_cli_title(cli_meta):
+            return
+        sid = str(getattr(session, "session_id", "") or "")
+        if not sid:
+            return
+
+        def _run() -> None:
+            try:
+                current = Session.load(sid)
+                if not current:
+                    return
+                current = _ensure_full_session_before_mutation(sid, current)
+                if getattr(current, "read_only", False):
+                    return
+                current_meta = {
+                    "title": getattr(current, "title", None),
+                    "source_tag": getattr(current, "source_tag", None),
+                    "raw_source": getattr(current, "raw_source", None),
+                    "session_source": getattr(current, "session_source", None),
+                    "source_label": getattr(current, "source_label", None),
+                }
+                if not _looks_like_default_cli_title(current_meta):
+                    return
+                next_title, _reason, _raw_preview = generate_session_title_for_session(current)
+                normalized_current = str(getattr(current, "title", "") or "").strip()
+                normalized_next = str(next_title or "").strip()
+                if not normalized_next or normalized_next == normalized_current:
+                    return
+                _persist_generated_session_title(
+                    current,
+                    normalized_next,
+                    event_reason="session_title_regenerate",
+                    require_default_title=True,
+                )
+            except Exception:
+                logger.debug("Failed to generate imported session title for %s", sid, exc_info=True)
+
+        threading.Thread(target=_run, daemon=True, name=f"imported-title-{sid}").start()
+    except Exception:
+        logger.debug(
+            "Failed to queue imported session title generation for %s",
+            getattr(session, "session_id", None),
+            exc_info=True,
+        )
+
+
 # ── Cron run tracking ────────────────────────────────────────────────────────
 # Track job IDs currently being executed so the frontend can poll status.
 _RUNNING_CRON_JOBS: dict[str, float] = {}  # job_id → start_timestamp
@@ -7591,27 +7718,6 @@ def handle_post(handler, parsed) -> bool:
     if parsed.path == "/api/sessions/cleanup_zero_message":
         return _handle_sessions_cleanup(handler, body, zero_only=True)
 
-    def _sync_session_title_to_insights(session):
-        """Write title-only session metadata updates through to state.db when enabled."""
-        try:
-            if not load_settings().get("sync_to_insights"):
-                return
-            from api.state_sync import sync_session_usage
-
-            messages = getattr(session, "messages", None) or []
-            sync_session_usage(
-                session_id=session.session_id,
-                input_tokens=getattr(session, "input_tokens", None) or 0,
-                output_tokens=getattr(session, "output_tokens", None) or 0,
-                estimated_cost=getattr(session, "estimated_cost", 0.0),
-                model=getattr(session, "model", ""),
-                title=session.title,
-                message_count=len(messages),
-                profile=getattr(session, "profile", None),
-            )
-        except Exception:
-            logger.debug("Failed to update session title in state.db", exc_info=True)
-
     if parsed.path == "/api/session/rename":
         try:
             require(body, "session_id", "title")
@@ -7644,19 +7750,12 @@ def handle_post(handler, parsed) -> bool:
             s = _ensure_full_session_before_mutation(sid, s)
         except KeyError:
             return bad(handler, "Session not found", 404)
-        if getattr(s, "read_only", False) or getattr(s, "is_imported", False):
+        if getattr(s, "read_only", False):
             return bad(handler, "Read-only imported sessions cannot be renamed", 403)
         next_title, reason, raw_preview = generate_session_title_for_session(s, prefer_latest=prefer_latest)
         if not next_title:
             return bad(handler, f"Could not generate a better title ({reason or 'empty'})", 422)
-        with _get_session_agent_lock(sid):
-            s.title = str(next_title).strip()[:80] or "Untitled"
-            from api.session_ops import mark_session_title_generated
-            # mark_session_title_generated sets s.llm_title_generated = True and clears manual_title.
-            mark_session_title_generated(s)
-            s.save(touch_updated_at=False)
-        _sync_session_title_to_insights(s)
-        publish_session_list_changed("session_title_regenerate", profile=getattr(s, "profile", None))
+        _persist_generated_session_title(s, next_title, event_reason="session_title_regenerate")
         return j(handler, {
             "session": s.compact(),
             "title": s.title,
@@ -16104,6 +16203,17 @@ def _handle_session_import_cli(handler, body):
         "session_import_cli",
         profile=getattr(s, "profile", None),
     )
+    _queue_generated_title_for_imported_session(
+        s,
+        {
+            "title": cli_title,
+            "source_tag": cli_source_tag,
+            "raw_source": cli_raw_source,
+            "session_source": cli_session_source,
+            "source_label": cli_source_label,
+            "read_only": cli_read_only,
+        },
+    )
     return j(
         handler,
         {

static/sessions.js

@@ -3181,39 +3181,33 @@ function _openSessionActionMenu(session, anchorEl){
       }
     ));
   }
-  // Title regeneration matches the backend guard (api/routes.py rejects
-  // read_only OR is_imported with 403). read_only sessions already bailed at
-  // the isReadOnly early-return above; skip imported sessions here so the
-  // action is hidden rather than failing with a 403 toast. This keeps the
-  // is_imported gate scoped to regenerate instead of broadening the shared
-  // _isReadOnlySession() helper (which gates rename/pin/archive/etc.).
-  if(!session.is_imported){
-    menu.appendChild(_buildSessionAction(
-      t('session_title_regenerate'),
-      t('session_title_regenerate_desc'),
-      ICONS.spark,
-      async()=>{
-        closeSessionActionMenu();
-        try{
-          if(typeof showToast==='function') showToast(t('session_title_regenerating'), 1600);
-          const response=await api('/api/session/title/regenerate',{method:'POST',body:JSON.stringify({session_id:session.session_id})});
-          const nextTitle=(response&&response.title)||(response&&response.session&&response.session.title)||'';
-          if(nextTitle){
-            session.title=nextTitle;
-            const cached=(_allSessions||[]).find(item=>item&&item.session_id===session.session_id);
-            if(cached) cached.title=nextTitle;
-            if(S.session&&S.session.session_id===session.session_id){S.session.title=nextTitle;syncTopbar();}
-            renderSessionListFromCache();
-          }
-          if(typeof showToast==='function') showToast(t('session_title_regenerated', nextTitle||t('untitled')), 2400);
-        }catch(err){
-          const msg=t('session_title_regenerate_failed')+(err&&err.message?err.message:String(err));
-          setStatus(msg);
-          if(typeof showToast==='function') showToast(msg,3000,'error');
+  // Title regeneration stays available for writable imported sessions.
+  // Read-only sessions return earlier through the shared action-menu guard.
+  menu.appendChild(_buildSessionAction(
+    t('session_title_regenerate'),
+    t('session_title_regenerate_desc'),
+    ICONS.spark,
+    async()=>{
+      closeSessionActionMenu();
+      try{
+        if(typeof showToast==='function') showToast(t('session_title_regenerating'), 1600);
+        const response=await api('/api/session/title/regenerate',{method:'POST',body:JSON.stringify({session_id:session.session_id})});
+        const nextTitle=(response&&response.title)||(response&&response.session&&response.session.title)||'';
+        if(nextTitle){
+          session.title=nextTitle;
+          const cached=(_allSessions||[]).find(item=>item&&item.session_id===session.session_id);
+          if(cached) cached.title=nextTitle;
+          if(S.session&&S.session.session_id===session.session_id){S.session.title=nextTitle;syncTopbar();}
+          renderSessionListFromCache();
         }
+        if(typeof showToast==='function') showToast(t('session_title_regenerated', nextTitle||t('untitled')), 2400);
+      }catch(err){
+        const msg=t('session_title_regenerate_failed')+(err&&err.message?err.message:String(err));
+        setStatus(msg);
+        if(typeof showToast==='function') showToast(msg,3000,'error');
       }
-    ));
-  }
+    }
+  ));
   if(!isExternalSession){
     if(session.worktree_path){
       menu.appendChild(_buildSessionAction(

tests/test_claude_code_session_import.py

@@ -219,6 +219,71 @@ def test_session_import_cli_returns_read_only_claude_code_payload(monkeypatch, t
     assert session["is_cli_session"] is True
 
 
+def test_session_import_cli_queues_generated_title_for_writable_default_cli_title(monkeypatch):
+    import api.routes as routes
+
+    sid = "cli_writable_default_title"
+    messages = [{"role": "user", "content": "Need a better imported title"}]
+    cli_meta = {
+        "session_id": sid,
+        "title": "CLI Session",
+        "model": "claude-code",
+        "created_at": 10.0,
+        "updated_at": 20.0,
+        "source_tag": "cli",
+        "raw_source": "cli",
+        "session_source": "external_agent",
+        "source_label": "CLI",
+        "is_cli_session": True,
+        "read_only": False,
+    }
+    persisted = {}
+    queued = []
+    published = []
+
+    class FakeImportedSession:
+        def __init__(self):
+            self.session_id = sid
+            self.title = "CLI Session"
+            self.messages = list(messages)
+            self.profile = "default"
+            self.model = "claude-code"
+            self.read_only = False
+            self.is_cli_session = True
+
+        def save(self, touch_updated_at=False):
+            persisted["saved"] = touch_updated_at
+
+        def compact(self):
+            return {"session_id": sid, "title": self.title}
+
+    imported = FakeImportedSession()
+
+    monkeypatch.setattr(routes.Session, "load", classmethod(lambda _cls, _sid: None))
+    monkeypatch.setattr(routes, "require", lambda body, *keys: None)
+    monkeypatch.setattr(routes, "bad", lambda _handler, msg, status=400: {"ok": False, "error": msg, "status": status})
+    monkeypatch.setattr(routes, "j", lambda _handler, payload, status=200, extra_headers=None: payload)
+    monkeypatch.setattr(routes, "get_cli_session_messages", lambda _sid: messages if _sid == sid else [])
+    monkeypatch.setattr(routes, "get_cli_sessions", lambda: [cli_meta])
+    monkeypatch.setattr(routes, "import_cli_session", lambda *args, **kwargs: imported)
+    monkeypatch.setattr(routes, "publish_session_list_changed", lambda reason, profile=None: published.append((reason, profile)))
+    monkeypatch.setattr(routes, "_queue_generated_title_for_imported_session", lambda session, meta: queued.append((session, meta.copy())))
+
+    response = routes._handle_session_import_cli(object(), {"session_id": sid})
+
+    assert response["imported"] is True
+    assert persisted["saved"] is False
+    assert published == [("session_import_cli", "default")]
+    assert queued == [(imported, {
+        "title": "CLI Session",
+        "source_tag": "cli",
+        "raw_source": "cli",
+        "session_source": "external_agent",
+        "source_label": "CLI",
+        "read_only": False,
+    })]
+
+
 def test_read_only_source_badge_ui_guards_are_present():
     sessions_js = (REPO_ROOT / "static" / "sessions.js").read_text(encoding="utf-8")
     messages_js = (REPO_ROOT / "static" / "messages.js").read_text(encoding="utf-8")

tests/test_issue3987_imported_session_titles.py

@@ -0,0 +1,224 @@
+"""Regression coverage for imported-session title generation after CLI import (#3987)."""
+
+from __future__ import annotations
+
+import io
+import json
+from collections import OrderedDict
+from pathlib import Path
+
+import api.models as models
+import api.routes as routes
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ROUTES_PY = (ROOT / "api" / "routes.py").read_text(encoding="utf-8")
+SESSIONS_JS = (ROOT / "static" / "sessions.js").read_text(encoding="utf-8")
+
+
+class _FakeHandler:
+    def __init__(self):
+        self.status = None
+        self.headers = {}
+        self.wfile = io.BytesIO()
+
+    def send_response(self, status):
+        self.status = status
+
+    def send_header(self, key, value):
+        self.headers[key] = value
+
+    def end_headers(self):
+        pass
+
+    def json_body(self):
+        return json.loads(self.wfile.getvalue().decode("utf-8"))
+
+
+def test_import_cli_handler_queues_default_titles_after_persisting_import():
+    handler_idx = ROUTES_PY.index("def _handle_session_import_cli")
+    next_handler_idx = ROUTES_PY.index("def _handle_session_import(", handler_idx)
+    block = ROUTES_PY[handler_idx:next_handler_idx]
+    queue_idx = block.index("_queue_generated_title_for_imported_session(")
+    publish_idx = block.index('publish_session_list_changed(\n        "session_import_cli",')
+    response_idx = block.index("return j(", queue_idx)
+    assert publish_idx < queue_idx < response_idx
+    queue_window = block[queue_idx:queue_idx + 400]
+    assert '"title": cli_title' in queue_window
+    assert '"read_only": cli_read_only' in queue_window
+
+
+def test_import_cli_queue_helper_is_guarded_and_runs_in_background():
+    helper_idx = ROUTES_PY.index("def _queue_generated_title_for_imported_session")
+    next_helper_idx = ROUTES_PY.index("def _gateway_sse_probe_payload", helper_idx)
+    block = ROUTES_PY[helper_idx:next_helper_idx]
+    assert "cli_meta.get(\"read_only\")" in block
+    assert "not _looks_like_default_cli_title(cli_meta)" in block
+    assert "if not _looks_like_default_cli_title(current_meta):" in block
+    assert "generate_session_title_for_session(current)" in block
+    assert "require_default_title=True" in block
+    assert "threading.Thread(target=_run, daemon=True" in block
+
+
+def test_import_cli_queue_helper_generates_title_once_for_placeholder_session(monkeypatch):
+    persisted = []
+    generated = []
+
+    class FakeSession:
+        def __init__(self, title):
+            self.session_id = "cli_queued_title"
+            self.title = title
+            self.source_tag = "cli"
+            self.raw_source = "cli"
+            self.session_source = "external_agent"
+            self.source_label = "CLI"
+            self.read_only = False
+
+    current = FakeSession("CLI Session")
+
+    class InlineThread:
+        def __init__(self, *, target, daemon, name):
+            self._target = target
+
+        def start(self):
+            self._target()
+
+    monkeypatch.setattr(routes.threading, "Thread", InlineThread)
+    monkeypatch.setattr(routes.Session, "load", classmethod(lambda _cls, sid: current if sid == current.session_id else None))
+    monkeypatch.setattr(routes, "_ensure_full_session_before_mutation", lambda sid, session: session)
+    monkeypatch.setattr(routes, "generate_session_title_for_session", lambda session: (generated.append(session.session_id) or "Better imported title", "llm", "raw"))
+    monkeypatch.setattr(
+        routes,
+        "_persist_generated_session_title",
+        lambda session, title, *, event_reason, require_default_title=False: persisted.append(
+            (session.session_id, title, event_reason, require_default_title)
+        ),
+    )
+
+    routes._queue_generated_title_for_imported_session(current, {"title": "CLI Session", "source_tag": "cli"})
+
+    assert generated == [current.session_id]
+    assert persisted == [(current.session_id, "Better imported title", "session_title_regenerate", True)]
+
+
+def test_import_cli_queue_helper_skips_sessions_that_already_have_real_titles(monkeypatch):
+    generated = []
+    persisted = []
+
+    class FakeSession:
+        def __init__(self):
+            self.session_id = "cli_real_title"
+            self.title = "Useful imported title"
+            self.source_tag = "cli"
+            self.raw_source = "cli"
+            self.session_source = "external_agent"
+            self.source_label = "CLI"
+            self.read_only = False
+
+    current = FakeSession()
+
+    class InlineThread:
+        def __init__(self, *, target, daemon, name):
+            self._target = target
+
+        def start(self):
+            self._target()
+
+    monkeypatch.setattr(routes.threading, "Thread", InlineThread)
+    monkeypatch.setattr(routes.Session, "load", classmethod(lambda _cls, sid: current if sid == current.session_id else None))
+    monkeypatch.setattr(routes, "_ensure_full_session_before_mutation", lambda sid, session: session)
+    monkeypatch.setattr(
+        routes,
+        "generate_session_title_for_session",
+        lambda session: generated.append(session.session_id) or ("Unexpected generated title", "llm", "raw"),
+    )
+    monkeypatch.setattr(
+        routes,
+        "_persist_generated_session_title",
+        lambda session, title, *, event_reason, require_default_title=False: persisted.append(
+            (session.session_id, title, event_reason, require_default_title)
+        ),
+    )
+
+    routes._queue_generated_title_for_imported_session(
+        current,
+        {"title": "CLI Session", "source_tag": "cli"},
+    )
+
+    assert generated == []
+    assert persisted == []
+
+
+def test_generated_title_persist_reloads_latest_session_before_saving(tmp_path, monkeypatch):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    cache = OrderedDict()
+
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSIONS", cache, raising=False)
+    monkeypatch.setattr(routes, "SESSIONS", cache, raising=False)
+    monkeypatch.setattr(routes, "_sync_session_title_to_insights", lambda session: None)
+    monkeypatch.setattr(routes, "_publish_session_list_changed", lambda *args, **kwargs: None)
+
+    stale = models.Session(
+        session_id="cli_import_race",
+        title="CLI Session",
+        workspace=".",
+        model="test-model",
+        messages=[{"role": "user", "content": "first"}],
+        source_tag="cli",
+        raw_source="cli",
+        session_source="external_agent",
+        source_label="CLI",
+    )
+    stale.save(skip_index=True)
+    stale_snapshot = models.Session.load(stale.session_id)
+
+    latest = models.Session(
+        session_id=stale.session_id,
+        title="CLI Session",
+        workspace=".",
+        model="test-model",
+        messages=[
+            {"role": "user", "content": "first"},
+            {"role": "assistant", "content": "reply"},
+            {"role": "user", "content": "second"},
+        ],
+        source_tag="cli",
+        raw_source="cli",
+        session_source="external_agent",
+        source_label="CLI",
+    )
+    latest.save(skip_index=True)
+    cache[latest.session_id] = latest
+
+    saved_title = routes._persist_generated_session_title(
+        stale_snapshot,
+        "Better imported title",
+        event_reason="session_title_regenerate",
+        require_default_title=True,
+    )
+
+    reloaded = models.Session.load(stale.session_id)
+    assert saved_title == "Better imported title"
+    assert reloaded.title == "Better imported title"
+    assert len(reloaded.messages) == 3
+    assert reloaded.llm_title_generated is True
+    assert reloaded.manual_title is False
+    assert len(cache[stale.session_id].messages) == 3
+    assert stale_snapshot.title == "Better imported title"
+
+
+def test_regenerate_endpoint_only_blocks_read_only_imported_sessions():
+    endpoint_idx = ROUTES_PY.index('"/api/session/title/regenerate"')
+    next_endpoint_idx = ROUTES_PY.index('"/api/personality/set"', endpoint_idx)
+    block = ROUTES_PY[endpoint_idx:next_endpoint_idx]
+    assert 'if getattr(s, "read_only", False):' in block
+    assert 'getattr(s, "is_imported", False)' not in block
+
+
+def test_sessions_ui_keeps_regenerate_action_for_writable_imports():
+    regen_idx = SESSIONS_JS.index("api('/api/session/title/regenerate'")
+    window = SESSIONS_JS[regen_idx - 500:regen_idx]
+    assert "session.is_imported" not in window
+    assert "_isReadOnlySession(session)" in SESSIONS_JS

tests/test_session_events.py

@@ -31,6 +31,8 @@ def test_session_events_publish_for_minimal_sidebar_mutations():
     ):
         if reason == "session_import_cli":
             assert f'publish_session_list_changed(\n        "{reason}",' in ROUTES, reason
+        elif reason == "session_title_regenerate":
+            assert '_persist_generated_session_title(s, next_title, event_reason="session_title_regenerate")' in ROUTES
         elif reason == "session_import":
             assert f'publish_session_list_changed("{reason}")' in ROUTES, reason
         else:
@@ -41,7 +43,8 @@ def test_session_events_publish_for_minimal_sidebar_mutations():
     assert 'if was_hidden_empty_session:\n        publish_session_list_changed("session_new", profile=getattr(s, "profile", None))' in ROUTES
     assert 'publish_session_list_changed("session_duplicate", profile=getattr(copied_session, "profile", None))' in ROUTES
     assert 'publish_session_list_changed("session_rename", profile=getattr(s, "profile", None))' in ROUTES
-    assert 'publish_session_list_changed("session_title_regenerate", profile=getattr(s, "profile", None))' in ROUTES
+    assert '_persist_generated_session_title(s, next_title, event_reason="session_title_regenerate")' in ROUTES
+    assert '_publish_session_list_changed(event_reason, profile=getattr(session, "profile", None))' in ROUTES
     assert 'event_profile = getattr(get_session(sid, metadata_only=True), "profile", None)' in ROUTES
     assert "Failed to resolve profile for deleted session" in ROUTES
     assert '_publish_session_list_changed("session_delete", profile=event_profile)' in ROUTES

tests/test_session_title_regenerate_controls.py

@@ -21,21 +21,19 @@ def test_session_action_menu_exposes_regenerate_title_control():
     assert "renderSessionListFromCache();" in SESSIONS_JS
 
 
-def test_imported_sessions_skip_regenerate_action_without_broadening_shared_gate():
+def test_writable_imported_sessions_keep_regenerate_action_without_broadening_shared_gate():
     # The shared _isReadOnlySession() helper must stay scoped to read_only flags
     # so it does not silently disable rename/pin/archive/etc. for imported
-    # sessions. The is_imported guard is scoped to the regenerate action only,
-    # matching the backend 403 guard in api/routes.py.
+    # sessions. Writable imported sessions should still expose regenerate.
     helper_idx = SESSIONS_JS.index("function _isReadOnlySession(session)")
     next_helper_idx = SESSIONS_JS.index("function _sourceKeyForSession", helper_idx)
     helper_block = SESSIONS_JS[helper_idx:next_helper_idx]
     assert "session.is_imported" not in helper_block, (
-        "_isReadOnlySession must not include is_imported — it gates rename/pin/archive too"
+        "_isReadOnlySession must not include is_imported; writable imports need regenerate"
     )
-    # The regenerate action is gated on !session.is_imported next to its api call.
     regen_idx = SESSIONS_JS.index("api('/api/session/title/regenerate'")
     guard_window = SESSIONS_JS[regen_idx - 600:regen_idx]
-    assert "if(!session.is_imported){" in guard_window
+    assert "session.is_imported" not in guard_window
 
 
 def test_regenerate_title_i18n_and_changelog_entries_exist():
@@ -56,13 +54,20 @@ def test_regenerate_endpoint_persists_generated_title_without_reordering_sidebar
     next_endpoint_idx = ROUTES_PY.index('"/api/personality/set"', endpoint_idx)
     block = ROUTES_PY[endpoint_idx:next_endpoint_idx]
     assert "generate_session_title_for_session" in block
-    assert "s.llm_title_generated = True" in block
-    assert "s.save(touch_updated_at=False)" in block
-    assert "_sync_session_title_to_insights(s)" in block
-    assert 'publish_session_list_changed("session_title_regenerate", profile=getattr(s, "profile", None))' in block
+    assert '_persist_generated_session_title(s, next_title, event_reason="session_title_regenerate")' in block
     assert "Read-only imported sessions cannot be renamed" in block
 
 
+def test_regenerate_helper_persists_generated_title_and_publishes_sidebar_refresh():
+    helper_idx = ROUTES_PY.index("def _persist_generated_session_title")
+    queue_idx = ROUTES_PY.index("def _queue_generated_title_for_imported_session", helper_idx)
+    helper_block = ROUTES_PY[helper_idx:queue_idx]
+    assert "mark_session_title_generated(session)" in helper_block
+    assert "session.save(touch_updated_at=False)" in helper_block
+    assert "_sync_session_title_to_insights(session)" in helper_block
+    assert '_publish_session_list_changed(event_reason, profile=getattr(session, "profile", None))' in helper_block
+
+
 def test_regenerate_endpoint_syncs_title_to_state_db_when_enabled():
     helper_idx = ROUTES_PY.index("def _sync_session_title_to_insights")
     endpoint_idx = ROUTES_PY.index('"/api/session/title/regenerate"')