diff --git a/hermes_state.py b/hermes_state.py index 36e5c91fe8a1..8847593d47c1 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1836,6 +1836,43 @@ class SessionDB: return cleaned + def _is_compression_ancestor( + self, conn, *, ancestor_id: str, descendant_id: str + ) -> bool: + """Return True if *ancestor_id* is a compression predecessor of + *descendant_id* (walking parent links up the continuation chain). + + 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 + from delegate subagents / branch children that also carry a + ``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 + # 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. @@ -1854,9 +1891,29 @@ class SessionDB: ) conflict = cursor.fetchone() if conflict: - raise ValueError( - f"Title '{title}' is already in use by session {conflict['id']}" - ) + conflict_id = conflict["id"] + # A compression continuation is the live, projected-forward + # head of its conversation; its compressed predecessors are + # ended and hidden from the session list (list_sessions_rich + # projects roots → tip). When the title that "conflicts" is + # held by such a hidden ancestor, the user has no way to free + # it — renaming the visible tip back to the base name would + # dead-end with "already in use by ". + # Treat this as a transfer: move the title off the ancestor + # onto the continuation. Uniqueness is preserved (still only + # one session carries the exact title) and the parent-link + # lineage is untouched. + if self._is_compression_ancestor( + conn, ancestor_id=conflict_id, descendant_id=session_id + ): + conn.execute( + "UPDATE sessions SET title = NULL WHERE id = ?", + (conflict_id,), + ) + else: + raise ValueError( + f"Title '{title}' is already in use by session {conflict_id}" + ) cursor = conn.execute( "UPDATE sessions SET title = ? WHERE id = ?", (title, session_id), diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index e4650ed5dc79..1d727132a8c3 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -2065,6 +2065,89 @@ class TestSessionTitle: assert session["ended_at"] is not None +class TestSessionTitleLineage: + """Renaming a compression continuation back to its base title must succeed + by transferring the title off the ended, hidden predecessor. + + After a context compaction the original session is ended and projected + behind its live tip in the session list (list_sessions_rich), so the user + cannot see or free it. Without lineage-aware handling, renaming the visible + tip back to the base name dead-ends with "already in use by ". + """ + + def _make_compression_chain(self, db, t0, *, root="root", tip="tip"): + db.create_session(root, "cli") + db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t0, root)) + db._conn.execute( + "UPDATE sessions SET ended_at=?, end_reason='compression' WHERE id=?", + (t0 + 100, root), + ) + db.create_session(tip, "cli", parent_session_id=root) + db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t0 + 200, tip)) + db._conn.commit() + + def test_rename_continuation_back_to_base_transfers_title(self, db): + import time as _time + self._make_compression_chain(db, _time.time() - 3600) + db.set_session_title("root", "fingerprint-scanner") + db.set_session_title("tip", "fingerprint-scanner #2") + + # User renames the visible tip back to the base name — must succeed. + assert db.set_session_title("tip", "fingerprint-scanner") is True + assert db.get_session("tip")["title"] == "fingerprint-scanner" + # Title transferred off the hidden ancestor — no duplicate titles. + assert db.get_session("root")["title"] is None + + def test_transfer_walks_multi_level_chain(self, db): + import time as _time + t0 = _time.time() - 7200 + # root (compression) -> mid (compression) -> tip + self._make_compression_chain(db, t0, root="root", tip="mid") + db._conn.execute( + "UPDATE sessions SET ended_at=?, end_reason='compression' WHERE id=?", + (t0 + 300, "mid"), + ) + db.create_session("tip", "cli", parent_session_id="mid") + db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t0 + 400, "tip")) + db._conn.commit() + + db.set_session_title("root", "deep-dive") + assert db.set_session_title("tip", "deep-dive") is True + assert db.get_session("tip")["title"] == "deep-dive" + assert db.get_session("root")["title"] is None + + def test_unrelated_session_still_conflicts(self, db): + db.create_session("a", "cli") + db.create_session("b", "cli") + db.set_session_title("a", "shared") + with pytest.raises(ValueError, match="already in use"): + db.set_session_title("b", "shared") + # The unrelated holder keeps its title. + assert db.get_session("a")["title"] == "shared" + + def test_non_compression_child_still_conflicts(self, db): + """A child whose parent did NOT end via compression (delegate/branch + spawned while the parent was live) is not a continuation, so renaming it + to the parent's title must still raise.""" + import time as _time + t0 = _time.time() - 3600 + db.create_session("parent", "cli") + db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t0, "parent")) + db.create_session("child", "cli", parent_session_id="parent") + # Child started BEFORE parent ended, and parent ended for a non- + # compression reason — not a continuation edge. + db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t0 + 10, "child")) + db._conn.execute( + "UPDATE sessions SET ended_at=?, end_reason='user_exit' WHERE id=?", + (t0 + 100, "parent"), + ) + db._conn.commit() + db.set_session_title("parent", "shared") + with pytest.raises(ValueError, match="already in use"): + db.set_session_title("child", "shared") + + class TestSanitizeTitle: """Tests for SessionDB.sanitize_title() validation and cleaning."""