"""Session lifecycle mixin for SessionDB: row upsert/inheritance, lifecycle flags (end/reopen/archive/pin/hide/read), model_config patching, listing and counting, delete cascades, and the auto-archive sweep.""" import json import logging import re import sqlite3 import time from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple from agent.session_activity import ( ActivityProvenance, bound_activity_description, normalize_activity_provenance, ) from hermes_startup_watchdog import report_startup_progress from hermes_state_common import ( _LISTABLE_CHILD_SQL, _PREVIEW_ELIGIBLE_SQL, _PREVIEW_RAW_SELECT, _RECOVERABLE_END_REASONS, _RECOVERABLE_END_REASONS_SQL, _RESET_END_REASONS, _legacy_reset_child_sql, _shape_preview, _sql_json_extract, _sql_session_last_active, _sql_session_last_active_by_id, escape_like as _escape_like, _SQL_IN_CHUNK, _id_chunks, _placeholders as _session_ids_placeholders, ) # caplog tests pin the "hermes_state" logger name. logger = logging.getLogger("hermes_state") def workspace_key(row: Dict[str, Any]) -> Optional[str]: """Workspace grouping key: git repo root, else cwd, else None (branch excluded: a checkout must not fragment history).""" return (row.get("git_repo_root") or "").strip() or (row.get("cwd") or "").strip() or None def _delegate_from_json(col: str = "model_config") -> str: return _sql_json_extract(col, "$._delegate_from") # _merge_model_config_json's "no such row" result — distinct from the legal None # ("merged config is empty → store NULL"). _MODEL_CONFIG_ROW_MISSING = object() def _parse_model_config(raw: Any) -> Dict[str, Any]: """Tolerant ``model_config`` decode: JSON text or dict -> dict copy; anything else -> {}.""" if isinstance(raw, str) and raw.strip(): try: raw = json.loads(raw) except (json.JSONDecodeError, TypeError): return {} return dict(raw) if isinstance(raw, dict) else {} def _cwd_prefix_clause(cwd_prefix: str) -> Tuple[str, List[str]]: prefix = cwd_prefix.rstrip("/\\") or cwd_prefix # ``_``/``%`` are LIKE wildcards but ordinary path characters: unescaped, a # prefix also matches sibling directories. The ``=`` arm keeps the raw prefix. esc = _escape_like(prefix) return ( "(s.cwd = ? OR s.cwd LIKE ? ESCAPE '\\' OR s.cwd LIKE ? ESCAPE '\\')", [prefix, f"{esc}/%", f"{esc}\\\\%"], ) def _workspace_key_clause(key: str) -> Tuple[str, List[str]]: """WHERE for ``workspace_key(row) == key``: git_repo_root equals ``key``, or (rows predating per-session git metadata) cwd is at/under ``key``.""" prefix = key.rstrip("/\\") or key cwd_clause, cwd_params = _cwd_prefix_clause(prefix) return ( f"(s.git_repo_root = ? OR (COALESCE(s.git_repo_root, '') = '' AND {cwd_clause}))", [prefix, *cwd_params], ) # First user message of a session, shaped by _shape_preview() in Python. # The indentation is part of the list_sessions_rich SQL text. _PREVIEW_COL_SQL = f"""COALESCE( (SELECT {_PREVIEW_RAW_SELECT} FROM messages m WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL AND {_PREVIEW_ELIGIBLE_SQL} ORDER BY m.timestamp, m.id LIMIT 1), '' ) AS _preview_raw""" def _where_sql(clauses: List[str], lead: str = "") -> str: """``WHERE a AND b`` (with *lead* prefix) or "" when there are no clauses.""" return f"{lead}WHERE {' AND '.join(clauses)}" if clauses else "" def _session_filter_where( *, exclude_children: bool = False, source: str = None, sources: List[str] = None, session_key: str = None, exclude_sources: List[str] = None, cwd_prefix: str = None, min_message_count: int = 0, archived_only: bool = False, include_archived: bool = False, ) -> Tuple[List[str], List[Any]]: """Shared ``sessions s`` WHERE builder so counts line up with listed rows. ``exclude_children`` hides sub-agent runs and compression continuations but keeps branch/reset children (``_LISTABLE_CHILD_SQL``). Clause order is part of the SQL text contract.""" where: List[str] = [] params: List[Any] = [] if exclude_children: where += [_LISTABLE_CHILD_SQL, f"{_delegate_from_json('s.model_config')} IS NULL"] # Show roots and user-visible branch/reset sessions, while still hiding sub-agent runs and compression # continuations. All four carry parent_session_id, so the shared predicate classifies the edge from # stable markers plus legacy-compatible parent metadata. Branch sessions are identified two ways, OR'd # for robustness: 1. A stable ``_branched_from`` marker in model_config, written by /branch at creation # time. This survives the parent being reopened and re-ended with a different end_reason (e.g. # tui_shutdown overwriting 'branched'), which otherwise hides the branch — see issue #20856. 2. The # legacy heuristic (parent ended with 'branched' before the child started), covering branch sessions # created before the marker existed. include_sources = [source] if source else list(sources or []) for clause, values in ( (f"s.source IN ({_session_ids_placeholders(include_sources)})", include_sources), ("s.session_key = ?", [session_key] if session_key else []), (f"s.source NOT IN ({_session_ids_placeholders(exclude_sources or ())})", exclude_sources or []), (_cwd_prefix_clause(cwd_prefix) if cwd_prefix else ("", [])), ("s.message_count >= ?", [min_message_count] if min_message_count > 0 else []), ): if values: where.append(clause) params.extend(values) if archived_only: where.append("s.archived = 1") elif not include_archived: where.append("s.archived = 0") return where, params def _collect_delegate_child_ids(conn, parent_ids: List[str]) -> List[str]: """Delegate-subagent ids (``_delegate_from`` marker, walked recursively) to cascade-delete with *parent_ids*; untagged children stay orphaned, not deleted.""" df = _delegate_from_json() seeds = {sid for sid in parent_ids if sid} # Seed visited with the parents: a marker chain can loop back onto a parent, # which would then be collected as its own descendant. Never return parents. # A delegation marker chain can loop back onto a parent — a cycle, or a parent that is also another # parent's delegate child when several ids are deleted at once — and without this guard that parent # would be collected as one of its own descendants and cascade-deleted along with all of its messages. # Callers delete the parents separately, so parents must never appear in the returned child set. # (#49148) found: set[str] = set(seeds) frontier = list(seeds) while frontier: next_frontier: List[str] = [] for chunk in _id_chunks(frontier, _SQL_IN_CHUNK // 2): # each id is bound twice below ph = _session_ids_placeholders(chunk) cursor = conn.execute( f"SELECT id FROM sessions WHERE {df} IN ({ph}) " f"OR (parent_session_id IN ({ph}) AND {df} IS NOT NULL)", chunk + chunk, ) for row in cursor.fetchall(): if row["id"] not in found: found.add(row["id"]) next_frontier.append(row["id"]) frontier = next_frontier return [sid for sid in found if sid not in seeds] def _delete_delegate_children(conn, parent_ids: List[str]) -> List[str]: ids = _collect_delegate_child_ids(conn, parent_ids) for chunk in _id_chunks(ids): ph = _session_ids_placeholders(chunk) conn.execute(f"DELETE FROM messages WHERE session_id IN ({ph})", chunk) # FK safety: orphan any untagged stragglers pointing at a doomed row. conn.execute(f"UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id IN ({ph})", chunk) conn.execute(f"DELETE FROM sessions WHERE id IN ({ph})", chunk) return ids # Lifecycle statuses surfaced by session pickers; classified from the final # message row ONLY so it stays O(1) per session. SESSION_STATUS_COMPLETE = "complete" SESSION_STATUS_INTERRUPTED = "interrupted" SESSION_STATUS_ERROR = "error" SESSION_STATUS_EMPTY = "empty" # finish_reason values meaning the turn ended in a provider/agent error. _ERROR_FINISH_REASONS = frozenset({"error", "agent_error", "content_filter"}) def classify_session_status(role: Optional[str], has_tool_calls: bool, finish_reason: Optional[str]) -> str: """Error finish → ``error``; assistant with pending tool_calls or a trailing user/tool row → ``interrupted``; otherwise ``complete`` (benign default: pickers must not alarm on unknown shapes).""" if (finish_reason or "").strip().lower() in _ERROR_FINISH_REASONS: return SESSION_STATUS_ERROR r = (role or "").strip().lower() if r in {"user", "tool"} or (r == "assistant" and has_tool_calls): return SESSION_STATUS_INTERRUPTED return SESSION_STATUS_COMPLETE # Parent→child profile_name inheritance fence: keyless rows inherit freely; two # ``agent::...`` keyed rows must agree on the namespace. # ``agent::...`` gateway keys encode the profile namespace; a keyless row (CLI / subagent lineage) # carries none and inherits freely. Two keyed rows must agree on ``agent::`` — a default child # (``agent:main:``) forked from a sibling profile's row must not be durably mislabelled as that profile's. # See #88381. _SAME_KEY_NAMESPACE_SQL = ( "p.session_key IS NULL OR sessions.session_key IS NULL" " OR substr(p.session_key, 1, instr(substr(p.session_key, 7), ':') + 6)" " = substr(sessions.session_key, 1, instr(substr(sessions.session_key, 7), ':') + 6)" ) # Upsert tail of _insert_session_row: every routing/metadata column keeps the value an # earlier writer set (whitespace is part of the SQL text). _UPSERT_KEEP_EXISTING_SQL = ",\n".join( f" {col} = COALESCE(sessions.{col}, excluded.{col})" for col in ( "session_key", "chat_id", "chat_type", "thread_id", "parent_session_id", "cwd", "profile_name", "git_repo_root", "origin_json", "display_name", ) ) def _inherit_col_sql(col: str, extra: str = "") -> str: """``col = COALESCE(sessions.col, (SELECT p.col FROM parent))`` (whitespace is part of the SQL text).""" pad = " " * (30 + len(col)) return ( f"{col} = COALESCE(sessions.{col},\n{pad}(SELECT p.{col} FROM sessions p\n" f"{pad} WHERE p.id = sessions.parent_session_id{extra}))" ) _INHERIT_SEP = ",\n" + " " * 27 _INHERIT_PARENT_META_SQL = ( "UPDATE sessions\n SET " + _INHERIT_SEP.join(( *(_inherit_col_sql(c) for c in ("cwd", "git_repo_root", "git_branch")), _inherit_col_sql("profile_name", "\n" + " " * 46 + f"AND ({_SAME_KEY_NAMESPACE_SQL})"), )) + "\n WHERE id = ? AND parent_session_id IS NOT NULL" ) _INHERIT_PARENT_ROUTING_SQL = ( "UPDATE sessions\n SET " + _INHERIT_SEP.join(_inherit_col_sql(c) for c in ( "user_id", "session_key", "chat_id", "chat_type", "thread_id", "display_name", "origin_json", )) + "\n WHERE id = ? AND parent_session_id IS NOT NULL\n" " AND EXISTS (\n" " SELECT 1 FROM sessions p\n" " WHERE p.id = sessions.parent_session_id\n" " AND p.end_reason = 'compression'\n" " )" ) class SessionSessionsMixin: """Session rows: create/inherit, lifecycle flags, model_config, listing, deletion.""" def _own_profile_name(self) -> Optional[str]: """The profile owning THIS store, from ``db_path`` alone (``/state.db`` → default, ``/profiles//state.db`` → name): a gateway serving a NON-launch profile opens that profile's store. None outside the profile tree — NULL beats a fabricated owner.""" try: from hermes_constants import get_default_hermes_root root = get_default_hermes_root().resolve() parent = Path(self.db_path).resolve().parent if parent == root: return "default" is_profile_dir = parent.parent == root / "profiles" if is_profile_dir and re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,63}", parent.name): return parent.name except Exception: logger.debug("own-profile derivation failed", exc_info=True) return None @staticmethod def _inherit_parent_session_metadata(conn, session_id: str) -> None: """NULL-fill a child's cwd/git/profile from its parent (profile_name only within the same ``agent::`` namespace). Gateway routing columns are inherited ONLY by compression forks (a crash before the gateway re-records the peer would strand the child unroutable); delegate children must NOT inherit them (peer recovery could repoint traffic into a subagent's session).""" conn.execute(_INHERIT_PARENT_META_SQL, (session_id,)) conn.execute(_INHERIT_PARENT_ROUTING_SQL, (session_id,)) def _insert_session_row( self, session_id: str, source: str, model: str = None, model_config: Dict[str, Any] = None, system_prompt: str = None, user_id: str = None, session_key: Optional[str] = None, chat_id: str = None, chat_type: str = None, thread_id: str = None, parent_session_id: str = None, cwd: str = None, profile_name: Optional[str] = None, git_repo_root: str = None, origin_json: str = None, display_name: str = None, ) -> None: """Upsert a session row, never overwriting what an earlier writer set (the gateway creates a bare row before create_session carries the real model/prompt) — the one exception is the token-accounting guard's placeholder ``source='unknown'``, which a later writer's real surface replaces (#111999): once minted, that placeholder otherwise labelled a real session anonymous for life, because this upsert is the only writer that could correct it. chat_id/thread_id scope gateway /resume (IDOR). Children backfill from the parent; a missing profile_name is stamped with THIS store's own (NULL reads as unowned). When ``parent_session_id`` is set (compression fork, delegate/subagent spawn, branch continuation) and this row's own ``cwd``/``git_repo_root``/ ``git_branch``/``profile_name`` are still NULL after the insert, they are backfilled from the parent row. Callers of ``create_session`` for a child session historically didn't propagate these fields themselves (e.g. the compression-fork path), so a lineage could silently lose its working directory and drop out of the project sidebar every time it forked (#64709), or lose its owning profile and be aggregated as "default" every time it rotated or branched (the cross-profile session-jump bug). This only fills NULLs — an explicit value on the child is never overwritten. For compression forks specifically (parent ended with ``end_reason='compression'``), the gateway origin columns (``user_id``/``session_key``/``chat_id``/``chat_type``/ ``thread_id``/``display_name``/``origin_json``) are inherited too, so a crash before the gateway re-records the peer can't strand the child without a recoverable routing mapping (#59527). When the caller passes no ``profile_name`` at all, the row is stamped with THIS store's own profile (:meth:`_own_profile_name`) instead of NULL. Every ``state.db`` belongs to exactly one profile — the same single-match contract :meth:`backfill_null_session_profiles` relies on — so the stamp is derivation, not a guess. Rows minted NULL after that one-shot #94724 backfill ran stayed NULL forever, and profile-keyed consumers (desktop sidebar scope matching, ``@session:/`` deep links, the fail-closed owner ladder) treat NULL as unowned: the session vanishes from the sidebar even though its transcript is intact (#99222). Stores outside the profile tree (explicit ``db_path`` in tests, ad-hoc copies) derive nothing and keep NULL — never guess. """ if not (profile_name or "").strip(): profile_name = self._own_profile_name() def _do(conn): system_prompt_hash = self._store_system_prompt(conn, system_prompt) conn.execute( """INSERT INTO sessions ( id, source, user_id, session_key, chat_id, chat_type, thread_id, model, model_config, system_prompt, system_prompt_hash, parent_session_id, cwd, profile_name, git_repo_root, origin_json, display_name, started_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET source = CASE WHEN sessions.source = 'unknown' THEN COALESCE(excluded.source, 'unknown') ELSE sessions.source END, model = COALESCE(sessions.model, excluded.model), model_config = CASE WHEN excluded.model_config IS NOT NULL AND json_type( sessions.model_config, '$._reset_from' ) IS NOT NULL AND json_remove( sessions.model_config, '$._reset_from' ) = '{}' THEN json_set( excluded.model_config, '$._reset_from', json_extract( sessions.model_config, '$._reset_from' ) ) ELSE COALESCE( sessions.model_config, excluded.model_config ) END, system_prompt_hash = COALESCE( sessions.system_prompt_hash, excluded.system_prompt_hash ), system_prompt = CASE WHEN sessions.system_prompt_hash IS NULL AND excluded.system_prompt_hash IS NOT NULL THEN NULL ELSE sessions.system_prompt END, """ + _UPSERT_KEEP_EXISTING_SQL, ( session_id, source, user_id, session_key, chat_id, chat_type, thread_id, model, json.dumps(model_config) if model_config else None, system_prompt_hash, parent_session_id, cwd, profile_name, git_repo_root, origin_json, display_name, time.time(), ), ) if system_prompt_hash is not None: self._delete_unreferenced_system_prompts(conn) if parent_session_id: self._inherit_parent_session_metadata(conn, session_id) # Transcript-critical: a failed row creation aborts the turn. self._execute_write(_do, patience_s=self._TRANSCRIPT_WRITE_PATIENCE_S) def create_session(self, session_id: str, source: str, **kwargs) -> str: """Create (upsert) a session record. Returns the session_id.""" self._insert_session_row(session_id, source, **kwargs) return session_id def ensure_session(self, session_id: str, source: str = "unknown", model: str = None, **kwargs) -> str: """Ensure a session row exists (upsert). Accepts optional kwargs.""" self._insert_session_row(session_id, source, model=model, **kwargs) return session_id def set_expiry_finalized(self, session_id: str, finalized: bool = True) -> None: """Mirror ``SessionEntry.expiry_finalized`` so it survives a lost sessions.json. See #9006. """ if not session_id: return self._write_sql( "UPDATE sessions SET expiry_finalized = ? WHERE id = ?", (1 if finalized else 0, session_id), ) def find_session_by_origin( self, *, platform: str, chat_id: str, thread_id: Optional[str] = None, user_id: Optional[str] = None, ) -> Optional[str]: """Most recent live session_id for source + chat_id (+ thread_id). With ``user_id`` exact sender matches win; several distinct users and no match → None (never another participant's session).""" if not platform or chat_id in (None, ""): return None query = """ SELECT id, user_id, started_at FROM sessions WHERE LOWER(source) = LOWER(?) AND session_key IS NOT NULL AND chat_id = ? AND ended_at IS NULL """ params: list = [platform, str(chat_id)] if thread_id is not None: query += " AND COALESCE(thread_id, '') = ?" params.append(str(thread_id)) rows = [dict(r) for r in self._read_all(query + " ORDER BY started_at DESC", params)] if not rows: return None if user_id: exact = [r for r in rows if str(r.get("user_id") or "") == str(user_id)] if exact: return str(exact[0]["id"]) if len(rows) > 1: return None elif len({u for u in (str(r.get("user_id") or "").strip() for r in rows) if u}) > 1: return None return str(rows[0]["id"]) # Orphaned gateway-session repair: widest plausible gap between a keyed predecessor going # quiet and its unkeyed successor (incident was ~60s; 15 min without spanning conversations). _ORPHAN_ADOPTION_MAX_GAP_S = 900.0 # Children that are NOT compression continuations (branches, delegates, tool sessions). Markers # are bound to the queried parent id: continuations inherit model_config verbatim, so # presence-matching misclassified them as delegates. _NON_CONTINUATION_CHILD_FILTER_SQL = ( f" AND COALESCE({_sql_json_extract('{alias}model_config', '$._branched_from')}, '') != ?\n" f" AND COALESCE({_sql_json_extract('{alias}model_config', '$._delegate_from')}, '') != ?\n" " AND COALESCE({alias}source, '') != 'tool'\n" ) def end_session(self, session_id: str, end_reason: str) -> None: """Mark a session ended; the first end_reason wins (a compression split must keep ``'compression'`` even if a stale end_session() lands later); reopen_session() to re-end.""" self._execute_write(lambda conn: self._end_and_bump( conn, "UPDATE sessions SET ended_at = ?, end_reason = ? WHERE id = ? AND ended_at IS NULL", (time.time(), end_reason, session_id), session_id, end_reason, )) def _end_and_bump(self, conn, sql: str, params: tuple, session_id: str, reason: str) -> int: """Run an end-stamp UPDATE; only a boundary this call actually wrote advances the conversation generation (a no-op must not rotate the peer). Returns rowcount.""" changed = conn.execute(sql, params).rowcount if changed: self._bump_conversation_generation(conn, session_id, reason) return changed def reopen_session(self, session_id: str) -> None: """Clear ended_at/end_reason so a session can be resumed; first freeze markerless legacy reset children, skipping explicit fork/delegate provenance and children that predate the parent itself. The guard compares against the parent's started_at, not its current ended_at: a parent that was reopened and re-ended later still owns reset children from its earlier boundaries.""" def _do(conn): conn.execute( "UPDATE sessions AS child SET model_config = json_set(" "COALESCE(child.model_config, '{}'), '$._reset_from', child.parent_session_id) " f"WHERE child.parent_session_id = ? AND {_sql_json_extract('child.model_config', '$._reset_from')} IS NULL " f"AND {_sql_json_extract('child.model_config', '$._branched_from')} IS NULL " f"AND {_sql_json_extract('child.model_config', '$._delegate_from')} IS NULL " "AND COALESCE(child.source, '') != 'tool' " "AND child.started_at >= (SELECT p.started_at FROM sessions p WHERE p.id = child.parent_session_id) " f"AND {_legacy_reset_child_sql('child', _session_ids_placeholders(_RESET_END_REASONS))}", (session_id, *_RESET_END_REASONS), ) conn.execute( "UPDATE sessions SET ended_at = NULL, end_reason = NULL WHERE id = ?", (session_id,), ) self._execute_write(_do) def promote_to_session_reset(self, session_id: str, reason: str = "session_reset") -> bool: """Durably mark an intentional reset boundary on live rows or rows with a *recoverable* accidental end_reason (explicit boundaries are preserved): an ``agent_close`` row left recoverable would be resurrected by stale-route recovery. Keep in sync with find_latest_gateway_session_for_peer. Plain ``end_session()`` is NOT sufficient for reset boundaries: it no-ops on an already-ended row, so a row that agent cleanup already closed as ``agent_close`` would stay recoverable and stale-route recovery would resurrect the reset session with its full history (#61220, #61993, #63539). """ if not session_id: return False now = time.time() # /new and policy auto-resets promote rather than end_session, so the # generation advances here too — same transaction, only when written. try: return bool(self._execute_write(lambda conn: self._end_and_bump( conn, "UPDATE sessions SET ended_at = ?, end_reason = ? WHERE id = ? AND (ended_at IS NULL " f"OR end_reason IN ({_RECOVERABLE_END_REASONS_SQL}))", (now, reason, session_id), session_id, reason, ))) except Exception: return False def update_session_cwd( self, session_id: str, cwd: str, git_branch: Optional[str] = None, git_repo_root: Optional[str] = None, replace_git_meta: bool = False, ) -> Optional[int]: """Persist the authoritative cwd and claim a Git metadata generation. git fields are written only when non-empty (a probe failure never clobbers a value) except under ``replace_git_meta`` (a workspace MOVE overwrites the old repo identity). Async probes publish with the returned generation so an older worker cannot overwrite a newer claim (A -> B -> A).""" if not session_id or not cwd: return None branch = (git_branch or "").strip() repo_root = (git_repo_root or "").strip() def _do(conn): current = conn.execute("SELECT cwd FROM sessions WHERE id = ?", (session_id,)).fetchone() if current is None: return None sets = ["cwd = ?", "git_metadata_generation = COALESCE(git_metadata_generation, 0) + 1"] params: List[Any] = [cwd] if current[0] != cwd or replace_git_meta: sets.extend(("git_branch = ?", "git_repo_root = ?")) params.extend((branch or None, repo_root or None)) else: # same cwd: only overwrite with captured (non-empty) values for col, val in (("git_branch", branch), ("git_repo_root", repo_root)): if val: sets.append(f"{col} = ?") params.append(val) conn.execute(f"UPDATE sessions SET {', '.join(sets)} WHERE id = ?", [*params, session_id]) row = conn.execute( "SELECT git_metadata_generation FROM sessions WHERE id = ?", (session_id,), ).fetchone() return None if row is None else int(row[0]) return self._execute_write(_do) def publish_session_git_metadata( self, session_id: str, cwd: str, generation: int, git_branch: Optional[str] = None, git_repo_root: Optional[str] = None, ) -> bool: """Publish async Git enrichment only while its cwd claim is current.""" valid_generation = isinstance(generation, int) and not isinstance(generation, bool) and generation >= 1 if not session_id or not cwd or not valid_generation: return False fields = [ (col, val) for col, val in ( ("git_branch", (git_branch or "").strip()), ("git_repo_root", (git_repo_root or "").strip()), ) if val ] if not fields: return False return self._write_rowcount( f"UPDATE sessions SET {', '.join(f'{col} = ?' for col, _ in fields)} " "WHERE id = ? AND cwd = ? AND git_metadata_generation = ?", [val for _, val in fields] + [session_id, cwd, generation], ) == 1 def backfill_repo_roots(self, cwd_to_root: Dict[str, str]) -> None: """Backfill git repo roots for cwds without one; never clobbers a recorded root.""" pairs = [(root, cwd) for cwd, root in cwd_to_root.items() if root and cwd] if pairs: self._write_sql( "UPDATE sessions SET git_repo_root = ? WHERE cwd = ? AND COALESCE(git_repo_root, '') = ''", pairs, many=True, ) def touch_session_activity( self, session_id: str, ts: Optional[float] = None, *, description: Optional[str] = None, provenance: Optional[ActivityProvenance] = None, ) -> None: """Stamp durable mid-turn activity (observation-only; rate-limited by the caller) so surfaces see activity before any message row lands. Never moves ``last_activity_at`` backwards. Called (rate-limited) from ``AIAgent._touch_activity`` so gateway/CLI surfaces and stall consumers observe API/tool/compaction activity even when no new message row has been written yet (#72016 / #72039). """ if not session_id: return when = float(ts if ts is not None else time.time()) self._write_sql( "UPDATE sessions SET last_activity_at = ?, " "last_activity_description = ?, last_activity_provenance = ? " "WHERE id = ? AND (last_activity_at IS NULL OR last_activity_at < ?)", ( when, bound_activity_description(description), normalize_activity_provenance(provenance).value, session_id, when, ), patience_s=self._ACTIVITY_WRITE_PATIENCE_S, ) # Observation-only write: never let it ride the full routine write-patience budget (#76354 review S1). # Under contention a heartbeat that waits ~20s would delay the response-critical path it is merely # observing; give up after a sub-second budget instead (the next due window retries naturally). def clear_session_activity_labels(self, session_id: str) -> None: """Clear activity labels after a turn (``last_activity_at`` is kept so idle / watchdog clocks stay continuous). A no-op clear skips the write transaction. Description and provenance are observation labels for *what was happening at* that timestamp during an active turn; once the turn is idle they must not keep advertising "compressing" / "executing tool" (#72039). Response-critical-path contract (#76354 review S1): runs in the turn's ``finally``; a no-op clear (labels already empty) skips the write transaction entirely, and a real clear uses the same short sub-second busy budget as :meth:`touch_session_activity` instead of the full routine write patience. """ if not session_id: return try: row = self._read_one( "SELECT last_activity_description, last_activity_provenance FROM sessions WHERE id = ?", (session_id,), ) except sqlite3.Error: row = None if row is not None and not row[0] and (not row[1] or row[1] == ActivityProvenance.UNKNOWN.value): return self._write_sql( "UPDATE sessions SET last_activity_description = ?, last_activity_provenance = ? WHERE id = ?", ("", ActivityProvenance.UNKNOWN.value, session_id), patience_s=self._ACTIVITY_WRITE_PATIENCE_S, ) def update_session_meta( self, session_id: str, model_config_json: str, model: Optional[str] = None, ) -> None: """Update model_config and (COALESCE) optionally model.""" self.flush_token_counts() # barrier against queued token deltas — see update_session_model self._write_sql( "UPDATE sessions SET model_config = ?, model = COALESCE(?, model) WHERE id = ?", (model_config_json, model, session_id), ) def update_system_prompt(self, session_id: str, system_prompt: Optional[str]) -> None: """Store the full assembled system prompt snapshot.""" def _do(conn): conn.execute( "UPDATE sessions SET system_prompt_hash = ?, system_prompt = NULL WHERE id = ?", (self._store_system_prompt(conn, system_prompt), session_id), ) self._delete_unreferenced_system_prompts(conn) self._execute_write(_do) def update_session_tool_names(self, session_id: str, tool_names: Optional[List[str]]) -> None: """Persist the resolved ``tools[]`` name order so a rebuilt AIAgent can't fork the cached tool prefix on a flipped check_fn verdict; ``None`` clears.""" payload = json.dumps(list(tool_names)) if tool_names is not None else None self._write_sql("UPDATE sessions SET tool_names = ? WHERE id = ?", (payload, session_id)) def update_session_model(self, session_id: str, model: str, provider: Optional[str] = None) -> None: """Set the model after a mid-session /model switch (unconditionally), null system_prompt so stale Model:/Provider: footers rebuild, and drop any Browser runtime lock (lineage markers survive). *provider* is merged into model_config so resume recombines model and provider. When *provider* is given, it is merged into ``model_config`` alongside the model (``$.model`` / ``$.provider``) so a later resume recombines the persisted model with the provider that actually serves it instead of the config.yaml primary provider (#79536). Callers without provider knowledge leave any stored provider untouched. """ # Flush first: a still-queued pre-switch delta applied after this UPDATE would trip the # first_accounted_route overwrite and resurrect the old route. self.flush_token_counts() patch: Dict[str, Any] = {"browser_model_lock": None} if model: patch["model"] = model if provider: patch["provider"] = provider self._write_model_config_patch( session_id, patch, "UPDATE sessions SET model = ?, model_config = ?, " "system_prompt = NULL, system_prompt_hash = NULL WHERE id = ?", lambda merged: (model, merged, session_id), ) def _write_model_config_patch( self, session_id: str, patch: Dict[str, Any], sql: str = "UPDATE sessions SET model_config = ? WHERE id = ?", params: Optional[Callable[[Optional[str]], tuple]] = None, ) -> None: """Merge ``patch`` into model_config then run ``sql`` with ``params(merged)`` in one write transaction; no-op when the row doesn't exist. Custom ``sql`` (prompt-nulling) also GCs prompts.""" def _do(conn): merged = self._merge_model_config_json(conn, session_id, patch) if merged is _MODEL_CONFIG_ROW_MISSING: return conn.execute(sql, params(merged) if params else (merged, session_id)) if params is not None: self._delete_unreferenced_system_prompts(conn) self._execute_write(_do) def _merge_model_config_json( self, conn, session_id: str, patch: Dict[str, Any], *, on_missing: str = "skip", ): """SELECT + tolerant-parse + merge ``patch`` into model_config (the one place that keeps ``_branched_from``/``_delegate_from`` alive); ``None`` deletes a key. Returns serialized JSON (``None`` when empty) or ``_MODEL_CONFIG_ROW_MISSING`` (``on_missing="raise"`` → ValueError).""" row = conn.execute("SELECT model_config FROM sessions WHERE id = ?", (session_id,)).fetchone() if row is None: if on_missing == "raise": raise ValueError(f"Session not found: {session_id}") return _MODEL_CONFIG_ROW_MISSING config = _parse_model_config(row[0]) for key, value in patch.items(): if value is None: config.pop(key, None) else: config[key] = value return json.dumps(config) if config else None def patch_session_model_config(self, session_id: str, patch: Dict[str, Any]) -> None: """Merge ``patch`` into model_config atomically (``None`` removes a key); no-op when the row or patch is empty.""" if not session_id or not patch: return self._write_model_config_patch(session_id, patch) def get_session_model_config_value(self, session_id: str, key: str, default: Any = None) -> Any: """Read one key out of a session's model_config JSON (tolerant parse).""" session = self.get_session(session_id) or {} return _parse_model_config(session.get("model_config")).get(key, default) def update_session_runtime_lock( self, session_id: str, *, model: Optional[str] = None, provider: Optional[str] = None, model_options: Optional[Dict[str, Any]] = None, route_source: Optional[str] = None, confirmed: bool = False, ) -> None: """Persist a Browser / API-client runtime lock into model_config (lineage markers survive); null system_prompt so cached footers cannot lie.""" lock = { "provider": provider or "", "model": model or "", "model_options": model_options or {}, "route_source": route_source or "", "confirmed": bool(confirmed), "updated_at": time.time(), } self._write_model_config_patch( session_id, {"browser_model_lock": lock}, """UPDATE sessions SET model_config = ?, model = COALESCE(?, model), system_prompt = NULL, system_prompt_hash = NULL WHERE id = ?""", lambda merged: (merged, model, session_id), ) def set_session_yolo(self, session_id: str, enabled: bool) -> None: """Persist the per-session YOLO flag so ``/yolo`` survives ``--resume``; no-op without a row.""" if not session_id: return self._write_model_config_patch(session_id, {"yolo_mode": bool(enabled)}) @staticmethod def session_yolo_enabled(session_meta: Optional[Dict[str, Any]]) -> bool: """Persisted YOLO flag; False on any parse failure (resume must never enable the bypass).""" return bool(_parse_model_config((session_meta or {}).get("model_config")).get("yolo_mode")) def get_session(self, session_id: str) -> Optional[Dict[str, Any]]: """Get a session by ID (drains queued token deltas first so cost readers see exact totals).""" self.flush_token_counts() row = self._read_one( "SELECT s.*, COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved " "FROM sessions s LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash WHERE s.id = ?", (session_id,), ) return self._session_row_dict(row) if row else None def get_recent_session_model_route(self, session_id: str) -> Optional[Dict[str, Any]]: """Most recently used main-loop model route as one coherent per-call tuple (``session_model_usage`` keeps model+provider together; ``sessions`` mixes route changes). Recency, not lifetime call count: on a long session a route retired weeks ago can hold the highest ``api_call_count`` forever, and /status and /usage would keep calling it current. ``rowid DESC`` breaks same-timestamp ties toward the route that first appeared later; without it SQLite's temp-sort order is unspecified and the retired route can win.""" self.flush_token_counts() row = self._read_one( """SELECT model, billing_provider, billing_base_url, billing_mode, api_call_count FROM session_model_usage WHERE session_id = ? AND task = '' AND model <> 'unknown' AND billing_provider <> '' ORDER BY last_seen DESC, rowid DESC LIMIT 1""", (session_id,), ) return dict(row) if row else None def resolve_session_id(self, session_id_or_prefix: str) -> Optional[str]: """Exact id, else the single unambiguous prefix match, else None.""" exact = self.get_session(session_id_or_prefix) if exact: return exact["id"] matches = self._read_all( "SELECT id FROM sessions WHERE id LIKE ? ESCAPE '\\' ORDER BY started_at DESC LIMIT 2", (f"{_escape_like(session_id_or_prefix)}%",), ) return matches[0]["id"] if len(matches) == 1 else None def backfill_null_session_profiles(self, profile_name: str) -> int: """Stamp this store's own profile onto legacy ``profile_name IS NULL`` rows, which the fail-closed owner ladder cannot route. Never overwrites a non-NULL owner. Returns rows stamped. Sessions created before the durable-ownership work (#95407 lineage) carry ``profile_name = NULL``. On single-backend installs that was harmless, but once a Desktop registers a second connection the fail-closed owner ladder (which is correct for new sessions) can no longer route those rows anywhere — every pre-campaign session becomes unresumable after upgrade (#94724, field report). """ stamp = (profile_name or "").strip() if not stamp: return 0 return int(self._write_rowcount( """UPDATE sessions SET profile_name = ? WHERE profile_name IS NULL OR TRIM(profile_name) = ''""", (stamp,), ) or 0) def _set_lineage_column(self, column: str, session_id: str, value: Any) -> bool: """Set one ``sessions`` column across a whole compression lineage: Desktop projects roots forward to their tip, so updating only the tip would let the root resurrect it on refresh.""" return self._write_rowcount( 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 parent.end_reason = 'compression' ), descendants(id) AS ( SELECT ? UNION SELECT child.id FROM descendants d JOIN sessions parent ON parent.id = d.id JOIN sessions child ON child.parent_session_id = parent.id WHERE parent.end_reason = 'compression' ), lineage(id) AS ( SELECT id FROM ancestors UNION SELECT id FROM descendants ) UPDATE sessions SET {column} = ? WHERE id IN (SELECT id FROM lineage) """, (session_id, session_id, value), ) > 0 def set_session_archived(self, session_id: str, archived: bool) -> bool: """Soft-hide (or unhide) a session and its compression lineage; messages are kept.""" return self._set_lineage_column("archived", session_id, int(archived)) # Accidental end reasons recovery treats as resumable (also interpolated into # the recovery/promotion SQL so literals cannot drift). RECOVERABLE_END_REASONS = _RECOVERABLE_END_REASONS def unarchive_recoverable_session(self, session_id: str) -> bool: """Un-archive a session archived by a recoverable accident (ws_orphan_reap, agent_close); deliberate archives are left alone. True when un-archived. Registry-style lookups (Bot Mode's canonical "Bot Chat") use this to resurrect a row the ws-orphan reaper (``ws_orphan_reap``) or older agent cleanup (``agent_close``) archived: those ends are accidents, not user intent, so the identity-scoped canonical chat must survive them (#92687). Sessions archived with no end_reason or an explicit boundary reason (user archived deliberately, ``session_reset``, …) are left untouched — returns ``False`` for those, ``True`` only when the row was archived for a recoverable reason and is now un-archived (whole compression lineage, via :meth:`set_session_archived`). """ if not session_id: return False try: row = self.get_session(session_id) except Exception: return False if not row or not row.get("archived"): return False # The accidental stamp lives on the live TIP; judge recoverability there. tip = row try: tip_id = self.get_compression_tip(session_id) or session_id if tip_id != session_id: tip = self.get_session(tip_id) or row except Exception: pass if (tip.get("end_reason") or "") not in self.RECOVERABLE_END_REASONS: return False if not self.set_session_archived(session_id, False): return False # Clear the accidental end stamp, or a LATER deliberate archive (which never # writes end_reason) would auto-resurrect on the next lookup. self._write_sql( "UPDATE sessions SET ended_at = NULL, end_reason = NULL WHERE id = ?", (tip["id"],), ) return True def set_session_pinned(self, session_id: str, pinned: bool) -> bool: """Pin/unpin a session and its compression lineage (pins are exempt from the auto_archive sweep). Pinning also clears ``hidden``: a pin means "keep this visible", and a hidden+pinned row is otherwise absent from both the default listing and the pinned back-fill (see #106171). Exempt the canonical Bot Chat (hidden + exact registry title): the desktop contract keeps it hidden and reachable only through the bot row, and unhiding it would also disable the rename guard in ``_set_session_title`` that protects its identity (see review on #106180).""" result = self._set_lineage_column("pinned", session_id, int(pinned)) if pinned: row = self.get_session(session_id) is_canonical_bot_chat = bool(row) and bool(row.get("hidden")) and ( (row.get("title") or "") == self.CANONICAL_BOT_CHAT_TITLE ) if not is_canonical_bot_chat: self._set_lineage_column("hidden", session_id, 0) return result def set_session_hidden(self, session_id: str, hidden: bool) -> bool: """Hide/unhide a session and its compression lineage from the default listing; still resumable.""" return self._set_lineage_column("hidden", session_id, int(hidden)) def set_session_read(self, session_id: str, read: bool = True) -> bool: """Mark read/unread across the compression lineage. ``last_read_at`` is a watermark: unread when activity postdates it (no write on the message path). NULL = never tracked = read; 0 = unread.""" return self._set_lineage_column("last_read_at", session_id, time.time() if read else 0.0) @staticmethod def session_unread(session_row: Dict[str, Any]) -> bool: """Unread = activity postdates the ``last_read_at`` watermark (NULL = read).""" last_read = session_row.get("last_read_at") if last_read is None: return False last_active = session_row.get("last_active") or session_row.get("started_at") return float(last_active or 0) > float(last_read) # compact_rows excludes only payload-heavy blobs no list consumer renders. _SESSION_COMPACT_EXCLUDED = frozenset( {"system_prompt", "system_prompt_hash", "git_metadata_generation"} ) _session_compact_cols_sql: Optional[str] = None @staticmethod def _chain_search_where(where_sql: str, id_needle: str, search_needle: str) -> Tuple[str, List[Any]]: """Extend ``where_sql`` with the id_query / search_query filters: a row is admitted when its own id or any id in its forward compression chain matches (search also matches titles and a punctuation-stripped form so ``an94`` finds ``AN-94``); chain membership bounds the LIKE.""" params: List[Any] = [] clauses: List[str] = [] def like(needle: str) -> str: return f"%{_escape_like(needle)}%" if id_needle: clauses.append( "EXISTS (SELECT 1 FROM chain cq WHERE cq.root_id = s.id" " AND LOWER(cq.cur_id) LIKE ? ESCAPE '\\')" ) params.append(like(id_needle)) if search_needle: compact_needle = re.sub(r"[\W_]+", "", search_needle) search_clause = ( "EXISTS (SELECT 1 FROM chain cq JOIN sessions cs ON cs.id = cq.cur_id" " WHERE cq.root_id = s.id AND (LOWER(COALESCE(cs.title, '')) LIKE ? ESCAPE '\\'" " OR LOWER(cq.cur_id) LIKE ? ESCAPE '\\'" ) params.extend([like(search_needle)] * 2) if compact_needle: search_clause += ( " OR REPLACE(REPLACE(REPLACE(REPLACE(LOWER(COALESCE(cs.title, ''))," " '-', ''), '_', ''), '.', ''), ' ', '') LIKE ? ESCAPE '\\'" ) params.append(like(compact_needle)) clauses.append(search_clause + "))") if not clauses: return where_sql, params combined = " AND ".join(clauses) return (f"{where_sql} AND {combined}" if where_sql else f"WHERE {combined}"), params def _project_compression_tips(self, sessions: List[Dict[str, Any]], compact_rows: bool) -> List[Dict[str, Any]]: """Replace each compression root's surfaced fields with its live tip's (root ``started_at`` kept for stable ordering), one batched query. ``_lineage_ids`` carries every chain id (a tile may hold a MIDDLE segment's id).""" chain_by_root: Dict[str, List[str]] = {} # only roots whose tip differs from themselves for s in sessions: if s.get("end_reason") == "compression": chain = self.get_compression_chain(s["id"]) if chain and chain[-1] != s["id"]: chain_by_root[s["id"]] = chain tip_rows = ( self._get_session_rich_rows_batch( {chain[-1] for chain in chain_by_root.values()}, compact_rows=compact_rows, ) if chain_by_root else {} ) projected = [] for s in sessions: chain = chain_by_root.get(s["id"]) tip_row = tip_rows.get(chain[-1]) if chain else None if not tip_row: projected.append(s) continue merged = dict(s) for key in ( "id", "ended_at", "end_reason", "message_count", "tool_call_count", "title", "last_active", "preview", "model", "system_prompt", "cwd", "git_branch", "git_repo_root", ): if key in tip_row: merged[key] = tip_row[key] if merged.get("title") is None: # The title is carried root->tip AFTER the publish transaction; a rotation cut off in # between leaves it on the ended root, and exact-title lookups (`hermes peer dm` -> # canonical "Bot Chat") must still see the lineage under its name (#106165). merged["title"] = s.get("title") merged["_lineage_root_id"] = s["id"] merged["_lineage_ids"] = chain projected.append(merged) return projected def list_recent_sessions_bounded( self, *, limit: int = 20, exclude_sources: List[str] = None, timeout_seconds: float = 3.0, candidate_limit: int = None, lineage_limit: int = None, ) -> List[Dict[str, Any]]: """Latency-bounded recent-conversation browse (``session_search()``): preselect a small candidate set from the indexed durable activity timestamp (fallback ``started_at``), resolve only those across compression ancestry/chains, then hydrate activity/previews for that bounded set. Lineage traversal uses ``UNION`` plus a total-row ceiling so a corrupt cycle or a deep/branching lineage cannot defeat the bound; a lineage that hits the ceiling before a terminal root/tip is omitted, not expanded. A cooperative SQLite progress deadline interrupts sustained work past ``timeout_seconds`` and raises ``TimeoutError`` (cheap statements may finish between callbacks). Supports only the agent-tool browse filters; rich callers keep using :meth:`list_sessions_rich`.""" limit = max(1, int(limit)) timeout_seconds = max(0.0, float(timeout_seconds)) if candidate_limit is None: candidate_limit = max(128, limit * 8) candidate_limit = max(limit, min(int(candidate_limit), 2048)) if lineage_limit is None: lineage_limit = min(8192, candidate_limit * 8) lineage_limit = max(candidate_limit, min(int(lineage_limit), 8192)) candidate_clauses = [ "s.archived = 0", "s.hidden = 0", f"{_delegate_from_json('s.model_config')} IS NULL", ] candidate_params: List[Any] = [] if exclude_sources: placeholders = ",".join("?" for _ in exclude_sources) candidate_clauses.append(f"s.source NOT IN ({placeholders})") candidate_params.extend(exclude_sources) candidate_where = " AND ".join(candidate_clauses) # A compression continuation is an implementation edge, unlike /new # reset and /branch children which are independent user-visible # conversations. The same predicate is used in both directions so a # candidate tip maps to its logical root and the root maps back to the # freshest live tip. compression_parent_edge = f""" parent.end_reason = 'compression' AND child.parent_session_id = parent.id AND json_extract( COALESCE(child.model_config, '{{}}'), '$._branched_from' ) IS NULL AND {_delegate_from_json('child.model_config')} IS NULL AND COALESCE(child.source, '') != 'tool' """ query = f""" WITH RECURSIVE recent_candidates(id) AS ( SELECT s.id FROM sessions s WHERE {candidate_where} ORDER BY COALESCE(s.last_activity_at, s.started_at) DESC, s.started_at DESC, s.id DESC LIMIT ? ), ancestors(candidate_id, cur_id) AS ( SELECT id, id FROM recent_candidates UNION SELECT a.candidate_id, parent.id FROM ancestors a JOIN sessions child ON child.id = a.cur_id JOIN sessions parent ON {compression_parent_edge} LIMIT ? ), candidate_roots(root_id) AS ( SELECT DISTINCT a.cur_id FROM ancestors a JOIN sessions child ON child.id = a.cur_id WHERE NOT EXISTS ( SELECT 1 FROM sessions parent WHERE {compression_parent_edge} ) ), chain(root_id, cur_id) AS ( SELECT root_id, root_id FROM candidate_roots UNION SELECT c.root_id, child.id FROM chain c JOIN sessions parent ON parent.id = c.cur_id JOIN sessions child ON {compression_parent_edge} LIMIT ? ), chain_rows AS ( SELECT c.root_id, c.cur_id, {_sql_session_last_active_by_id('c.cur_id')} AS activity, CASE WHEN EXISTS ( SELECT 1 FROM sessions parent JOIN sessions child ON {compression_parent_edge} WHERE parent.id = c.cur_id ) THEN 0 ELSE 1 END AS is_tip FROM chain c ), ranked_tips AS ( SELECT root_id, cur_id, activity, ROW_NUMBER() OVER ( PARTITION BY root_id ORDER BY activity DESC, cur_id DESC ) AS rank_in_root FROM chain_rows WHERE is_tip = 1 ) SELECT tip.id, tip.source, tip.model, COALESCE(tip.title, s.title) AS title, s.started_at AS started_at, tip.ended_at, tip.end_reason, tip.message_count, tip.tool_call_count, rt.activity AS last_active, COALESCE( (SELECT {_PREVIEW_RAW_SELECT} FROM messages m WHERE m.session_id = tip.id AND m.role = 'user' AND m.content IS NOT NULL AND {_PREVIEW_ELIGIBLE_SQL} ORDER BY m.timestamp, m.id LIMIT 1), '' ) AS _preview_raw, CASE WHEN s.id != tip.id THEN s.id ELSE NULL END AS _lineage_root_id FROM ranked_tips rt JOIN sessions s ON s.id = rt.root_id JOIN sessions tip ON tip.id = rt.cur_id WHERE rt.rank_in_root = 1 AND s.archived = 0 AND s.hidden = 0 AND {_LISTABLE_CHILD_SQL} AND {_delegate_from_json('s.model_config')} IS NULL ORDER BY rt.activity DESC, s.started_at DESC, tip.id DESC LIMIT ? """ params = candidate_params + [ candidate_limit, lineage_limit, lineage_limit, limit, ] deadline = time.monotonic() + timeout_seconds interrupted_by_deadline = False def _deadline_progress_handler() -> int: nonlocal interrupted_by_deadline if time.monotonic() >= deadline: interrupted_by_deadline = True return 1 return 0 try: with self._read_ctx() as conn: conn.set_progress_handler(_deadline_progress_handler, 1000) try: rows = conn.execute(query, params).fetchall() finally: conn.set_progress_handler(None, 0) except sqlite3.OperationalError as exc: if interrupted_by_deadline and "interrupt" in str(exc).lower(): raise TimeoutError( f"recent-session browse exceeded {timeout_seconds:g}s deadline" ) from exc raise sessions = [] for row in rows: session = self._session_row_dict(row) session["preview"] = _shape_preview(session.pop("_preview_raw", "")) session["unread"] = self.session_unread(session) sessions.append(session) return sessions @classmethod def _list_row(cls, row: sqlite3.Row) -> Dict[str, Any]: """Project a list_sessions_rich row: shape the preview, drop internal ordering columns.""" s = cls._session_row_dict(row) s["preview"] = _shape_preview(s.pop("_preview_raw", "")) s.pop("_effective_last_active", None) return s def list_sessions_rich( self, source: str = None, sources: List[str] = None, exclude_sources: List[str] = None, cwd_prefix: str = None, limit: int = 20, offset: int = 0, include_children: bool = False, min_message_count: int = 0, project_compression_tips: bool = True, order_by_last_active: bool = False, include_archived: bool = False, archived_only: bool = False, id_query: str = None, search_query: str = None, compact_rows: bool = False, include_pinned: bool = False, session_key: str = None, include_hidden: bool = False, ) -> List[Dict[str, Any]]: """List sessions with preview and ``last_active`` in one query. ``order_by_last_active`` sorts by the chain TIP via a recursive CTE (the only path honouring ``id_query`` / ``search_query``); ``include_pinned`` back-fills pins the page missed, still obeying the other filters.""" self.flush_token_counts() # rows carry token/cost totals where_clauses, params = _session_filter_where( exclude_children=not include_children, source=source, sources=sources, session_key=session_key, exclude_sources=exclude_sources, cwd_prefix=cwd_prefix, min_message_count=min_message_count, archived_only=archived_only, include_archived=include_archived, ) if not include_hidden: where_clauses.append("s.hidden = 0") where_sql = _where_sql(where_clauses) base_where_params = list(params) # pinned back-fill reuses the WHERE before LIMIT/OFFSET # Shared projection head of the three list queries (whitespace is part of the SQL text). select_head = ( f"SELECT {self._compact_session_cols() if compact_rows else 's.*'}" + ("" if compact_rows else ", COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved") + f",\n {_PREVIEW_COL_SQL},\n " ) prompt_join = ( "" if compact_rows else "LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash" ) from_sessions = f"FROM sessions s\n {prompt_join}" if order_by_last_active: # The CTE walks compression-continuation edges forward from the admitted # rows; MAX over the chain gives effective_last_active in SQL. Do NOT # require child.started_at >= parent.ended_at: races insert the # continuation before ended_at is written. outer_where, id_params = self._chain_search_where( where_sql, (id_query or "").strip().lower(), (search_query or "").strip().lower(), ) query = f""" WITH RECURSIVE chain(root_id, cur_id) AS ( SELECT s.id, s.id FROM sessions s {where_sql} UNION ALL SELECT c.root_id, child.id FROM chain c JOIN sessions parent ON parent.id = c.cur_id JOIN sessions child ON child.parent_session_id = c.cur_id WHERE parent.end_reason = 'compression' AND {_sql_json_extract('child.model_config', '$._branched_from')} IS NULL AND {_sql_json_extract('child.model_config', '$._delegate_from')} IS NULL AND COALESCE(child.source, '') != 'tool' ), chain_max AS ( SELECT root_id, MAX({_sql_session_last_active_by_id("cur_id")}) AS effective_last_active FROM chain GROUP BY root_id ) {select_head}{_sql_session_last_active("s")} AS last_active, COALESCE(cm.effective_last_active, s.started_at) AS _effective_last_active FROM sessions s LEFT JOIN chain_max cm ON cm.root_id = s.id {prompt_join} {outer_where} ORDER BY _effective_last_active DESC, s.started_at DESC, s.id DESC LIMIT ? OFFSET ? """ params = params + params + id_params + [limit, offset] # WHERE binds twice (seed + outer) else: query = f""" {select_head}{_sql_session_last_active("s")} AS last_active {from_sessions} {where_sql} ORDER BY s.started_at DESC LIMIT ? OFFSET ? """ params.extend([limit, offset]) sessions = [self._list_row(row) for row in self._read_all(query, params)] # Pinned back-fill runs BEFORE compression projection so a back-filled root # projects to its tip like any other row. if include_pinned: seen_ids = {s["id"] for s in sessions} pinned_where = f"{where_sql} AND s.pinned = 1" if where_sql else "WHERE s.pinned = 1" pinned_query = f""" {select_head}COALESCE( (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id), s.started_at ) AS last_active {from_sessions} {pinned_where} ORDER BY s.started_at DESC """ for row in self._read_all(pinned_query, base_where_params): s = self._list_row(row) if s["id"] not in seen_ids: seen_ids.add(s["id"]) sessions.append(s) if project_compression_tips and not include_children: sessions = self._project_compression_tips(sessions, compact_rows) # last_read_at is lineage-stamped, so root and tip watermarks agree. for s in sessions: s["unread"] = self.session_unread(s) return sessions def session_lifecycle_statuses(self, session_ids: List[str]) -> Dict[str, str]: """``{session_id: status}`` from each session's LAST message row (``'empty'`` when none); one query, MAX(id) per session joined back — never scans transcripts.""" ids = [sid for sid in (session_ids or []) if sid] if not ids: return {} statuses: Dict[str, str] = {sid: "empty" for sid in ids} rows = self._read_all(f""" SELECT m.session_id, m.role, m.tool_calls IS NOT NULL AS has_tool_calls, m.finish_reason FROM messages m JOIN ( SELECT session_id, MAX(id) AS max_id FROM messages WHERE session_id IN ({_session_ids_placeholders(ids)}) GROUP BY session_id ) latest ON m.id = latest.max_id """, ids) for row in rows: statuses[row["session_id"]] = classify_session_status( role=row["role"], has_tool_calls=bool(row["has_tool_calls"]), finish_reason=row["finish_reason"], ) return statuses def assert_export_safe(self, session_id: str, max_messages: Optional[int] = None) -> int: """Active row count of this segment, or raise SessionExportTooLargeError (the LIMITed subquery stops once the bound is exceeded). ``None`` resolves ``sessions.max_export_messages``; 0 disables the guard.""" from hermes_state import SessionExportTooLargeError, resolved_max_export_messages if max_messages is None: max_messages = resolved_max_export_messages() if max_messages < 0: raise ValueError("max_messages must be non-negative") if max_messages == 0: return 0 row = self._read_one( "SELECT COUNT(*) FROM (SELECT 1 FROM messages WHERE session_id = ? AND active = 1 LIMIT ?)", (session_id, max_messages + 1), ) message_count = int(row[0] if row else 0) if message_count > max_messages: raise SessionExportTooLargeError(session_id, message_count, max_messages) return message_count def _is_explicit_branch_session(self, session_id: str) -> bool: """Copied user-facing branch (``_branched_from``)? Branches own a copied transcript; compression continuations need the parent's archived rows.""" if not session_id: return False row = self._read_one("SELECT model_config FROM sessions WHERE id = ?", (session_id,)) return row is not None and bool(_parse_model_config(row[0]).get("_branched_from")) def _session_lineage_root_to_tip(self, session_id: str) -> List[str]: if not session_id: return [session_id] chain: List[str] = [] current = session_id with self._read_ctx() as conn: while current and current not in chain and len(chain) < 100: chain.append(current) row = conn.execute( "SELECT parent_session_id FROM sessions WHERE id = ?", (current,), ).fetchone() if row is None: break current = row[0] return list(reversed(chain)) or [session_id] def search_sessions( self, source: str = None, limit: int = 20, offset: int = 0, workspace_key: str = None, ) -> List[Dict[str, Any]]: """Sessions MRU-first with a computed ``last_active``; ``workspace_key`` scopes to one workspace so ``hermes -c``/``--resume`` picks its last session.""" where_clauses = [] params: list = [] if source: where_clauses.append("s.source = ?") params.append(source) if workspace_key: ws_clause, ws_params = _workspace_key_clause(workspace_key) where_clauses.append(ws_clause) params.extend(ws_params) return [self._session_row_dict(row) for row in self._read_all( "SELECT s.*, COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved, " f"{_sql_session_last_active('s')} AS last_active " "FROM sessions s LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash " f"{_where_sql(where_clauses, ' ')} " "ORDER BY last_active DESC, s.started_at DESC, s.id DESC LIMIT ? OFFSET ?", [*params, limit, offset], )] def session_count( self, source: str = None, sources: List[str] = None, cwd_prefix: str = None, min_message_count: int = 0, include_archived: bool = False, archived_only: bool = False, exclude_children: bool = False, exclude_sources: List[str] = None, ) -> int: """Count sessions with list_sessions_rich's filters so a paired "load more" total matches.""" where_clauses, params = _session_filter_where( exclude_children=exclude_children, source=source, sources=sources, exclude_sources=exclude_sources, cwd_prefix=cwd_prefix, min_message_count=min_message_count, archived_only=archived_only, include_archived=include_archived, ) return self._read_one(f"SELECT COUNT(*) FROM sessions s{_where_sql(where_clauses, ' ')}", params)[0] def session_count_ge(self, n: int = 1) -> bool: """At least N sessions exist (archived included); LIMIT short-circuits session_count()'s scan.""" return len(self._read_all("SELECT 1 FROM sessions LIMIT ?", (n,))) >= n def session_count_by_source( self, *, include_archived: bool = False, archived_only: bool = False, exclude_children: bool = False, ) -> Dict[str, int]: """``{source: count}`` via one GROUP BY; ``exclude_children`` mirrors listing visibility.""" where_clauses, params = _session_filter_where( exclude_children=exclude_children, archived_only=archived_only, include_archived=include_archived, ) with self._read_ctx() as conn: if self._conn is None: raise RuntimeError("SessionDB connection is closed") rows = conn.execute( "SELECT COALESCE(NULLIF(s.source, ''), 'cli') AS source, COUNT(*) AS count " f"FROM sessions s{_where_sql(where_clauses, ' ')} " "GROUP BY COALESCE(NULLIF(s.source, ''), 'cli') ORDER BY count DESC", params, ).fetchall() return {str(row["source"]): int(row["count"] or 0) for row in rows} def declared_scope_identity(self, session_id: str) -> Tuple[bool, str]: """(is_fork_child, source) in ONE read (prompt_cache_scope needs both from the same row). Missing row → (False, ""); DB errors propagate (fail closed). ``agent/prompt_cache_scope.py`` needs both to resolve a host-declared conversation scope, and both live on the same ``sessions`` row; asking for them separately read that row twice per resolution (@teknium1 on 98811). The marker rules stay here, beside :meth:`is_explicit_fork_child`, instead of being re-implemented by the caller. See #98811. """ session = self.get_session(session_id) if not session: return False, "" return self._is_explicit_fork_child_row(session), str(session.get("source") or "").strip() @staticmethod def _remove_session_files(sessions_dir: Optional[Path], session_id: str) -> None: """Remove ``.json``/``.jsonl`` and gateway ``request_dump__*.json``; OSError is swallowed so a filesystem hiccup never blocks a DB operation.""" if sessions_dir is None: return targets = [sessions_dir / f"{session_id}{suffix}" for suffix in (".json", ".jsonl")] try: targets.extend(sessions_dir.glob(f"request_dump_{session_id}_*.json")) except OSError: pass for p in targets: try: p.unlink(missing_ok=True) except OSError: pass def get_session_delete_targets(self, session_id: str) -> List[str]: """Rows :meth:`delete_session` would remove: the session, then its recursive delegate children (branch/compression children are orphaned, not deleted).""" with self._read_ctx() as conn: if not conn.execute("SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,)).fetchone(): return [] # Use the borrowed read connection, never self._conn: handing the shared writer connection to a # helper here executes on it without self._lock — the same unsynchronized-read class as # #99349/#90734. delegate_ids = _collect_delegate_child_ids(conn, [session_id]) return [session_id, *sorted(delegate_ids)] def delete_session( self, session_id: str, sessions_dir: Optional[Path] = None, expected_delete_ids: Optional[List[str]] = None, ) -> bool: """Delete a session and its messages; delegate children cascade, branch/compression children are orphaned. *expected_delete_ids*: proceed only if parent + delegate cascade still equals that set (re-walked inside the transaction on purpose: export-before-delete fails closed).""" removed_ids: List[str] = [] expected_ids = set(expected_delete_ids) if expected_delete_ids is not None else None def _do(conn): if conn.execute("SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,)).fetchone() is None: return False if expected_ids is not None and expected_ids != { session_id, *_collect_delegate_child_ids(conn, [session_id]) }: return False removed_ids.extend(_delete_delegate_children(conn, [session_id])) conn.execute( # orphan remaining children (branches) so FK is satisfied "UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id = ?", (session_id,), ) conn.execute("DELETE FROM messages WHERE session_id = ?", (session_id,)) conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,)) self._delete_unreferenced_system_prompts(conn) removed_ids.append(session_id) return True deleted = self._execute_write(_do) for sid in removed_ids: self._remove_session_files(sessions_dir, sid) return bool(deleted) def delete_session_if_empty(self, session_id: str, sessions_dir: Optional[Path] = None) -> bool: """Delete *session_id* only if it has no messages, no title and no children; check and delete share one transaction so a concurrent flush can't be lost.""" def _do(conn): cursor = conn.execute( """ DELETE FROM sessions WHERE id = ? AND title IS NULL AND NOT EXISTS ( SELECT 1 FROM messages WHERE messages.session_id = sessions.id ) AND NOT EXISTS ( SELECT 1 FROM sessions child WHERE child.parent_session_id = sessions.id ) """, (session_id,), ) if cursor.rowcount > 0: self._delete_unreferenced_system_prompts(conn) return cursor.rowcount > 0 deleted = self._execute_write(_do) if deleted: self._remove_session_files(sessions_dir, session_id) return deleted def delete_sessions(self, session_ids: List[str], sessions_dir: Optional[Path] = None) -> int: """Bulk delete with :meth:`delete_session` semantics per row, in ONE transaction. Unknown ids are skipped (UI selection can race another tab's delete). Returns the number deleted.""" unique_ids = list({sid for sid in session_ids or () if isinstance(sid, str) and sid}) if not unique_ids: return 0 removed_ids: list[str] = [] def _do(conn): existing = [row["id"] for chunk in _id_chunks(unique_ids) for row in conn.execute( f"SELECT id FROM sessions WHERE id IN ({_session_ids_placeholders(chunk)})", chunk, ).fetchall()] if not existing: return 0 removed_ids.extend(_delete_delegate_children(conn, existing)) for chunk in _id_chunks(existing): ph = _session_ids_placeholders(chunk) conn.execute( # orphan children whose parent is in the kill list (FK) f"UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id IN ({ph})", chunk, ) conn.execute(f"DELETE FROM messages WHERE session_id IN ({ph})", chunk) conn.execute(f"DELETE FROM sessions WHERE id IN ({ph})", chunk) self._delete_unreferenced_system_prompts(conn) removed_ids.extend(existing) return len(existing) count = self._execute_write(_do) for sid in removed_ids: self._remove_session_files(sessions_dir, sid) return count # Shared by count_empty_sessions / delete_empty_sessions so badge and sweep agree. message_count # counts live rows only (rewind/compaction keep dropped turns as active = 0): NOT EXISTS is authority. # The ``NOT EXISTS`` probe is the authority; : ``message_count = 0`` stays as a cheap prefilter. Same # shape as every : other emptiness guard in this module. (#95868) _EMPTY_SESSION_WHERE = ( "message_count = 0 AND ended_at IS NOT NULL AND archived = 0 AND NOT EXISTS (" "SELECT 1 FROM messages WHERE messages.session_id = sessions.id)" ) def count_empty_sessions(self) -> int: """Count of empty, ended, non-archived sessions; ended_at guards a fresh session's first message.""" return self._read_one(f"SELECT COUNT(*) FROM sessions WHERE {self._EMPTY_SESSION_WHERE}")[0] def delete_empty_sessions(self, sessions_dir: Optional[Path] = None) -> int: """Delete every empty, ended, non-archived session in one transaction, orphaning (not cascading) children; transcript files are swept too.""" removed_ids: list[str] = [] def _do(conn): session_ids = {row["id"] for row in conn.execute( f"SELECT id FROM sessions WHERE {self._EMPTY_SESSION_WHERE}" ).fetchall()} if not session_ids: return 0 for chunk in _id_chunks(session_ids): ph = _session_ids_placeholders(chunk) conn.execute(f"UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id IN ({ph})", chunk) # DELETE FROM messages: a row inserted between the SELECT and here # would otherwise dangle (clean FK state). conn.execute(f"DELETE FROM messages WHERE session_id IN ({ph})", chunk) conn.execute(f"DELETE FROM sessions WHERE id IN ({ph})", chunk) removed_ids.extend(chunk) self._delete_unreferenced_system_prompts(conn) return len(session_ids) count = self._execute_write(_do) for sid in removed_ids: self._remove_session_files(sessions_dir, sid) return count def archive_sessions( self, older_than_days: Optional[float] = None, source: str = None, **filters, ) -> int: """Bulk soft-hide with prune_sessions' filter surface, via set_session_archived so each lineage flips as a unit; idempotent. Returns matches.""" filters.setdefault("archived", False) rows = self.list_prune_candidates(older_than_days=older_than_days, source=source, **filters) for row in rows: self.set_session_archived(row["id"], True) return len(rows) def maybe_auto_archive( self, idle_days: float = 3, min_interval_hours: int = 24, exclude_pinned: bool = True, ) -> Dict[str, Any]: """Idempotent, non-destructive auto-archive of sessions idle for ``idle_days``; state_meta ``last_auto_archive`` gates runs within ``min_interval_hours``. Never raises.""" result: Dict[str, Any] = {"skipped": False, "archived": 0} try: now = time.time() try: last = float(self.get_meta("last_auto_archive") or 0.0) except (TypeError, ValueError): last = 0.0 # corrupt meta; treat as no prior run if last and now - last < min_interval_hours * 3600: result["skipped"] = True return result # Startup-watchdog lease: the archive sweep is I/O-bound (near-zero CPU), # which the watchdog's CPU fallback misreads as a parked deadlock. # No-op when the watchdog is not armed; never raises. report_startup_progress(900.0, phase="state_db_auto_archive") archived = result["archived"] = self.archive_stale_sessions(idle_days, exclude_pinned=exclude_pinned) # Record even a zero-archive run so we don't re-sweep every call. self.set_meta("last_auto_archive", str(now)) if archived > 0: logger.info( "state.db auto-archive: archived %d session(s) idle >= %s days", archived, idle_days, ) except Exception as exc: logger.warning("state.db auto-archive failed: %s", exc) result["error"] = str(exc) return result