From 8c70346e33e34d204ecf9ef1c29e8d374182d56c Mon Sep 17 00:00:00 2001
From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com>
Date: Fri, 19 Jun 2026 17:37:39 +0530
Subject: [PATCH] refactor(sessions): express compression-ancestor check as one
 recursive CTE

_is_compression_ancestor walked parent links in a 100-hop Python loop
issuing two SELECTs per hop and hand-re-encoded the compression
continuation edge a fourth time. Collapse it into a single recursive CTE
that reuses the canonical _COMPRESSION_CHILD_SQL fragment (already shared
by _ephemeral_child_sql and set_session_archived), so the edge definition
lives in exactly one place. The UNION recursion also dedups visited nodes,
making it cycle-safe without the defensive hop cap. Behavior is unchanged
(all TestSessionTitleLineage + existing title-command tests pass).
---
 hermes_state.py | 55 ++++++++++++++++++++++---------------------------
 1 file changed, 25 insertions(+), 30 deletions(-)

diff --git a/hermes_state.py b/hermes_state.py
index 2ca3c657d13..8847593d47c 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -1842,41 +1842,36 @@ class SessionDB:
         """Return True if *ancestor_id* is a compression predecessor of
         *descendant_id* (walking parent links up the continuation chain).
 
-        Uses the same edge definition as :meth:`get_compression_tip`: a
-        parent → child edge counts as a compression continuation only when the
+        The continuation edge is the canonical one shared with
+        :func:`_ephemeral_child_sql` / :meth:`set_session_archived`
+        (``_COMPRESSION_CHILD_SQL``): a parent → child edge counts only when the
         parent ended with ``end_reason = 'compression'`` and the child started
-        at or after the parent's ``ended_at`` (which distinguishes continuations
+        at or after the parent's ``ended_at``, which distinguishes continuations
         from delegate subagents / branch children that also carry a
-        ``parent_session_id``).
+        ``parent_session_id``. Expressed as a single recursive CTE rather than a
+        per-hop Python walk so the edge definition lives in exactly one place.
         """
         if not ancestor_id or not descendant_id or ancestor_id == descendant_id:
             return False
-        current = descendant_id
-        # Bound the walk defensively, mirroring get_compression_tip.
-        for _ in range(100):
-            row = conn.execute(
-                "SELECT parent_session_id, started_at FROM sessions WHERE id = ?",
-                (current,),
-            ).fetchone()
-            if row is None or not row["parent_session_id"]:
-                return False
-            parent_id = row["parent_session_id"]
-            parent = conn.execute(
-                "SELECT ended_at, end_reason FROM sessions WHERE id = ?",
-                (parent_id,),
-            ).fetchone()
-            if (
-                parent is None
-                or parent["end_reason"] != "compression"
-                or parent["ended_at"] is None
-                or row["started_at"] is None
-                or row["started_at"] < parent["ended_at"]
-            ):
-                return False
-            if parent_id == ancestor_id:
-                return True
-            current = parent_id
-        return False
+        # Walk parent links up from the descendant, following only compression
+        # continuation edges, and check whether ancestor_id is reached.
+        edge = _COMPRESSION_CHILD_SQL.format(a="child")
+        row = conn.execute(
+            f"""
+            WITH RECURSIVE ancestors(id) AS (
+                SELECT ?
+                UNION
+                SELECT parent.id
+                FROM ancestors a
+                JOIN sessions child ON child.id = a.id
+                JOIN sessions parent ON parent.id = child.parent_session_id
+                WHERE {edge}
+            )
+            SELECT 1 FROM ancestors WHERE id = ? AND id != ? LIMIT 1
+            """,
+            (descendant_id, ancestor_id, descendant_id),
+        ).fetchone()
+        return row is not None
 
     def set_session_title(self, session_id: str, title: str) -> bool:
         """Set or update a session's title.