fix(gateway): scope session lists before limiting

gateway/session.py

@@ -1831,6 +1831,7 @@ class SessionStore:
         session_key: str,
         source: Optional[SessionSource],
         display_name: Optional[str] = None,
+        include_compression_ancestors: bool = False,
     ) -> None:
         """Persist the routing peer for an existing gateway session row."""
         if not self._db or not source:
@@ -1854,6 +1855,7 @@ class SessionStore:
                 thread_id=source.thread_id,
                 display_name=display_name or source.chat_name,
                 origin_json=origin_json,
+                include_compression_ancestors=include_compression_ancestors,
             )
         except TypeError:
             # Older SessionDB without display_name/origin_json kwargs.
@@ -2873,6 +2875,7 @@ class SessionStore:
                 session_key,
                 new_entry.origin if new_entry else None,
                 display_name=new_entry.display_name if new_entry else None,
+                include_compression_ancestors=True,
             )
 
         return new_entry

gateway/slash_commands.py

@@ -4307,7 +4307,9 @@ class GatewaySlashCommandsMixin:
             from hermes_state import format_session_db_unavailable
             return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix"))
 
-        source = event.source
+        source = await asyncio.to_thread(
+            self._normalize_source_for_session_key, event.source
+        )
         session_key = self._session_key_for_source(source)
         raw_args = event.get_command_args().strip()
         try:
@@ -4330,7 +4332,12 @@ class GatewaySlashCommandsMixin:
 
         async def _list_titled_sessions() -> list[dict]:
             user_source = source.platform.value if source.platform else None
-            sessions = await self._session_db.list_sessions_rich(source=user_source, limit=10)
+            widen = allow_all and self._resume_caller_is_admin(source)
+            sessions = await self._session_db.list_sessions_rich(
+                source=user_source,
+                session_key=None if widen else session_key,
+                limit=10,
+            )
             return [s for s in sessions if s.get("title")][:10]
 
         if not name:
@@ -4474,7 +4481,6 @@ class GatewaySlashCommandsMixin:
             query_session_listing,
         )
 
