diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index a0b077e3e589..731201ccc300 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -516,6 +516,12 @@ def _release_lock() -> None: source=agent.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"), model=agent.model, model_config=agent._session_init_model_config, + user_id=getattr(agent, "_user_id", None), + user_id_alt=getattr(agent, "_user_id_alt", None), + chat_type=getattr(agent, "_chat_type", None), + chat_id=getattr(agent, "_chat_id", None), + thread_id=getattr(agent, "_thread_id", None), + session_key=getattr(agent, "_gateway_session_key", None), parent_session_id=old_session_id, ) agent._session_db_created = True diff --git a/agent/tool_executor.py b/agent/tool_executor.py index bbbd239dff9a..5aad581dca1f 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -717,8 +717,17 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe around_message_id=function_args.get("around_message_id"), window=function_args.get("window", 5), sort=function_args.get("sort"), + mode=function_args.get("mode"), + scope=function_args.get("scope"), db=session_db, current_session_id=agent.session_id, + current_source=agent.platform or getattr(agent, "_platform", None), + current_user_id=getattr(agent, "_user_id", None), + current_user_id_alt=getattr(agent, "_user_id_alt", None), + current_chat_type=getattr(agent, "_chat_type", None), + current_chat_id=getattr(agent, "_chat_id", None), + current_thread_id=getattr(agent, "_thread_id", None), + current_session_key=getattr(agent, "_gateway_session_key", None), ) tool_duration = time.time() - tool_start_time if agent._should_emit_quiet_tool_messages(): diff --git a/docs/plans/2026-05-31-session-search-scope-handoff-contract.md b/docs/plans/2026-05-31-session-search-scope-handoff-contract.md new file mode 100644 index 000000000000..166046abbc97 --- /dev/null +++ b/docs/plans/2026-05-31-session-search-scope-handoff-contract.md @@ -0,0 +1,138 @@ +# Session Search Scope + Previous/Handoff Contract Implementation Plan + +> **For Hermes:** This plan is the implementation contract for fixing gateway session-search cross-contamination after `/new`. + +**Goal:** Prevent `/new` handoff/previous-session requests from retrieving unrelated sessions, especially across QQ users or adjacent projects. + +**Architecture:** Store stable session scope metadata in `sessions`, propagate current gateway scope into `session_search`, and add an explicit `previous`/`handoff` retrieval path that does not rely on keyword search. Default gateway recall is scoped; explicit global search remains available for debug/admin use. + +**Tech Stack:** Python, SQLite SessionDB, Hermes gateway `SessionSource`, tool executor, `session_search` tool. + +--- + +## Merged Review Contract + +This merges Codex design review and the user's follow-up review. + +### 1. Dedicated previous/handoff mode + +`session_search(mode="previous" | "handoff", scope="current")` + +Behavior: + +- Only uses the current platform/chat scope. +- Excludes the current session lineage. +- Picks the most recent ended session first (`ended_at DESC`). +- Falls back to last message activity only after `ended_at` ordering. +- Never performs keyword/global discovery. +- If scoped lookup finds nothing, returns an empty result; it must not silently fall back to global search. + +This is the core fix for “刚才那个会话 / 交接信息”. + +### 2. Stable scope fields + +`session_key` may be persisted and used as auxiliary evidence, but default isolation is based on stable business fields: + +- `source` — canonical platform/source (`qqbot`, `telegram`, `cli`, `webui`, `cron`, etc.) +- `chat_type` +- `chat_id` +- `thread_id` +- `user_id` +- `session_key` + +QQ DM default scope is `source + chat_type + chat_id`; `user_id` is stored but not the primary QQ DM isolation key. + +### 3. Scope propagation chain + +The current scope must be available from gateway to tool execution: + +- gateway adapter creates `SessionSource` +- gateway `SessionStore` persists scope on new/reset session creation +- `run_agent.AIAgent` receives `platform/user_id/chat_id/chat_type/thread_id/gateway_session_key` +- `agent/tool_executor.py` passes those as hidden current-scope kwargs to `session_search` +- `tools/session_search_tool.py` applies scope defaults and filters + +### 4. Legacy compatibility with strict fallback + +New sessions write full scope fields. Old sessions may have null scope fields. + +Rules: + +- `previous`/`handoff`: primary path is scoped new fields. Legacy fallback is allowed only within the same `source`, excluding current lineage, bounded to recent/ended ordering, and marked in the response. No cross-source/global fallback. +- Ordinary search/browse: gateway sessions default to current scope. CLI remains broad/legacy-friendly unless explicit scope is provided. +- Global search must be explicit: `scope="global"`. + +### 5. Reliable ended_at + +`/new`, auto reset, session switch, and compression split should mark old sessions ended. This change depends on existing `SessionStore.reset_session()` and `SessionDB.end_session()` behavior; tests must cover `/new`-style ended-session selection. + +### 6. Behavior-level regression tests + +Required tests: + +- QQ user A/B both mention “新增功能”; A scoped search does not see B. +- A `/new` then `mode="handoff"` returns A's just-ended admissions session. +- Adjacent admissions/tutoring sessions: `mode="handoff"` returns admissions and not tutoring/OCR/PDF/学生档案. +- Current lineage is excluded. +- Scoped no-result does not global fallback unless `scope="global"`. +- Legacy null-scope sessions are not lost, but fallback is source-bounded and flagged. +- CLI search is not accidentally constrained by QQ scope rules. + +## Implementation Tasks + +### Task 1: Add scope columns and write paths + +Modify `hermes_state.py`: + +- Add nullable columns to `sessions`: `chat_type`, `chat_id`, `thread_id`, `session_key`, `user_id_alt`. +- Keep indexes referencing new columns after `_reconcile_columns()`. +- Extend `_insert_session_row()` / `create_session()` to accept those kwargs. + +Modify `gateway/session.py` and `run_agent.py`: + +- Pass scope metadata when creating DB sessions. + +### Task 2: Add scope helper + scoped filtering + +Modify `tools/session_search_tool.py`: + +- Add helper to resolve current scope from hidden kwargs. +- Add helper to decide default `scope`: + - gateway source (`qqbot`, `telegram`, `discord`, `slack`, etc.) + chat scope => current + - CLI/local with no chat scope => legacy/global-ish + - explicit `scope="global"` bypasses scope filters +- Add shared session scope matcher. + +### Task 3: Add previous/handoff mode + +Modify `tools/session_search_tool.py`: + +- Add `mode` schema enum: `previous`, `handoff`. +- Implement previous/handoff selection by current scope, excluding current lineage. +- Return recent session metadata + bookend start/end + messages; no FTS keyword search. +- Return empty when scoped none. + +### Task 4: Preserve search behavior with scoped default + +Modify discovery/browse paths: + +- Apply scope filters when default/current scoped. +- Keep explicit global mode. +- Preserve CLI broad search by default. + +### Task 5: Verify + +Run focused tests: + +```bash +python -m pytest tests/tools/test_session_search.py -o addopts='' -q +python -m pytest tests/gateway/test_session*.py tests/test_hermes_state*.py -o addopts='' -q +``` + +Then inspect: + +```bash +git status --short --branch --untracked-files=all +git diff --stat +git diff --check +``` diff --git a/gateway/session.py b/gateway/session.py index 5f6fcb9a62fa..075be618675c 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -937,6 +937,11 @@ def get_or_create_session( "session_id": session_id, "source": source.platform.value, "user_id": source.user_id, + "user_id_alt": source.user_id_alt, + "chat_type": source.chat_type, + "chat_id": str(source.chat_id) if source.chat_id is not None else None, + "thread_id": str(source.thread_id) if source.thread_id is not None else None, + "session_key": session_key, } # SQLite operations outside the lock @@ -1163,6 +1168,11 @@ def reset_session(self, session_key: str, display_name: Optional[str] = None) -> "session_id": session_id, "source": old_entry.platform.value if old_entry.platform else "unknown", "user_id": old_entry.origin.user_id if old_entry.origin else None, + "user_id_alt": old_entry.origin.user_id_alt if old_entry.origin else None, + "chat_type": old_entry.origin.chat_type if old_entry.origin else old_entry.chat_type, + "chat_id": str(old_entry.origin.chat_id) if old_entry.origin and old_entry.origin.chat_id is not None else None, + "thread_id": str(old_entry.origin.thread_id) if old_entry.origin and old_entry.origin.thread_id is not None else None, + "session_key": session_key, } if self._db and db_end_session_id: diff --git a/hermes_state.py b/hermes_state.py index 5122c69b9396..c79b8385ae7e 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -33,7 +33,7 @@ DEFAULT_DB_PATH = get_hermes_home() / "state.db" -SCHEMA_VERSION = 14 +SCHEMA_VERSION = 15 # --------------------------------------------------------------------------- # WAL-compatibility fallback @@ -235,6 +235,11 @@ def _log_wal_fallback_once(db_label: str, exc: Exception) -> None: id TEXT PRIMARY KEY, source TEXT NOT NULL, user_id TEXT, + user_id_alt TEXT, + chat_type TEXT, + chat_id TEXT, + thread_id TEXT, + session_key TEXT, model TEXT, model_config TEXT, system_prompt TEXT, @@ -754,6 +759,21 @@ def _init_schema(self): # recreates them. self._drop_fts_triggers(cursor) + try: + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_sessions_scope_recent " + "ON sessions(source, chat_type, chat_id, thread_id, ended_at, started_at)" + ) + except sqlite3.OperationalError as exc: + logger.debug("idx_sessions_scope_recent create skipped: %s", exc) + try: + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_sessions_session_key " + "ON sessions(session_key) WHERE session_key IS NOT NULL" + ) + except sqlite3.OperationalError as exc: + logger.debug("idx_sessions_session_key create skipped: %s", exc) + # ── Schema version bookkeeping ───────────────────────────────── # Bump to current so future data migrations (if any) can gate on # version. No version-gated column additions remain. @@ -889,18 +909,29 @@ def _insert_session_row( model_config: Dict[str, Any] = None, system_prompt: str = None, user_id: str = None, + user_id_alt: str = None, + chat_type: str = None, + chat_id: str = None, + thread_id: str = None, + session_key: str = None, parent_session_id: str = None, ) -> None: """Shared INSERT OR IGNORE for session rows.""" def _do(conn): conn.execute( - """INSERT OR IGNORE INTO sessions (id, source, user_id, model, model_config, + """INSERT OR IGNORE INTO sessions (id, source, user_id, user_id_alt, + chat_type, chat_id, thread_id, session_key, model, model_config, system_prompt, parent_session_id, started_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( session_id, source, user_id, + user_id_alt, + chat_type, + chat_id, + thread_id, + session_key, model, json.dumps(model_config) if model_config else None, system_prompt, diff --git a/run_agent.py b/run_agent.py index 18ca748908d0..9b826c79b0d0 100644 --- a/run_agent.py +++ b/run_agent.py @@ -483,7 +483,12 @@ def _ensure_db_session(self) -> None: model=self.model, model_config=self._session_init_model_config, system_prompt=self._cached_system_prompt, - user_id=None, + user_id=self._user_id, + user_id_alt=self._user_id_alt, + chat_type=self._chat_type, + chat_id=self._chat_id, + thread_id=self._thread_id, + session_key=self._gateway_session_key, parent_session_id=self._parent_session_id, ) self._session_db_created = True diff --git a/tests/agent/test_compression_concurrent_fork.py b/tests/agent/test_compression_concurrent_fork.py index 76e8a459258a..d7f3533ae45f 100644 --- a/tests/agent/test_compression_concurrent_fork.py +++ b/tests/agent/test_compression_concurrent_fork.py @@ -39,7 +39,7 @@ from hermes_state import SessionDB -def _build_agent_with_db(db: SessionDB, session_id: str): +def _build_agent_with_db(db: SessionDB, session_id: str, **agent_kwargs): """Build an AIAgent that's wired to ``db`` and pinned to ``session_id``.""" with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}): from run_agent import AIAgent @@ -53,6 +53,7 @@ def _build_agent_with_db(db: SessionDB, session_id: str): session_id=session_id, skip_context_files=True, skip_memory=True, + **agent_kwargs, ) # Stub the compressor so it returns deterministic output and DOESN'T make @@ -173,6 +174,47 @@ def test_skipped_compression_returns_messages_unchanged(tmp_path: Path) -> None: agent.context_compressor.compress.assert_not_called() +def test_compression_child_inherits_gateway_scope(tmp_path: Path) -> None: + """Compression-created child sessions must stay in the same gateway scope.""" + db = SessionDB(db_path=tmp_path / "state.db") + parent_sid = "SCOPED_PARENT" + db.create_session( + parent_sid, + source="discord", + user_id="user-a", + user_id_alt="alt-a", + chat_type="group", + chat_id="chat-1", + thread_id="thread-1", + session_key="agent:main:discord:group:chat-1:user-a", + ) + + agent = _build_agent_with_db( + db, + parent_sid, + platform="discord", + user_id="user-a", + user_id_alt="alt-a", + chat_type="group", + chat_id="chat-1", + thread_id="thread-1", + gateway_session_key="agent:main:discord:group:chat-1:user-a", + ) + messages = [{"role": "user", "content": f"m{i}"} for i in range(20)] + + agent._compress_context(messages, "sys", approx_tokens=120_000) + + child = db.get_session(agent.session_id) + assert child["parent_session_id"] == parent_sid + assert child["source"] == "discord" + assert child["user_id"] == "user-a" + assert child["user_id_alt"] == "alt-a" + assert child["chat_type"] == "group" + assert child["chat_id"] == "chat-1" + assert child["thread_id"] == "thread-1" + assert child["session_key"] == "agent:main:discord:group:chat-1:user-a" + + class _NoLockSubsystemDB: """Wraps a real SessionDB but simulates a pre-#34351 version skew. diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 99a8616e2e61..e7c98967c9a1 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -1791,9 +1791,13 @@ def test_topic_mode_schema_is_not_auto_migrated_on_open(self, tmp_path): conn.close() db = SessionDB(db_path=old_db) - cursor = db._conn.execute("PRAGMA table_info(sessions)") - columns = {row[1] for row in cursor.fetchall()} - assert {"chat_id", "chat_type", "thread_id", "session_key"}.isdisjoint(columns) + tables = { + row[0] + for row in db._conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ).fetchall() + } + assert "telegram_dm_topic_bindings" not in tables db.close() def test_apply_telegram_topic_migration_creates_topic_tables_explicitly(self, tmp_path): diff --git a/tests/tools/test_session_search.py b/tests/tools/test_session_search.py index 3f517aa1a4b6..534abd11d628 100644 --- a/tests/tools/test_session_search.py +++ b/tests/tools/test_session_search.py @@ -76,10 +76,10 @@ def test_schema_has_required_params(self): # Shared assert "role_filter" in params - def test_no_mode_parameter(self): - # Mode is inferred from which args are set — no explicit mode param + def test_mode_parameter_includes_previous_handoff(self): params = SESSION_SEARCH_SCHEMA["parameters"]["properties"] - assert "mode" not in params + assert params["mode"]["enum"] == ["previous", "handoff"] + assert params["scope"]["enum"] == ["current", "global"] def test_sort_enum(self): params = SESSION_SEARCH_SCHEMA["parameters"]["properties"] @@ -142,6 +142,247 @@ def test_browse_returns_titles(self, db): assert any("Modpack" in (t or "") for t in titles) +# ========================================================================= +# Scoped gateway recall + previous/handoff mode +# ========================================================================= + +class TestScopedGatewayRecall: + def _create_scoped_session( + self, + db, + session_id, + *, + source="qqbot", + chat_type="dm", + chat_id="user-a", + user_id="user-a", + session_key=None, + started_at=None, + ended_at=None, + text="新增功能", + ): + db.create_session( + session_id, + source=source, + user_id=user_id, + chat_type=chat_type, + chat_id=chat_id, + thread_id=None, + session_key=session_key or f"agent:main:{source}:{chat_type}:{chat_id}", + ) + if started_at is not None or ended_at is not None: + db._conn.execute( + "UPDATE sessions SET started_at = COALESCE(?, started_at), ended_at = ? WHERE id = ?", + (started_at, ended_at, session_id), + ) + db._conn.commit() + db.append_message(session_id, role="user", content=text) + db.append_message(session_id, role="assistant", content=f"已记录：{text}") + + def test_default_gateway_search_is_scoped_to_current_chat(self, db): + self._create_scoped_session(db, "a_old", chat_id="qq-a", user_id="qq-a", text="新增功能 admissions") + self._create_scoped_session(db, "b_old", chat_id="qq-b", user_id="qq-b", text="新增功能 tutoring") + db.create_session("a_current", source="qqbot", user_id="qq-a", chat_type="dm", chat_id="qq-a") + + result = json.loads(session_search( + query="新增功能", + db=db, + current_session_id="a_current", + current_source="qqbot", + current_chat_type="dm", + current_chat_id="qq-a", + current_user_id="qq-a", + )) + sids = [r["session_id"] for r in result["results"]] + assert "a_old" in sids + assert "b_old" not in sids + + def test_handoff_returns_recent_ended_session_not_adjacent_project(self, db): + now = time.time() + self._create_scoped_session( + db, + "admissions_done", + chat_id="qq-a", + user_id="qq-a", + started_at=now - 200, + ended_at=now - 5, + text="Stage 52 admissions-sales-workbench PR #16 已 merge", + ) + self._create_scoped_session( + db, + "tutoring_neighbor", + chat_id="qq-a", + user_id="qq-a", + started_at=now - 100, + ended_at=now - 30, + text="/workspace/tutoring-exam-analysis OCR PDF 学生档案", + ) + db.create_session("a_current", source="qqbot", user_id="qq-a", chat_type="dm", chat_id="qq-a") + + result = json.loads(session_search( + mode="handoff", + db=db, + current_session_id="a_current", + current_source="qqbot", + current_chat_type="dm", + current_chat_id="qq-a", + current_user_id="qq-a", + )) + assert result["success"] is True + assert result["mode"] == "handoff" + assert result["results"][0]["session_id"] == "admissions_done" + payload = json.dumps(result, ensure_ascii=False) + assert "/workspace/tutoring-exam-analysis" not in payload + assert "OCR" not in payload + assert "PDF" not in payload + assert "学生档案" not in payload + + def test_scoped_no_result_does_not_global_fallback(self, db): + self._create_scoped_session(db, "b_old", chat_id="qq-b", user_id="qq-b", text="新增功能 only b") + db.create_session("a_current", source="qqbot", user_id="qq-a", chat_type="dm", chat_id="qq-a") + + scoped = json.loads(session_search( + query="新增功能", + db=db, + current_session_id="a_current", + current_source="qqbot", + current_chat_type="dm", + current_chat_id="qq-a", + current_user_id="qq-a", + )) + assert scoped["results"] == [] + + global_result = json.loads(session_search( + query="新增功能", + scope="global", + db=db, + current_session_id="a_current", + current_source="qqbot", + current_chat_type="dm", + current_chat_id="qq-a", + current_user_id="qq-a", + )) + assert [r["session_id"] for r in global_result["results"]] == ["b_old"] + + def test_group_chat_recall_is_isolated_by_user_and_session_key(self, db): + now = time.time() + shared_chat = "group-1" + key_a = "agent:main:qqbot:group:group-1:user-a" + key_b = "agent:main:qqbot:group:group-1:user-b" + self._create_scoped_session( + db, + "group_a_old", + chat_type="group", + chat_id=shared_chat, + user_id="user-a", + session_key=key_a, + started_at=now - 200, + ended_at=now - 100, + text="shared chat scoped secret alpha", + ) + self._create_scoped_session( + db, + "group_b_old", + chat_type="group", + chat_id=shared_chat, + user_id="user-b", + session_key=key_b, + started_at=now - 100, + ended_at=now - 10, + text="shared chat scoped secret beta", + ) + self._create_scoped_session( + db, + "group_missing_user", + chat_type="group", + chat_id=shared_chat, + user_id=None, + session_key=None, + started_at=now - 80, + ended_at=now - 20, + text="shared chat scoped secret missing-user", + ) + db._conn.execute( + "UPDATE sessions SET user_id = NULL, user_id_alt = NULL, session_key = NULL WHERE id = ?", + ("group_missing_user",), + ) + db._conn.commit() + db.create_session("group_legacy_missing_user", source="qqbot") + db._conn.execute( + "UPDATE sessions SET started_at = ?, ended_at = ? WHERE id = ?", + (now - 70, now - 15, "group_legacy_missing_user"), + ) + db.append_message( + "group_legacy_missing_user", + role="user", + content="shared chat scoped secret legacy-missing-user", + ) + db.create_session( + "group_a_current", + source="qqbot", + user_id="user-a", + chat_type="group", + chat_id=shared_chat, + session_key=key_a, + ) + + common_scope = dict( + db=db, + current_session_id="group_a_current", + current_source="qqbot", + current_chat_type="group", + current_chat_id=shared_chat, + current_user_id="user-a", + current_session_key=key_a, + ) + search = json.loads(session_search(query="shared chat scoped secret", **common_scope)) + assert [r["session_id"] for r in search["results"]] == ["group_a_old"] + payload = json.dumps(search, ensure_ascii=False) + assert "group_b_old" not in payload + assert "group_missing_user" not in payload + assert "group_legacy_missing_user" not in payload + + previous = json.loads(session_search(mode="previous", **common_scope)) + assert previous["results"][0]["session_id"] == "group_a_old" + assert "group_b_old" not in json.dumps(previous, ensure_ascii=False) + + handoff = json.loads(session_search(mode="handoff", **common_scope)) + assert handoff["results"][0]["session_id"] == "group_a_old" + assert "group_b_old" not in json.dumps(handoff, ensure_ascii=False) + + def test_handoff_legacy_fallback_is_source_bounded_and_marked(self, db): + now = time.time() + db.create_session("legacy_qq", source="qqbot", user_id="qq-a") + db._conn.execute("UPDATE sessions SET started_at = ?, ended_at = ? WHERE id = ?", (now - 50, now - 10, "legacy_qq")) + db.append_message("legacy_qq", role="user", content="legacy admissions handoff") + db.create_session("legacy_telegram", source="telegram", user_id="qq-a") + db._conn.execute("UPDATE sessions SET started_at = ?, ended_at = ? WHERE id = ?", (now - 40, now - 5, "legacy_telegram")) + db.append_message("legacy_telegram", role="user", content="legacy telegram should not leak") + db.create_session("a_current", source="qqbot", user_id="qq-a", chat_type="dm", chat_id="qq-a") + + result = json.loads(session_search( + mode="previous", + db=db, + current_session_id="a_current", + current_source="qqbot", + current_chat_type="dm", + current_chat_id="qq-a", + current_user_id="qq-a", + )) + assert result["results"][0]["session_id"] == "legacy_qq" + assert result["results"][0].get("legacy_scope_fallback") is True + assert "legacy_telegram" not in json.dumps(result, ensure_ascii=False) + + def test_cli_search_remains_broad_by_default(self, db): + self._create_scoped_session(db, "qq_old", chat_id="qq-a", user_id="qq-a", text="modpack qq") + db.create_session("cli_old", source="cli") + db.append_message("cli_old", role="user", content="modpack cli") + + result = json.loads(session_search(query="modpack", db=db, current_source="cli")) + sids = {r["session_id"] for r in result["results"]} + assert {"qq_old", "cli_old"}.issubset(sids) + + # ========================================================================= # Discovery shape (with query) # ========================================================================= @@ -347,6 +588,79 @@ def test_scroll_invalid_around_message_id_errors(self, db): )) assert result["success"] is False + def test_scroll_rejects_out_of_scope_session_and_lineage_rebind(self, db): + shared_chat = "group-1" + key_a = "agent:main:qqbot:group:group-1:user-a" + key_b = "agent:main:qqbot:group:group-1:user-b" + db.create_session( + "owner_parent", + source="qqbot", + user_id="user-a", + chat_type="group", + chat_id=shared_chat, + session_key=key_a, + ) + parent_mid = db.append_message("owner_parent", role="user", content="owner parent") + db.create_session( + "other_parent", + source="qqbot", + user_id="user-b", + chat_type="group", + chat_id=shared_chat, + session_key=key_b, + ) + other_mid = db.append_message("other_parent", role="user", content="other parent") + db.create_session( + "other_child", + source="qqbot", + user_id="user-b", + chat_type="group", + chat_id=shared_chat, + session_key=key_b, + parent_session_id="owner_parent", + ) + child_mid = db.append_message("other_child", role="user", content="other child") + db.create_session( + "owner_current", + source="qqbot", + user_id="user-a", + chat_type="group", + chat_id=shared_chat, + session_key=key_a, + ) + + common_scope = dict( + db=db, + current_session_id="owner_current", + current_source="qqbot", + current_chat_type="group", + current_chat_id=shared_chat, + current_user_id="user-a", + current_session_key=key_a, + ) + own = json.loads(session_search( + session_id="owner_parent", + around_message_id=parent_mid, + **common_scope, + )) + assert own["success"] is True + + direct = json.loads(session_search( + session_id="other_parent", + around_message_id=other_mid, + **common_scope, + )) + assert direct["success"] is False + assert "outside the current scope" in direct.get("error", "") + + rebound = json.loads(session_search( + session_id="owner_parent", + around_message_id=child_mid, + **common_scope, + )) + assert rebound["success"] is False + assert "outside the current scope" in rebound.get("error", "") + class TestScrollPattern: """The forward/backward scroll loop using tool output.""" diff --git a/tools/session_search_tool.py b/tools/session_search_tool.py index 65b9d32f1f70..7234e32e7127 100644 --- a/tools/session_search_tool.py +++ b/tools/session_search_tool.py @@ -37,6 +37,193 @@ # Third-party integrations tag their sessions with HERMES_SESSION_SOURCE=tool # so they don't clutter the user's session history. _HIDDEN_SESSION_SOURCES = ("tool",) +_GATEWAY_SCOPED_SOURCES = { + "qqbot", + "telegram", + "discord", + "slack", + "whatsapp", + "signal", + "matrix", + "mattermost", + "feishu", + "wecom", + "dingding", + "bluebubbles", + "yuanbao", + "api_server", + "webui", +} + + +def _clean(value: Any) -> str: + return str(value or "").strip() + + +def _build_current_scope( + *, + current_source: str = None, + current_user_id: str = None, + current_user_id_alt: str = None, + current_chat_type: str = None, + current_chat_id: str = None, + current_thread_id: str = None, + current_session_key: str = None, +) -> Dict[str, str]: + """Normalize hidden current-session scope passed by the agent runtime.""" + return { + "source": _clean(current_source).lower(), + "user_id": _clean(current_user_id), + "user_id_alt": _clean(current_user_id_alt), + "chat_type": _clean(current_chat_type).lower(), + "chat_id": _clean(current_chat_id), + "thread_id": _clean(current_thread_id), + "session_key": _clean(current_session_key), + } + + +def _should_scope_to_current(scope: str, current_scope: Dict[str, str]) -> bool: + scope_norm = _clean(scope).lower() + if scope_norm == "global": + return False + if scope_norm == "current": + return bool(current_scope.get("source") and current_scope.get("chat_id")) + source = current_scope.get("source") or "" + if source in {"cli", "local", "tool"}: + return False + return bool(source in _GATEWAY_SCOPED_SOURCES and current_scope.get("chat_id")) + + +def _current_user_values(current_scope: Dict[str, str]) -> set[str]: + return { + value + for value in ( + current_scope.get("user_id") or "", + current_scope.get("user_id_alt") or "", + ) + if value + } + + +def _row_user_values(session_meta: Dict[str, Any]) -> set[str]: + return { + value + for value in ( + _clean(session_meta.get("user_id")), + _clean(session_meta.get("user_id_alt")), + ) + if value + } + + +def _requires_user_isolation(current_scope: Dict[str, str]) -> bool: + chat_type = (current_scope.get("chat_type") or "").lower() + return bool(chat_type and chat_type not in {"dm", "direct", "private"}) + + +def _session_matches_current_scope(session_meta: Dict[str, Any], current_scope: Dict[str, str]) -> bool: + """Strict current-scope match for new sessions with persisted scope fields.""" + if not session_meta: + return False + if _clean(session_meta.get("source")).lower() != (current_scope.get("source") or ""): + return False + # Gateway chat scope is authoritative. For QQ DM this is source+chat_type+chat_id. + if current_scope.get("chat_type"): + if _clean(session_meta.get("chat_type")).lower() != current_scope.get("chat_type"): + return False + if current_scope.get("chat_id"): + if _clean(session_meta.get("chat_id")) != current_scope.get("chat_id"): + return False + if current_scope.get("thread_id"): + if _clean(session_meta.get("thread_id")) != current_scope.get("thread_id"): + return False + else: + # A non-thread current chat should not see thread-specific sessions. + if _clean(session_meta.get("thread_id")): + return False + + # Modern scoped rows also carry per-user / stable session identity. In + # group/channel chats the chat_id is shared, so these fields prevent one + # sender from seeing another sender's scoped recall just because they share + # a channel. If the row has both user IDs, either may match the current + # primary/alternate identity. + current_session_key = current_scope.get("session_key") or "" + row_session_key = _clean(session_meta.get("session_key")) + if current_session_key and row_session_key and row_session_key != current_session_key: + return False + current_users = _current_user_values(current_scope) + row_users = _row_user_values(session_meta) + if current_users and row_users and current_users.isdisjoint(row_users): + return False + if _requires_user_isolation(current_scope): + # Shared chats must have at least one positive per-user signal: either + # a matching stable session key or an intersecting user identity. + if current_session_key and row_session_key == current_session_key: + return True + if current_users and row_users and not current_users.isdisjoint(row_users): + return True + return False + return True + + +def _session_allowed_for_current_scope( + session_meta: Dict[str, Any], + current_scope: Dict[str, str], + *, + include_legacy: bool = False, +) -> bool: + if _session_matches_current_scope(session_meta, current_scope): + return True + return include_legacy and _session_is_legacy_scope_candidate(session_meta, current_scope) + + +def _session_is_legacy_scope_candidate(session_meta: Dict[str, Any], current_scope: Dict[str, str]) -> bool: + """Narrow legacy fallback for rows that predate persisted chat scope fields.""" + if not session_meta: + return False + if _clean(session_meta.get("source")).lower() != (current_scope.get("source") or ""): + return False + if _clean(session_meta.get("chat_id")) or _clean(session_meta.get("thread_id")) or _clean(session_meta.get("session_key")): + return False + row_user = _clean(session_meta.get("user_id")) + cur_users = _current_user_values(current_scope) + if _requires_user_isolation(current_scope) and cur_users: + return bool(row_user and row_user in cur_users) + return not row_user or not cur_users or row_user in cur_users + + +def _filter_sessions_for_scope( + db, + sessions: List[Dict[str, Any]], + *, + current_scope: Dict[str, str], + apply_scope: bool, + current_session_id: str = None, + include_legacy: bool = False, +) -> List[Dict[str, Any]]: + current_root = _resolve_to_parent(db, current_session_id) if current_session_id else None + filtered: List[Dict[str, Any]] = [] + for row in sessions: + sid = row.get("id") or row.get("session_id") or "" + lineage = _resolve_to_parent(db, sid) if sid else sid + if current_root and lineage == current_root: + continue + meta = row + if "chat_id" not in meta and sid: + try: + meta = db.get_session(sid) or row + except Exception: + meta = row + if apply_scope: + if _session_matches_current_scope(meta, current_scope): + filtered.append(row) + elif include_legacy and _session_is_legacy_scope_candidate(meta, current_scope): + legacy_row = dict(row) + legacy_row["legacy_scope_fallback"] = True + filtered.append(legacy_row) + else: + filtered.append(row) + return filtered def _format_timestamp(ts: Union[int, float, str, None]) -> str: @@ -107,7 +294,14 @@ 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 _list_recent_sessions(db, limit: int, current_session_id: str = None) -> str: +def _list_recent_sessions( + db, + limit: int, + current_session_id: str = None, + *, + current_scope: Optional[Dict[str, str]] = None, + apply_scope: bool = False, +) -> str: """Return metadata for the most recent sessions (no LLM calls, no FTS5).""" try: sessions = db.list_sessions_rich( @@ -116,6 +310,15 @@ def _list_recent_sessions(db, limit: int, current_session_id: str = None) -> str order_by_last_active=True, ) # fetch extra so we can skip current + sessions = _filter_sessions_for_scope( + db, + sessions, + current_scope=current_scope or {}, + apply_scope=apply_scope, + current_session_id=current_session_id, + include_legacy=True, + ) + current_root = _resolve_to_parent(db, current_session_id) if current_session_id else None results = [] @@ -126,7 +329,7 @@ def _list_recent_sessions(db, limit: int, current_session_id: str = None) -> str # Skip child / delegation sessions if s.get("parent_session_id"): continue - results.append({ + entry = { "session_id": sid, "title": s.get("title") or None, "source": s.get("source", ""), @@ -134,7 +337,10 @@ def _list_recent_sessions(db, limit: int, current_session_id: str = None) -> str "last_active": s.get("last_active", ""), "message_count": s.get("message_count", 0), "preview": s.get("preview", ""), - }) + } + if s.get("legacy_scope_fallback"): + entry["legacy_scope_fallback"] = True + results.append(entry) if len(results) >= limit: break @@ -156,6 +362,9 @@ def _scroll( around_message_id: int, window: int = 5, current_session_id: str = None, + *, + current_scope: Optional[Dict[str, str]] = None, + apply_scope: bool = False, ) -> str: """Scroll shape: return a window of messages centered on an anchor. @@ -199,6 +408,13 @@ def _scroll( session_meta = {} if not session_meta: return tool_error(f"session_id not found: {session_id}", success=False) + current_scope = current_scope or {} + if apply_scope and not _session_allowed_for_current_scope( + session_meta, + current_scope, + include_legacy=True, + ): + return tool_error("scroll rejected: session_id is outside the current scope", success=False) # Fetch the window try: @@ -230,6 +446,19 @@ def _scroll( a_root = _resolve_to_parent(db, session_id) o_root = _resolve_to_parent(db, owning) if a_root and o_root and a_root == o_root: + try: + owning_meta = db.get_session(owning) or {} + except Exception: + owning_meta = {} + if apply_scope and not _session_allowed_for_current_scope( + owning_meta, + current_scope, + include_legacy=True, + ): + return tool_error( + "scroll rejected: around_message_id is outside the current scope", + success=False, + ) try: rebind_view = db.get_messages_around(owning, around_message_id, window=window) messages = rebind_view.get("window") or [] @@ -281,6 +510,9 @@ def _discover( limit: int, sort: Optional[str], current_session_id: str = None, + *, + current_scope: Optional[Dict[str, str]] = None, + apply_scope: bool = False, ) -> str: """Discovery shape: FTS5 + anchored window + bookends per hit. Single call.""" role_list = role_filter if role_filter else ["user", "assistant"] @@ -290,7 +522,7 @@ def _discover( query=query, role_filter=role_list, exclude_sources=list(_HIDDEN_SESSION_SOURCES), - limit=50, # widen so dedup-by-lineage can find distinct sessions + limit=200, # widen so scoped filtering + dedup-by-lineage can find distinct sessions offset=0, sort=sort, ) @@ -309,6 +541,7 @@ def _discover( }, ensure_ascii=False) current_lineage_root = _resolve_to_parent(db, current_session_id) if current_session_id else None + current_scope = current_scope or {} # Dedupe by lineage. Keep the raw owning session_id on the surviving # row — only that pairs validly with the FTS5 match id for the anchored @@ -322,8 +555,20 @@ def _discover( continue if current_session_id and raw_sid == current_session_id: continue + if apply_scope: + try: + meta = db.get_session(raw_sid) or {} + except Exception: + meta = {} + legacy_match = _session_is_legacy_scope_candidate(meta, current_scope) + if not _session_matches_current_scope(meta, current_scope) and not legacy_match: + continue + else: + legacy_match = False if resolved_sid not in seen_sessions: row = dict(r) + if legacy_match: + row["legacy_scope_fallback"] = True row["_lineage_root"] = resolved_sid seen_sessions[resolved_sid] = row if len(seen_sessions) >= limit: @@ -363,6 +608,8 @@ def _discover( } if lineage_root and lineage_root != hit_sid: entry["parent_session_id"] = lineage_root + if match_info.get("legacy_scope_fallback"): + entry["legacy_scope_fallback"] = True results.append(entry) return json.dumps({ @@ -375,6 +622,124 @@ def _discover( }, ensure_ascii=False) +def _shape_handoff_session(db, session_meta: Dict[str, Any], *, legacy_scope_fallback: bool = False) -> Dict[str, Any]: + sid = session_meta.get("id") or session_meta.get("session_id") + messages = [] + try: + messages = db.get_messages(sid) if sid else [] + except Exception: + messages = [] + visible = [m for m in messages if m.get("role") in ("user", "assistant")] + entry = { + "session_id": sid, + "title": session_meta.get("title") or None, + "source": session_meta.get("source", ""), + "started_at": session_meta.get("started_at"), + "ended_at": session_meta.get("ended_at"), + "end_reason": session_meta.get("end_reason"), + "last_active": session_meta.get("last_active"), + "message_count": session_meta.get("message_count", 0), + "preview": session_meta.get("preview", ""), + "bookend_start": [_shape_message(m) for m in visible[:3]], + "bookend_end": [_shape_message(m) for m in visible[-5:]], + } + if legacy_scope_fallback: + entry["legacy_scope_fallback"] = True + return entry + + +def _previous_or_handoff( + db, + *, + mode: str, + limit: int, + current_session_id: str = None, + current_scope: Optional[Dict[str, str]] = None, + apply_scope: bool = True, +) -> str: + """Return previous/handoff sessions for the current scope without keyword search.""" + limit = 1 # previous/handoff is a single immediate predecessor by contract. + current_scope = current_scope or {} + if not apply_scope: + return json.dumps({ + "success": True, + "mode": mode, + "scope": "current", + "results": [], + "count": 0, + "message": "No current chat scope available for previous/handoff lookup.", + }, ensure_ascii=False) + + try: + sessions = db.list_sessions_rich( + limit=200, + exclude_sources=list(_HIDDEN_SESSION_SOURCES), + order_by_last_active=True, + ) + except Exception as e: + logging.error("previous/handoff list failed: %s", e, exc_info=True) + return tool_error(f"Failed to list sessions: {e}", success=False) + + # Primary scoped pass. + primary = _filter_sessions_for_scope( + db, + sessions, + current_scope=current_scope, + apply_scope=True, + current_session_id=current_session_id, + include_legacy=False, + ) + + def sort_key(row: Dict[str, Any]): + ended = row.get("ended_at") or 0 + last_active = row.get("last_active") or row.get("started_at") or 0 + started = row.get("started_at") or 0 + return (1 if ended else 0, ended, last_active, started) + + primary = sorted(primary, key=sort_key, reverse=True) + picked = primary[:limit] + legacy_used = False + + if not picked: + legacy = _filter_sessions_for_scope( + db, + sessions, + current_scope=current_scope, + apply_scope=True, + current_session_id=current_session_id, + include_legacy=True, + ) + legacy = [s for s in legacy if s.get("legacy_scope_fallback")] + legacy = sorted(legacy, key=sort_key, reverse=True) + picked = legacy[:limit] + legacy_used = bool(picked) + if legacy_used: + logging.debug( + "session_search %s using legacy source-bounded fallback for source=%s chat_type=%s chat_id=%s", + mode, + current_scope.get("source"), + current_scope.get("chat_type"), + current_scope.get("chat_id"), + ) + + results = [ + _shape_handoff_session(db, s, legacy_scope_fallback=bool(s.get("legacy_scope_fallback"))) + for s in picked + ] + return json.dumps({ + "success": True, + "mode": mode, + "scope": "current", + "results": results, + "count": len(results), + "legacy_scope_fallback": legacy_used, + "message": ( + f"Found {len(results)} previous session(s) in current scope." + if results else "No previous session found in current scope." + ), + }, ensure_ascii=False) + + def session_search( query: str = "", role_filter: str = None, @@ -387,6 +752,17 @@ def session_search( window: int = 5, # Discovery shape sort: str = None, + # Explicit previous/handoff shape and scope controls + mode: str = None, + scope: str = None, + # Hidden runtime scope, supplied by tool executor (not model-authored) + current_source: str = None, + current_user_id: str = None, + current_user_id_alt: str = None, + current_chat_type: str = None, + current_chat_id: str = None, + current_thread_id: str = None, + current_session_key: str = None, ) -> str: """Single-shape tool. Mode inferred from which args are set. @@ -406,6 +782,20 @@ def session_search( from hermes_state import format_session_db_unavailable return tool_error(format_session_db_unavailable(), success=False) + current_scope = _build_current_scope( + current_source=current_source, + current_user_id=current_user_id, + current_user_id_alt=current_user_id_alt, + current_chat_type=current_chat_type, + current_chat_id=current_chat_id, + current_thread_id=current_thread_id, + current_session_key=current_session_key, + ) + apply_scope = _should_scope_to_current(scope, current_scope) + mode_norm = _clean(mode).lower() + if mode_norm not in {"previous", "handoff"}: + mode_norm = "" + # Scroll shape takes precedence — explicit anchor beats any query. if (isinstance(session_id, str) and session_id.strip()) and around_message_id is not None: return _scroll( @@ -414,6 +804,8 @@ def session_search( around_message_id=around_message_id, window=window, current_session_id=current_session_id, + current_scope=current_scope, + apply_scope=apply_scope, ) # Limit clamp [1, 10] @@ -424,9 +816,25 @@ def session_search( limit = 3 limit = max(1, min(limit, 10)) + if mode_norm in {"previous", "handoff"}: + return _previous_or_handoff( + db, + mode=mode_norm, + limit=limit, + current_session_id=current_session_id, + current_scope=current_scope, + apply_scope=apply_scope, + ) + # Browse shape: no query → recent sessions. if not query or not isinstance(query, str) or not query.strip(): - return _list_recent_sessions(db, limit, current_session_id) + return _list_recent_sessions( + db, + limit, + current_session_id, + current_scope=current_scope, + apply_scope=apply_scope, + ) # Parse role_filter role_list: Optional[List[str]] = None @@ -447,6 +855,8 @@ def session_search( limit=limit, sort=sort_norm, current_session_id=current_session_id, + current_scope=current_scope, + apply_scope=apply_scope, ) @@ -539,6 +949,24 @@ def check_session_search_requirements() -> bool: "and browse shapes." ), }, + "mode": { + "type": "string", + "enum": ["previous", "handoff"], + "description": ( + "Dedicated previous-session shape. Use 'previous' or 'handoff' " + "when the user asks for the immediately previous/current-chat " + "conversation, e.g. '刚才那个会话' or '交接信息'. This does not do " + "keyword search or global fallback." + ), + }, + "scope": { + "type": "string", + "enum": ["current", "global"], + "description": ( + "Recall scope. Gateway sessions default to 'current' chat scope. " + "Use 'global' only for explicit debug/admin cross-session search." + ), + }, "session_id": { "type": "string", "description": ( @@ -594,8 +1022,17 @@ def check_session_search_requirements() -> bool: around_message_id=args.get("around_message_id"), window=args.get("window", 5), sort=args.get("sort"), + mode=args.get("mode"), + scope=args.get("scope"), db=kw.get("db"), current_session_id=kw.get("current_session_id"), + current_source=kw.get("current_source"), + current_user_id=kw.get("current_user_id"), + current_user_id_alt=kw.get("current_user_id_alt"), + current_chat_type=kw.get("current_chat_type"), + current_chat_id=kw.get("current_chat_id"), + current_thread_id=kw.get("current_thread_id"), + current_session_key=kw.get("current_session_key"), ), check_fn=check_session_search_requirements, emoji="🔍",