fix(tui): drop cross-session events during null-sid switch window (#51058)

agent/conversation_compression.py

@@ -543,7 +543,33 @@ def _release_lock() -> None:
 
     todo_snapshot = agent._todo_store.format_for_injection()
     if todo_snapshot:
-        compressed.append({"role": "user", "content": todo_snapshot})
+        # Inject the preserved task list so the model keeps its plan across the
+        # compaction boundary. When the compressed transcript already ends with
+        # a user message, fold the snapshot into that message (blank-line
+        # separated) instead of appending a second standalone user message —
+        # consecutive user/user turns are a content-ordering violation some
+        # providers reject, and a merged turn keeps the latest user content
+        # carrying both the original text and the task list. Only plain string
+        # content is mergeable; an empty transcript, a non-user tail, or
+        # structured (list) content falls back to the append path so image/tool
+        # parts aren't corrupted.
+        _merged_into_tail = False
+        if compressed:
+            _tail = compressed[-1]
+            if (
+                isinstance(_tail, dict)
+                and _tail.get("role") == "user"
+                and isinstance(_tail.get("content"), str)
+            ):
+                _tail_content = _tail["content"]
+                _tail["content"] = (
+                    f"{_tail_content}\n\n{todo_snapshot}"
+                    if _tail_content
+                    else todo_snapshot
+                )
+                _merged_into_tail = True
+        if not _merged_into_tail:
+            compressed.append({"role": "user", "content": todo_snapshot})
 
     agent._invalidate_system_prompt()
     new_system_prompt = agent._build_system_prompt(system_message)

cron/scheduler.py

@@ -1983,9 +1983,13 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
     # stdout to telegram" watchdog pattern. The agent path is skipped
     # entirely: no AIAgent, no prompt, no tool loop, no token spend.
     #
-    # We check this BEFORE importing run_agent / constructing SessionDB so
-    # a pure-script tick never pays for the agent machinery it isn't going
-    # to use. Keep this block self-contained.
+    # We check this BEFORE importing run_agent so a pure-script tick never
+    # pays for the agent machinery it isn't going to use. The run is still
+    # recorded as a ``cron_{job_id}_{ts}`` session (a cheap sqlite write,
+    # same shape as agent runs) so it appears in the run-history endpoint
+    # (``/api/cron/jobs/{id}/runs``) and the desktop GUI — without it,
+    # no_agent runs are invisible after a manual trigger (#44080). Keep
+    # this block self-contained.
     #
     # Semantics:
     #   - script stdout (trimmed) → delivered verbatim as the final message
@@ -2001,6 +2005,63 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             logger.error("Job '%s': %s", job_id, err)
             return False, "", "", err
 
+        # Record this run as a session BEFORE executing the script: the run
+        # session's open/ended state is what the desktop run-history endpoint
+        # uses for its "running" indicator (``is_active``), and the session
+        # row is the run record itself. Best-effort — a missing/broken state
+        # store degrades to the old no-record behaviour, never blocks the run.
+        _session_db = None
+        _run_session_id = None
+        try:
+            from hermes_state import SessionDB
+            _session_db = SessionDB()
+            _run_session_id = f"cron_{job_id}_{_hermes_now().strftime('%Y%m%d_%H%M%S')}"
+            _session_db.create_session(_run_session_id, source="cron")
+            # First user message becomes the run's preview in run history.
+            _session_db.append_message(
+                _run_session_id, "user", f"no_agent script: {script_path}"
+            )
+        except (Exception, KeyboardInterrupt) as e:
+            logger.debug(
+                "Job '%s': SQLite session store not available for no_agent run: %s",
+                job_id, e,
+            )
+            _session_db = None
+
+        def _record_run(run_doc: str, success: bool = True) -> None:
+            """Persist the outcome and close out this run's session record."""
+            if not _session_db:
+                return
+            try:
+                _session_db.append_message(_run_session_id, "assistant", run_doc)
+            except (Exception, KeyboardInterrupt) as e:
+                logger.debug(
+                    "Job '%s': failed to record no_agent run output: %s", job_id, e
+                )
+            # Same titling scheme as the agent path so sidebars/history show a
+            # meaningful label; the run-time suffix keeps it unique against
+            # the sessions.title index across runs.
+            try:
+                _title_base = " ".join(job_name.split())[:60].strip() or f"cron {job_id}"
+                _session_db.set_session_title(
+                    _run_session_id,
+                    f"{_title_base} · {_hermes_now().strftime('%b %d %H:%M')}",
+                )
+            except (Exception, KeyboardInterrupt) as e:
+                logger.debug("Job '%s': failed to set cron session title: %s", job_id, e)
+            try:
+                _session_db.end_session(
+                    _run_session_id, "cron_complete" if success else "cron_failed"
+                )
+            except (Exception, KeyboardInterrupt) as e:
+                logger.debug("Job '%s': failed to end session: %s", job_id, e)
+            try:
+                _session_db.close()
+            except (Exception, KeyboardInterrupt) as e:
+                logger.debug(
+                    "Job '%s': failed to close SQLite session store: %s", job_id, e
+                )
+
         # Apply workdir if configured — lets scripts use predictable relative
         # paths. For no_agent jobs this is just the subprocess cwd (not an
         # agent TERMINAL_CWD bridge).
@@ -2041,6 +2102,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
                 f"**Status:** script failed\n\n"
                 f"{output}\n"
             )
