Session list cleanup for subagents and compression

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #5309 fix: link subagent sessions to parent and hide from session list

  merged to main

  files: hermes_state.py, run_agent.py, tools/delegate_tool.py

  Show the diff

  • hermes_state.py

    @@ -787,6 +787,7 @@ def list_sessions_rich(
             exclude_sources: List[str] = None,
             limit: int = 20,
             offset: int = 0,
    +        include_children: bool = False,
         ) -> List[Dict[str, Any]]:
             """List sessions with preview (first user message) and last active timestamp.
     
    @@ -795,10 +796,16 @@ def list_sessions_rich(
             last_active (timestamp of last message).
     
             Uses a single query with correlated subqueries instead of N+2 queries.
    +
    +        By default, child sessions (subagent runs, compression continuations)
    +        are excluded.  Pass ``include_children=True`` to include them.
             """
             where_clauses = []
             params = []
     
    +        if not include_children:
    +            where_clauses.append("s.parent_session_id IS NULL")
    +
             if source:
                 where_clauses.append("s.source = ?")
                 params.append(source)
    @@ -1229,22 +1236,38 @@ def _do(conn):
             self._execute_write(_do)
     
         def delete_session(self, session_id: str) -> bool:
    -        """Delete a session and all its messages. Returns True if found."""
    +        """Delete a session, its child sessions, and all their messages.
    +
    +        Child sessions (subagent runs, compression continuations) are deleted
    +        first to satisfy the ``parent_session_id`` foreign key constraint.
    +        Returns True if the session was found and deleted.
    +        """
             def _do(conn):
                 cursor = conn.execute(
                     "SELECT COUNT(*) FROM sessions WHERE id = ?", (session_id,)
                 )
                 if cursor.fetchone()[0] == 0:
                     return False
    +            # Delete child sessions first (FK constraint)
    +            child_ids = [r[0] for r in conn.execute(
    +                "SELECT id FROM sessions WHERE parent_session_id = ?",
    +                (session_id,),
    +            ).fetchall()]
    +            for cid in child_ids:
    +                conn.execute("DELETE FROM messages WHERE session_id = ?", (cid,))
    +                conn.execute("DELETE FROM sessions WHERE id = ?", (cid,))
    +            # Delete the session itself
                 conn.execute("DELETE FROM messages WHERE session_id = ?", (session_id,))
                 conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
                 return True
             return self._execute_write(_do)
     
         def prune_sessions(self, older_than_days: int = 90, source: str = None) -> int:
    -        """
    -        Delete sessions older than N days. Returns count of deleted sessions.
    -        Only prunes ended sessions (not active ones).
    +        """Delete sessions older than N days. Returns count of deleted sessions.
    +
    +        Only prunes ended sessions (not active ones).  Child sessions whose
    +        parents are being pruned are deleted first to satisfy the
    +        ``parent_session_id`` foreign key constraint.
             """
             cutoff = time.time() - (older_than_days * 86400)
     
    @@ -1260,7 +1283,19 @@ def _do(conn):
                         "SELECT id FROM sessions WHERE started_at < ? AND ended_at IS NOT NULL",
                         (cutoff,),
                     )
    -            session_ids = [row["id"] for row in cursor.fetchall()]
    +            session_ids = set(row["id"] for row in cursor.fetchall())
    +
    +            # Delete children first whose parents are in the prune set
    +            # (avoids FK constraint errors)
    +            for sid in list(session_ids):
    +                child_ids = [r[0] for r in conn.execute(
    +                    "SELECT id FROM sessions WHERE parent_session_id = ?",
    +                    (sid,),
    +                ).fetchall()]
    +                for cid in child_ids:
    +                    conn.execute("DELETE FROM messages WHERE session_id = ?", (cid,))
    +                    conn.execute("DELETE FROM sessions WHERE id = ?", (cid,))
    +                    session_ids.discard(cid)  # don't double-delete
     
                 for sid in session_ids:
                     conn.execute("DELETE FROM messages WHERE session_id = ?", (sid,))

  • run_agent.py

    @@ -530,6 +530,7 @@ def __init__(
             skip_context_files: bool = False,
             skip_memory: bool = False,
             session_db=None,
    +        parent_session_id: str = None,
             iteration_budget: "IterationBudget" = None,
             fallback_model: Dict[str, Any] = None,
             credential_pool=None,
    @@ -1025,6 +1026,7 @@ def __init__(
             
             # SQLite session store (optional -- provided by CLI or gateway)
             self._session_db = session_db
    +        self._parent_session_id = parent_session_id
             self._last_flushed_db_idx = 0  # tracks DB-write cursor to prevent duplicate writes
             if self._session_db:
                 try:
    @@ -1038,6 +1040,7 @@ def __init__(
                             "max_tokens": max_tokens,
                         },
                         user_id=None,
    +                    parent_session_id=self._parent_session_id,
                     )
                 except Exception as e:
                     # Transient SQLite lock contention (e.g. CLI and gateway writing

  • tools/delegate_tool.py

    @@ -251,6 +251,7 @@ def _child_thinking(text: str) -> None:
             clarify_callback=None,
             thinking_callback=child_thinking_cb,
             session_db=getattr(parent_agent, '_session_db', None),
    +        parent_session_id=getattr(parent_agent, 'session_id', None),
             providers_allowed=parent_agent.providers_allowed,
             providers_ignored=parent_agent.providers_ignored,
             providers_order=parent_agent.providers_order,

• #12960 fix(sessions): surface compression tips in session lists and resume lookups

  merged to main

  files: hermes_cli/main.py, hermes_state.py, tests/test_hermes_state.py

  Show the diff

  • hermes_cli/main.py

    @@ -693,6 +693,10 @@ def _resolve_session_by_name_or_id(name_or_id: str) -> Optional[str]:
         - If it looks like a session ID (contains underscore + hex), try direct lookup first.
         - Otherwise, treat it as a title and use resolve_session_by_title (auto-latest).
         - Falls back to the other method if the first doesn't match.
    +    - If the resolved session is a compression root, follow the chain forward
    +      to the latest continuation. Users who remember the old root ID (e.g.
    +      from an exit summary printed before the bug fix, or from notes) get
    +      resumed at the live tip instead of a stale parent with no messages.
         """
         try:
             from hermes_state import SessionDB
    @@ -701,14 +705,23 @@ def _resolve_session_by_name_or_id(name_or_id: str) -> Optional[str]:
     
             # Try as exact session ID first
             session = db.get_session(name_or_id)
    +        resolved_id: Optional[str] = None
             if session:
    -            db.close()
    -            return session["id"]
    +            resolved_id = session["id"]
    +        else:
    +            # Try as title (with auto-latest for lineage)
    +            resolved_id = db.resolve_session_by_title(name_or_id)
    +
    +        if resolved_id:
    +            # Project forward through compression chain so resumes land on
    +            # the live tip instead of a dead compressed parent.
    +            try:
    +                resolved_id = db.get_compression_tip(resolved_id) or resolved_id
    +            except Exception:
    +                pass
     
    -        # Try as title (with auto-latest for lineage)
    -        session_id = db.resolve_session_by_title(name_or_id)
             db.close()
    -        return session_id
    +        return resolved_id
         except Exception:
             pass
         return None

  • hermes_state.py

    @@ -723,6 +723,42 @@ def get_next_title_in_lineage(self, base_title: str) -> str:
     
             return f"{base} #{max_num + 1}"
     
    +    def get_compression_tip(self, session_id: str) -> Optional[str]:
    +        """Walk the compression-continuation chain forward and return the tip.
    +
    +        A compression continuation is a child session where:
    +        1. The parent's ``end_reason = 'compression'``
    +        2. The child was created AFTER the parent was ended (started_at >= ended_at)
    +
    +        The second condition distinguishes compression continuations from
    +        delegate subagents or branch children, which can also have a
    +        ``parent_session_id`` but were created while the parent was still live.
    +
    +        Returns the session_id of the latest continuation in the chain, or the
    +        input ``session_id`` if it isn't part of a compression chain (or if the
    +        input itself doesn't exist).
    +        """
    +        current = session_id
    +        # Bound the walk defensively — compression chains this deep are
    +        # pathological and shouldn't happen in practice. 100 = plenty.
    +        for _ in range(100):
    +            with self._lock:
    +                cursor = self._conn.execute(
    +                    "SELECT id FROM sessions "
    +                    "WHERE parent_session_id = ? "
    +                    "  AND started_at >= ("
    +                    "      SELECT ended_at FROM sessions "
    +                    "      WHERE id = ? AND end_reason = 'compression'"
    +                    "  ) "
    +                    "ORDER BY started_at DESC LIMIT 1",
    +                    (current, current),
    +                )
    +                row = cursor.fetchone()
    +            if row is None:
    +                return current
    +            current = row["id"]
    +        return current
    +
         def list_sessions_rich(
             self,
             source: str = None,
    @@ -730,6 +766,7 @@ def list_sessions_rich(
             limit: int = 20,
             offset: int = 0,
             include_children: bool = False,
    +        project_compression_tips: bool = True,
         ) -> List[Dict[str, Any]]:
             """List sessions with preview (first user message) and last active timestamp.
     
    @@ -741,6 +778,14 @@ def list_sessions_rich(
     
             By default, child sessions (subagent runs, compression continuations)
             are excluded.  Pass ``include_children=True`` to include them.
    +
    +        With ``project_compression_tips=True`` (default), sessions that are
    +        roots of compression chains are projected forward to their latest
    +        continuation — one logical conversation = one list entry, showing the
    +        live continuation's id/message_count/title/last_active. This prevents
    +        compressed continuations from being invisible to users while keeping
    +        delegate subagents and branches hidden. Pass ``False`` to return the
    +        raw root rows (useful for admin/debug UIs).
             """
             where_clauses = []
             params = []
    @@ -791,8 +836,77 @@ def list_sessions_rich(
                     s["preview"] = ""
                 sessions.append(s)
     
    +        # Project compression roots forward to their tips. Each row whose
    +        # end_reason is 'compression' has a continuation child; replace the
    +        # surfaced fields (id, message_count, title, last_active, ended_at,
    +        # end_reason, preview) with the tip's values so the list entry acts
    +        # as the live conversation. Keep the root's started_at to preserve
    +        # chronological ordering by original conversation start.
    +        if project_compression_tips and not include_children:
    +            projected = []
    +            for s in sessions:
    +                if s.get("end_reason") != "compression":
    +                    projected.append(s)
    +                    continue
    +                tip_id = self.get_compression_tip(s["id"])
    +                if tip_id == s["id"]:
    +                    projected.append(s)
    +                    continue
    +                tip_row = self._get_session_rich_row(tip_id)
    +                if not tip_row:
    +                    projected.append(s)
    +                    continue
    +                # Preserve the root's started_at for stable sort order, but
    +                # surface the tip's identity and activity data.
    +                merged = dict(s)
    +                for key in (
    +                    "id", "ended_at", "end_reason", "message_count",
    +                    "tool_call_count", "title", "last_active", "preview",
    +                    "model", "system_prompt",
    +                ):
    +                    if key in tip_row:
    +                        merged[key] = tip_row[key]
    +                merged["_lineage_root_id"] = s["id"]
    +                projected.append(merged)
    +            sessions = projected
    +
             return sessions
     
    +    def _get_session_rich_row(self, session_id: str) -> Optional[Dict[str, Any]]:
    +        """Fetch a single session with the same enriched columns as
    +        ``list_sessions_rich`` (preview + last_active). Returns None if the
    +        session doesn't exist.
    +        """
    +        query = """
    +            SELECT s.*,
    +                COALESCE(
    +                    (SELECT SUBSTR(REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' '), 1, 63)
    +                     FROM messages m
    +                     WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
    +                     ORDER BY m.timestamp, m.id LIMIT 1),
    +                    ''
    +                ) AS _preview_raw,
    +                COALESCE(
    +                    (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
    +                    s.started_at
    +                ) AS last_active
    +            FROM sessions s
    +            WHERE s.id = ?
    +        """
    +        with self._lock:
    +            cursor = self._conn.execute(query, (session_id,))
    +            row = cursor.fetchone()
    +        if not row:
    +            return None
    +        s = dict(row)
    +        raw = s.pop("_preview_raw", "").strip()
    +        if raw:
    +            text = raw[:60]
    +            s["preview"] = text + ("..." if len(raw) > 60 else "")
    +        else:
    +            s["preview"] = ""
    +        return s
    +
         # =========================================================================
         # Message storage
         # =========================================================================

  • tests/test_hermes_state.py

    @@ -1381,6 +1381,178 @@ def test_preview_newlines_collapsed(self, db):
             assert "Line one Line two" in sessions[0]["preview"]
     
     
    +class TestCompressionChainProjection:
    +    """Tests for lineage-aware list_sessions_rich — compressed conversations
    +    surface as their live continuation tip, not the dead parent root.
    +    """
    +
    +    def _build_compression_chain(self, db, t0: float):
    +        """Helper: builds root -> delegate -> compression-child -> tip chain.
    +
    +        Returns (root_id, delegate_id, mid_id, tip_id).
    +        """
    +        import time as _time
    +        # Root that gets compressed
    +        db.create_session("root1", "cli")
    +        db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t0, "root1"))
    +        db.append_message("root1", "user", "help me refactor auth")
    +
    +        # Delegate subagent spawned while root1 was live (before it ended)
    +        db.create_session("delegate1", "cli", parent_session_id="root1")
    +        db._conn.execute(
    +            "UPDATE sessions SET started_at=?, ended_at=? WHERE id=?",
    +            (t0 + 600, t0 + 650, "delegate1"),
    +        )
    +        db.append_message("delegate1", "user", "delegate task")
    +
    +        # root1 compressed at t0+1800
    +        t_compress_root = t0 + 1800
    +        db._conn.execute(
    +            "UPDATE sessions SET ended_at=?, end_reason=? WHERE id=?",
    +            (t_compress_root, "compression", "root1"),
    +        )
    +
    +        # Continuation mid created 1s after parent ended
    +        db.create_session("mid1", "cli", parent_session_id="root1")
    +        db._conn.execute(
    +            "UPDATE sessions SET started_at=? WHERE id=?",
    +            (t_compress_root + 1, "mid1"),
    +        )
    +        db.append_message("mid1", "user", "continuing")
    +
    +        # mid1 also compressed
    +        t_compress_mid = t_compress_root + 1800
    +        db._conn.execute(
    +            "UPDATE sessions SET ended_at=?, end_reason=? WHERE id=?",
    +            (t_compress_mid, "compression", "mid1"),
    +        )
    +
    +        # Tip — latest continuation
    +        db.create_session("tip1", "cli", parent_session_id="mid1")
    +        db._conn.execute(
    +            "UPDATE sessions SET started_at=? WHERE id=?",
    +            (t_compress_mid + 1, "tip1"),
    +        )
    +        db.append_message("tip1", "user", "latest message")
    +
    +        db._conn.commit()
    +        return ("root1", "delegate1", "mid1", "tip1")
    +
    +    def test_get_compression_tip_walks_full_chain(self, db):
    +        import time as _time
    +        self._build_compression_chain(db, _time.time() - 3600)
    +        assert db.get_compression_tip("root1") == "tip1"
    +        assert db.get_compression_tip("mid1") == "tip1"
    +        assert db.get_compression_tip("tip1") == "tip1"
    +
    +    def test_get_compression_tip_returns_self_for_uncompressed(self, db):
    +        db.create_session("solo", "cli")
    +        assert db.get_compression_tip("solo") == "solo"
    +
    +    def test_get_compression_tip_skips_delegate_children(self, db):
    +        """Delegate subagents have parent_session_id set but were created
    +        BEFORE the parent ended. They must not be followed as compression
    +        continuations — the started_at >= ended_at guard handles this.
    +        """
    +        import time as _time
    +        self._build_compression_chain(db, _time.time() - 3600)
    +        # delegate1 is a child of root1 but NOT a compression continuation.
    +        # root1's tip must be tip1 (via mid1), not delegate1.
    +        assert db.get_compression_tip("root1") == "tip1"
    +
    +    def test_list_surfaces_tip_for_compressed_root(self, db):
    +        """The list must show the tip's id/message_count/preview in place of
    +        the root row, so users can see and resume the live conversation.
    +        """
    +        import time as _time
    +        self._build_compression_chain(db, _time.time() - 3600)
    +        # Add an uncompressed root for comparison.
    +        db.create_session("solo", "cli")
    +        db.append_message("solo", "user", "standalone")
    +        db._conn.commit()
    +
    +        sessions = db.list_sessions_rich(source="cli", limit=20)
    +        ids = [s["id"] for s in sessions]
    +        # Only top-level conversations appear: tip1 (projected from root1) + solo.
    +        # Delegate children, mid1, and the dead root1 must NOT be in the list.
    +        assert "tip1" in ids
    +        assert "solo" in ids
    +        assert "root1" not in ids
    +        assert "mid1" not in ids
    +        assert "delegate1" not in ids
    +
    +        tip_row = next(s for s in sessions if s["id"] == "tip1")
    +        # The row surfaces the tip's identity but preserves the root's start
    +        # timestamp for stable ordering and lineage tracking.
    +        assert tip_row["_lineage_root_id"] == "root1"
    +        assert tip_row["preview"].startswith("latest message")
    +        assert tip_row["ended_at"] is None  # tip is still live
    +        assert tip_row["end_reason"] is None
    +
    +    def test_list_without_projection_returns_raw_root(self, db):
    +        """project_compression_tips=False returns the raw parent-NULL root
    +        rows — useful for admin/debug UIs.
    +        """
    +        import time as _time
    +        self._build_compression_chain(db, _time.time() - 3600)
    +        sessions = db.list_sessions_rich(
    +            source="cli", limit=20, project_compression_tips=False
    +        )
    +        ids = [s["id"] for s in sessions]
    +        assert "root1" in ids
    +        assert "tip1" not in ids
    +
    +        root_row = next(s for s in sessions if s["id"] == "root1")
    +        assert root_row["end_reason"] == "compression"
    +        assert "_lineage_root_id" not in root_row
    +
    +    def test_list_preserves_sort_by_started_at(self, db):
    +        """Chronological ordering uses the ROOT's started_at (conversation
    +        start), not the tip's. This keeps lineage entries stable in the list
    +        even as new compressions push the tip forward in time.
    +        """
    +        import time as _time
    +        t0 = _time.time() - 3600
    +        self._build_compression_chain(db, t0)
    +
    +        # Create a newer standalone session that should sort above the lineage
    +        # if we used tip.started_at, but below if we correctly use root.started_at.
    +        t_between = t0 + 120  # between root1 and its compression
    +        db.create_session("newer", "cli")
    +        db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t_between, "newer"))
    +        db.append_message("newer", "user", "newer session started after root1")
    +        db._conn.commit()
    +
    +        sessions = db.list_sessions_rich(source="cli", limit=20)
    +        ids_in_order = [s["id"] for s in sessions]
    +        # 'newer' started AFTER root1 but BEFORE tip1's actual started_at.
    +        # Correct ordering (by root started_at): newer > tip1's lineage entry.
    +        assert ids_in_order.index("newer") < ids_in_order.index("tip1")
    +
    +    def test_list_handles_broken_chain_gracefully(self, db):
    +        """A compression root with no child (e.g. DB corruption or a partial
    +        end_session call that didn't finish creating the child) must not
    +        crash the list — it should fall back to surfacing the root as-is.
    +        """
    +        import time as _time
    +        t0 = _time.time() - 100
    +        db.create_session("orphan", "cli")
    +        db._conn.execute("UPDATE sessions SET started_at=? WHERE id=?", (t0, "orphan"))
    +        db._conn.execute(
    +            "UPDATE sessions SET ended_at=?, end_reason=? WHERE id=?",
    +            (t0 + 10, "compression", "orphan"),
    +        )
    +        db._conn.commit()
    +
    +        sessions = db.list_sessions_rich(source="cli", limit=10)
    +        ids = [s["id"] for s in sessions]
    +        assert "orphan" in ids
    +        row = next(s for s in sessions if s["id"] == "orphan")
    +        # No tip means no projection — row stays raw.
    +        assert "_lineage_root_id" not in row
    +        assert row["end_reason"] == "compression"
    +
    +
     # =========================================================================
     # Session source exclusion (--source flag for third-party isolation)
     # =========================================================================