fix(session-search): allow scrolling compacted lineage history

tests/tools/test_session_search.py

@@ -381,18 +381,98 @@ def test_scroll_missing_session_errors(self, db):
         assert result["success"] is False
 
     def test_scroll_rejects_current_session_lineage(self, db):
-        _seed_modpack_sessions(db)
-        # Grab some valid id from s_oldest
-        disc = json.loads(session_search(query="modpack", limit=3, db=db))
-        match = [r for r in disc["results"] if r["session_id"] == "s_oldest"]
-        if match:
-            mid = match[0]["match_message_id"]
-            result = json.loads(session_search(
-                session_id="s_oldest", around_message_id=mid, db=db,
-                current_session_id="s_oldest",
-            ))
-            assert result["success"] is False
-            assert "current session" in result.get("error", "").lower()
+        db.create_session("s_current", source="cli")
+        mid = db.append_message("s_current", role="user", content="still live")
+
+        result = json.loads(session_search(
+            session_id="s_current", around_message_id=mid, db=db,
+            current_session_id="s_current",
+        ))
+
+        assert result["success"] is False
+        assert "current session" in result.get("error", "").lower()
+
+    def test_scroll_allows_compacted_anchor_in_current_session(self, db):
+        db.create_session("s_current", source="cli")
+        db.append_message(
+            "s_current", role="user", content="history removed from live context"
+        )
+        db.archive_and_compact("s_current", [
+            {"role": "assistant", "content": "Compacted history summary"},
+        ])
+
+        discovery = json.loads(session_search(
+            query="history removed", db=db, current_session_id="s_current",
+        ))
+        assert discovery["count"] == 1
+        anchor = discovery["results"][0]
+
+        result = json.loads(session_search(
+            session_id=anchor["session_id"],
+            around_message_id=anchor["match_message_id"],
+            db=db,
+            current_session_id="s_current",
+        ))
+
+        assert result["success"] is True
+        assert any(
+            message["id"] == anchor["match_message_id"]
+            for message in result["messages"]
+        )
+
+    def test_scroll_allows_compression_ended_parent_from_continuation(self, db):
+        db.create_session("s_parent", source="cli")
+        mid = db.append_message(
+            "s_parent", role="user", content="history summarized into child"
+        )
+        db.end_session("s_parent", "compression")
+        db.create_session("s_current", source="cli", parent_session_id="s_parent")
+
+        result = json.loads(session_search(
+            session_id="s_parent", around_message_id=mid, db=db,
+            current_session_id="s_current",
+        ))
+
+        assert result["success"] is True
+        assert any(message["id"] == mid for message in result["messages"])
+
+    def test_scroll_rejects_active_delegation_child_in_current_lineage(self, db):
+        db.create_session("s_current", source="cli")
+        db.create_session(
+            "s_delegate", source="delegate", parent_session_id="s_current"
+        )
+        mid = db.append_message(
+            "s_delegate", role="assistant", content="live delegated result"
+        )
+
+        result = json.loads(session_search(
+            session_id="s_delegate", around_message_id=mid, db=db,
+            current_session_id="s_current",
+        ))
+
+        assert result["success"] is False
+        assert "current session" in result.get("error", "").lower()
+
+    def test_scroll_rejects_rewound_anchor_in_compression_parent(self, db):
+        db.create_session("s_parent", source="cli")
+        mid = db.append_message(
+            "s_parent", role="user", content="message removed by rewind"
+        )
+        db._conn.execute(
+            "UPDATE messages SET active = 0, compacted = 0 WHERE id = ?",
+            (mid,),
+        )
+        db._conn.commit()
+        db.end_session("s_parent", "compression")
+        db.create_session("s_current", source="cli", parent_session_id="s_parent")
+
+        result = json.loads(session_search(
+            session_id="s_parent", around_message_id=mid, db=db,
+            current_session_id="s_current",
+        ))
+
+        assert result["success"] is False
+        assert "current session" in result.get("error", "").lower()
 
     def test_scroll_invalid_around_message_id_errors(self, db):
         _seed_modpack_sessions(db)

tools/session_search_tool.py