+            _record_run(doc, success=False)
             return False, doc, alert, output
 
         # Honour the wakeAgent gate as a silent signal — `wakeAgent: false`
@@ -2056,6 +2118,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
                 f"**Mode:** no_agent (script)\n"
                 f"**Status:** silent (wakeAgent=false)\n"
             )
+            _record_run(silent_doc)
             return True, silent_doc, SILENT_MARKER, None
 
         if not output.strip():
@@ -2067,6 +2130,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
                 f"**Mode:** no_agent (script)\n"
                 f"**Status:** silent (empty output)\n"
             )
+            _record_run(silent_doc)
             return True, silent_doc, SILENT_MARKER, None
 
         doc = (
@@ -2077,6 +2141,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             f"---\n\n"
             f"{output}\n"
         )
+        _record_run(doc)
         return True, doc, output, None
 
     # ---------------------------------------------------------------

hermes_state.py

@@ -116,6 +116,75 @@ def _delete_delegate_children(conn, parent_ids: List[str]) -> List[str]:
         conn.execute(f"DELETE FROM sessions WHERE id IN ({ph})", ids)
     return ids
 
+
+# A child is a *compression continuation* of its parent (not a branch,
+# subagent, or tool-spawned run) when the parent ended via 'compression' and
+# the child carries no _branched_from / _delegate_from marker and is not a
+# tool-source row. This is the same edge definition get_compression_tip and
+# list_sessions_rich use to project roots forward to their tips, so a delete
+# that walks this graph can never leave a row that resurfaces on refresh.
+_COMPRESSION_CONTINUATION_PREDICATE = (
+    "parent.end_reason = 'compression'"
+    " AND json_extract(COALESCE(child.model_config, '{}'), '$._branched_from') IS NULL"
+    " AND json_extract(COALESCE(child.model_config, '{}'), '$._delegate_from') IS NULL"
+    " AND COALESCE(child.source, '') != 'tool'"
+)
+
+
+def _compression_lineage_ids(conn, session_ids: List[str]) -> List[str]:
+    """All ids in the compression-continuation lineages touching *session_ids*.
+
+    Walks both directions along the compression-continuation edge (parent ended
+    ``end_reason='compression'``, child is not a branch / delegate / tool row),
+    matching the projection that ``get_compression_tip`` and
+    ``list_sessions_rich`` apply. Deleting any member of a lineage must remove
+    the whole lineage, or the orphaned root (or a sibling continuation)
+    resurfaces in session pickers on the next refresh.
+
+    The seeds themselves are always included. Returns ids in deterministic
+    order; callers orphan children before deleting the returned rows, so the
+    order is not used for FK safety.
+    """
+    seeds = {sid for sid in session_ids if sid}
+    if not seeds:
+        return []
+
+    found: set[str] = set(seeds)
+    frontier: set[str] = set(seeds)
+
+    # Walk up (ancestors) and down (descendants) in lockstep. A seed may sit in
+    # the middle of a chain, so both directions are needed to cover the whole
+    # lineage. Bounded by the reachable set, which is naturally finite.
+    while frontier:
+        ph = ",".join("?" * len(frontier))
+        params = list(frontier)
+        rows = conn.execute(
+            f"""
+            SELECT child.id AS id
+            FROM sessions parent
+            JOIN sessions child ON child.parent_session_id = parent.id
+            WHERE parent.id IN ({ph})
+              AND {_COMPRESSION_CONTINUATION_PREDICATE}
+            UNION ALL
+            SELECT parent.id AS id
+            FROM sessions child
+            JOIN sessions parent ON parent.id = child.parent_session_id
+            WHERE child.id IN ({ph})
+              AND {_COMPRESSION_CONTINUATION_PREDICATE}
+            """,
+            params + params,
+        ).fetchall()
+        next_frontier: set[str] = set()
+        for row in rows:
+            rid = row["id"]
+            if rid and rid not in found:
+                found.add(rid)
+                next_frontier.add(rid)
+        frontier = next_frontier
+
+    return sorted(found)
+
+
 T = TypeVar("T")
 
 DEFAULT_DB_PATH = get_hermes_home() / "state.db"
