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).
This commit is contained in:
kshitijk4poor 2026-06-19 17:37:39 +05:30
parent 65d050cf0e
commit 8c70346e33

View file

@ -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.