-        source = event.source
         raw_args = event.get_command_args().strip()
         try:
             include_all, include_unnamed, target, search_query = (
@@ -4490,6 +4496,11 @@ class GatewaySlashCommandsMixin:
             resume_event = dataclasses.replace(event, text=f"/resume {target}")
             return await self._handle_resume_command(resume_event)
 
+        source = await asyncio.to_thread(
+            self._normalize_source_for_session_key, event.source
+        )
+        session_key = self._session_key_for_source(source)
+
         # A cross-origin listing (`/sessions all`) is honored only for an
         # admin, mirroring the `/resume --all` override. `all` is just a parsed
         # user argument, so without this gate any caller could run
@@ -4501,6 +4512,7 @@ class GatewaySlashCommandsMixin:
             query_session_listing,
             getattr(self._session_db, "_db", self._session_db),
             source=source.platform.value if source.platform else None,
+            session_key=None if cross_origin else session_key,
             current_session_id=current_entry.session_id,
             include_all_sources=cross_origin,
             include_unnamed=include_unnamed,

hermes_cli/session_listing.py

@@ -46,6 +46,7 @@ def query_session_listing(
     session_db: Any,
     *,
     source: str | None,
+    session_key: str | None = None,
     current_session_id: str | None = None,
     include_all_sources: bool = False,
     include_unnamed: bool = False,
@@ -58,6 +59,8 @@ 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.
+    ``session_key`` further restricts gateway callers to one exact conversation
+    lane before the database applies its result limit.
     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.
@@ -67,6 +70,7 @@ def query_session_listing(
     search = (search_query or "").strip()
     rows = session_db.list_sessions_rich(
         source=query_source,
+        session_key=session_key,
         exclude_sources=exclude_sources,
         limit=fetch_limit,
         search_query=search or None,

hermes_state.py

@@ -2718,7 +2718,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         model_config: Dict[str, Any] = None,
         system_prompt: str = None,
         user_id: str = None,
-        session_key: str = None,
+        session_key: Optional[str] = None,
         chat_id: str = None,
         chat_type: str = None,
         thread_id: str = None,
@@ -2882,6 +2882,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         thread_id: str = None,
         display_name: str = None,
         origin_json: str = None,
+        include_compression_ancestors: bool = False,
     ) -> None:
         """Persist the gateway routing peer for an existing session row.
 
@@ -2890,18 +2891,43 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         channel directory) can read routing data from state.db instead of
         sessions.json.  They are COALESCE'd only in the sense that ``None``
         leaves the existing value untouched.
+
+        ``include_compression_ancestors`` keeps a logical compression lineage
+        on one routing peer when an explicit gateway resume moves its tip to a
+        different lane. Normal per-turn metadata refreshes update only the
+        supplied row.
         """
         if not session_id or not session_key:
             return
 
         def _do(conn):
-            conn.execute(
-                """UPDATE sessions
-                   SET session_key = ?, source = ?, user_id = ?, chat_id = ?,
-                       chat_type = ?, thread_id = ?,
-                       display_name = COALESCE(?, display_name),
-                       origin_json = COALESCE(?, origin_json)
-                   WHERE id = ?""",
+            lineage_cte = ""
+            target_clause = "WHERE id = ?"
+            query_params = []
+            if include_compression_ancestors:
+                lineage_cte = """
+                    WITH RECURSIVE compression_lineage(id) AS (
+                        SELECT ?
+                        UNION
+                        SELECT parent.id
+                        FROM compression_lineage lineage
+                        JOIN sessions child ON child.id = lineage.id
+                        JOIN sessions parent ON parent.id = child.parent_session_id
+                        WHERE parent.end_reason = 'compression'
+                          AND json_extract(
+                              COALESCE(child.model_config, '{}'),
+                              '$._branched_from'
+                          ) IS NULL
+                          AND json_extract(
+                              COALESCE(child.model_config, '{}'),
+                              '$._delegate_from'
+                          ) IS NULL
+                          AND COALESCE(child.source, '') != 'tool'
+                    )
+                """
+                target_clause = "WHERE id IN (SELECT id FROM compression_lineage)"
+                query_params.append(session_id)
+            query_params.extend(
                 (
                     session_key,
                     source,
@@ -2911,8 +2937,19 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                     thread_id,
                     display_name,
                     origin_json,
-                    session_id,
-                ),
+                )
+            )
+            if not include_compression_ancestors:
+                query_params.append(session_id)
+            conn.execute(
+                f"""{lineage_cte}
+                   UPDATE sessions
+                   SET session_key = ?, source = ?, user_id = ?, chat_id = ?,
+                       chat_type = ?, thread_id = ?,
+                       display_name = COALESCE(?, display_name),
+                       origin_json = COALESCE(?, origin_json)
+                   {target_clause}""",
+                query_params,
             )
 
         self._execute_write(_do)
@@ -5105,6 +5142,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         search_query: str = None,
         compact_rows: bool = False,
         include_pinned: bool = False,
+        session_key: str = None,
     ) -> List[Dict[str, Any]]:
         """List sessions with preview (first user message) and last active timestamp.
 
@@ -5152,6 +5190,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         desktop sidebar would render an empty Pinned section. Back-filled rows
         obey the same filters (source, archived, min_message_count) as the
         page: an archived or filtered-out conversation stays out.
+
+        Pass ``session_key`` to restrict results to one stable gateway
+        conversation scope (DM, group, channel, or thread, including the
+        configured per-user isolation policy).
         """
         # Rows carry token/cost totals — drain queued deltas first so
         # listings (sidebar, /resume, dashboards) show exact counters.
@@ -5182,6 +5224,9 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             placeholders = ",".join("?" for _ in include_sources)
             where_clauses.append(f"s.source IN ({placeholders})")
             params.extend(include_sources)
+        if session_key:
+            where_clauses.append("s.session_key = ?")
+            params.append(session_key)
         if exclude_sources:
             placeholders = ",".join("?" for _ in exclude_sources)
             where_clauses.append(f"s.source NOT IN ({placeholders})")

tests/gateway/test_resume_command.py

@@ -84,12 +84,19 @@ class TestHandleResumeCommand:
         """With no argument, lists recently titled sessions."""
         from hermes_state import SessionDB
         db = SessionDB(db_path=tmp_path / "state.db")
-        db.create_session("sess_001", "telegram", user_id="12345", chat_id="67890")
-        db.create_session("sess_002", "telegram", user_id="12345", chat_id="67890")
+        event = _make_event(text="/resume")
+        lane_key = _session_key_for_event(event)
+        db.create_session(
+            "sess_001", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
+        db.create_session(
+            "sess_002", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
         db.set_session_title("sess_001", "Research")
         db.set_session_title("sess_002", "Coding")
 
-        event = _make_event(text="/resume")
         runner = _make_runner(session_db=db, event=event)
         result = await runner._handle_resume_command(event)
         assert "Research" in result
@@ -228,9 +235,267 @@ class TestHandleResumeCommand:
         db.close()
 
 
+    @pytest.mark.asyncio
+    async def test_bare_resume_lists_exact_lane_before_limit(self, tmp_path):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/resume")
+        lane_key = _session_key_for_event(event)
+        for i in range(3):
+            sid = f"lane_{i}"
+            db.create_session(
+                sid, "telegram", session_key=lane_key,
+                user_id="12345", chat_id="67890",
+            )
+            db.set_session_title(sid, f"Lane Work {i}")
+        for i in range(12):
+            sid = f"foreign_{i}"
+            db.create_session(
+                sid, "telegram",
+                session_key=f"agent:main:telegram:dm:foreign-{i}",
+                user_id=f"foreign-user-{i}", chat_id=f"foreign-{i}",
+            )
+            db.set_session_title(sid, f"Foreign Work {i}")
+
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_resume_command(event)
+
+        assert "Lane Work 0" in result
+        assert "Lane Work 1" in result
+        assert "Lane Work 2" in result
+        assert "Foreign Work" not in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_bare_resume_admin_all_preserves_same_platform_widening(self, tmp_path):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/resume --all")
+        db.create_session(
+            "other_lane", "telegram",
+            session_key="agent:main:telegram:dm:other",
+            user_id="other-user", chat_id="other",
+        )
+        db.set_session_title("other_lane", "Other Lane Work")
+
+        runner = _make_runner(session_db=db, event=event)
+        runner._resume_caller_is_admin = lambda _source: True
+        result = await runner._handle_resume_command(event)
+
+        assert "Other Lane Work" in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_numeric_resume_fallback_uses_exact_lane_candidates(self, tmp_path):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/resume 2")
+        lane_key = _session_key_for_event(event)
+        db.create_session(
+            "lane_older", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
+        db.set_session_title("lane_older", "Lane Older")
+        db.create_session(
+            "lane_newer", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
+        db.set_session_title("lane_newer", "Lane Newer")
+        for i in range(12):
+            sid = f"foreign_{i}"
+            db.create_session(
+                sid, "telegram",
+                session_key=f"agent:main:telegram:dm:foreign-{i}",
+                user_id=f"foreign-user-{i}", chat_id=f"foreign-{i}",
+            )
+            db.set_session_title(sid, f"Foreign Work {i}")
+        db.create_session(
+            "current_session_001", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
+
+        runner = _make_runner(
+            session_db=db, current_session_id="current_session_001", event=event
+        )
+        result = await runner._handle_resume_command(event)
+
+        assert "Resumed" in result
+        runner.session_store.switch_session.assert_called_once()
+        assert runner.session_store.switch_session.call_args[0][1] == "lane_older"
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_bare_resume_normalizes_telegram_lobby_source_to_bound_topic(
+        self, tmp_path
+    ):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/resume")
+        topic_source = SessionSource(
+            platform=Platform.TELEGRAM,
+            user_id="12345",
+            chat_id="67890",
+            chat_type="dm",
+            thread_id="topic-42",
+        )
+        topic_key = build_session_key(topic_source)
+        db.create_session(
+            "topic_session", "telegram", session_key=topic_key,
+            user_id="12345", chat_id="67890", chat_type="dm",
+            thread_id="topic-42",
+        )
+        db.set_session_title("topic_session", "Recovered Topic Work")
+        db.enable_telegram_topic_mode(chat_id="67890", user_id="12345")
+        db.bind_telegram_topic(
+            chat_id="67890",
+            thread_id="topic-42",
+            user_id="12345",
+            session_key=topic_key,
+            session_id="topic_session",
+        )
+        lobby_key = _session_key_for_event(event)
+        db.create_session(
+            "lobby_session", "telegram", session_key=lobby_key,
+            user_id="12345", chat_id="67890", chat_type="dm",
+        )
+        db.set_session_title("lobby_session", "Lobby Work")
+
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_resume_command(event)
+
+        assert "Recovered Topic Work" in result
+        assert "Lobby Work" not in result
+        db.close()
+
+
 class TestHandleSessionsCommand:
     """Tests for GatewayRunner._handle_sessions_command."""
 
+    @pytest.mark.asyncio
+    async def test_sessions_busy_platform_lists_exact_lane_and_excludes_current_tip(
+        self, tmp_path
+    ):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/sessions")
+        lane_key = _session_key_for_event(event)
+        for i in range(11):
+            sid = f"lane_root_{i}"
+            db.create_session(
+                sid, "telegram", session_key=lane_key,
+                user_id="12345", chat_id="67890",
+            )
+            db.set_session_title(sid, f"Lane Work {i}")
+
+        db.create_session(
+            "current_root", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
+        db.set_session_title("current_root", "Current compressed root")
+        db.end_session("current_root", "compression")
+        db.create_session(
+            "current_tip", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890", parent_session_id="current_root",
+        )
+        db.set_session_title("current_tip", "Current compressed tip")
+
+        for i in range(60):
+            sid = f"foreign_{i}"
+            db.create_session(
+                sid, "telegram",
+                session_key=f"agent:main:telegram:dm:foreign-{i}",
+                user_id=f"foreign-user-{i}", chat_id=f"foreign-{i}",
+            )
+            db.set_session_title(sid, f"Foreign Work {i}")
+
+        runner = _make_runner(
+            session_db=db, current_session_id="current_tip", event=event
+        )
+        result = await runner._handle_sessions_command(event)
+
+        assert result.count("Lane Work") == 10
+        assert "Lane Work 1" in result
+        assert "Lane Work 0" not in result
+        assert "Foreign Work" not in result
+        assert "current_tip" not in result
+        assert "current_root" not in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_sessions_admin_all_preserves_cross_origin_widening(self, tmp_path):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/sessions all")
+        lane_key = _session_key_for_event(event)
+        db.create_session(
+            "tg_named", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
+        db.set_session_title("tg_named", "Telegram Work")
+        db.create_session(
+            "discord_named", "discord",
+            session_key="agent:main:discord:dm:other",
+            user_id="other-user", chat_id="other",
+        )
+        db.set_session_title("discord_named", "Discord Work")
+
+        runner = _make_runner(session_db=db, event=event)
+        runner._resume_caller_is_admin = lambda _source: True
+        result = await runner._handle_sessions_command(event)
+
+        assert "Telegram Work" in result
+        assert "Discord Work" in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_sessions_normalizes_telegram_lobby_source_to_bound_topic(self, tmp_path):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/sessions")
+        topic_source = SessionSource(
+            platform=Platform.TELEGRAM,
+            user_id="12345",
+            chat_id="67890",
+            chat_type="dm",
+            thread_id="topic-42",
+        )
+        topic_key = build_session_key(topic_source)
+        db.create_session(
+            "topic_session", "telegram", session_key=topic_key,
+            user_id="12345", chat_id="67890", chat_type="dm",
+            thread_id="topic-42",
+        )
+        db.set_session_title("topic_session", "Recovered Topic Work")
+        db.enable_telegram_topic_mode(chat_id="67890", user_id="12345")
+        db.bind_telegram_topic(
+            chat_id="67890",
+            thread_id="topic-42",
+            user_id="12345",
+            session_key=topic_key,
+            session_id="topic_session",
+        )
+        lobby_key = _session_key_for_event(event)
+        db.create_session(
+            "lobby_session", "telegram", session_key=lobby_key,
+            user_id="12345", chat_id="67890", chat_type="dm",
+        )
+        db.set_session_title("lobby_session", "Lobby Work")
+
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_sessions_command(event)
+
+        assert "Recovered Topic Work" in result
+        assert "Lobby Work" not in result
+        db.close()
+
+
 
     @pytest.mark.asyncio
     async def test_sessions_all_does_not_leak_cross_origin_for_non_admin(self, tmp_path):
@@ -241,12 +506,16 @@ class TestHandleSessionsCommand:
         config is not."""
         from hermes_state import SessionDB
         db = SessionDB(db_path=tmp_path / "state.db")
-        db.create_session("tg_named", "telegram", user_id="12345", chat_id="67890")
+        event = _make_event(text="/sessions all full")
+        lane_key = _session_key_for_event(event)
+        db.create_session(
+            "tg_named", "telegram", session_key=lane_key,
+            user_id="12345", chat_id="67890",
+        )
         db.set_session_title("tg_named", "Telegram Work")
         db.create_session("discord_unnamed", "discord")  # other origin
         db.append_message("discord_unnamed", "user", "discord first prompt")
 
-        event = _make_event(text="/sessions all full")
         runner = _make_runner(session_db=db, event=event)
 
         result = await runner._handle_sessions_command(event)
@@ -264,15 +533,22 @@ class TestHandleSessionsCommand:
         and orders by activity, keeping the caller's own scope."""
         from hermes_state import SessionDB
         db = SessionDB(db_path=tmp_path / "state.db")
+        event = _make_event(text="/sessions search an94")
+        lane_key = _session_key_for_event(event)
         # 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.create_session(
+            "target_an94", "telegram", session_key=lane_key,
+            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.create_session(
+                sid, "telegram", session_key=lane_key,
+                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)
 
@@ -288,12 +564,20 @@ class TestHandleSessionsCommand:
         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")
+        event = _make_event(text="/sessions search an94")
+        lane_key = _session_key_for_event(event)
+        db.create_session(
+            "mine", "telegram", session_key=lane_key,
+            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.create_session(
+            "theirs", "telegram",
+            session_key="agent:main:telegram:dm:55555",
+            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)
 

tests/gateway/test_session.py

@@ -491,6 +491,57 @@ class TestSessionStoreSwitchSession:
         assert resumed["end_reason"] is None
         db.close()
 
+    def test_switch_session_rebinds_full_compression_lineage(self, tmp_path):
+        from hermes_state import SessionDB
+
+        config = GatewayConfig()
+        with patch("gateway.session.SessionStore._ensure_loaded"):
+            store = SessionStore(sessions_dir=tmp_path / "sessions", config=config)
+        db = SessionDB(db_path=tmp_path / "state.db")
+        store._db = db
+        store._loaded = True
+
+        destination = SessionSource(
+            platform=Platform.TELEGRAM,
+            chat_id="destination-chat",
+            chat_type="dm",
+            user_id="destination-user",
+        )
+        current_entry = store.get_or_create_session(destination)
+        destination_key = current_entry.session_key
+        original_key = "agent:main:telegram:dm:original-chat"
+
+        db.create_session(
+            "compressed_root", "telegram", session_key=original_key,
+            user_id="original-user", chat_id="original-chat",
+        )
+        db.end_session("compressed_root", "compression")
+        db.create_session(
+            "compressed_tip", "telegram", session_key=original_key,
+            user_id="original-user", chat_id="original-chat",
+            parent_session_id="compressed_root",
+        )
+        db.end_session("compressed_tip", "session_reset")
+
+        switched = store.switch_session(destination_key, "compressed_tip")
+
+        assert switched is not None
+        assert db.get_session("compressed_root")["session_key"] == destination_key
+        assert db.get_session("compressed_tip")["session_key"] == destination_key
+        assert [
+            row["id"] for row in db.list_sessions_rich(
+                source="telegram", session_key=destination_key, limit=10
+            )
+            if row["id"] == "compressed_tip"
+        ] == ["compressed_tip"]
+        assert not any(
+            row["id"] == "compressed_tip"
+            for row in db.list_sessions_rich(
+                source="telegram", session_key=original_key, limit=10
+            )
+        )
+        db.close()
+
 
 class TestSessionStoreLookupBySessionId:
     @pytest.fixture()

tests/hermes_cli/test_session_listing.py

@@ -56,3 +56,82 @@ class TestQuerySessionListingSearch:
         finally:
             db.close()
 
+
+class TestQuerySessionListingLaneScope:
+    @pytest.fixture
+    def db(self, tmp_path):
+        from hermes_state import SessionDB
+
+        db = SessionDB(db_path=tmp_path / "state.db")
+        lane_key = "agent:main:telegram:dm:lane"
+        db.create_session(
+            "lane_current", "telegram", session_key=lane_key,
+            user_id="lane-user", chat_id="lane",
+        )
+        db.set_session_title("lane_current", "Current lane")
+        db.create_session(
+            "lane_named", "telegram", session_key=lane_key,
+            user_id="lane-user", chat_id="lane",
+        )
+        db.set_session_title("lane_named", "Needle lane")
+        db.create_session(
+            "lane_unnamed", "telegram", session_key=lane_key,
+            user_id="lane-user", chat_id="lane",
+        )
+        for i in range(60):
+            db.create_session(
+                f"foreign_{i}", "telegram",
+                session_key=f"agent:main:telegram:dm:foreign-{i}",
+                user_id=f"foreign-user-{i}", chat_id=f"foreign-{i}",
+            )
+            db.set_session_title(f"foreign_{i}", f"Needle foreign {i}")
+        yield db, lane_key
+        db.close()
+
+    def test_exact_lane_precedes_limit_and_current_session_exclusion(self, db):
+        session_db, lane_key = db
+
+        rows = query_session_listing(
+            session_db,
+            source="telegram",
+            session_key=lane_key,
+            current_session_id="lane_current",
+            limit=1,
+        )
+
+        assert [row["id"] for row in rows] == ["lane_named"]
+
+    def test_exact_lane_preserves_full_and_search_modes(self, db):
+        session_db, lane_key = db
+
+        full_rows = query_session_listing(
+            session_db,
+            source="telegram",
+            session_key=lane_key,
+            include_unnamed=True,
+            limit=10,
+        )
+        search_rows = query_session_listing(
+            session_db,
+            source="telegram",
+            session_key=lane_key,
+            search_query="needle",
+            limit=10,
+        )
+
+        assert {row["id"] for row in full_rows} == {
+            "lane_current", "lane_named", "lane_unnamed",
+        }
+        assert [row["id"] for row in search_rows] == ["lane_named"]
+
+    def test_omitted_session_key_keeps_source_scope(self, db):
+        session_db, _lane_key = db
+
+        rows = query_session_listing(
+            session_db,
+            source="telegram",
+            search_query="needle foreign 59",
+            limit=10,
+        )
+
+        assert [row["id"] for row in rows] == ["foreign_59"]

tests/test_hermes_state.py

@@ -1388,6 +1388,77 @@ class TestListSessionsRich:
 
 
 
+    def test_rich_list_session_key_filter_precedes_limit(self, db):
+        lane_key = "agent:main:telegram:dm:lane"
+        db.create_session(
+            "lane_oldest", "telegram", session_key=lane_key,
+            user_id="lane-user", chat_id="lane",
+        )
+        db.create_session(
+            "lane_newest", "telegram", session_key=lane_key,
+            user_id="lane-user", chat_id="lane",
+        )
+        for i in range(60):
+            db.create_session(
+                f"foreign_{i}", "telegram",
+                session_key=f"agent:main:telegram:dm:foreign-{i}",
+                user_id=f"foreign-user-{i}", chat_id=f"foreign-{i}",
+            )
+        db.create_session(
+            "legacy_null_key", "telegram", user_id="lane-user", chat_id="lane"
+        )
+
+        sessions = db.list_sessions_rich(
+            source="telegram", session_key=lane_key, limit=2
+        )
+
+        assert [session["id"] for session in sessions] == [
+            "lane_newest", "lane_oldest",
+        ]
+
+    def test_rich_list_session_key_scopes_search_and_projects_compression(self, db):
+        lane_key = "agent:main:telegram:dm:lane"
+        db.create_session(
+            "lane_root", "telegram", session_key=lane_key,
+            user_id="lane-user", chat_id="lane",
+        )
+        db.set_session_title("lane_root", "Needle root")
+        db.end_session("lane_root", "compression")
+        db.create_session(
+            "lane_tip", "telegram", session_key=lane_key,
+            user_id="lane-user", chat_id="lane", parent_session_id="lane_root",
+        )
+        db.set_session_title("lane_tip", "Needle continuation")
+        db.append_message("lane_tip", "user", "latest lane activity")
+        db.create_session(
+            "foreign_match", "telegram",
+            session_key="agent:main:telegram:dm:foreign",
+            user_id="foreign-user", chat_id="foreign",
+        )
+        db.set_session_title("foreign_match", "Needle foreign")
+
+        sessions = db.list_sessions_rich(
+            source="telegram",
+            session_key=lane_key,
+            search_query="needle",
+            order_by_last_active=True,
+            limit=1,
+        )
+
+        assert [session["id"] for session in sessions] == ["lane_tip"]
+        assert sessions[0]["_lineage_root_id"] == "lane_root"
+
+    def test_session_key_predicate_can_use_session_key_index(self, db):
+        plan = db._conn.execute(
+            "EXPLAIN QUERY PLAN "
+            "SELECT s.id FROM sessions s WHERE s.session_key = ? "
+            "ORDER BY s.started_at DESC LIMIT 10",
+            ("agent:main:telegram:dm:lane",),
+        ).fetchall()
+
+        detail = " ".join(row[-1] for row in plan)
+        assert "idx_sessions_session_key" in detail, detail
+
     def test_delegate_subagent_marker_hides_orphaned_row(self, db):
         """``_delegate_from`` keeps delegate rows out of pickers after orphaning."""
         db.create_session("parent", "cli")