@@ -4255,12 +4324,25 @@ def delete_session(
 
         Delegate subagent children (``model_config._delegate_from``) are
         cascade-deleted with the parent so they never resurface in session
-        pickers as orphaned rows. Branch / compression children are orphaned
-        (``parent_session_id → NULL``) so they remain accessible independently.
+        pickers as orphaned rows. Branch children (``_branched_from``) are
+        orphaned (``parent_session_id → NULL``) so they remain accessible
+        independently.
+
+        The entire **compression lineage** of *session_id* is deleted with it:
+        every ancestor and descendant linked by a compression-continuation
+        edge (parent ended ``end_reason='compression'``, child is not a
+        branch / delegate / tool row — the same edge ``get_compression_tip``
+        and ``list_sessions_rich`` use to project roots forward to their tips).
+        Otherwise deleting the visible tip orphans the old compression root,
+        which then resurfaces in session pickers on the next refresh as a
+        "deleted session resurrected". Deleting any member removes the whole
+        logical conversation.
+
         When *sessions_dir* is provided, also removes on-disk transcript
         files (``.json`` / ``.jsonl`` / ``request_dump_*``) for every deleted
         session. Returns True if the session was found and deleted.
         """
+        removed_lineage_ids: List[str] = []
         removed_delegate_ids: List[str] = []
 
         def _do(conn):
@@ -4269,22 +4351,34 @@ def _do(conn):
             )
             if cursor.fetchone()[0] == 0:
                 return False
-            removed_delegate_ids.extend(_delete_delegate_children(conn, [session_id]))
-            # Orphan remaining child sessions (branches, etc.) so FK is satisfied.
+            # Expand to the whole compression lineage so neither the old root
+            # nor a sibling continuation row can resurface after the tip goes.
+            kill_ids = _compression_lineage_ids(conn, [session_id])
+            removed_lineage_ids.extend(kill_ids)
+            kill_ph = ",".join("?" * len(kill_ids))
+            removed_delegate_ids.extend(_delete_delegate_children(conn, kill_ids))
+            # Orphan remaining children (branches, etc.) of every doomed row so
+            # the FK stays satisfied. Compression/delegate children are in the
+            # kill set already; branches must survive and stay accessible.
             conn.execute(
-                "UPDATE sessions SET parent_session_id = NULL "
-                "WHERE parent_session_id = ?",
-                (session_id,),
+                f"UPDATE sessions SET parent_session_id = NULL "
+                f"WHERE parent_session_id IN ({kill_ph})",
+                kill_ids,
+            )
+            conn.execute(
+                f"DELETE FROM messages WHERE session_id IN ({kill_ph})", kill_ids
+            )
+            conn.execute(
+                f"DELETE FROM sessions WHERE id IN ({kill_ph})", kill_ids
             )
-            conn.execute("DELETE FROM messages WHERE session_id = ?", (session_id,))
-            conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
             return True
 
         deleted = self._execute_write(_do)
         if deleted:
             for delegate_id in removed_delegate_ids:
                 self._remove_session_files(sessions_dir, delegate_id)
-            self._remove_session_files(sessions_dir, session_id)
+            for lineage_id in removed_lineage_ids:
+                self._remove_session_files(sessions_dir, lineage_id)
         return bool(deleted)
 
     def delete_session_if_empty(
@@ -4343,6 +4437,9 @@ def delete_sessions(
         * Unknown IDs are silently skipped (no 404) — selection state
           in the UI can race against another tab's delete, and we'd
           rather succeed-on-the-rest than fail-the-whole-batch.
+        * The full compression lineage of every selected row is deleted so
+          visible continuation tips cannot leave old roots behind that later
+          resurface in session lists.
         * Delegate subagent children (``model_config._delegate_from``) are
           cascade-deleted with their parent; branch children are orphaned
           (``parent_session_id → NULL``) so they stay accessible.
@@ -4382,27 +4479,28 @@ def _do(conn):
             if not existing:
                 return 0
 
-            existing_placeholders = ",".join("?" * len(existing))
-            removed_delegate_ids.extend(_delete_delegate_children(conn, existing))
+            kill_ids = _compression_lineage_ids(conn, existing)
+            kill_placeholders = ",".join("?" * len(kill_ids))
+            removed_delegate_ids.extend(_delete_delegate_children(conn, kill_ids))
             # Orphan remaining children whose parent is in the kill list so the
             # FK constraint stays satisfied. Pin children whose parent
             # is itself in the kill list rather than NULL-ing parents
             # of survivors — the IN list on ``parent_session_id`` does
             # exactly this.
             conn.execute(
                 f"UPDATE sessions SET parent_session_id = NULL "
-                f"WHERE parent_session_id IN ({existing_placeholders})",
-                existing,
+                f"WHERE parent_session_id IN ({kill_placeholders})",
+                kill_ids,
             )
             conn.execute(
-                f"DELETE FROM messages WHERE session_id IN ({existing_placeholders})",
-                existing,
+                f"DELETE FROM messages WHERE session_id IN ({kill_placeholders})",
+                kill_ids,
             )
             conn.execute(
-                f"DELETE FROM sessions WHERE id IN ({existing_placeholders})",
-                existing,
+                f"DELETE FROM sessions WHERE id IN ({kill_placeholders})",
+                kill_ids,
             )
-            removed_ids.extend(existing)
+            removed_ids.extend(kill_ids)
             return len(existing)
 
         count = self._execute_write(_do)