feat(gateway): add /sessions search <query> (#57685)

Gateway users can now search resumable sessions from messaging surfaces:
/sessions search <query> (alias: find) matches titles and session ids —
including every title/id in a row's forward compression chain, so a
compressed-away title still surfaces its live tip — plus a
punctuation-normalized variant so 'an94' matches 'AN-94'.

Implemented by generalizing the existing id_query chain-filter in
SessionDB.list_sessions_rich into a combined SQL-level filter (search
stays ORDER BY last-active + LIMIT at SQL level), threading a
search_query through the shared query_session_listing helper, and
teaching parse_session_listing_args to split off a search query.

Search results pass through the existing _resume_row_visible guard
unchanged: origin scoping, admin-only 'all', and the fail-closed
legacy-row posture from the July 1 hardening are preserved exactly.
Over-fetch (50) before the visibility cut so origin-invisible matches
can't starve the page.

Salvages the feature direction of PR #57595 by @GodsBoy with a minimal
implementation that keeps the resume authorization surface untouched.

gateway/slash_commands.py

@@ -3666,10 +3666,15 @@ class GatewaySlashCommandsMixin:
         source = event.source
         raw_args = event.get_command_args().strip()
         try:
-            include_all, include_unnamed, target = parse_session_listing_args(raw_args)
+            include_all, include_unnamed, target, search_query = (
+                parse_session_listing_args(raw_args)
+            )
         except ValueError as exc:
             return t("gateway.resume.parse_error", error=exc)
 
+        if search_query == "":
+            return "Usage: `/sessions search <query>`"
+
         if target:
             resume_event = dataclasses.replace(event, text=f"/resume {target}")
             return await self._handle_resume_command(resume_event)
@@ -3688,7 +3693,10 @@ class GatewaySlashCommandsMixin:
             current_session_id=current_entry.session_id,
             include_all_sources=cross_origin,
             include_unnamed=include_unnamed,
-            limit=10,
+            search_query=search_query,
+            # Search filters at SQL level, so over-fetch before the visibility
+            # cut: origin-invisible matches would otherwise consume the page.
+            limit=50 if search_query else 10,
             exclude_sources=["tool"],
         )
         if not cross_origin:
@@ -3698,10 +3706,15 @@ class GatewaySlashCommandsMixin:
                 row for row in rows
                 if await self._resume_row_visible(source, row, allow_all=False)
             ]
+        rows = rows[:10]
+        if search_query:
+            title = f"Sessions matching “{search_query}”"
+        else:
+            title = "Sessions" if include_unnamed else "Named Sessions"
         return format_gateway_session_listing(
             rows,
             include_source=cross_origin,
-            title="Sessions" if include_unnamed else "Named Sessions",
+            title=title,
         )
 
     async def _handle_branch_command(self, event: MessageEvent) -> str:

hermes_cli/session_listing.py

@@ -5,13 +5,18 @@ from __future__ import annotations
 from typing import Any
 
 
-def parse_session_listing_args(raw_args: str) -> tuple[bool, bool, str]:
-    """Parse `/sessions`-style args into listing flags plus a resume target.
+def parse_session_listing_args(raw_args: str) -> tuple[bool, bool, str, str | None]:
+    """Parse `/sessions`-style args into listing flags, a resume target, and a search query.
 
