feat(cron): pass job name as session title for cron sessions

cron/scheduler.py

@@ -925,6 +925,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             platform="cron",
             session_id=_cron_session_id,
             session_db=_session_db,
+            session_title=job_name,
         )
 
         # Run the agent with an *inactivity*-based timeout: the job can run

hermes_state.py

@@ -388,13 +388,15 @@ def create_session(
         system_prompt: str = None,
         user_id: str = None,
         parent_session_id: str = None,
+        title: str = None,
     ) -> str:
         """Create a new session record. Returns the session_id."""
         def _do(conn):
+            resolved_title = self._allocate_unique_title(conn, title, session_id=session_id)
             conn.execute(
                 """INSERT OR IGNORE INTO sessions (id, source, user_id, model, model_config,
-                   system_prompt, parent_session_id, started_at)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
+                   system_prompt, parent_session_id, title, started_at)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                 (
                     session_id,
                     source,
@@ -403,6 +405,7 @@ def _do(conn):
                     json.dumps(model_config) if model_config else None,
                     system_prompt,
                     parent_session_id,
+                    resolved_title,
                     time.time(),
                 ),
             )
@@ -544,6 +547,7 @@ def ensure_session(
         session_id: str,
         source: str = "unknown",
         model: str = None,
+        title: str = None,
     ) -> None:
         """Ensure a session row exists, creating it with minimal metadata if absent.
 
@@ -552,11 +556,12 @@ def ensure_session(
         INSERT OR IGNORE is safe to call even when the row already exists.
         """
         def _do(conn):
+            resolved_title = self._allocate_unique_title(conn, title, session_id=session_id)
             conn.execute(
                 """INSERT OR IGNORE INTO sessions
-                   (id, source, model, started_at)
-                   VALUES (?, ?, ?, ?)""",
-                (session_id, source, model, time.time()),
+                   (id, source, model, title, started_at)
+                   VALUES (?, ?, ?, ?, ?)""",
+                (session_id, source, model, resolved_title, time.time()),
             )
         self._execute_write(_do)
 
@@ -643,6 +648,48 @@ def sanitize_title(title: Optional[str]) -> Optional[str]:
 
         return cleaned
 
+    def _allocate_unique_title(
+        self,
+        conn: sqlite3.Connection,
+        title: Optional[str],
+        session_id: Optional[str] = None,
+    ) -> Optional[str]:
+        """Sanitize a title and allocate a unique lineage variant when needed."""
+        sanitized = self.sanitize_title(title)
+        if sanitized is None:
+            return None
+
+        params = [sanitized]
+        sql = "SELECT 1 FROM sessions WHERE title = ?"
+        if session_id is not None:
+            sql += " AND id != ?"
+            params.append(session_id)
+
+        if conn.execute(sql, tuple(params)).fetchone() is None:
+            return sanitized
+
+        match = re.match(r'^(.*?) #(\d+)$', sanitized)
+        base = match.group(1) if match else sanitized
+        escaped = base.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+        params = [base, f"{escaped} #%"]
+        sql = "SELECT title FROM sessions WHERE (title = ? OR title LIKE ? ESCAPE '\\')"
+        if session_id is not None:
+            sql += " AND id != ?"
+            params.append(session_id)
+
+        existing = [row["title"] for row in conn.execute(sql, tuple(params)).fetchall()]
+        if not existing:
+            return base
+
+        max_num = 1
+        for existing_title in existing:
+            match = re.match(rf'^{re.escape(base)} #(\d+)$', existing_title)
+            if match:
+                max_num = max(max_num, int(match.group(1)))
+
+        return f"{base} #{max_num + 1}"
+
     def set_session_title(self, session_id: str, title: str) -> bool:
         """Set or update a session's title.
 

run_agent.py

@@ -751,6 +751,7 @@ def __init__(
         skip_memory: bool = False,
         session_db=None,
         parent_session_id: str = None,
+        session_title: str = None,
         iteration_budget: "IterationBudget" = None,
         fallback_model: Dict[str, Any] = None,
         credential_pool=None,
@@ -1381,6 +1382,7 @@ def __init__(
         # SQLite session store (optional -- provided by CLI or gateway)
         self._session_db = session_db
         self._parent_session_id = parent_session_id
+        self._session_title = session_title
         self._last_flushed_db_idx = 0  # tracks DB-write cursor to prevent duplicate writes
         if self._session_db:
             try:
@@ -1395,6 +1397,7 @@ def __init__(
                     },
                     user_id=None,
                     parent_session_id=self._parent_session_id,
+                    title=self._session_title,
                 )
             except Exception as e:
                 # Transient SQLite lock contention (e.g. CLI and gateway writing
@@ -3026,6 +3029,7 @@ def _flush_messages_to_session_db(self, messages: List[Dict], conversation_histo
                 self.session_id,
                 source=self.platform or "cli",
                 model=self.model,
+                title=self._session_title,
             )
             start_idx = len(conversation_history) if conversation_history else 0
             flush_from = max(start_idx, self._last_flushed_db_idx)

tests/run_agent/test_run_agent.py

@@ -3468,6 +3468,29 @@ def test_fresh_build_when_db_has_no_prompt(self, agent):
         # Empty string is falsy, so should fall through to fresh build
         assert "Hermes Agent" in agent._cached_system_prompt
 
+
+class TestSessionPersistence:
+    def test_flush_messages_to_session_db_passes_session_title_to_ensure_session(self, agent):
+        mock_db = MagicMock()
+        agent._session_db = mock_db
+        agent._session_title = "wiki-auto-ingest"
+        agent.session_id = "cron-session-1"
+        agent.platform = "cron"
+        agent.model = "test-model"
+        agent._last_flushed_db_idx = 0
+
+        messages = [{"role": "user", "content": "hello"}]
+
+        agent._flush_messages_to_session_db(messages, conversation_history=[])
+
+        mock_db.ensure_session.assert_called_once_with(
+            "cron-session-1",
+            source="cron",
+            model="test-model",
+            title="wiki-auto-ingest",
+        )
+        mock_db.append_message.assert_called_once()
+
 class TestBudgetPressure:
     """Budget exhaustion grace call system."""
 

tests/test_hermes_state.py

@@ -989,6 +989,18 @@ def test_delete_orphans_children(self, db):
 # =========================================================================
 
 class TestSessionTitle:
+    def test_create_session_with_duplicate_title_gets_numbered_variant(self, db):
+        db.create_session(session_id="s1", source="cron", title="wiki-auto-ingest")
+        db.create_session(session_id="s2", source="cron", title="wiki-auto-ingest")
+
+        assert db.get_session("s1")["title"] == "wiki-auto-ingest"
+        assert db.get_session("s2")["title"] == "wiki-auto-ingest #2"
+
+    def test_create_session_sanitizes_title_input(self, db):
+        db.create_session(session_id="s1", source="cron", title="  hello\t\nworld  ")
+
+        assert db.get_session("s1")["title"] == "hello world"
+
     def test_set_and_get_title(self, db):
         db.create_session(session_id="s1", source="cli")
         assert db.set_session_title("s1", "My Session") is True
@@ -1754,6 +1766,19 @@ def test_ensure_session_creates_missing_row(self, db):
         assert row["source"] == "gateway"
         assert row["model"] == "test-model"
 
+    def test_ensure_session_preserves_requested_title(self, db):
+        """Late recovery path should still keep the caller's requested title."""
+        db.ensure_session(
+            "late-session",
+            source="cron",
+            model="gpt-4",
+            title="wiki-auto-ingest",
+        )
+
+        row = db.get_session("late-session")
+        assert row is not None
+        assert row["title"] == "wiki-auto-ingest"
+
     def test_ensure_session_is_idempotent(self, db):
         """ensure_session on an existing row must be a no-op (no overwrite)."""
         db.create_session(session_id="existing", source="cli", model="original-model")