fix(agent): make auto-title write atomic

agent/title_generator.py

@@ -237,14 +237,11 @@ def _auto_title_session(
         return
 
     try:
-        latest = session_db.get_session_title(session_id)
-        if latest:
+        if not session_db.set_auto_title_if_empty(session_id, title):
             logger.debug(
-                "Skipping auto-generated session title because a title was set while generation was in flight: %s",
-                latest,
+                "Skipping auto-generated session title because a title was set while generation was in flight"
             )
             return
-        session_db.set_session_title(session_id, title)
         logger.debug("Auto-generated session title: %s", title)
         if title_callback is not None:
             try:

hermes_state.py

@@ -3222,16 +3222,24 @@ class SessionDB:
         ).fetchone()
         return row is not None
 
-    def set_session_title(self, session_id: str, title: str) -> bool:
-        """Set or update a session's title.
-
-        Returns True if session was found and title was set.
-        Raises ValueError if title is already in use by another session,
-        or if the title fails validation (too long, invalid characters).
-        Empty/whitespace-only strings are normalized to None (clearing the title).
-        """
+    def _set_session_title(
+        self,
+        session_id: str,
+        title: str,
+        *,
+        only_if_empty: bool,
+    ) -> bool:
         title = self.sanitize_title(title)
+
         def _do(conn):
+            if only_if_empty:
+                current = conn.execute(
+                    "SELECT title FROM sessions WHERE id = ?",
+                    (session_id,),
+                ).fetchone()
+                if current is None or current["title"] is not None:
+                    return 0
+
             if title:
                 # Check uniqueness (allow the same session to keep its own title)
                 cursor = conn.execute(
@@ -3263,14 +3271,35 @@ class SessionDB:
                         raise ValueError(
                             f"Title '{title}' is already in use by session {conflict_id}"
                         )
+            predicate = " AND title IS NULL" if only_if_empty else ""
             cursor = conn.execute(
-                "UPDATE sessions SET title = ? WHERE id = ?",
+                f"UPDATE sessions SET title = ? WHERE id = ?{predicate}",
                 (title, session_id),
             )
             return cursor.rowcount
+
         rowcount = self._execute_write(_do)
         return rowcount > 0
 
+    def set_session_title(self, session_id: str, title: str) -> bool:
+        """Set or update a session's title.
+
+        Returns True if session was found and title was set.
+        Raises ValueError if title is already in use by another session,
+        or if the title fails validation (too long, invalid characters).
+        Empty/whitespace-only strings are normalized to None (clearing the title).
+        """
+        return self._set_session_title(session_id, title, only_if_empty=False)
+
+    def set_auto_title_if_empty(self, session_id: str, title: str) -> bool:
+        """Set an auto-generated title only when the current title is NULL.
+
+        The predicate and write run in one transaction so a concurrent manual
+        rename cannot be overwritten. Validation and uniqueness behavior match
+        :meth:`set_session_title`.
+        """
+        return self._set_session_title(session_id, title, only_if_empty=True)
+
     def get_session_title(self, session_id: str) -> Optional[str]:
         """Get the title for a session, or None."""
         with self._lock:

tests/agent/test_title_generator.py

@@ -9,6 +9,7 @@ from agent.title_generator import (
     maybe_auto_title,
     _title_language,
 )
+from hermes_state import SessionDB
 
 
 class TestGenerateTitle:
@@ -240,17 +241,27 @@ class TestAutoTitleSession:
     def test_generates_and_sets_title(self):
         db = MagicMock()
         db.get_session_title.return_value = None
+        db.set_auto_title_if_empty.return_value = True
 
         with patch("agent.title_generator.generate_title", return_value="New Title"):
             auto_title_session(db, "sess-1", "hi", "hello")
-            db.set_session_title.assert_called_once_with("sess-1", "New Title")
+            db.set_auto_title_if_empty.assert_called_once_with("sess-1", "New Title")
 
-    def test_does_not_overwrite_title_set_while_generation_was_in_flight(self):
-        db = MagicMock()
-        db.get_session_title.side_effect = [None, "Manual Title"]
+    def test_does_not_overwrite_title_set_immediately_before_conditional_write(
+        self, tmp_path
+    ):
+        db = SessionDB(tmp_path / "state.db")
+        db.create_session(session_id="sess-1", source="cli")
         seen = []
 
-        with patch("agent.title_generator.generate_title", return_value="Auto Title"):
+        def generate_after_manual_title(*_args, **_kwargs):
+            db.set_session_title("sess-1", "Manual Title")
+            return "Auto Title"
+
+        with patch(
+            "agent.title_generator.generate_title",
+            side_effect=generate_after_manual_title,
+        ):
             auto_title_session(
                 db,
                 "sess-1",
@@ -259,12 +270,13 @@ class TestAutoTitleSession:
                 title_callback=seen.append,
             )
 
-        db.set_session_title.assert_not_called()
+        assert db.get_session_title("sess-1") == "Manual Title"
         assert seen == []
 
     def test_invokes_title_callback_after_setting_title(self):
         db = MagicMock()
         db.get_session_title.return_value = None
+        db.set_auto_title_if_empty.return_value = True
         seen = []
         with patch("agent.title_generator.generate_title", return_value="Readable Session"):
             auto_title_session(
@@ -274,7 +286,7 @@ class TestAutoTitleSession:
                 "hi there",
                 title_callback=seen.append,
             )
-        db.set_session_title.assert_called_once_with("sess-1", "Readable Session")
+        db.set_auto_title_if_empty.assert_called_once_with("sess-1", "Readable Session")
         assert seen == ["Readable Session"]
 
     def test_skips_if_generation_fails(self):
@@ -283,7 +295,7 @@ class TestAutoTitleSession:
 
         with patch("agent.title_generator.generate_title", return_value=None):
             auto_title_session(db, "sess-1", "hi", "hello")
-            db.set_session_title.assert_not_called()
+            db.set_auto_title_if_empty.assert_not_called()
 
     def test_never_raises_when_body_throws(self):
         """Daemon-thread target must swallow ALL exceptions (e.g. the

tests/test_hermes_state.py

@@ -2990,6 +2990,12 @@ class TestSessionTitle:
         session = db.get_session("s1")
         assert session["title"] == "Updated Title"
 
+    def test_auto_title_only_sets_an_empty_title(self, db):
+        db.create_session(session_id="s1", source="cli")
+        assert db.set_auto_title_if_empty("s1", "Generated Title") is True
+        assert db.set_auto_title_if_empty("s1", "Replacement Title") is False
+        assert db.get_session_title("s1") == "Generated Title"
+
     def test_title_in_search_sessions(self, db):
         db.create_session(session_id="s1", source="cli")
         db.set_session_title("s1", "Debugging Auth")