-    Returns ``(include_all_sources, include_unnamed, target)``. ``list``/``ls``
-    and ``browse`` are display aliases; ``all``/``--all`` widens source scope;
-    ``full``/``--full`` keeps unnamed sessions in the listing. Anything else is
-    treated as a target so `/sessions <id-or-title>` can delegate to `/resume`.
+    Returns ``(include_all_sources, include_unnamed, target, search_query)``.
+    ``list``/``ls`` and ``browse`` are display aliases; ``all``/``--all`` widens
+    source scope; ``full``/``--full`` keeps unnamed sessions in the listing.
+    ``search``/``find`` makes the remaining words a search query —
+    ``search_query`` is ``None`` when search wasn't requested and ``""`` when it
+    was requested without a query. Flags are only honored before the first
+    positional word, so titles containing e.g. "all" aren't misparsed. Anything
+    else is treated as a target so `/sessions <id-or-title>` can delegate to
+    `/resume`.
     """
     import shlex
 
@@ -19,18 +24,22 @@ def parse_session_listing_args(raw_args: str) -> tuple[bool, bool, str]:
     include_all = False
     include_unnamed = False
     target_parts: list[str] = []
-    for part in parts:
+    for i, part in enumerate(parts):
         lower = part.strip().lower()
-        if lower in {"list", "ls", "browse"}:
-            continue
-        if lower in {"all", "--all"}:
-            include_all = True
-            continue
-        if lower in {"full", "--full"}:
-            include_unnamed = True
-            continue
+        if not target_parts:
+            if lower in {"list", "ls", "browse"}:
+                continue
+            if lower in {"all", "--all"}:
+                include_all = True
+                continue
+            if lower in {"full", "--full"}:
+                include_unnamed = True
+                continue
+            if lower in {"search", "find"}:
+                query = " ".join(parts[i + 1:]).strip()
+                return include_all, include_unnamed, "", query
         target_parts.append(part)
-    return include_all, include_unnamed, " ".join(target_parts).strip()
+    return include_all, include_unnamed, " ".join(target_parts).strip(), None
 
 
 def query_session_listing(
@@ -40,6 +49,7 @@ def query_session_listing(
     current_session_id: str | None = None,
     include_all_sources: bool = False,
     include_unnamed: bool = False,
+    search_query: str | None = None,
     limit: int = 10,
     exclude_sources: list[str] | None = None,
 ) -> list[dict[str, Any]]:
@@ -48,19 +58,25 @@ def query_session_listing(
     This is the shared selection policy behind CLI/gateway session browsing:
     source-scoped by default, optionally global, hide unnamed sessions unless
     the caller asks for a full listing, and never include the current session.
+    With ``search_query``, rows are filtered by title/id match (SQL-level, see
+    ``SessionDB.list_sessions_rich``) and ordered by most-recent activity;
+    unnamed sessions stay visible since an id match may be the only handle.
     """
     query_source = None if include_all_sources else source
     fetch_limit = max(limit * 4, limit)
+    search = (search_query or "").strip()
     rows = session_db.list_sessions_rich(
         source=query_source,
         exclude_sources=exclude_sources,
         limit=fetch_limit,
+        search_query=search or None,
+        order_by_last_active=bool(search),
     )
     result: list[dict[str, Any]] = []
     for row in rows:
         if current_session_id and row.get("id") == current_session_id:
             continue
-        if not include_unnamed and not row.get("title"):
+        if not include_unnamed and not row.get("title") and not search:
             continue
         result.append(row)
         if len(result) >= limit:
@@ -93,5 +109,5 @@ def format_gateway_session_listing(
         lines.append(f"{idx}. **{title_text}**{source_part} — `{session_id}`{preview_part}")
     lines.append("")
     lines.append("Resume: `/resume <session id>` or `/resume <number>` from `/resume`.")
-    lines.append("More: `/sessions all`, `/sessions full`, `/sessions all full`.")
+    lines.append("More: `/sessions all`, `/sessions full`, `/sessions search <query>`.")
     return "\n".join(lines)

hermes_state.py

@@ -2712,6 +2712,7 @@ class SessionDB:
         include_archived: bool = False,
         archived_only: bool = False,
         id_query: str = None,
+        search_query: str = None,
     ) -> List[Dict[str, Any]]:
         """List sessions with preview (first user message) and last active timestamp.
 
@@ -2739,6 +2740,12 @@ class SessionDB:
         surfaces in the correct slot. Ordering is computed at SQL level via
         a recursive CTE that walks compression-continuation edges, so LIMIT
         and OFFSET still apply efficiently.
+
+        ``search_query`` matches case-insensitive substrings against each
+        surfaced row's title and id (and, like ``id_query``, every title/id in
+        its forward compression chain). A punctuation-stripped variant is also
+        matched so e.g. ``an94`` finds ``AN-94``. Only honored in the
+        ``order_by_last_active`` path.
         """
         where_clauses = []
         params = []
@@ -2791,6 +2798,7 @@ class SessionDB:
         # order_by_last_active path (which builds the chain CTE); other callers
         # pass id_query=None.
         id_needle = (id_query or "").strip().lower()
+        search_needle = (search_query or "").strip().lower()
         if order_by_last_active:
             # Compute effective_last_active by walking each surfaced session's
             # compression-continuation chain forward in SQL and taking the MAX
@@ -2807,25 +2815,53 @@ class SessionDB:
             # the timestamp test and hijack resume/list projection.
             outer_where = where_sql
             id_params: List[Any] = []
