Teknium
1588 lines
78 KiB
Python
1588 lines
78 KiB
Python
"""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
|
|
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_session_last_active, _sql_session_last_active_by_id, escape_like as _escape_like,
|
|
)
|
|
|
|
# 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 when known, else cwd, else None.
|
|
Branch is deliberately excluded so a checkout doesn't 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 f"json_extract(COALESCE({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:
|
|
parsed = json.loads(raw)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return {}
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
if isinstance(raw, dict):
|
|
return dict(raw)
|
|
return {}
|
|
|
|
|
|
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 (``my_project``):
|
|
# unescaped, a prefix also matches sibling directories. The ``=`` arm is an
|
|
# exact compare and keeps the raw prefix; the Windows separator backslash
|
|
# in the LIKE pattern needs escaping too.
|
|
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``. Used by
|
|
``hermes -c``/``--resume`` to pick the current workspace's MRU, not the global one."""
|
|
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 _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 session counts line up with the
|
|
listed rows. ``exclude_children`` hides sub-agent runs and compression
|
|
continuations but keeps branch/reset children: ``_LISTABLE_CHILD_SQL`` uses
|
|
the stable ``_branched_from`` marker (survives a re-ended parent) OR'd with
|
|
the legacy parent-ended-'branched' heuristic for pre-marker rows. Clause
|
|
order is part of the SQL text contract."""
|
|
where: List[str] = []
|
|
params: List[Any] = []
|
|
if exclude_children:
|
|
where.append(_LISTABLE_CHILD_SQL)
|
|
where.append(f"{_delegate_from_json('s.model_config')} IS NULL")
|
|
include_sources = [source] if source else list(sources or [])
|
|
if include_sources:
|
|
where.append(f"s.source IN ({','.join('?' for _ in include_sources)})")
|
|
params.extend(include_sources)
|
|
if session_key:
|
|
where.append("s.session_key = ?")
|
|
params.append(session_key)
|
|
if exclude_sources:
|
|
where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})")
|
|
params.extend(exclude_sources)
|
|
if cwd_prefix:
|
|
clause, clause_params = _cwd_prefix_clause(cwd_prefix)
|
|
where.append(clause)
|
|
params.extend(clause_params)
|
|
if min_message_count > 0:
|
|
where.append("s.message_count >= ?")
|
|
params.append(min_message_count)
|
|
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) to cascade-delete with
|
|
*parent_ids*; untagged children keep the orphan-don't-delete contract.
|
|
Walks marker chains recursively so an orchestrator's own delegates go too."""
|
|
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
|
|
# (cycle, or a parent that is another parent's delegate child in one batch)
|
|
# and it would be collected as its own descendant and cascade-deleted.
|
|
# Callers delete parents separately; never return them as children.
|
|
found: set[str] = set(seeds)
|
|
frontier = list(seeds)
|
|
while frontier:
|
|
ph = ",".join("?" * len(frontier))
|
|
cursor = conn.execute(
|
|
f"SELECT id FROM sessions WHERE {df} IN ({ph}) "
|
|
f"OR (parent_session_id IN ({ph}) AND {df} IS NOT NULL)",
|
|
frontier + frontier,
|
|
)
|
|
frontier = [row["id"] for row in cursor.fetchall() if row["id"] not in found]
|
|
found.update(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)
|
|
if ids:
|
|
ph = ",".join("?" * len(ids))
|
|
conn.execute(f"DELETE FROM messages WHERE session_id IN ({ph})", ids)
|
|
# 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})", ids,
|
|
)
|
|
conn.execute(f"DELETE FROM sessions WHERE id IN ({ph})", ids)
|
|
return ids
|
|
|
|
|
|
# Lifecycle statuses surfaced by session pickers; classified from the final
|
|
# message row ONLY (role, tool_calls, finish_reason) 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:
|
|
"""Lifecycle from the final message: error finish → ``error``; assistant
|
|
with pending tool_calls (result never landed), or a trailing user/tool row →
|
|
``interrupted``; normal assistant finish or unknown shape → ``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 == "assistant":
|
|
return SESSION_STATUS_INTERRUPTED if has_tool_calls else SESSION_STATUS_COMPLETE
|
|
if r in {"user", "tool"}:
|
|
return SESSION_STATUS_INTERRUPTED
|
|
return SESSION_STATUS_COMPLETE
|
|
|
|
|
|
# Parent→child profile_name inheritance fence: keyless rows (CLI / subagent)
|
|
# inherit freely; two ``agent:<ns>:...`` keyed rows must agree on the namespace
|
|
# so a default child forked from a sibling profile's row isn't mislabelled.
|
|
_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)"
|
|
)
|
|
|
|
|
|
class SessionSessionsMixin:
|
|
"""Session rows: create/inherit, lifecycle flags, model_config, listing, deletion."""
|
|
|
|
_PROFILE_DIR_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
|
|
|
|
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). Path-based, not
|
|
get_active_profile_name(): a gateway serving a NON-launch profile opens
|
|
that profile's store and rows must carry the store's owner. None outside
|
|
the profile tree — keep NULL rather than 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"
|
|
if parent.parent == root / "profiles" and self._PROFILE_DIR_RE.match(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 (child creators
|
|
didn't propagate them, so lineages dropped out of the project sidebar);
|
|
profile_name only within the same ``agent:<ns>:`` namespace. The second
|
|
UPDATE inherits gateway routing columns ONLY for compression forks: a
|
|
crash before the gateway re-records the peer would otherwise strand the
|
|
child unroutable, while delegate children are spawned under a live
|
|
parent and must NOT inherit routing keys (peer recovery could repoint
|
|
gateway traffic into a subagent's session)."""
|
|
conn.execute(
|
|
f"""UPDATE sessions
|
|
SET cwd = COALESCE(sessions.cwd,
|
|
(SELECT p.cwd FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
git_repo_root = COALESCE(sessions.git_repo_root,
|
|
(SELECT p.git_repo_root FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
git_branch = COALESCE(sessions.git_branch,
|
|
(SELECT p.git_branch FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
profile_name = COALESCE(sessions.profile_name,
|
|
(SELECT p.profile_name FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id
|
|
AND ({_SAME_KEY_NAMESPACE_SQL})))
|
|
WHERE id = ? AND parent_session_id IS NOT NULL""",
|
|
(session_id,),
|
|
)
|
|
conn.execute(
|
|
"""UPDATE sessions
|
|
SET user_id = COALESCE(sessions.user_id,
|
|
(SELECT p.user_id FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
session_key = COALESCE(sessions.session_key,
|
|
(SELECT p.session_key FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
chat_id = COALESCE(sessions.chat_id,
|
|
(SELECT p.chat_id FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
chat_type = COALESCE(sessions.chat_type,
|
|
(SELECT p.chat_type FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
thread_id = COALESCE(sessions.thread_id,
|
|
(SELECT p.thread_id FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
display_name = COALESCE(sessions.display_name,
|
|
(SELECT p.display_name FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id)),
|
|
origin_json = COALESCE(sessions.origin_json,
|
|
(SELECT p.origin_json FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id))
|
|
WHERE id = ? AND parent_session_id IS NOT NULL
|
|
AND EXISTS (
|
|
SELECT 1 FROM sessions p
|
|
WHERE p.id = sessions.parent_session_id
|
|
AND p.end_reason = 'compression'
|
|
)""",
|
|
(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, COALESCE-filling NULL columns and never
|
|
overwriting what an earlier writer set (the gateway creates a bare row
|
|
before the agent's create_session carries the real model/prompt; a later
|
|
bare source="unknown" cannot clobber it). chat_id/thread_id scope gateway
|
|
/resume (IDOR). Children backfill from the parent
|
|
(:meth:`_inherit_parent_session_metadata`); a missing profile_name is
|
|
stamped with THIS store's own profile (NULL reads as unowned)."""
|
|
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
|
|
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,
|
|
session_key = COALESCE(sessions.session_key, excluded.session_key),
|
|
chat_id = COALESCE(sessions.chat_id, excluded.chat_id),
|
|
chat_type = COALESCE(sessions.chat_type, excluded.chat_type),
|
|
thread_id = COALESCE(sessions.thread_id, excluded.thread_id),
|
|
parent_session_id = COALESCE(sessions.parent_session_id, excluded.parent_session_id),
|
|
cwd = COALESCE(sessions.cwd, excluded.cwd),
|
|
profile_name = COALESCE(sessions.profile_name, excluded.profile_name),
|
|
git_repo_root = COALESCE(sessions.git_repo_root, excluded.git_repo_root),
|
|
origin_json = COALESCE(sessions.origin_json, excluded.origin_json),
|
|
display_name = COALESCE(sessions.display_name, excluded.display_name)""",
|
|
(
|
|
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. Ride out long holds.
|
|
self._execute_write(_do, patience_s=self._TRANSCRIPT_WRITE_PATIENCE_S)
|
|
|
|
def create_session(self, session_id: str, source: str, **kwargs) -> str:
|
|
"""Create a new session record. Returns the session_id."""
|
|
self._insert_session_row(session_id, source, **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."""
|
|
if not session_id:
|
|
return
|
|
self._write_sql(
|
|
"UPDATE sessions SET expiry_finalized = ? WHERE id = ?",
|
|
(1 if finalized else 0, session_id),
|
|
)
|
|
|
|
# ── Gateway routing index (replaces sessions.json) ────
|
|
|
|
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; if several distinct users share
|
|
the chat and none matches, None rather than contaminating 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))
|
|
query += " ORDER BY started_at DESC"
|
|
rows = [dict(r) for r in self._read_all(query, 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(rows) > 1:
|
|
distinct_users = {u for u in (str(r.get("user_id") or "").strip() for r in rows) if u}
|
|
if len(distinct_users) > 1:
|
|
return None
|
|
return str(rows[0]["id"])
|
|
|
|
# ── Orphaned gateway-session repair (``hermes sessions repair-routing``) ──
|
|
# A write-path failure between routing publication and row creation leaves
|
|
# the live transcript in a row without identity columns, invisible to
|
|
# recovery (the chat resolves to a days-older keyed row). Widest plausible
|
|
# gap between a keyed predecessor going quiet and its unkeyed successor:
|
|
# the reported incident was ~60s; 15 minutes stays generous without
|
|
# spanning unrelated conversations.
|
|
_ORPHAN_ADOPTION_MAX_GAP_S = 900.0
|
|
|
|
# Children with a ``parent_session_id`` that are NOT compression
|
|
# continuations (branches, delegate runs, tool sessions). Markers are bound
|
|
# to the queried parent id: compression continuations inherit the rotated
|
|
# agent's model_config verbatim, so a delegate's continuation carries
|
|
# ``_delegate_from=<the delegate's own parent>`` and presence-matching
|
|
# misclassified real continuations as delegate children.
|
|
_NON_CONTINUATION_CHILD_FILTER_SQL = (
|
|
" AND COALESCE(json_extract(COALESCE({alias}model_config, '{{}}'),"
|
|
" '$._branched_from'), '') != ?\n"
|
|
" AND COALESCE(json_extract(COALESCE({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 (no-op when already
|
|
ended): a compression split must keep ``'compression'`` even if a stale
|
|
desynced-CLI end_session() targets it later. reopen_session() first to
|
|
deliberately re-end with a new reason."""
|
|
def _do(conn):
|
|
changed = conn.execute(
|
|
"UPDATE sessions SET ended_at = ?, end_reason = ? "
|
|
"WHERE id = ? AND ended_at IS NULL",
|
|
(time.time(), end_reason, session_id),
|
|
).rowcount
|
|
# Only a boundary this call wrote advances the generation (a no-op must not rotate the peer).
|
|
if changed:
|
|
self._bump_conversation_generation(conn, session_id, end_reason)
|
|
self._execute_write(_do)
|
|
|
|
def reopen_session(self, session_id: str) -> None:
|
|
"""Clear ended_at/end_reason so a session can be resumed. First stamp
|
|
markerless legacy reset children that depend on the parent's mutable
|
|
end_reason (WHERE shared with the listing predicate via
|
|
_legacy_reset_child_sql so the two cannot drift)."""
|
|
def _do(conn):
|
|
placeholders = ",".join("?" for _ in _RESET_END_REASONS)
|
|
conn.execute(
|
|
"UPDATE sessions AS child SET model_config = json_set("
|
|
"COALESCE(child.model_config, '{}'), '$._reset_from', child.parent_session_id) "
|
|
"WHERE child.parent_session_id = ? "
|
|
"AND json_extract(COALESCE(child.model_config, '{}'), "
|
|
" '$._reset_from') IS NULL "
|
|
f"AND {_legacy_reset_child_sql('child', placeholders)}",
|
|
(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
|
|
(first writer wins). Plain end_session() no-ops on an ended row, so an
|
|
``agent_close`` row would stay recoverable and stale-route recovery would
|
|
resurrect the reset session. Keep in sync with
|
|
find_latest_gateway_session_for_peer. True when promoted."""
|
|
if not session_id:
|
|
return False
|
|
now = time.time()
|
|
def _do(conn):
|
|
cursor = conn.execute(
|
|
"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),
|
|
)
|
|
# /new and policy auto-resets promote rather than end_session, so the
|
|
# generation advances here too — same transaction, only when written.
|
|
if cursor.rowcount:
|
|
self._bump_conversation_generation(conn, session_id, reason)
|
|
return cursor.rowcount
|
|
try:
|
|
return bool(self._execute_write(_do))
|
|
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 captured value) except under ``replace_git_meta`` (a
|
|
workspace MOVE must overwrite the old repo identity even when the new
|
|
cwd has none). Each call bumps ``git_metadata_generation``; async probes
|
|
publish via :meth:`publish_session_git_metadata` with that 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
|
|
current_cwd = current[0]
|
|
sets = ["cwd = ?", "git_metadata_generation = COALESCE(git_metadata_generation, 0) + 1"]
|
|
params: List[Any] = [cwd]
|
|
if current_cwd != cwd or replace_git_meta:
|
|
sets.extend(("git_branch = ?", "git_repo_root = ?"))
|
|
params.extend((branch or None, repo_root or None))
|
|
elif branch:
|
|
sets.append("git_branch = ?")
|
|
params.append(branch)
|
|
if repo_root and current_cwd == cwd and not replace_git_meta:
|
|
sets.append("git_repo_root = ?")
|
|
params.append(repo_root)
|
|
params.append(session_id)
|
|
conn.execute(f"UPDATE sessions SET {', '.join(sets)} WHERE id = ?", params)
|
|
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."""
|
|
if (
|
|
not session_id
|
|
or not cwd
|
|
or isinstance(generation, bool)
|
|
or not isinstance(generation, int)
|
|
or generation < 1
|
|
):
|
|
return False
|
|
branch = (git_branch or "").strip()
|
|
repo_root = (git_repo_root or "").strip()
|
|
if not branch and not repo_root:
|
|
return False
|
|
sets: List[str] = []
|
|
params: List[Any] = []
|
|
if branch:
|
|
sets.append("git_branch = ?")
|
|
params.append(branch)
|
|
if repo_root:
|
|
sets.append("git_repo_root = ?")
|
|
params.append(repo_root)
|
|
params.extend((session_id, cwd, generation))
|
|
return self._write_rowcount(
|
|
f"UPDATE sessions SET {', '.join(sets)} "
|
|
"WHERE id = ? AND cwd = ? AND git_metadata_generation = ?",
|
|
params,
|
|
) == 1
|
|
|
|
def backfill_repo_roots(self, cwd_to_root: Dict[str, str]) -> None:
|
|
"""Backfill git repo roots for cwds without one (pre-column sessions);
|
|
never clobbers a recorded root; empty roots are skipped."""
|
|
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,
|
|
)
|
|
|
|
# Compression locks (atomic per-session, keyed by session_id, recovered via
|
|
# expires_at) live in SessionCompressionMixin; they stop two AIAgents that
|
|
# share a session_id from both rotating it into two orphan children.
|
|
|
|
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
|
|
AIAgent._touch_activity) so surfaces see API/tool/compaction activity
|
|
before any message row lands. Never moves ``last_activity_at`` backwards."""
|
|
if not session_id:
|
|
return
|
|
from agent.session_activity import (
|
|
bound_activity_description, normalize_activity_provenance,
|
|
)
|
|
when = float(ts if ts is not None else time.time())
|
|
desc = bound_activity_description(description)
|
|
prov = normalize_activity_provenance(provenance).value
|
|
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, desc, prov, session_id, when),
|
|
patience_s=self._ACTIVITY_WRITE_PATIENCE_S,
|
|
)
|
|
|
|
def clear_session_activity_labels(self, session_id: str) -> None:
|
|
"""Clear activity labels after a turn (keep ``last_activity_at`` so idle /
|
|
watchdog clocks stay continuous; an idle turn must not keep advertising
|
|
"compressing"). Runs in the turn's finally: a no-op clear skips the
|
|
write transaction, a real one uses the short activity budget."""
|
|
if not session_id:
|
|
return
|
|
from agent.session_activity import ActivityProvenance
|
|
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 get_session_activity(self, session_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Return the durable activity snapshot for *session_id*, or None."""
|
|
row = self.get_session(session_id) if session_id else None
|
|
if not row:
|
|
return None
|
|
from agent.session_activity import build_activity_snapshot
|
|
return build_activity_snapshot(
|
|
last_activity_at=row.get("last_activity_at"),
|
|
last_activity_description=row.get("last_activity_description"),
|
|
last_activity_provenance=row.get("last_activity_provenance"),
|
|
)
|
|
|
|
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):
|
|
system_prompt_hash = self._store_system_prompt(conn, system_prompt)
|
|
conn.execute(
|
|
"UPDATE sessions SET system_prompt_hash = ?, system_prompt = NULL WHERE id = ?",
|
|
(system_prompt_hash, 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
|
|
(agent-cache eviction) can't fork the cached tool prefix on a flipped
|
|
check_fn verdict. ``None`` clears the pin."""
|
|
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,
|
|
unlike update_token_counts' COALESCE), null system_prompt so stale
|
|
Model:/Provider: footers rebuild, and replace any confirmed Browser
|
|
runtime lock while keeping lineage markers. *provider* is merged into
|
|
model_config so resume recombines the model with the provider that
|
|
actually serves it, not the config.yaml primary."""
|
|
# This write bypasses the token queue: a still-queued first delta carries
|
|
# the pre-switch route and, applied after this UPDATE, would trip the
|
|
# first_accounted_route overwrite and resurrect the old model/provider.
|
|
self.flush_token_counts()
|
|
# browser_model_lock is deleted via a None patch value (same semantics
|
|
# as the old json_remove); lineage markers survive the merge.
|
|
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),
|
|
clear_prompts=True,
|
|
)
|
|
|
|
def _write_model_config_patch(
|
|
self, session_id: str, patch: Dict[str, Any], sql: str,
|
|
params: Callable[[Optional[str]], tuple], *, clear_prompts: bool = False,
|
|
) -> None:
|
|
"""Merge ``patch`` into model_config then run ``sql`` with ``params(merged)``.
|
|
|
|
One write transaction; no-op when the row doesn't exist. ``clear_prompts``
|
|
additionally garbage-collects unreferenced system_prompts (for writers
|
|
that NULL the row's system_prompt_hash).
|
|
"""
|
|
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 clear_prompts:
|
|
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 the merge discipline keeping ``_branched_from``/``_delegate_from``
|
|
alive lives. ``None`` deletes a key. Runs inside the caller's write
|
|
transaction. Returns serialized JSON (``None`` when empty, matching
|
|
create_session's NULL) or ``_MODEL_CONFIG_ROW_MISSING`` when the row
|
|
doesn't exist (``on_missing="raise"`` raises ValueError instead)."""
|
|
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. The transcript-coupled path is
|
|
archive_and_compact's ``model_config_patch``."""
|
|
if not session_id or not patch:
|
|
return
|
|
self._write_model_config_patch(
|
|
session_id, patch, "UPDATE sessions SET model_config = ? WHERE id = ?",
|
|
lambda merged: (merged, session_id),
|
|
)
|
|
|
|
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),
|
|
clear_prompts=True,
|
|
)
|
|
|
|
def set_session_yolo(self, session_id: str, enabled: bool) -> None:
|
|
"""Persist the per-session YOLO flag into model_config so ``/yolo`` or
|
|
``--yolo`` survives ``hermes --resume``. No-op when the row doesn't exist
|
|
yet (creation-time model_config carries the flag for --yolo launches)."""
|
|
if not session_id:
|
|
return
|
|
self._write_model_config_patch(
|
|
session_id, {"yolo_mode": bool(enabled)},
|
|
"UPDATE sessions SET model_config = ? WHERE id = ?",
|
|
lambda merged: (merged, session_id),
|
|
)
|
|
|
|
@staticmethod
|
|
def session_yolo_enabled(session_meta: Optional[Dict[str, Any]]) -> bool:
|
|
"""Persisted YOLO flag from a session row (JSON string or parsed dict);
|
|
False on any parse failure — resume must never enable the bypass by accident."""
|
|
return bool(_parse_model_config((session_meta or {}).get("model_config")).get("yolo_mode"))
|
|
|
|
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 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_dominant_session_model_route(self, session_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Main-loop model route that served most API calls. ``sessions`` is a
|
|
legacy aggregate mixing route changes; ``session_model_usage`` keeps the
|
|
coherent per-call tuple, so status/billing reads prefer it."""
|
|
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 api_call_count DESC,
|
|
(input_tokens + output_tokens + cache_read_tokens +
|
|
cache_write_tokens + reasoning_tokens) DESC,
|
|
last_seen 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"]
|
|
escaped = _escape_like(session_id_or_prefix)
|
|
matches = [row["id"] for row in self._read_all(
|
|
"SELECT id FROM sessions WHERE id LIKE ? ESCAPE '\\' ORDER BY started_at DESC LIMIT 2",
|
|
(f"{escaped}%",),
|
|
)]
|
|
return matches[0] 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 once a Desktop
|
|
registers a second connection (pre-ownership sessions became
|
|
unresumable). Single-match, not a guess: a store belongs to exactly one
|
|
profile. Never overwrites a non-NULL owner; idempotent. Returns rows stamped."""
|
|
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
|
|
(ancestors + descendants joined by end_reason='compression'): Desktop
|
|
projects roots forward to their tip, and updating only the displayed tip
|
|
would let the untouched root resurrect it on refresh. True if any row changed."""
|
|
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 whole compression lineage;
|
|
messages are kept. True when at least one row changed."""
|
|
return self._set_lineage_column('archived', session_id, 1 if archived else 0)
|
|
|
|
# Accidental end reasons recovery treats as resumable; the same constant is
|
|
# 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) — used by registry lookups like Bot Mode's canonical chat.
|
|
Deliberate archives (no end_reason, or an explicit boundary) are left
|
|
alone. True only when a recoverable row was un-archived (whole lineage)."""
|
|
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 (the registry row keeps
|
|
# end_reason='compression'); 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:
|
|
tip_id = session_id
|
|
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. Pinned sessions are
|
|
exempt from the ``sessions.auto_archive`` sweep; Desktop mirrors its
|
|
sidebar pins here so backend sweeps honour them."""
|
|
return self._set_lineage_column('pinned', session_id, 1 if pinned else 0)
|
|
|
|
def set_session_hidden(self, session_id: str, hidden: bool) -> bool:
|
|
"""Hide/unhide a session and its compression lineage from the default
|
|
list_sessions_rich listing; it stays resumable by the owning surface
|
|
(plugins such as kanban manage their own sessions)."""
|
|
return self._set_lineage_column('hidden', session_id, 1 if hidden else 0)
|
|
|
|
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, not a flag: unread when activity postdates it, so new
|
|
messages flip it back without any write on the message path. NULL =
|
|
never tracked = read (shipping the column doesn't badge all history);
|
|
0 = explicitly unread; timestamp = read up to then."""
|
|
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;
|
|
# the projection derives from SCHEMA_SQL so new columns join automatically.
|
|
_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``). Leading-wildcard LIKE can't use an index but
|
|
chain membership keeps it bounded — far cheaper than scanning in Python."""
|
|
params: List[Any] = []
|
|
clauses: List[str] = []
|
|
def _like_pattern(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_pattern(id_needle))
|
|
if search_needle:
|
|
compact_needle = re.sub(r"[\W_]+", "", search_needle)
|
|
compact_sql = (
|
|
"REPLACE(REPLACE(REPLACE(REPLACE(LOWER(COALESCE({0}, '')),"
|
|
" '-', ''), '_', ''), '.', ''), ' ', '')"
|
|
)
|
|
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_pattern(search_needle)] * 2)
|
|
if compact_needle:
|
|
search_clause += f" OR {compact_sql.format('cs.title')} LIKE ? ESCAPE '\\'"
|
|
params.append(_like_pattern(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); tip rows are fetched in
|
|
one batched query. ``_lineage_ids`` carries every id on the chain: a
|
|
persisted tile can hold a MIDDLE segment's id, and with only root/tip a
|
|
surface cannot prove it names this conversation (one chat open twice)."""
|
|
tip_ids_by_root: Dict[str, str] = {}
|
|
chain_by_root: Dict[str, List[str]] = {}
|
|
for s in sessions:
|
|
if s.get("end_reason") != "compression":
|
|
continue
|
|
chain = self.get_compression_chain(s["id"])
|
|
tip_id = chain[-1] if chain else s["id"]
|
|
if tip_id != s["id"]:
|
|
tip_ids_by_root[s["id"]] = tip_id
|
|
chain_by_root[s["id"]] = chain
|
|
tip_rows = (
|
|
self._get_session_rich_rows_batch(set(tip_ids_by_root.values()), compact_rows=compact_rows)
|
|
if tip_ids_by_root else {}
|
|
)
|
|
projected = []
|
|
for s in sessions:
|
|
tip_id = tip_ids_by_root.get(s["id"])
|
|
tip_row = tip_rows.get(tip_id) if tip_id 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]
|
|
merged["_lineage_root_id"] = s["id"]
|
|
merged["_lineage_ids"] = chain_by_root.get(s["id"]) or None
|
|
projected.append(merged)
|
|
return projected
|
|
|
|
@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. Subagent
|
|
runs / compression continuations are hidden unless ``include_children``;
|
|
``project_compression_tips`` shows each chain as its live tip;
|
|
``order_by_last_active`` sorts by the chain TIP via a recursive CTE (the
|
|
only path honouring ``id_query`` / ``search_query``); ``compact_rows``
|
|
omits the system_prompt blob; ``include_pinned`` back-fills pins the page
|
|
missed ("always reachable"), 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 = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
|
|
base_where_params = list(params) # pinned back-fill reuses the WHERE before LIMIT/OFFSET
|
|
prompt_select = (
|
|
"" if compact_rows
|
|
else ", COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved"
|
|
)
|
|
prompt_join = (
|
|
"" if compact_rows
|
|
else "LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash"
|
|
)
|
|
_sel = self._compact_session_cols() if compact_rows else "s.*"
|
|
id_needle = (id_query or "").strip().lower()
|
|
search_needle = (search_query or "").strip().lower()
|
|
if order_by_last_active:
|
|
# The CTE seeds from rows the outer WHERE admits and walks
|
|
# compression-continuation edges forward; MAX over the chain gives
|
|
# effective_last_active so ORDER BY + LIMIT happen in SQL. Do NOT
|
|
# require child.started_at >= parent.ended_at: races insert the
|
|
# continuation before the parent's ended_at is written, while stale
|
|
# websocket siblings could pass the timestamp test and hijack projection.
|
|
outer_where, id_params = self._chain_search_where(where_sql, id_needle, search_needle)
|
|
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 json_extract(COALESCE(child.model_config, '{{}}'), '$._branched_from') IS NULL
|
|
AND json_extract(COALESCE(child.model_config, '{{}}'), '$._delegate_from') IS NULL
|
|
AND COALESCE(child.source, '') != 'tool'
|
|
),
|
|
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 {_sel}{prompt_select},
|
|
{_PREVIEW_COL_SQL},
|
|
{_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 {_sel}{prompt_select},
|
|
{_PREVIEW_COL_SQL},
|
|
{_sql_session_last_active("s")} AS last_active
|
|
FROM sessions s
|
|
{prompt_join}
|
|
{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. One query, never N+1.
|
|
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 {_sel}{prompt_select},
|
|
{_PREVIEW_COL_SQL},
|
|
COALESCE(
|
|
(SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
|
|
s.started_at
|
|
) AS last_active
|
|
FROM sessions s
|
|
{prompt_join}
|
|
{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 (see
|
|
:func:`classify_session_status`; ``'empty'`` when no messages). One query:
|
|
MAX(id) per session (index seek) joined back for that row — 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}
|
|
placeholders = ",".join("?" for _ in ids)
|
|
query = 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 ({placeholders})
|
|
GROUP BY session_id
|
|
) latest ON m.id = latest.max_id
|
|
"""
|
|
rows = self._read_all(query, 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 (compression ancestors excluded), or
|
|
raise SessionExportTooLargeError. The LIMITed subquery stops once it
|
|
proves the bound is exceeded. ``None`` resolves ``sessions.max_export_messages``;
|
|
0 disables the guard (returns 0 without counting)."""
|
|
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`` marker)? 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,))
|
|
if row is None:
|
|
return False
|
|
return 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 = []
|
|
current = session_id
|
|
seen = set()
|
|
with self._read_ctx() as conn:
|
|
for _ in range(100):
|
|
if not current or current in seen:
|
|
break
|
|
seen.add(current)
|
|
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 (:func:`workspace_key` semantics) so
|
|
``hermes -c``/``--resume`` picks the current workspace's last session."""
|
|
select_with_last_active = (
|
|
"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 "
|
|
)
|
|
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)
|
|
where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
|
|
params.extend([limit, offset])
|
|
return [self._session_row_dict(row) for row in self._read_all(
|
|
f"{select_with_last_active}{where_sql} "
|
|
"ORDER BY last_active DESC, s.started_at DESC, s.id DESC LIMIT ? OFFSET ?",
|
|
params,
|
|
)]
|
|
|
|
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 the same filters as list_sessions_rich, so a
|
|
paired "load more" total matches the listable rows (children or a
|
|
cron-excluded page would otherwise inflate it and never settle)."""
|
|
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,
|
|
)
|
|
where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
|
|
return self._read_one(f"SELECT COUNT(*) FROM sessions s{where_sql}", params)[0]
|
|
|
|
def session_count_ge(self, n: int = 1) -> bool:
|
|
"""At least N sessions exist (archived included — "has this install ever
|
|
had sessions"). LIMIT short-circuits: 4us vs session_count()'s 543us
|
|
index scan on a 20k-session DB."""
|
|
rows = self._read_all("SELECT 1 FROM sessions LIMIT ?", (n,))
|
|
return len(rows) >= 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 (uses idx_sessions_source unless
|
|
``exclude_children``, whose predicates need a table scan like
|
|
list_sessions_rich). ``exclude_children`` mirrors listing visibility."""
|
|
where_clauses, params = _session_filter_where(
|
|
exclude_children=exclude_children,
|
|
archived_only=archived_only, include_archived=include_archived,
|
|
)
|
|
where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
|
|
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} "
|
|
"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) for *session_id* in ONE read — prompt_cache_scope
|
|
needs both from the same row. A missing row is (False, ""); DB errors
|
|
propagate so the caller fails closed."""
|
|
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`` and gateway ``request_dump_<id>_*.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:
|
|
# request_dump files use session_id as a prefix component
|
|
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 []
|
|
# The borrowed read connection, never self._conn (unlocked writer use).
|
|
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 (they'd
|
|
resurface as orphans in pickers); branch/compression children are
|
|
orphaned (parent -> NULL). *sessions_dir*: also remove transcript files.
|
|
*expected_delete_ids*: proceed only if parent + delegate cascade still
|
|
equals that set (export-before-delete fails closed if a new delegate
|
|
appeared); the tree is re-walked inside the transaction on purpose (TOCTOU)."""
|
|
removed_delegate_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_delegate_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)
|
|
return True
|
|
deleted = self._execute_write(_do)
|
|
if deleted:
|
|
for sid in removed_delegate_ids + [session_id]:
|
|
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 (a parent that spawned work is not "empty"), so start-and-quit
|
|
sessions don't pile up in /resume. Check and delete share one
|
|
transaction so a concurrently flushed message 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 bool(deleted)
|
|
|
|
def delete_sessions(self, session_ids: List[str], sessions_dir: Optional[Path] = None) -> int:
|
|
"""Bulk delete (dashboard multi-select) with :meth:`delete_session`
|
|
semantics per row, in ONE transaction so a partial failure can't leave
|
|
"messages gone, row still there". Unknown ids are skipped (UI selection
|
|
can race another tab's delete: succeed-on-the-rest). Returns the number
|
|
that actually existed and were deleted."""
|
|
if not session_ids:
|
|
return 0
|
|
unique_ids = list({sid for sid in session_ids if isinstance(sid, str) and sid})
|
|
if not unique_ids:
|
|
return 0
|
|
removed_ids: list[str] = []
|
|
removed_delegate_ids: list[str] = []
|
|
def _do(conn):
|
|
# Filter to IDs that actually exist: return the real deleted count.
|
|
existing = [row["id"] for row in conn.execute(
|
|
f"SELECT id FROM sessions WHERE id IN ({','.join('?' * len(unique_ids))})",
|
|
unique_ids,
|
|
).fetchall()]
|
|
if not existing:
|
|
return 0
|
|
existing_placeholders = ",".join("?" * len(existing))
|
|
removed_delegate_ids.extend(_delete_delegate_children(conn, existing))
|
|
conn.execute( # orphan children whose parent is in the kill list (FK)
|
|
f"UPDATE sessions SET parent_session_id = NULL "
|
|
f"WHERE parent_session_id IN ({existing_placeholders})",
|
|
existing,
|
|
)
|
|
conn.execute(
|
|
f"DELETE FROM messages WHERE session_id IN ({existing_placeholders})", existing,
|
|
)
|
|
conn.execute(f"DELETE FROM sessions WHERE id IN ({existing_placeholders})", existing)
|
|
self._delete_unreferenced_system_prompts(conn)
|
|
removed_ids.extend(existing)
|
|
return len(existing)
|
|
count = self._execute_write(_do)
|
|
for sid in removed_delegate_ids + 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 and compaction
|
|
#: reset it to 0 while keeping dropped turns as ``active = 0`` (the only
|
|
#: recoverable copy) — so NOT EXISTS is the authority; message_count = 0 is
|
|
#: a cheap prefilter.
|
|
_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 (:data:`_EMPTY_SESSION_WHERE`).
|
|
The ended_at guard matches prune_sessions: a fresh session whose first
|
|
message hasn't landed is never sniped out from under the runtime."""
|
|
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 (:data:`_EMPTY_SESSION_WHERE`)
|
|
in one transaction, orphaning (not cascading) children so branch/subagent
|
|
transcripts survive. Transcript files are swept too: the gateway can
|
|
leave a stub request_dump_* if it crashed before the first reply."""
|
|
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
|
|
conn.execute(
|
|
f"UPDATE sessions SET parent_session_id = NULL "
|
|
f"WHERE parent_session_id IN ({','.join('?' * len(session_ids))})",
|
|
list(session_ids),
|
|
)
|
|
for sid in session_ids:
|
|
# DELETE FROM messages is paranoia — the selector's NOT EXISTS
|
|
# probe proved these own no rows — but a row inserted between
|
|
# the SELECT and here would otherwise dangle (clean FK state).
|
|
conn.execute("DELETE FROM messages WHERE session_id = ?", (sid,))
|
|
conn.execute("DELETE FROM sessions WHERE id = ?", (sid,))
|
|
removed_ids.append(sid)
|
|
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. ``archived``
|
|
defaults to False so repeat runs are 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 auto-archive of sessions idle for ``idle_days`` (ages on last
|
|
activity, non-destructive). ``state_meta['last_auto_archive']`` gates
|
|
runs within ``min_interval_hours``; safe to call opportunistically.
|
|
Never raises: {"skipped", "archived", "error"?}."""
|
|
result: Dict[str, Any] = {"skipped": False, "archived": 0}
|
|
try:
|
|
last_raw = self.get_meta("last_auto_archive")
|
|
now = time.time()
|
|
if last_raw:
|
|
try:
|
|
if now - float(last_raw) < min_interval_hours * 3600:
|
|
result["skipped"] = True
|
|
return result
|
|
except (TypeError, ValueError):
|
|
pass # corrupt meta; treat as no prior run
|
|
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
|