feat(state): slow-query log for session search with routing-path attribution

One INFO line per slow search naming the path taken (fts_cjk / fts5 /
trigram / like_scan), elapsed time, row count, and the query. The 2026-07
session_search investigation needed turn-trace archaeology plus workload
replay to discover that short-CJK queries were full-scanning the table —
with this line the next routing regression is a journalctl grep.

Threshold: sessions.search_slow_ms (default 1000ms; 0 logs every call),
bridged to HERMES_SEARCH_SLOW_MS.

Salvaged from PR #65544 (adapted to the v23 schema in follow-up commits).

hermes_state.py

@@ -6300,6 +6300,76 @@ class SessionDB:
         offset: int = 0,
         sort: str = None,
         include_inactive: bool = False,
+    ) -> List[Dict[str, Any]]:
+        """Instrumented wrapper around :meth:`_search_messages_impl`.
+
+        Logs one line per slow search with the routing path taken, so
+        production latency stays attributable per query shape (the 2026-07
+        session_search investigation needed trace archaeology to discover
+        the LIKE full scans; this makes the next regression a grep).
+        Threshold: HERMES_SEARCH_SLOW_MS (default 1000; 0 logs every call).
+        """
+        started = time.time()
+        rows = None
+        try:
+            rows = self._search_messages_impl(
+                query,
+                source_filter=source_filter,
+                exclude_sources=exclude_sources,
+                role_filter=role_filter,
+                limit=limit,
+                offset=offset,
+                sort=sort,
+                include_inactive=include_inactive,
+            )
+            return rows
+        finally:
+            try:
+                threshold = float(os.getenv("HERMES_SEARCH_SLOW_MS", "1000"))
+            except (TypeError, ValueError):
+                threshold = 1000.0
+            elapsed_ms = (time.time() - started) * 1000.0
+            if elapsed_ms >= threshold:
+                logger.info(
+                    "slow session search: path=%s elapsed=%.0fms rows=%s query=%r",
+                    self._describe_search_path(query),
+                    elapsed_ms,
+                    len(rows) if rows is not None else "err",
+                    query[:200],
+                )
+
+    def _describe_search_path(self, query: str) -> str:
+        """Best-effort name of the routing path a query takes (log-only)."""
+        try:
+            sanitized = self._sanitize_fts5_query(query or "")
+            if not sanitized:
+                return "empty"
+            if self._fts_v2_query_allowed(sanitized):
+                return "fts_v2"
+            if not self._contains_cjk(sanitized):
+                return "fts5"
+            raw = sanitized.strip('"').strip()
+            tokens = [
+                t for t in raw.split()
+                if t.upper() not in {"AND", "OR", "NOT"} and self._contains_cjk(t)
+            ]
+            short = any(self._count_cjk(t) < 3 for t in tokens)
+            if self._count_cjk(raw) >= 3 and not short and self._trigram_available:
+                return "trigram"
+            return "like_scan"
+        except Exception:
+            return "unknown"
+
+    def _search_messages_impl(
+        self,
+        query: str,
+        source_filter: List[str] = None,
+        exclude_sources: List[str] = None,
+        role_filter: List[str] = None,
+        limit: int = 20,
+        offset: int = 0,
+        sort: str = None,
+        include_inactive: bool = False,
     ) -> List[Dict[str, Any]]:
         """
         Full-text search across session messages using FTS5.

tests/test_search_slow_query_log.py

@@ -0,0 +1,47 @@
+"""Tests for the session-search slow-query log (patch search-slow-query-log)."""
+
+import logging
+
+import pytest
+
+from hermes_state import SessionDB
+
+
+@pytest.fixture()
+def db(tmp_path):
+    d = SessionDB(db_path=tmp_path / "state.db")
+    d.create_session(session_id="s1", source="cli", model="m")
+    d.append_message("s1", role="user", content="hello graphiti 일본 MCP 정리")
+    yield d
+    d.close()
+
+
+def test_slow_log_emitted_at_zero_threshold(db, monkeypatch, caplog):
+    monkeypatch.setenv("HERMES_SEARCH_SLOW_MS", "0")
+    with caplog.at_level(logging.INFO, logger="hermes_state"):
+        rows = db.search_messages("graphiti", limit=5)
+    assert rows
+    slow = [r for r in caplog.records if "slow session search" in r.getMessage()]
+    assert slow, "threshold 0 must log every search"
+    msg = slow[0].getMessage()
+    assert "path=" in msg and "rows=1" in msg
+
+
+def test_no_log_under_threshold(db, monkeypatch, caplog):
+    monkeypatch.setenv("HERMES_SEARCH_SLOW_MS", "60000")
+    with caplog.at_level(logging.INFO, logger="hermes_state"):
+        db.search_messages("graphiti", limit=5)
+    assert not [r for r in caplog.records if "slow session search" in r.getMessage()]
+
+
+def test_path_attribution(db, monkeypatch):
+    monkeypatch.setenv("HERMES_FTS_V2_READ", "0")
+    assert db._describe_search_path("graphiti OR neo4j") == "fts5"
+    assert db._describe_search_path("우선순위 캘린더") == "trigram"
+    assert db._describe_search_path("일본 MCP") == "like_scan"
+
+
+def test_results_unchanged_by_wrapper(db, monkeypatch):
+    monkeypatch.setenv("HERMES_SEARCH_SLOW_MS", "0")
+    rows = db.search_messages("graphiti", limit=5)
+    assert rows and rows[0]["session_id"] == "s1"