feat(session-search): scope default discovery to the current shared chat

tests/tools/test_session_search.py

@@ -12,6 +12,7 @@
 
 import pytest
 
+from gateway.session_context import clear_session_vars, set_session_vars
 from hermes_state import SessionDB
 from tools.session_search_tool import (
     SESSION_SEARCH_SCHEMA,
@@ -254,6 +255,83 @@ def test_current_session_filtered_out(self, db):
         sids = [r["session_id"] for r in result["results"]]
         assert "s_newest" not in sids
 
+    def test_shared_context_defaults_discovery_to_current_chat_thread(self, db):
+        now = int(time.time())
+
+        db.create_session(
+            "s_signal_prior_same_scope",
+            source="signal",
+            chat_id="group-alpha",
+            chat_type="group",
+            thread_id="topic-1",
+        )
+        db._conn.execute(
+            "UPDATE sessions SET started_at = ?, title = ? WHERE id = ?",
+            (now - 120, "Alpha Topic Earlier", "s_signal_prior_same_scope"),
+        )
+        db.append_message("s_signal_prior_same_scope", role="user", content="Need the room inventory migration plan")
+
+        db.create_session(
+            "s_signal_current",
+            source="signal",
+            chat_id="group-alpha",
+            chat_type="group",
+            thread_id="topic-1",
+        )
+        db._conn.execute(
+            "UPDATE sessions SET started_at = ?, title = ? WHERE id = ?",
+            (now - 100, "Alpha Topic", "s_signal_current"),
+        )
+        db.append_message("s_signal_current", role="user", content="Need the room inventory migration plan")
+
+        db.create_session(
+            "s_signal_other_topic",
+            source="signal",
+            chat_id="group-alpha",
+            chat_type="group",
+            thread_id="topic-2",
+        )
+        db._conn.execute(
+            "UPDATE sessions SET started_at = ?, title = ? WHERE id = ?",
+            (now - 90, "Other Topic", "s_signal_other_topic"),
+        )
+        db.append_message("s_signal_other_topic", role="user", content="Need the room inventory migration plan")
+
+        db.create_session(
+            "s_signal_other_chat",
+            source="signal",
+            chat_id="group-beta",
+            chat_type="group",
+            thread_id="topic-1",
+        )
+        db._conn.execute(
+            "UPDATE sessions SET started_at = ?, title = ? WHERE id = ?",
+            (now - 80, "Other Chat", "s_signal_other_chat"),
+        )
+        db.append_message("s_signal_other_chat", role="user", content="Need the room inventory migration plan")
+        db._conn.commit()
+
+        tokens = set_session_vars(
+            platform="signal",
+            source="signal",
+            chat_id="group-alpha",
+            thread_id="topic-1",
+            session_id="s_signal_current",
+        )
+        try:
+            result = json.loads(
+                session_search(
+                    query="room inventory migration plan",
+                    db=db,
+                    current_session_id="s_signal_current",
+                )
+            )
+        finally:
+            clear_session_vars(tokens)
+
+        sids = [r["session_id"] for r in result["results"]]
+        assert sids == ["s_signal_prior_same_scope"]
+
 
 class TestDiscoverySort:
     def test_sort_newest_orders_by_recency(self, db):

tools/session_search_tool.py

@@ -33,6 +33,8 @@
 import logging
 from typing import Any, Dict, List, Optional, Union
 
+from gateway.session_context import get_session_env
+
 # Sources that are excluded from session browsing/searching by default.
 # Third-party integrations tag their sessions with HERMES_SESSION_SOURCE=tool;
 # delegate subagent runs are tagged "subagent" — neither belongs in the
@@ -141,6 +143,46 @@ def _shape_message(m: Dict[str, Any], anchor_id: Optional[int] = None) -> Dict[s
     return {k: v for k, v in entry.items() if v is not None or k in ("content",)}
 
 
+def _current_recall_scope() -> Optional[Dict[str, str]]:
+    """Return current shared-chat scope for default discovery filtering.
+
+    Only shared gateway contexts get implicit scoping. Local/CLI contexts and
+    sessions without a bound chat id remain global.
+    """
+    platform = (get_session_env("HERMES_SESSION_PLATFORM", "") or "").strip().lower()
+    source = (get_session_env("HERMES_SESSION_SOURCE", "") or platform).strip().lower()
+    chat_id = (get_session_env("HERMES_SESSION_CHAT_ID", "") or "").strip()
+    thread_id = (get_session_env("HERMES_SESSION_THREAD_ID", "") or "").strip()
+    if not source or source in {"local", "cli"}:
+        return None
+    if not chat_id:
+        return None
+    return {
+        "source": source,
+        "chat_id": chat_id,
+        "thread_id": thread_id,
+    }
+
+
+def _session_matches_scope(db, session_id: Optional[str], scope: Optional[Dict[str, str]]) -> bool:
+    """True when the session belongs to the current shared chat/thread scope."""
+    if not scope or not session_id:
+        return True
+    try:
+        meta = db.get_session(session_id) or {}
+    except Exception:
+        logging.debug("get_session failed for scope filter %s", session_id, exc_info=True)
+        return False
+    if (meta.get("source") or "").strip().lower() != scope["source"]:
+        return False
+    if (meta.get("chat_id") or "") != scope["chat_id"]:
+        return False
+    scope_thread = scope.get("thread_id") or ""
+    if scope_thread:
+        return (meta.get("thread_id") or "") == scope_thread
+    return True
+
+
 def _resolve_profile_db(profile: str):
     """Open another profile's ``state.db`` read-only, or None for the current one.
 
@@ -508,6 +550,7 @@ def _discover(
     role_list = role_filter if role_filter else ["user", "assistant"]
     current_lineage_root = _resolve_to_parent(db, current_session_id) if current_session_id else None
     title_result = _title_match_result(db, query, current_lineage_root)
+    current_scope = _current_recall_scope()
 
     try:
         raw_results = db.search_messages(
@@ -530,6 +573,11 @@ def _discover(
     # within each class.
     raw_results = _order_for_recall(raw_results)
 
+    if current_scope:
+        raw_results = [r for r in raw_results if _session_matches_scope(db, r.get("session_id"), current_scope)]
+        if title_result and not _session_matches_scope(db, title_result.get("session_id"), current_scope):
+            title_result = None
+
     if not raw_results and not title_result:
         return json.dumps({
             "success": True,