hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity
Files
main
hermes-agent/hermes_state_sessions.py
kshitijk4poor 230f89b47b fix(sessions): share the write-guard filter and trim repeated walks 
prune_sessions hand-rolled the same "guarded by a live lease/lock"
comprehension as the new _guarded_ids helper, so the two could drift on
the next guard change. Move _guarded_ids next to _write_guards_reject in
the maintenance mixin and have prune call it.

delete_session walked the delegate tree up to three times in one write
transaction; compute the target ids once for both the guard check and the
expected-ids fence. delete_sessions ran a per-root guard walk for every
selected id; do one batched check over all roots and their children first
and only attribute per root when something is actually guarded.

The CLI export --delete message repeated "session 'X'" because the
exception text already names the session.
2026-09-27 20:49:04 +05:30

1767 lines
91 KiB
Python
Raw Permalink Blame History

"""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 glob
import json
import logging
import re
import sqlite3
import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
from agent.session_activity import (
ActivityProvenance, bound_activity_description, normalize_activity_provenance,
)
from hermes_startup_watchdog import report_startup_progress
from hermes_state_errors import SessionActiveWriteGuardError
from hermes_state_common import (
_LISTABLE_CHILD_SQL, _PREVIEW_ELIGIBLE_SQL, _PREVIEW_RAW_SELECT, _RECOVERABLE_END_REASONS,
_RECOVERABLE_END_REASONS_SQL, _RESET_CHILD_SQL, _RESET_END_REASONS, _legacy_reset_child_sql, _non_continuation_child_sql,
_shape_preview,
_sql_in_window, _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,
include_subagents: 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``); ``include_subagents`` re-admits the sub-agent runs only
(``sessions.show_subagents``). Clause order is part of the SQL text contract."""
where: List[str] = []
params: List[Any] = []
if exclude_children and include_subagents:
where.append(f"({_LISTABLE_CHILD_SQL} OR {_delegate_from_json('s.model_config')} IS NOT NULL)")
elif 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.
# Sessions that are not human conversations (kanban workers, third-party tool integrations, finite one-shot
# runs): every human picker — TUI/Desktop session lists, ``/sessions`` in the CLI, ``sessions list`` in the
# console — excludes them. A deny-list, so new interactive platforms surface automatically.
INTERNAL_LISTING_SOURCES = ("kanban", "tool", "oneshot")
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:<ns>:...`` keyed rows must agree on the namespace.
# ``agent:<ns>:...`` gateway keys encode the profile namespace; a keyless row (CLI / subagent lineage)
# carries none and inherits freely. Two keyed rows must agree on ``agent:<ns>:`` — 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",
"transport_profile", "git_repo_root", "origin_json", "display_name",
# Immutable provenance (#56439): stamped once at first creation, never overwritten by later
# upserts — unlike ``source``, which stays live routing state.
"created_source",
)
)
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"
)
# A delegate/branch fork of a row that happens to have ended on compression is still not that
# conversation's continuation. Markers are matched against the QUERIED parent id rather than mere
# presence, for the same reason as _NON_CONTINUATION_CHILD_FILTER_SQL: a continuation inherits its
# parent's model_config verbatim, so presence-matching would misclassify it as a delegate.
_FORK_EDGE_EXCLUSION_SQL = "".join(
f"\n AND COALESCE({_sql_json_extract('model_config', f'$.{marker}')}, '')"
"\n != parent_session_id"
for marker in ("_delegate_from", "_branched_from")
)
_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",
"transport_profile",
))
+ "\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"
" )"
+ _FORK_EDGE_EXCLUSION_SQL
)
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 (``<root>/state.db`` → default,
``<root>/profiles/<name>/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:<ns>:`` 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
and branch children must NOT inherit them (peer recovery could repoint traffic into a
subagent's session), including when their parent row itself ended on compression — a long
batch outlives its coordinator's rotation, and two live rows holding one routing key is the
shape reported in #92859."""
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,
transport_profile: Optional[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:<profile>/<id>``
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, created_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, transport_profile, 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, 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, transport_profile, 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 (see _non_continuation_child_sql); presence-
# matching misclassified inherited markers as delegates. Callers bind the parent id three times.
_NON_CONTINUATION_CHILD_FILTER_SQL = _non_continuation_child_sql("{alias}")
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, pin: Any) -> None:
"""Persist the session's ``tools[]`` pin (JSON-serializable) so a rebuilt AIAgent sends the
same bytes; ``None`` clears. The array repeats across sessions like a system prompt does, so it
is stored in the same content-addressed ``system_prompts`` table and the column holds its hash
(legacy rows: an inline JSON name list); ``get_session`` resolves either."""
payload = json.dumps(pin) if pin is not None else None
def _do(conn):
conn.execute("UPDATE sessions SET tool_names = ? WHERE id = ?",
(self._store_system_prompt(conn, payload), session_id))
self._delete_unreferenced_system_prompts(conn)
self._execute_write(_do)
def update_session_model(
self, session_id: str, model: str, provider: Optional[str] = None, *,
base_url: Optional[str] = None, api_mode: Optional[str] = None,
) -> None:
"""Set the model after a mid-session /model switch (unconditionally) and drop any Browser
runtime lock (lineage markers survive).
Route writers never touch the stored prompt; ``_stored_prompt_matches_runtime`` decides staleness.
When *provider* is given the whole route is written, in both shapes resume reads (top-level
keys for the TUI/Desktop, ``gateway_runtime`` for the CLI), so a later resume recombines the
model with the provider that serves it (#79536). ``base_url``/``api_mode`` are always
replaced then (``None`` deletes): the previous provider's endpoint must not survive a switch,
or resume sends the new provider's model to the old host. Callers without provider knowledge
leave the stored route 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:
route = {"provider": provider, "base_url": base_url or None, "api_mode": api_mode or None}
patch.update(route, gateway_runtime=route)
self._write_model_config_patch(
session_id, patch, "UPDATE sessions SET model = ?, model_config = ? 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."""
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))
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).
Route writers never touch the stored prompt; ``_stored_prompt_matches_runtime`` decides staleness."""
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)
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, "
"COALESCE(tp.prompt, s.tool_names) AS _tool_names_resolved "
"FROM sessions s LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash "
"LEFT JOIN system_prompts tp ON tp.hash = s.tool_names 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 backfill_acp_session_cwd(self) -> int:
"""Promote ``model_config.cwd`` into the cwd column for ACP rows lacking one.
ACP sessions minted before the adapter populated the column still carry
their workspace inside ``model_config``, written by the same adapter that
knew the real directory — so this is a record being promoted, not a guess.
Only fills NULL/empty; an explicit column value always wins. Returns the
number of rows changed.
"""
return int(self._write_rowcount(
"""UPDATE sessions
SET cwd = json_extract(model_config, '$.cwd')
WHERE source = 'acp'
AND COALESCE(cwd, '') = ''
AND json_valid(model_config)
AND COALESCE(json_extract(model_config, '$.cwd'), '') != ''""",
(),
) 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
# #121148: the projected row IS an automatic continuation, not a fresh conversation and
# not a user branch. The server already classifies the edge (_COMPRESSION_CHILD_SQL) but
# only ever used it to hide rows — surface the kind so clients can label the provenance
# instead of rendering a sealed-and-rotated chat as a brand-new session.
merged["continuation_kind"] = "compression"
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,
include_subagents: 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 except archived: a pin is an explicit keep, so a pinned row stamped
archived must still return."""
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, include_subagents=include_subagents,
)
# The archived-only view is the recovery surface for rows that dropped out of every
# default list: a session that is archived AND hidden (Bot Mode marks its sessions
# hidden) must still be reachable there, or nothing but direct DB access can bring
# it back (#90946).
if not include_hidden and not archived_only:
where_clauses.append("s.hidden = 0")
where_sql = _where_sql(where_clauses)
# 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 NOT ({_RESET_CHILD_SQL.format(a='child')})
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, {_sql_in_window("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. Do not inherit the archived
# constraint: the sidebar lists with include_archived=False, and a pin
# must stay reachable even when that row is also archived.
if include_pinned:
seen_ids = {s["id"] for s in sessions}
pinned_clauses, pinned_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=False, include_archived=True,
include_subagents=include_subagents,
)
if not include_hidden and not archived_only:
pinned_clauses.append("s.hidden = 0")
pinned_clauses.append("s.pinned = 1")
pinned_where = _where_sql(pinned_clauses)
pinned_query = f"""
{select_head}{_sql_session_last_active("s")} AS last_active
{from_sessions}
{pinned_where}
ORDER BY s.started_at DESC
"""
for row in self._read_all(pinned_query, pinned_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:
"""Row count of this segment — every row, archived included, as the transfer export materializes
it — 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 = ? 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: Union[str, Sequence[str], None] = 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. ``source`` may be one label or several."""
where_clauses = []
params: list = []
if source:
sources = [source] if isinstance(source, str) else list(source)
where_clauses.append(f"s.source IN ({','.join('?' * len(sources))})")
params.extend(sources)
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, include_subagents: bool = False,
) -> 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, include_subagents=include_subagents,
)
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 ``<id>.json``/``.jsonl``, the legacy ``session_<id>.json`` snapshot, and gateway
``request_dump_<id>_*.json``; OSError is swallowed so a filesystem hiccup never blocks a
DB operation. Every historical writer name is swept because a "deleted" session's snapshot
can carry plaintext secrets (#20334, #60207)."""
if sessions_dir is None:
return
targets = [sessions_dir / f"{session_id}{suffix}" for suffix in (".json", ".jsonl")]
targets.append(sessions_dir / f"session_{session_id}.json")
try:
# glob.escape: a session id carrying ``[`` / ``?`` / ``*`` is a PATTERN otherwise, so the
# dump sweep either matches nothing or matches another session's files.
targets.extend(sessions_dir.glob(f"request_dump_{glob.escape(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,
expected_display_messages: Optional[Dict[str, List[Dict[str, Any]]]] = None,
exclude_active_write_guards: bool = False,
) -> bool:
"""Delete a session and its messages; delegate children cascade, branch/compression children
are orphaned. Optional expected ids fence delegate drift; expected display snapshots fence
transcript drift. Both checks run inside the same write transaction as deletion.
With ``exclude_active_write_guards``, raises :class:`SessionActiveWriteGuardError` if the row
is protected by an active turn lease or compression lock."""
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
target_ids = (
[session_id, *_collect_delegate_child_ids(conn, [session_id])]
if exclude_active_write_guards or expected_ids is not None else None
)
if exclude_active_write_guards and self._guarded_ids(conn, target_ids):
# Delegate children cascade with the root, so a guard on any of them refuses too.
raise SessionActiveWriteGuardError(
f"session '{session_id}' (or a delegate child) has an active turn lease or compression lock"
)
if expected_ids is not None and expected_ids != set(target_ids):
return False
if expected_display_messages is not None and any(
self._display_messages_from_conn(conn, covered_id) != expected
for covered_id, expected in expected_display_messages.items()
):
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,
exclude_active_write_guards: bool = False, skipped_ids: Optional[List[str]] = 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). With ``exclude_active_write_guards``,
rows protected by an active turn lease or compression lock are skipped and, when given, appended
to ``skipped_ids`` so callers can tell the user. 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
if exclude_active_write_guards:
# A root is skipped when it or any delegate child it would cascade is guarded, so the
# cascade below never deletes a guarded row reported back as kept.
# One batched check first; per-root attribution only when something is guarded.
active_ids: set = set()
if self._guarded_ids(conn, [*existing, *_collect_delegate_child_ids(conn, existing)]):
active_ids = {
sid for sid in existing
if self._guarded_ids(conn, [sid, *_collect_delegate_child_ids(conn, [sid])])
}
existing = [sid for sid in existing if sid not in active_ids]
if skipped_ids is not None:
skipped_ids.extend(sorted(active_ids))
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. A lineage is matched through its TIP only: an
old compression ancestor never qualifies on its own age, or the fan-out would hide an open,
recently active continuation (#115489)."""
filters.setdefault("archived", False)
filters["lineage_tips_only"] = True
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
Reference in New Issue View Git Blame Copy Permalink