+            filter_clauses: List[str] = []
+
+            def _like_pattern(needle: str) -> str:
+                escaped = (
+                    needle.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+                )
+                return f"%{escaped}%"
+
             if id_needle:
                 # Admit a surfaced row if its own id or any id in its forward
                 # compression chain matches the needle. LIKE with a leading
                 # wildcard can't use an index, but the chain membership and
                 # the small result set keep this bounded — far cheaper than
                 # fetching every session and scanning in Python.
-                id_clause = (
+                filter_clauses.append(
                     "EXISTS (SELECT 1 FROM chain cq"
                     "        WHERE cq.root_id = s.id"
                     "          AND LOWER(cq.cur_id) LIKE ? ESCAPE '\\')"
                 )
-                like_pattern = (
-                    "%"
-                    + id_needle.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
-                    + "%"
+                id_params.append(_like_pattern(id_needle))
+            if search_needle:
+                # Same chain-membership trick as id_query, but matching either
+                # the title or the id of any session in the chain. The compact
+                # (punctuation-stripped) variant lets `an94` match `AN-94`.
+                compact_needle = re.sub(r"[\W_]+", "", search_needle)
+                compact_sql = (
+                    "REPLACE(REPLACE(REPLACE(REPLACE(LOWER(COALESCE({0}, '')),"
+                    " '-', ''), '_', ''), '.', ''), ' ', '')"
                 )
-                id_params = [like_pattern]
+                search_clause = (
+                    "EXISTS (SELECT 1 FROM chain cq"
+                    " JOIN sessions cs ON cs.id = cq.cur_id"
+                    " WHERE cq.root_id = s.id"
+                    " AND (LOWER(COALESCE(cs.title, '')) LIKE ? ESCAPE '\\'"
+                    " OR LOWER(cq.cur_id) LIKE ? ESCAPE '\\'"
+                )
+                id_params.extend([_like_pattern(search_needle)] * 2)
+                if compact_needle:
+                    search_clause += (
+                        f" OR {compact_sql.format('cs.title')} LIKE ? ESCAPE '\\'"
+                    )
+                    id_params.append(_like_pattern(compact_needle))
+                filter_clauses.append(search_clause + "))")
+            if filter_clauses:
+                combined = " AND ".join(filter_clauses)
                 outer_where = (
-                    f"{where_sql} AND {id_clause}" if where_sql else f"WHERE {id_clause}"
+                    f"{where_sql} AND {combined}" if where_sql else f"WHERE {combined}"
                 )
             query = f"""
                 WITH RECURSIVE chain(root_id, cur_id) AS (

tests/gateway/test_resume_command.py

@@ -451,6 +451,60 @@ class TestHandleSessionsCommand:
         assert "Discord" not in result
         db.close()
 
+    @pytest.mark.asyncio
+    async def test_sessions_search_finds_older_titled_session(self, tmp_path):
+        """`/sessions search <query>` matches titles beyond the recent-10 list
+        and orders by activity, keeping the caller's own scope."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        # Bury the target under newer sessions so a plain listing misses it.
+        db.create_session("target_an94", "telegram", user_id="12345", chat_id="67890")
+        db.set_session_title("target_an94", "AN-94 Prestige Barrel Build #2")
+        for i in range(12):
+            sid = f"filler_{i}"
+            db.create_session(sid, "telegram", user_id="12345", chat_id="67890")
+            db.set_session_title(sid, f"Filler {i}")
+
+        event = _make_event(text="/sessions search an94")
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_sessions_command(event)
+
+        assert "AN-94 Prestige Barrel Build #2" in result
+        assert "target_an94" in result
+        assert "Filler" not in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_sessions_search_missing_query_shows_usage(self, tmp_path):
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/sessions search")
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_sessions_command(event)
+        assert "Usage" in result
+        assert "/sessions search" in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_sessions_search_does_not_leak_other_users_sessions(self, tmp_path):
+        """Search results honor the same owner-scoping guard as listing —
+        a matching title owned by a different user/chat must not surface."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("mine", "telegram", user_id="12345", chat_id="67890")
+        db.set_session_title("mine", "AN-94 mine")
+        db.create_session("theirs", "telegram", user_id="99999", chat_id="55555")
+        db.set_session_title("theirs", "AN-94 someone else's secret")
+
+        event = _make_event(text="/sessions search an94")
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_sessions_command(event)
+
+        assert "AN-94 mine" in result
+        assert "theirs" not in result
+        assert "secret" not in result
+        db.close()
+
     @pytest.mark.asyncio
     async def test_resume_blocks_cross_user_and_unowned_rows(self, tmp_path):
         """An identity-bearing caller cannot resume a session it can't prove it

tests/hermes_cli/test_session_listing.py

@@ -0,0 +1,99 @@
+"""Tests for the shared session-listing helpers (hermes_cli/session_listing.py)."""
+
+import pytest
+
+from hermes_cli.session_listing import (
+    parse_session_listing_args,
+    query_session_listing,
+)
+
+
+class TestParseSessionListingArgs:
+    def test_plain_listing(self):
+        assert parse_session_listing_args("") == (False, False, "", None)
+
+    def test_flags(self):
+        assert parse_session_listing_args("all full") == (True, True, "", None)
+
+    def test_target_passthrough(self):
+        assert parse_session_listing_args("My Cool Session") == (
+            False, False, "My Cool Session", None,
+        )
+
+    def test_search_query(self):
+        assert parse_session_listing_args("search an94") == (False, False, "", "an94")
+
+    def test_find_alias_multiword(self):
+        assert parse_session_listing_args("find winton email") == (
+            False, False, "", "winton email",
+        )
+
+    def test_all_search(self):
+        assert parse_session_listing_args("all search cod") == (True, False, "", "cod")
+
+    def test_search_without_query_is_empty_string(self):
+        assert parse_session_listing_args("search") == (False, False, "", "")
+
+    def test_search_word_inside_target_is_not_a_flag(self):
+        # Flags/keywords only apply before the first positional word.
+        assert parse_session_listing_args("deep search notes") == (
+            False, False, "deep search notes", None,
+        )
+
+
+class TestQuerySessionListingSearch:
+    @pytest.fixture
+    def db(self, tmp_path):
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("sess_an94", "telegram", user_id="1", chat_id="2")
+        db.set_session_title("sess_an94", "AN-94 Prestige Barrel Build #2")
+        db.create_session("sess_winton", "whatsapp", user_id="1", chat_id="2")
+        db.set_session_title("sess_winton", "Winton Email Sheet Update #3")
+        db.create_session("sess_untitled", "telegram", user_id="1", chat_id="2")
+        yield db
+        db.close()
+
+    def _ids(self, db, **kw):
+        return [r["id"] for r in query_session_listing(db, **kw)]
+
+    def test_title_substring_match(self, db):
+        assert self._ids(db, source="telegram", search_query="prestige") == ["sess_an94"]
+
+    def test_punctuation_normalized_match(self, db):
+        # "an94" should match the title "AN-94 ..." via compact matching.
+        assert self._ids(db, source="telegram", search_query="an94") == ["sess_an94"]
+
+    def test_id_substring_match_includes_unnamed(self, db):
+        assert self._ids(db, source="telegram", search_query="untitled") == ["sess_untitled"]
+
+    def test_source_scoping(self, db):
+        assert self._ids(db, source="telegram", search_query="winton") == []
+        assert self._ids(db, source="whatsapp", search_query="winton") == ["sess_winton"]
+
+    def test_no_match(self, db):
+        assert self._ids(db, source="telegram", search_query="zzz-nope") == []
+
+    def test_like_wildcards_are_literal(self, db):
+        assert self._ids(db, source="telegram", search_query="%") == []
+
+    def test_search_matches_compression_root_title(self, tmp_path):
+        """Searching an old (compressed-away) title surfaces the live tip."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "chain.db")
+        db.create_session("root_1", "telegram", user_id="1", chat_id="2")
+        db.set_session_title("root_1", "Old Chat")
+        db.end_session("root_1", end_reason="compression")
+        db.create_session(
+            "tip_1", "telegram", user_id="1", chat_id="2", parent_session_id="root_1"
+        )
+        db.set_session_title("tip_1", "AN-94 Build")
+        try:
+            for query in ("old chat", "root_1", "an94"):
+                rows = query_session_listing(db, source="telegram", search_query=query)
+                assert [r["id"] for r in rows] == ["tip_1"], query
+        finally:
+            db.close()
+
+    def test_plain_listing_still_hides_unnamed(self, db):
+        assert self._ids(db, source="telegram") == ["sess_an94"]