@@ -160,6 +160,25 @@ def _is_compression_ended(db, session_id: str) -> bool:
         return False
 
 
+def _get_message_storage_state(db, message_id) -> Optional[Dict[str, Any]]:
+    """Return the owning session and visibility flags for *message_id*."""
+    if not message_id:
+        return None
+    try:
+        with db._lock:
+            cursor = db._conn.execute(
+                "SELECT session_id, active, compacted FROM messages WHERE id = ?",
+                (message_id,),
+            )
+            row = cursor.fetchone()
+    except Exception:
+        logging.debug(
+            "message storage-state lookup failed for %s", message_id, exc_info=True
+        )
+        return None
+    return dict(row) if row is not None else None
+
+
 def _is_compacted_message(db, message_id) -> bool:
     """Return True if *message_id* is a compaction-archived row.
 
@@ -173,18 +192,8 @@ def _is_compacted_message(db, message_id) -> bool:
     Returns False on any error so the caller falls back to the safe default
     (skip the current session).
     """
-    if not message_id:
-        return False
-    try:
-        with db._lock:
-            cursor = db._conn.execute(
-                "SELECT active, compacted FROM messages WHERE id = ?", (message_id,)
-            )
-            row = cursor.fetchone()
-    except Exception:
-        logging.debug("is_compacted_message lookup failed for %s", message_id, exc_info=True)
-        return False
-    return row is not None and row["active"] == 0 and row["compacted"] == 1
+    state = _get_message_storage_state(db, message_id)
+    return state is not None and state["active"] == 0 and state["compacted"] == 1
 
 
 def _annotate_rebuild_status(db, payload: Dict[str, Any]) -> None:
@@ -481,16 +490,40 @@ def _scroll(
             window = 5
     window = max(1, min(window, 20))
 
-    # Reject scrolling inside the active session lineage — those messages are
-    # already in context.
+    # Locate the anchor before applying the current-lineage guard. Discovery
+    # intentionally surfaces two kinds of same-lineage history that are no
+    # longer in live context: in-place compacted rows, and rows owned by a
+    # legacy session that ended via compression. Scroll must preserve that
+    # distinction instead of rejecting the discovery result it just returned.
+    anchor_state = _get_message_storage_state(db, around_message_id)
+    owning_session_id = (
+        anchor_state.get("session_id") if anchor_state is not None else None
+    )
+
     if current_session_id:
-        a_root = _resolve_lineage(db, session_id)
+        anchor_session_id = owning_session_id or session_id
+        a_root = _resolve_lineage(db, anchor_session_id)
         c_root = _resolve_lineage(db, current_session_id)
         if a_root and c_root and a_root == c_root:
-            return tool_error(
-                "scroll rejected: anchor lives in the current session lineage (already in your active context)",
-                success=False,
+            is_compacted_anchor = (
+                anchor_state is not None
+                and anchor_state["active"] == 0
+                and anchor_state["compacted"] == 1
+            )
+            is_inactive_non_compacted_anchor = (
+                anchor_state is not None
+                and anchor_state["active"] == 0
+                and anchor_state["compacted"] != 1
             )
+            is_compression_history = (
+                not is_inactive_non_compacted_anchor
+                and _is_compression_ended(db, anchor_session_id)
+            )
+            if not (is_compacted_anchor or is_compression_history):
+                return tool_error(
+                    "scroll rejected: anchor lives in the current session lineage (already in your active context)",
+                    success=False,
+                )
 
     # Session existence check
     try:
@@ -515,18 +548,7 @@ def _scroll(
     # child sessions). Locate the real owning session and refetch.
     rebind_warning = None
     if not messages:
-        owning = None
-        try:
-            conn = getattr(db, "_conn", None)
-            if conn is not None:
-                row = conn.execute(
-                    "SELECT session_id FROM messages WHERE id = ?",
-                    (around_message_id,),
-                ).fetchone()
-                owning = row[0] if row else None
-        except Exception as e:
-            logging.debug("owning-session lookup failed: %s", e, exc_info=True)
-            owning = None
+        owning = owning_session_id
         if owning and owning != session_id:
             a_root = _resolve_lineage(db, session_id)
             o_root = _resolve_lineage(db, owning)