From 1a6a9b66b9d52085972ed5c06bdd611031c56875 Mon Sep 17 00:00:00 2001
From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com>
Date: Sun, 27 Sep 2026 15:57:48 +0530
Subject: [PATCH] fix(sessions): guard delegate children that a delete would
 cascade

delete_session/delete_sessions cascade-delete delegate children, but the
write-guard check only looked at the root. A guarded delegate child could be
removed out from under its live turn, and in bulk delete an active id that
was also another selected root's delegate child was reported in
skipped_active while the cascade deleted it anyway.

Check {root, *delegate children} via a small _guarded_ids helper: single
delete refuses, bulk delete skips the root, so the cascade never touches a
guarded row. Ports the delegate-protection idea from #124496.

Co-authored-by: JoaoMarcos44 <joaomarcosdias444@gmail.com>
---
 hermes_state_sessions.py                        | 16 ++++++++++++----
 .../test_delete_session_write_guards.py         | 17 +++++++++++++++++
 2 files changed, 29 insertions(+), 4 deletions(-)

diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py
index 2b8e0ec045..18e8eb8744 100644
--- a/hermes_state_sessions.py
+++ b/hermes_state_sessions.py
@@ -1576,11 +1576,12 @@ class SessionSessionsMixin:
         def _do(conn):
             if conn.execute("SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,)).fetchone() is None:
                 return False
-            if exclude_active_write_guards and self._write_guards_reject(
-                conn, session_id, allow_closed_compression_parent=True,
+            if exclude_active_write_guards and self._guarded_ids(
+                conn, [session_id, *_collect_delegate_child_ids(conn, [session_id])],
             ):
+                # Delegate children cascade with the root, so a guard on any of them refuses too.
                 raise SessionActiveWriteGuardError(
-                    f"session '{session_id}' has an active turn lease or compression lock"
+                    f"session '{session_id}' (or a delegate child) has an active turn lease or compression lock"
                 )
             if expected_ids is not None and expected_ids != {
                 session_id, *_collect_delegate_child_ids(conn, [session_id])
@@ -1632,6 +1633,11 @@ class SessionSessionsMixin:
             self._remove_session_files(sessions_dir, session_id)
         return deleted
 
+    def _guarded_ids(self, conn, ids: List[str]) -> set:
+        """Ids in *ids* protected by a live turn lease / compression lock. Idle compression-ended
+        parents are closed, not live, so they are not guarded (matches prune)."""
+        return {sid for sid in ids if self._write_guards_reject(conn, sid, allow_closed_compression_parent=True)}
+
     def delete_sessions(
         self, session_ids: List[str], sessions_dir: Optional[Path] = None,
         exclude_active_write_guards: bool = False, skipped_ids: Optional[List[str]] = None,
@@ -1651,9 +1657,11 @@ class SessionSessionsMixin:
             if not existing:
                 return 0
             if exclude_active_write_guards:
+                # A root is skipped when it or any delegate child it would cascade is guarded, so the
+                # cascade below never deletes a guarded row reported back as kept.
                 active_ids = {
                     sid for sid in existing
-                    if self._write_guards_reject(conn, sid, allow_closed_compression_parent=True)
+                    if self._guarded_ids(conn, [sid, *_collect_delegate_child_ids(conn, [sid])])
                 }
                 existing = [sid for sid in existing if sid not in active_ids]
                 if skipped_ids is not None:
diff --git a/tests/hermes_state/test_delete_session_write_guards.py b/tests/hermes_state/test_delete_session_write_guards.py
index 208ded057c..461db91e55 100644
--- a/tests/hermes_state/test_delete_session_write_guards.py
+++ b/tests/hermes_state/test_delete_session_write_guards.py
@@ -71,4 +71,21 @@ def test_delete_sessions_bulk_skips_active_write_guards(tmp_path):
     assert db.get_session("bulk-idle") is None
 
     db.release_session_turn_lease("bulk-active", turn_holder)
+
+    # Delegate children cascade with their root, so a guarded child protects the root too:
+    # single delete refuses, bulk delete skips the root and never cascades the guarded child away.
+    db.create_session("deleg-root", source="test")
+    db.create_session(
+        "deleg-child", source="test", parent_session_id="deleg-root",
+        model_config={"_delegate_from": "deleg-root"},
+    )
+    child_holder = f"pid={os.getpid()}:turn=child"
+    assert db.try_acquire_session_turn_lease("deleg-child", child_holder, ttl_seconds=300.0) is True
+    with pytest.raises(SessionActiveWriteGuardError):
+        db.delete_session("deleg-root", exclude_active_write_guards=True)
+    skipped = []
+    assert db.delete_sessions(["deleg-root"], exclude_active_write_guards=True, skipped_ids=skipped) == 0
+    assert skipped == ["deleg-root"]
+    assert db.get_session("deleg-root") is not None and db.get_session("deleg-child") is not None
+    db.release_session_turn_lease("deleg-child", child_holder)
     db.close()