"""Shared helpers for reading Hermes Agent sessions from state.db.""" import json import logging import os import sqlite3 import sys from contextlib import closing from pathlib import Path, PurePosixPath, PureWindowsPath from urllib.parse import quote, quote_from_bytes logger = logging.getLogger(__name__) # state.db paths that already produced the "no 'source' column" warning below. # ``get_cli_sessions(all_profiles=True)`` re-reads every profile DB on every # sidebar poll (behind a 5 s cache), so a single pre-``source`` profile DB would # otherwise re-emit the identical WARNING line every ~15 s for the life of the # process. The condition is a property of the DB file, not of the poll, so it # is reported once per path. Process-lifetime only: a restart warns again, # which is the desired behaviour (the log line is the operator's cue that the # agent still needs upgrading). Plain ``set`` mutation under the GIL is # sufficient here; a duplicate line from two racing first calls is harmless. _SOURCE_COLUMN_WARNED_DB_PATHS: set[str] = set() def state_db_file_uri(db_path, platform: str | None = None) -> str: """Build the ``file:`` URI (no query string) for an absolute ``state.db`` path. ``Path.as_uri()`` is not usable here: for a UNC share it emits ``file://server/share/state.db`` and SQLite rejects any non-local URI authority unless compiled with ``SQLITE_ALLOW_URI_AUTHORITY`` (the CPython 3.11-3.13 Windows builds are not). SQLite instead accepts the *empty*-authority spelling ``file:////server/share/state.db``, whose path ``//server/share/...`` the Windows VFS opens as the UNC name. Drive-letter paths keep the documented ``file:///C:/...`` form and POSIX paths are unchanged. Only ``/`` survives unescaped, so ``?`` and ``#`` in path components cannot leak into the query string. ``platform`` defaults to the running interpreter; tests pass it explicitly so the Windows shapes are checked from any host. """ platform = platform or sys.platform if platform == "win32": win = PureWindowsPath(str(db_path)) drive = win.drive if drive.startswith("\\\\?\\"): # Extended-length prefix: "\\?\C:" or "\\?\UNC\server\share". drive = drive[4:] if drive.upper().startswith("UNC\\"): drive = "\\\\" + drive[4:] parts = win.parts[1:] # drop the anchor, keep the path components if drive.startswith("\\\\"): host_share = drive[2:].replace("\\", "/") posix_path = "//" + "/".join((host_share, *parts)) else: posix_path = "/" + "/".join((drive, *parts)) # ``:`` stays literal so the drive letter keeps SQLite's documented # ``file:///C:/...`` shape (``Path.as_uri()`` leaves it unescaped too). return "file://" + quote(posix_path, safe="/:") posix_path = PurePosixPath(str(db_path)).as_posix() # Quote the filesystem BYTES, not the str: a POSIX path component that is # not valid UTF-8 is carried in the str as ``surrogateescape`` code points, # which ``quote(str)`` rejects with UnicodeEncodeError (the caller then # treats the db as unreadable and every agent-backed session vanishes). # ``Path.as_uri()`` — what master used — percent-encodes os.fsencode() # bytes; this keeps that behavior. return "file://" + quote_from_bytes(os.fsencode(posix_path), safe="/") def state_db_readonly_uri(db_path, platform: str | None = None) -> str: """Strict read-only (``mode=ro``) form of :func:`state_db_file_uri`.""" return state_db_file_uri(db_path, platform=platform) + "?mode=ro" def open_state_db_readonly(db_path: Path, log: logging.Logger | None = None) -> sqlite3.Connection: """Open the live agent ``state.db`` read-only for a pure-read projection. Same rationale as the session-listing path (#5455): a write-capable handle on the multi-GB, WAL ``state.db`` while the agent streams into it adds needless checkpoint/lock surface. The read-only ``file:...?mode=ro`` URI avoids that. Read failures propagate; a reader never upgrades to a writer. The caller must ensure ``db_path`` exists — this raises ``FileNotFoundError`` for a missing path rather than creating a ghost database. Callers own the returned connection (wrap it in ``contextlib.closing``). """ if not db_path.exists(): raise FileNotFoundError(f"agent state.db not found: {db_path}") return sqlite3.connect(state_db_readonly_uri(db_path.resolve()), uri=True) MESSAGING_SOURCES = { 'discord', 'email', 'wecom', 'wecom_callback', 'slack', 'telegram', 'weixin', 'matrix', 'signal', } CLI_MIN_UNTITLED_MESSAGE_COUNT = 6 CLI_MIN_UNTITLED_USER_MESSAGE_COUNT = 2 # Sub-second scheduling/write-order races during a compression/cli_close # handoff can persist the continuation row with ``started_at`` a few # milliseconds BEFORE the parent's ``ended_at`` lands (#6931). The tolerance # below lets those rows still classify as the next segment; the fork / # model_config branch-marker / cross-source / end_reason guards in # ``_is_continuation_session`` already rule out unrelated rows, so a small # window cannot collapse genuinely concurrently started children. CONTINUATION_STARTED_AT_TOLERANCE_SECONDS = 2.0 # Accepted ``project_assignment`` values for read_importable_agent_session_rows. PROJECT_ASSIGNMENT_FILTERS = frozenset({'assigned', 'unassigned'}) # Raw-row oversample factors for the bounded candidate window. ``limit`` counts # LOGICAL conversations, but compression segments and the post-projection # visibility filters are spent from the raw window first, so a single 8x pass can # come up short on a lineage-heavy profile. The second factor is only paid when # the first window was fully consumed AND still under-delivered. CANDIDATE_WINDOW_MULTIPLIERS = (8, 32) SOURCE_LABELS = { 'acp': 'ACP', 'api_server': 'API', 'cli': 'CLI', 'cron': 'Cron', 'discord': 'Discord', 'email': 'Email', 'kanban': 'Kanban', 'wecom': 'WeCom', 'wecom_callback': 'WeCom Callback', 'slack': 'Slack', 'telegram': 'Telegram', 'tool': 'Tool', 'tui': 'TUI', 'webhook': 'Webhook', 'webui': 'WebUI', 'weixin': 'Weixin', 'matrix': 'Matrix', 'signal': 'Signal', } def normalize_agent_session_source(raw_source: str | None) -> dict: """Return stable source metadata for Hermes Agent session rows. ``sessions.source`` is an Agent-level raw value. WebUI needs a smaller, durable contract so routes, SSE snapshots, and future sidebar policies do not each reimplement raw-source checks. """ raw = str(raw_source or '').strip().lower() or 'unknown' if raw == 'webui': session_source = 'webui' elif raw in {'acp', 'cli', 'tui'}: # 'acp' (Agent Client Protocol adapter — Zed, external device bridges) # is a local interactive agent client like the CLI/TUI: its sessions # live only in state.db, so classifying it 'other' would leave them # invisible in both sidebar buckets (webui skips the state.db # projection; cli keeps only CLI-classified rows). session_source = 'cli' elif raw in MESSAGING_SOURCES: session_source = 'messaging' elif raw == 'cron': session_source = 'cron' elif raw == 'webhook': session_source = 'webhook' elif raw == 'kanban': session_source = 'kanban' elif raw == 'tool': session_source = 'tool' elif raw == 'api_server': session_source = 'api' else: session_source = 'other' label = SOURCE_LABELS.get(raw) if not label: label = raw.replace('_', ' ').title() if raw != 'unknown' else 'Agent' return { 'raw_source': None if raw == 'unknown' else raw, 'session_source': session_source, 'source_label': label, } def _with_normalized_source(row: dict) -> dict: normalized = normalize_agent_session_source(row.get('source')) return {**row, **normalized} def _optional_col(name: str, columns: set[str], fallback: str = "NULL") -> str: return f"s.{name}" if name in columns else f"{fallback} AS {name}" def _safe_lower(value) -> str: return str(value or "").strip().lower() def _normalize_source_name(value: object) -> str: source = _safe_lower(value) if not source: return "" if source.endswith(" session"): source = source[:-len(" session")].strip() return source def _looks_like_default_cli_title(row: dict) -> bool: """Return True when a CLI row looks like framework-generated metadata.""" title = _safe_lower(row.get("title")) if not title or title == "untitled": return True if title in {"cli", "cli session"}: return True source_candidates = { _normalize_source_name(row.get("source")), _normalize_source_name(row.get("session_source")), _normalize_source_name(row.get("source_tag")), _normalize_source_name(row.get("raw_source")), _normalize_source_name(row.get("source_label")), } source_candidates.discard("") source_candidates.add("cli") return any(title == f"{candidate} session" for candidate in source_candidates) def _as_positive_int(value) -> int: try: return max(0, int(float(value))) except (TypeError, ValueError): return 0 def _as_score(*values) -> float: """First numerically-coercible value as a float, else 0.0. Used to score lineage tips by recency. ``last_message_at`` comes from ``MAX(timestamp)`` and is normally a numeric epoch, but older/non-standard state.db schemas can store an ISO-8601 *text* timestamp. Rather than letting a non-numeric value raise ValueError (which previously escaped the DB try-block and dropped all lineage metadata), fall through to the next candidate (e.g. ``started_at``). """ for value in values: if value in (None, ""): continue try: return float(value) except (TypeError, ValueError): continue return 0.0 def _count_user_turns(row: dict) -> int: user_turns = row.get("actual_user_message_count") if user_turns is None: user_turns = row.get("user_message_count") if user_turns is None: messages = row.get("messages") or [] if isinstance(messages, list): return sum( 1 for msg in messages if _safe_lower(msg.get("role") if isinstance(msg, dict) else msg) == "user" ) return 0 return _as_positive_int(user_turns) def _has_cli_lineage(row: dict) -> bool: segment_count = _as_positive_int(row.get("_compression_segment_count")) return segment_count > 1 or bool(row.get("_lineage_root_id")) def is_cli_session_row(row: dict) -> bool: """Return True for rows that should be treated as CLI-imported sessions.""" if not isinstance(row, dict): return False source = _safe_lower(row.get("session_source")) source_tag = _safe_lower(row.get("source_tag")) raw_source = _safe_lower(row.get("raw_source")) source_name = _safe_lower(row.get("source")) source_label = _safe_lower(row.get("source_label")) if "webui" in {source, source_tag, raw_source, source_name, source_label}: return False # 'subagent' is a delegated delegate_task child: view-only, owned by the # runner, never a writable WebUI/CLI session (#5307). Classify it non-CLI so # sidebar rows and every is_cli_session_row() consumer keep it out of the # CLI/writable treatment. non_cli_sources = MESSAGING_SOURCES | {"cron", "webhook", "kanban", "tool", "api", "api_server", "subagent"} if {source, source_tag, raw_source, source_name, source_label} & non_cli_sources: return False if source == "messaging": return False if source == "cli": return True # External-agent imports (Claude Code, Codex, etc.) are read-only sessions # that Hermes discovers on disk and lists alongside CLI/TUI sessions. The # client renderer (static/sessions.js: _isCliSession) files them in the CLI # bucket via the is_cli_session fallthrough, so the server session-count # classifier MUST agree — otherwise the server counts them under # webui_session_count while the client renders them under CLI, and the WebUI # filter shows a non-zero count with an empty list (#5831). These carry a # real title, so they'd otherwise fall through to the conservative # default-title gate below and be misclassified as non-CLI. if source in {"external_agent", "external-agent"}: return True if ( source_tag in {"acp", "cli", "tui"} or raw_source in {"acp", "cli", "tui"} or source_name in {"acp", "cli", "tui"} or source_label in {"acp", "cli", "tui"} ): return True # Legacy imported CLI rows may only be marked as CLI in sidebar metadata. # Keep this conservative to avoid treating messaging sessions as CLI. return bool( row.get("is_cli_session") and source not in MESSAGING_SOURCES and source_tag not in MESSAGING_SOURCES and raw_source not in MESSAGING_SOURCES and source_name not in MESSAGING_SOURCES and _looks_like_default_cli_title(row) ) def is_cli_session_row_visible(row: dict) -> bool: """Return whether a CLI-related row should remain visible in the sidebar.""" if not isinstance(row, dict): return False if not is_cli_session_row(row): return True actual_message_count = _as_positive_int(row.get("actual_message_count")) message_count = actual_message_count or _as_positive_int(row.get("message_count")) if message_count <= 0: return False if ( actual_message_count > 0 and _count_user_turns(row) > 0 and row.get("ended_at") is None and not row.get("end_reason") ): return True interactive_sources = { _normalize_source_name(row.get("source")), _normalize_source_name(row.get("source_tag")), _normalize_source_name(row.get("raw_source")), _normalize_source_name(row.get("source_label")), } if "tui" in interactive_sources: return True if "acp" in interactive_sources: # Like TUI rows, user-driven ACP sessions stay visible even when # ended/untitled. Unlike TUI, an ACP connection can record only # assistant/tool/system rows (e.g. a replayed or aborted turn), so # require at least one user turn before surfacing the row. return _count_user_turns(row) > 0 if _has_cli_lineage(row): return True if not _looks_like_default_cli_title(row): return True return _count_user_turns(row) >= CLI_MIN_UNTITLED_USER_MESSAGE_COUNT # Every model_config marker Hermes Agent binds to ``parent_session_id`` in its # non-continuation child predicate (``_NON_CONTINUATION_CHILD_FILTER_SQL``): # ``_delegate_from`` (delegate_task), ``_branched_from`` (/branch) and # ``_reset_from`` (gateway reset children, stamped at creation or durably by # ``reopen_session()`` in ``gateway/session_recovery.py``). _MODEL_CONFIG_LINEAGE_KEYS = ('_delegate_from', '_branched_from', '_reset_from') def _branch_markers(row: dict | None) -> tuple[str, dict[str, str]]: """Return ``(state, markers)`` for a row's ``model_config`` lineage identity. Hermes Agent stamps explicit branches, delegate/subagent runs and gateway reset children in the ``model_config`` JSON column (see ``_MODEL_CONFIG_LINEAGE_KEYS``); ``session_source='fork'`` alone only covers WebUI-created forks. ``state`` is one of: - ``'none'``: no ``model_config``, or an object carrying no lineage marker; - ``'markers'``: every non-null marker is a non-empty string, returned by key; - ``'unknown'``: identity evidence exists but cannot be trusted (payload is not JSON, not a JSON object, too deeply nested to decode, or a marker is not a non-empty string). Callers must fail closed on this state and treat the row as a lineage boundary, never as a continuation. """ if not row: return 'none', {} raw = row.get('model_config') if raw is None: return 'none', {} if isinstance(raw, (str, bytes, bytearray)): if not raw.strip(): return 'none', {} try: config = json.loads(raw) except (TypeError, ValueError, RecursionError): # RecursionError: a valid but pathologically deep payload exceeds # the decoder's recursion limit; it is untrusted identity evidence # like any other undecodable payload, never an escaping crash. return 'unknown', {} else: config = raw if not isinstance(config, dict): return 'unknown', {} markers: dict[str, str] = {} for key in _MODEL_CONFIG_LINEAGE_KEYS: value = config.get(key) if value is None: continue if not isinstance(value, str) or not value.strip(): return 'unknown', {} markers[key] = value.strip() return ('markers', markers) if markers else ('none', {}) def _is_continuation_session(parent: dict | None, child: dict | None) -> bool: """Return True when ``child`` is the next segment of the same conversation. Compression rotates session ids automatically. A manual CLI close followed by ``hermes -c`` also records a new child session; for sidebar projection it should continue the same visible conversation rather than becoming a separate child-session row. Plain parent/child links that started before the parent's ended boundary remain child sessions. A small scheduling/write-order overlap between the handoff boundary timestamps is tolerated (#6931). Do not collapse lineage across raw sources. A WebUI session that continues from a Telegram/CLI/etc. parent must remain visible as its own surface-owned conversation; otherwise the tip inherits the root's title/source metadata and can disappear under messaging/sidebar policies. Explicit branches/delegates/resets never continue: a child whose ``model_config`` ``_branched_from`` / ``_delegate_from`` / ``_reset_from`` points at this parent is a separate conversation and must stay visible. Unparsable or non-string marker evidence fails closed as a boundary. Beyond the bounded tolerance window there is no reliable continuation signal — titles are user-controlled and non-unique, so they are never used to widen the window (#7021 re-gate). """ if not parent or not child: return False if str(child.get('session_source') or '').strip().lower() == 'fork': return False # Real Agent branches/delegates/resets are marked in model_config (not # session_source, which only WebUI-created forks carry). The marker must # reference THIS parent: compression continuations inherit the rotated # agent's model_config verbatim, so presence alone (a delegate's # continuation still carries the delegate's own ``_delegate_from``) would # misclassify real continuations. Untrusted identity evidence fails closed. marker_state, child_markers = _branch_markers(child) if marker_state == 'unknown': return False parent_id = str(parent.get('id') or '').strip() if parent_id and parent_id in child_markers.values(): return False parent_source = str(parent.get('source') or '').strip().lower() child_source = str(child.get('source') or '').strip().lower() # Agent lineage excludes tool children even when the parent has the same # source (or no source); a timestamp overlap cannot override that boundary. if child_source == 'tool': return False if parent_source and child_source and parent_source != child_source: return False if parent.get('end_reason') not in {'compression', 'cli_close'}: return False ended_at = parent.get('ended_at') if ended_at is None: # Older state.db rows/tests may not have ended_at populated. Preserve # the historical contract that compression/cli_close parent links are # continuations when no boundary timestamp is available. return True try: child_started = float(child.get('started_at') or 0) parent_ended = float(ended_at) except (TypeError, ValueError): return False if child_started >= parent_ended - CONTINUATION_STARTED_AT_TOLERANCE_SECONDS: return True # Beyond the bounded early-side window the child is a genuine concurrent # session — an exact title match is not evidence (titles are user-controlled # and non-unique, so it cannot extend the tolerance). return False def _continuation_root_id(rows_by_id: dict[str, dict], session_id: str | None) -> str | None: """Return the visible lineage root for ``session_id`` by walking continuations.""" if not session_id: return None root_id = str(session_id) current_id = root_id seen = {current_id} for _ in range(len(rows_by_id) + 1): current = rows_by_id.get(current_id) parent_id = current.get('parent_session_id') if current else None parent = rows_by_id.get(parent_id) if parent_id else None if not parent or not _is_continuation_session(parent, current): return root_id if parent_id in seen: return root_id root_id = str(parent_id) current_id = str(parent_id) seen.add(current_id) return root_id def _project_agent_session_rows(rows: list[dict]) -> list[dict]: """Collapse compression chains into one logical sidebar row. The visible conversation should still look like the original chain head (title and timestamps), while importing should use the latest importable segment so the user continues from the current compressed state. """ rows_by_id = {row['id']: row for row in rows} children_by_parent: dict[str, list[dict]] = {} continuation_child_ids = set() for row in rows: parent_id = row.get('parent_session_id') if not parent_id: continue children_by_parent.setdefault(parent_id, []).append(row) parent = rows_by_id.get(parent_id) if _is_continuation_session(parent, row): continuation_child_ids.add(row['id']) else: row['relationship_type'] = 'child_session' row['parent_title'] = parent.get('title') if parent else None row['parent_source'] = parent.get('source') if parent else None parent_root = _continuation_root_id(rows_by_id, parent_id) if parent_root: row['_parent_lineage_root_id'] = parent_root for children in children_by_parent.values(): children.sort(key=lambda row: row.get('started_at') or 0, reverse=True) def compression_tip(row: dict) -> tuple[dict | None, int, str | None]: """Return the freshest importable continuation descendant for ``row``. Compression parents can have multiple continuation-looking children when a stale segment is resumed after a newer compressed branch already exists. Picking the newest *direct* child can hide the branch whose deeper descendant has the actual latest activity. Walk all reachable continuation descendants and select by real message activity instead. """ latest_importable = row if (row.get('actual_message_count') or 0) > 0 else None segment_count = 0 best_depth = 1 latest_project_id: str | None = None project_depth = 0 project_score = -1.0 best_score = ( _as_score(latest_importable.get('last_activity'), latest_importable.get('started_at')) if latest_importable else 0 ) stack: list[tuple[dict, int]] = [(row, 1)] seen: set[str] = set() while stack: current, depth = stack.pop() current_id = current.get('id') if not current_id or current_id in seen: continue seen.add(current_id) segment_count += 1 current_score = _as_score(current.get('last_activity'), current.get('started_at')) current_project_id = str(current.get('project_id') or '').strip() if ( current_project_id and ( current_score > project_score or (current_score == project_score and depth >= project_depth) ) ): latest_project_id = current_project_id project_depth = depth project_score = current_score if ( (current.get('actual_message_count') or 0) > 0 and (current_score > best_score or (current_score == best_score and depth >= best_depth)) ): latest_importable = current best_depth = depth best_score = current_score for child in children_by_parent.get(current_id, []): child_id = child.get('id') if not child_id or child_id in seen: continue if not _is_continuation_session(current, child): continue stack.append((child, depth + 1)) return latest_importable, max(segment_count, 1), latest_project_id projected = [] for row in rows: if row['id'] in continuation_child_ids: continue segment_count = 1 tip = row lineage_project_id = str(row.get('project_id') or '').strip() or None if row.get('end_reason') in {'compression', 'cli_close'}: tip, segment_count, lineage_project_id = compression_tip(row) if not tip or (tip.get('actual_message_count') or 0) <= 0: continue if tip is row: # The root can still be the freshest *importable* segment while a # newer EMPTY continuation carries the project assignment. Applying # the resolved lineage id here too keeps that assignment instead of # silently dropping it on the early-return path (#6659). root_only = dict(row) if lineage_project_id: root_only['project_id'] = lineage_project_id projected.append(root_only) continue merged = dict(row) # Keep the chain head's visible identity (title, started_at), but # point the row at the latest importable segment for navigation AND # surface the tip's recency so an actively-used chain bubbles to the # top of the sidebar by its true last activity. Without overriding # last_activity, a long-lived chain whose tip is being edited NOW # would sort by the root's old timestamp and fall below recently # touched standalone sessions — exactly the inverse of what a user # expects from "Show agent sessions" sorted by activity. for key in ( 'id', 'model', 'message_count', 'actual_message_count', 'actual_user_message_count', 'ended_at', 'end_reason', 'last_activity', 'archived', ): if key in tip: merged[key] = tip[key] if lineage_project_id: merged['project_id'] = lineage_project_id if str(tip.get('source') or '').strip().lower() == 'tui': # TUI continuation rows are user-visible session segments (#6, #17, # ...), not opaque compression snapshots. Keep navigation pointed at # the latest tip and show that tip's title so the newest conversation # can be found by its visible TUI name. if tip.get('title'): merged['title'] = tip.get('title') if tip.get('source'): merged['source'] = tip.get('source') else: if not merged.get('title'): merged['title'] = tip.get('title') if not merged.get('source'): merged['source'] = tip.get('source') merged['_lineage_root_id'] = row['id'] merged['_lineage_tip_id'] = tip['id'] merged['_compression_segment_count'] = segment_count projected.append(merged) projected.sort( key=lambda row: _as_score(row.get('last_activity'), row.get('started_at')), reverse=True, ) return projected def read_importable_agent_session_rows( db_path: Path, limit: int | None = 200, log=None, exclude_sources: tuple[str, ...] | None = ("cron", "webui"), include_sources: tuple[str, ...] | None = None, project_assignment: str | None = None, project_ids: tuple[str, ...] | None = None, return_window_exhaustion: bool = False, ) -> list[dict] | tuple[list[dict], bool]: """Return agent sessions projected as importable conversations. Hermes Agent can create rows in ``state.db.sessions`` before a session has any messages, and long conversations can be split into compression-linked rows. WebUI cannot import empty rows and should not show compression segments as separate conversations, so both the regular ``/api/sessions`` path and the gateway SSE watcher use this shared projection. By default, omit background/internal sources such as ``cron`` from the WebUI sidebar. This mirrors Hermes Agent CLI's session-list behaviour: interactive views should stay focused on user-facing conversations, while callers that need a source-specific diagnostic view can opt out by passing ``exclude_sources=None``. ``include_sources`` is an additional narrowing filter; callers that want an include-only query should explicitly pass ``exclude_sources=None`` so the default exclusions do not also apply. ``project_assignment`` narrows to project-assigned (``'assigned'``) or project-free (``'unassigned'``) conversations; ``None`` returns both. The two are exact complements and key on the whole compression lineage, so a caller can budget assigned and unassigned conversations independently without either filter double-counting a lineage (#6659). ``project_ids`` narrows the assigned set further, to conversations whose LINEAGE is assigned to one of the given projects. It exists so a caller can give each project its own ``limit`` instead of sharing one global window: a single ``project_assignment='assigned'`` query orders by recency, so one busy project can consume the whole window and leave quieter projects unreachable (greptile P1 on #6659). Only meaningful together with ``project_assignment='assigned'``; any other combination raises ValueError rather than silently returning an unnarrowed (and therefore starvable) set. ``limit`` bounds LOGICAL conversations, not raw rows: the slice is applied after lineage projection, and the raw candidate window is re-widened once if compression segments consumed it before the projection could fill the request. ``limit`` bounds the *recency slice*, not the returned row count. Subagent rows only render as children when their parent row is in the same payload, so subagent ancestors of selected rows are re-added afterwards and the result can exceed ``limit`` by the number of such anchors. Callers must therefore iterate the result rather than assume ``len(rows) <= limit``. ``return_window_exhaustion=True`` returns ``(rows, exhausted)`` instead of just ``rows``. ``exhausted`` is true only when the final bounded raw-candidate window was full while its logical projection still under-delivered ``limit``. That lets a caller distinguish a genuinely short database from a compressed or filtered projection that needs a narrow recovery query, without paying a count probe for ordinary short windows. That recovery is deliberately bounded by the oversampled candidate set (the ``CANDIDATE_WINDOW_MULTIPLIERS`` window that actually ran — ``limit * 8``, or ``limit * 32`` when the first window was consumed and re-widened): it re-uses rows the projection already fetched and never issues an extra query, so an ancestor older than the oversample stays unresolved and its children render top-level, exactly as they did before. Widening that window is a ``CANDIDATE_WINDOW_MULTIPLIERS`` change, not a change to this walk. """ def _result(rows: list[dict], exhausted: bool = False): """Preserve the list API unless this caller needs raw-window status.""" return (rows, exhausted) if return_window_exhaustion else rows wanted_project_ids: tuple[str, ...] = () if project_ids is not None: if str(project_assignment or '').strip().lower() != 'assigned': raise ValueError( "project_ids requires project_assignment='assigned'" ) wanted_project_ids = tuple( cleaned for value in project_ids if (cleaned := str(value or '').strip()) ) if not wanted_project_ids: # An empty narrowing selects nothing. Returning [] keeps that exact # instead of degrading to "every assigned conversation". return _result([]) db_path = Path(db_path) if not db_path.exists(): return _result([]) log = log or logger conn = open_state_db_readonly(db_path, log=log) with closing(conn): conn.row_factory = sqlite3.Row cur = conn.cursor() # Older Hermes Agent versions may not have source tracking. Without a # source column we cannot safely distinguish WebUI rows from agent rows. cur.execute("PRAGMA table_info(sessions)") session_cols = {row[1] for row in cur.fetchall()} cur.execute("PRAGMA table_info(messages)") message_cols = {row[1] for row in cur.fetchall()} if 'source' not in session_cols: warned_key = str(db_path.resolve()) if warned_key not in _SOURCE_COLUMN_WARNED_DB_PATHS: _SOURCE_COLUMN_WARNED_DB_PATHS.add(warned_key) log.warning( "agent session listing skipped: state.db at %s has no 'source' column " "(older hermes-agent?). Agent sessions unavailable. " "Upgrade hermes-agent to fix this.", db_path, ) return _result([]) parent_expr = _optional_col('parent_session_id', session_cols) session_source_expr = _optional_col('session_source', session_cols) model_config_expr = _optional_col('model_config', session_cols) ended_expr = _optional_col('ended_at', session_cols) end_reason_expr = _optional_col('end_reason', session_cols) user_id_expr = _optional_col('user_id', session_cols) chat_id_expr = _optional_col('chat_id', session_cols) chat_type_expr = _optional_col('chat_type', session_cols) thread_id_expr = _optional_col('thread_id', session_cols) session_key_expr = _optional_col('session_key', session_cols) origin_chat_id_expr = _optional_col('origin_chat_id', session_cols) origin_user_id_expr = _optional_col('origin_user_id', session_cols) platform_expr = _optional_col('platform', session_cols) archived_expr = _optional_col('archived', session_cols) project_id_expr = _optional_col('project_id', session_cols) # Older/minimal state.db schemas can have NO ``messages`` table at all, # or a ``messages`` table without a ``session_id`` / ``timestamp`` column. # The projection SQL below joins ``messages`` and aggregates # ``MAX(m.timestamp)`` unconditionally, so on those schemas the query # raised ``sqlite3.OperationalError`` — which the caller # (``get_cli_sessions``) swallows into an empty list, silently hiding # ALL imported/CLI/agent sessions from the sidebar. Detect the columns # and degrade gracefully (mirrors ``read_session_lineage_metadata``): # only join/aggregate ``messages`` when it's actually usable, otherwise # fall back to the per-session ``s.message_count`` / ``s.started_at``. (#3762) messages_has_session_id = 'session_id' in message_cols messages_has_timestamp = 'timestamp' in message_cols use_messages_join = messages_has_session_id count_col = 'id' if 'id' in message_cols else 'session_id' # Index creation belongs to explicit drained maintenance, never a read. messages_index_present = False if messages_has_session_id and messages_has_timestamp: try: cur.execute("PRAGMA index_list(messages)") messages_index_present = any(str(row[1]) == "idx_messages_session" for row in cur.fetchall()) except sqlite3.Error: messages_index_present = False if use_messages_join: actual_count_expr = f"COUNT(m.{count_col})" if 'role' in message_cols: user_message_count_expr = "COUNT(CASE WHEN LOWER(m.role) = 'user' THEN 1 END)" else: user_message_count_expr = f"COUNT(m.{count_col})" last_activity_expr = "MAX(m.timestamp)" if messages_has_timestamp else "NULL" join_clause = "LEFT JOIN messages m ON m.session_id = s.id" group_by_clause = "GROUP BY s.id" else: # No usable messages table: use the denormalized per-session counts # and ``started_at`` so the rows still surface in the sidebar. actual_count_expr = "s.message_count" user_message_count_expr = "s.message_count" last_activity_expr = "NULL" join_clause = "" group_by_clause = "" order_by_clause = "ORDER BY s.started_at DESC" latest_messages_cte = None candidate_order_clause = "ORDER BY s.started_at DESC" where_clauses = ["s.source IS NOT NULL"] params: list[object] = [] if include_sources: included = tuple(str(source) for source in include_sources if source) if included: placeholders = ", ".join("?" for _ in included) where_clauses.append(f"s.source IN ({placeholders})") params.extend(included) if exclude_sources: excluded = tuple(str(source) for source in exclude_sources if source) if excluded: placeholders = ", ".join("?" for _ in excluded) where_clauses.append(f"s.source NOT IN ({placeholders})") params.extend(excluded) if project_assignment is not None: wanted = str(project_assignment).strip().lower() if wanted not in PROJECT_ASSIGNMENT_FILTERS: raise ValueError( "project_assignment must be one of " f"{sorted(PROJECT_ASSIGNMENT_FILTERS)} or None" ) if 'project_id' not in session_cols: # A schema that cannot persist assignments has no assigned rows, # and every row it does have is unassigned. Keep both answers # exact instead of emitting SQL against a missing column. if wanted == 'assigned': return _result([]) elif {'parent_session_id', 'end_reason'} <= session_cols: continuation_checks = [ "parent.end_reason IN ('compression', 'cli_close')", "(parent.source IS NULL OR child.source IS NULL " "OR LOWER(TRIM(parent.source)) = LOWER(TRIM(child.source)))", ] if 'ended_at' in session_cols: continuation_checks.append( "(parent.ended_at IS NULL OR child.started_at >= parent.ended_at)" ) if 'session_source' in session_cols: continuation_checks.append( "LOWER(TRIM(COALESCE(child.session_source, ''))) != 'fork'" ) continuation_where = " AND ".join(continuation_checks) # An assignment anywhere in a compression lineage assigns the # whole logical conversation, so both filters must key on the # lineage, not the individual row: 'unassigned' is the exact # complement of 'assigned' (#6659). membership = "IN" if wanted == 'assigned' else "NOT IN" # ``project_ids`` narrows only the SEED: the lineage walk still # pulls in the whole compression chain around a seeded row, so a # per-project query returns the same logical conversations the # global one would have, minus the other projects'. seed_project_filter = "" if wanted_project_ids: placeholders = ", ".join("?" for _ in wanted_project_ids) seed_project_filter = ( f"\n AND TRIM(seed.project_id) IN ({placeholders})" ) where_clauses.append( f""" s.id {membership} ( WITH RECURSIVE project_lineage(id) AS ( SELECT seed.id FROM sessions seed WHERE seed.project_id IS NOT NULL AND TRIM(seed.project_id) != ''{seed_project_filter} UNION SELECT child.id FROM sessions child JOIN project_lineage lineage ON child.parent_session_id = lineage.id JOIN sessions parent ON parent.id = lineage.id WHERE {continuation_where} UNION SELECT parent.id FROM sessions child JOIN project_lineage lineage ON child.id = lineage.id JOIN sessions parent ON parent.id = child.parent_session_id WHERE {continuation_where} ) SELECT id FROM project_lineage ) """ ) params.extend(wanted_project_ids) elif wanted == 'assigned': if wanted_project_ids: placeholders = ", ".join("?" for _ in wanted_project_ids) where_clauses.append( f"TRIM(COALESCE(s.project_id, '')) IN ({placeholders})" ) params.extend(wanted_project_ids) else: where_clauses.append( "s.project_id IS NOT NULL AND TRIM(s.project_id) != ''" ) else: where_clauses.append( "(s.project_id IS NULL OR TRIM(s.project_id) = '')" ) # Without ``idx_messages_session`` the correlated ``MAX(mx.timestamp)`` # candidate ordering rescans ``messages`` once per session (seconds on # a few thousand sessions). Every missing-index projection — default # sidebar, gateway watcher, cron/webhook/kanban views — orders through # one read-only pre-aggregation pass instead; the listing never creates # the index itself (that is drained maintenance, see # ``scripts/ensure_state_db_read_indexes.py``). use_preaggregated_candidate_order = ( use_messages_join and messages_has_timestamp and not messages_index_present ) if use_preaggregated_candidate_order: order_by_clause = "ORDER BY COALESCE(MAX(m.timestamp), s.started_at) DESC" latest_messages_cte = ( "latest_messages AS (\n" " SELECT mx.session_id AS session_id, MAX(mx.timestamp) AS last_message_at\n" " FROM messages mx\n" " GROUP BY mx.session_id\n" " )" ) candidate_order_clause = "ORDER BY COALESCE(lm.last_message_at, s.started_at) DESC, s.started_at DESC" elif use_messages_join and messages_has_timestamp: order_by_clause = "ORDER BY COALESCE(MAX(m.timestamp), s.started_at) DESC" candidate_order_clause = ( "ORDER BY COALESCE(\n" " (SELECT MAX(mx.timestamp) FROM messages mx WHERE mx.session_id = s.id),\n" " s.started_at\n" " ) DESC,\n" " s.started_at DESC" ) select_sql = f""" SELECT s.id, s.title, s.model, s.message_count, s.started_at, s.source, {session_source_expr}, {model_config_expr}, {user_id_expr}, {chat_id_expr}, {chat_type_expr}, {thread_id_expr}, {session_key_expr}, {origin_chat_id_expr}, {origin_user_id_expr}, {platform_expr}, {project_id_expr}, {parent_expr}, {ended_expr}, {end_reason_expr}, {archived_expr}, {actual_count_expr} AS actual_message_count, {user_message_count_expr} AS actual_user_message_count, {last_activity_expr} AS last_activity """ def _project(raw_rows: list[dict]) -> list[dict]: projected = _project_agent_session_rows(raw_rows) # model_config is selected only so the continuation classifier can # read the real _branched_from/_delegate_from markers; it is internal # agent state and must not leak into the /api/sessions response. for row in projected: row.pop('model_config', None) projected = [_with_normalized_source(row) for row in projected] return [row for row in projected if is_cli_session_row_visible(row)] if limit is not None: result_limit = max(0, int(limit)) if result_limit == 0: return _result([]) # The sidebar only needs a small visible window. Bound the expensive # messages join to a recent-activity candidate set instead of # aggregating every historical Hermes state.db session before # slicing in Python. The candidate ordering must include the latest # message timestamp, not only ``started_at``: long-lived CLI sessions # can be resumed days later and should still surface at the top. # Oversampling preserves room for hidden compression segments or # other rows filtered after projection. if latest_messages_cte: candidate_cte = ( "WITH {latest_messages_cte}, candidates AS (\n" " SELECT s.id\n" " FROM sessions s\n" " LEFT JOIN latest_messages lm ON lm.session_id = s.id\n" " WHERE {where_clause}\n" " {candidate_order_clause}\n" " LIMIT ?\n" " )" ).format( latest_messages_cte=latest_messages_cte, where_clause=" AND ".join(where_clauses), candidate_order_clause=candidate_order_clause, ) else: candidate_cte = ( "WITH candidates AS (\n" " SELECT s.id\n" " FROM sessions s\n" " WHERE {where_clause}\n" " {candidate_order_clause}\n" " LIMIT ?\n" " )" ).format( where_clause=" AND ".join(where_clauses), candidate_order_clause=candidate_order_clause, ) candidate_sql = f""" {candidate_cte} {select_sql} FROM sessions s JOIN candidates c ON c.id = s.id {join_clause} {group_by_clause} {order_by_clause} """ # Compression segments and the post-projection visibility filters are # spent from the RAW candidate window before the logical slice below, # so a lineage-heavy profile can return far fewer conversations than # asked for (greptile P1 on #6659). Widen the window once when it was # fully consumed and still came up short instead of silently # truncating the caller's request. projected: list[dict] = [] window_exhausted = False for multiplier in CANDIDATE_WINDOW_MULTIPLIERS: candidate_limit = max(result_limit * multiplier, result_limit) cur.execute(candidate_sql, [*params, candidate_limit]) raw_rows = [dict(row) for row in cur.fetchall()] projected = _project(raw_rows) window_exhausted = ( len(raw_rows) >= candidate_limit and len(projected) < result_limit ) if not window_exhausted: # Either the request is satisfied or the raw candidate window # was not binding, so projection has seen every candidate this # query can reveal. break selected = projected[:result_limit] # The recency slice is per-row, but subagent rows are only renderable as # children: the sidebar nests a child under its parent solely when that # parent row is present in the same payload. A frozen orchestrator stops # writing while its leaves keep streaming, so the leaves win the recency # race and the parent falls outside the window — leaving the leaves to be # promoted to top-level sidebar rows. Re-add subagent parents that the # oversampled candidate set already projected (no extra query); webui # ancestors are left out because that sidebar bucket already has them. # # Bounded by construction: ``by_id`` only holds the candidates of the # window that was actually executed (``limit * 8``, or ``limit * 32`` # when the re-widening pass above had to run), so an ancestor older # than that oversample is not recovered and its children stay # top-level — the pre-existing behaviour, narrowed rather than fixed. # Resolving those would need an unbounded per-row ancestor query on # the hot sidebar path; widen ``CANDIDATE_WINDOW_MULTIPLIERS`` instead # if the window proves too tight. # NOTE: this can return more than ``limit`` rows (see docstring). have = {row.get('id') for row in selected} by_id = {row.get('id'): row for row in projected if row.get('id')} pending = list(selected) while pending: row = pending.pop() if str(row.get('raw_source') or row.get('source') or '').strip().lower() != 'subagent': continue parent_id = row.get('parent_session_id') if not parent_id or parent_id in have: continue parent = by_id.get(parent_id) if parent is None: continue if str(parent.get('raw_source') or parent.get('source') or '').strip().lower() != 'subagent': continue selected.append(parent) have.add(parent_id) pending.append(parent) return _result(selected, window_exhausted) cur.execute( f""" {select_sql} FROM sessions s {join_clause} WHERE {' AND '.join(where_clauses)} {group_by_clause} {order_by_clause} """, params, ) return _result(_project([dict(row) for row in cur.fetchall()])) def read_assigned_project_row_counts( db_path: Path, log=None, exclude_sources: tuple[str, ...] | None = None, ) -> dict[str, int]: """Return ``{project_id: assigned raw row count}`` from ``state.db``. A single GROUP BY over ``sessions.project_id``: no messages join, no lineage recursion, and an answer bounded by the number of distinct project ids. It exists so the assigned-recovery pass can tell WHICH projects still have rows it has not delivered, and pay a per-project follow-up query only for those, instead of one query per registered project (greptile P1 on #6659). The counts are RAW rows, so a compression lineage counts once per segment. That makes the signal deliberately conservative — it can overestimate what a project still owes, never underestimate it, so a starved project is always detected while at worst one follow-up query comes back empty. """ db_path = Path(db_path) log = log or logger if not db_path.exists(): return {} try: with closing(open_state_db_readonly(db_path, log)) as conn: cur = conn.cursor() cur.execute("PRAGMA table_info(sessions)") session_cols = {row[1] for row in cur.fetchall()} if not {'project_id', 'source'} <= session_cols: return {} where_clauses = [ "s.source IS NOT NULL", "s.project_id IS NOT NULL", "TRIM(s.project_id) != ''", ] params: list[object] = [] excluded = tuple( str(source) for source in (exclude_sources or ()) if source ) if excluded: placeholders = ", ".join("?" for _ in excluded) where_clauses.append(f"s.source NOT IN ({placeholders})") params.extend(excluded) cur.execute( f""" SELECT TRIM(s.project_id) AS project_id, COUNT(*) AS row_count FROM sessions s WHERE {' AND '.join(where_clauses)} GROUP BY TRIM(s.project_id) """, params, ) return { str(row[0]): int(row[1] or 0) for row in cur.fetchall() if str(row[0] or '').strip() } except Exception: log.debug("assigned project row-count probe failed", exc_info=True) return {} def _lineage_report_row(row: dict, role: str) -> dict: updated_at = row.get('ended_at') if row.get('ended_at') is not None else row.get('started_at') return { 'session_id': row.get('id'), 'role': role, 'title': row.get('title'), 'source': row.get('source'), 'started_at': row.get('started_at'), 'updated_at': updated_at, 'end_reason': row.get('end_reason'), 'active': row.get('ended_at') is None, 'archived': False, } def _empty_lineage_report(session_id: str, *, found: bool = False) -> dict: return { 'mutation': False, 'found': found, 'session_id': session_id, 'lineage_key': session_id, 'tip_session_id': session_id, 'total_segments': 0, 'materialized_segments': 0, 'segments': [], 'children': [], 'manual_review': False, } def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops: int = 20) -> dict: """Return a bounded, read-only lifecycle report for a session lineage. This helper intentionally reports only facts that can be derived from ``state.db.sessions`` without mutating WebUI JSON, archiving rows, or deleting historical segments. It mirrors the sidebar continuation rules so a future UI/PR can explain which rows are hidden compression/cli-close segments and which child-session branches remain distinct. """ sid = str(session_id or '').strip() if not sid: return _empty_lineage_report('') db_path = Path(db_path) if not db_path.exists(): return _empty_lineage_report(sid) try: with closing(open_state_db_readonly(db_path)) as conn: conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("PRAGMA table_info(sessions)") session_cols = {row[1] for row in cur.fetchall()} required = {'id', 'parent_session_id', 'end_reason'} if not required.issubset(session_cols): return _empty_lineage_report(sid) source_expr = _optional_col('source', session_cols) session_source_expr = _optional_col('session_source', session_cols) model_config_expr = _optional_col('model_config', session_cols) title_expr = _optional_col('title', session_cols) started_expr = _optional_col('started_at', session_cols, '0') ended_expr = _optional_col('ended_at', session_cols) end_reason_expr = _optional_col('end_reason', session_cols) parent_expr = _optional_col('parent_session_id', session_cols) def fetch_one(row_id: str | None) -> dict | None: if not row_id: return None cur.execute( f""" SELECT s.id, {source_expr}, {session_source_expr}, {model_config_expr}, {title_expr}, {started_expr}, {parent_expr}, {ended_expr}, {end_reason_expr} FROM sessions s WHERE s.id = ? """, (row_id,), ) row = cur.fetchone() return dict(row) if row else None target = fetch_one(sid) if not target: return _empty_lineage_report(sid) segments = [target] current = target seen = {sid} manual_review = False for _hop in range(max(0, int(max_hops))): parent_id = current.get('parent_session_id') parent = fetch_one(parent_id) if not parent or parent_id in seen: manual_review = bool(parent_id and parent_id in seen) break if not _is_continuation_session(parent, current): break segments.append(parent) seen.add(parent_id) current = parent else: manual_review = True segment_ids = {row['id'] for row in segments} child_rows: list[dict] = [] parent_ids = [row['id'] for row in segments] children_by_parent: dict[str, list[dict]] = {pid: [] for pid in parent_ids} if parent_ids: placeholders = ','.join('?' * len(parent_ids)) cur.execute( f""" SELECT s.id, {source_expr}, {session_source_expr}, {model_config_expr}, {title_expr}, {started_expr}, {parent_expr}, {ended_expr}, {end_reason_expr} FROM sessions s WHERE s.parent_session_id IN ({placeholders}) """, parent_ids, ) for child_row in cur.fetchall(): child = dict(child_row) parent_id = child.get('parent_session_id') if parent_id in children_by_parent: children_by_parent[parent_id].append(child) for parent in segments: parent_children = children_by_parent.get(parent['id'], []) parent_children.sort(key=lambda row: row.get('started_at') or 0, reverse=True) for child in parent_children: if child['id'] in segment_ids: continue if _is_continuation_session(parent, child): # A continuation outside the selected path means the # lineage is branched or the caller selected an older # segment. Report manual review rather than proposing # destructive cleanup candidates. manual_review = True continue child_rows.append(child) except Exception: return _empty_lineage_report(sid) root_id = segments[-1]['id'] if segments else sid tip_id = segments[0]['id'] if segments else sid return { 'mutation': False, 'found': True, 'session_id': sid, 'lineage_key': root_id, 'tip_session_id': tip_id, 'total_segments': len(segments), 'materialized_segments': len(segments), 'segments': [ _lineage_report_row(row, 'tip' if idx == 0 else 'hidden_segment') for idx, row in enumerate(segments) ], 'children': [_lineage_report_row(row, 'child_session') for row in child_rows], 'manual_review': manual_review, } def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[str]) -> dict[str, dict]: """Return compression-lineage metadata for known WebUI sidebar sessions. WebUI sessions are persisted as JSON files, but Hermes Agent also mirrors them into ``state.db.sessions`` for insights/session history. Compression and cross-surface continuation create parent chains there. ``/api/sessions`` needs to surface that lineage to the sidebar so client-side collapse can group logical continuations without mutating or deleting any session files. Missing DBs, old schemas, or incomplete rows degrade to an empty mapping. """ wanted = {str(sid) for sid in (session_ids or []) if sid} db_path = Path(db_path) if not wanted or not db_path.exists(): return {} try: with closing(open_state_db_readonly(db_path)) as conn: conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("PRAGMA table_info(sessions)") session_cols = {row[1] for row in cur.fetchall()} if 'parent_session_id' not in session_cols or 'end_reason' not in session_cols: return {} session_source_expr = _optional_col('session_source', session_cols) source_expr = _optional_col('source', session_cols) model_config_expr = _optional_col('model_config', session_cols) message_count_expr = _optional_col('message_count', session_cols, '0') # Scoped fetch via PRIMARY KEY + idx_sessions_parent rather than a # full table scan. The sessions table grows unbounded over time # (1000+ rows is normal, 10000+ for power users), and this function # runs on every sidebar refresh — a full SELECT was ~50x slower # than the indexed lookup at 1000 rows and scales linearly. # # Fetch the wanted ids first, then chase parent_session_id chains # in batches until no new ids appear. Each batch hits PRIMARY KEY # so it's effectively O(N) lookups. Then walk continuation children # from the materialized ancestors so branchy compression lineages can # mark the real freshest tip, not just the newest direct sibling. # # IN-clause is chunked to 500 to stay under SQLITE_MAX_VARIABLE_NUMBER # on older sqlite (Python 3.9 ships sqlite 3.31 which defaults to 999; # newer Python ships sqlite 3.32+ at 32766). On a power user with # 2000+ sessions in the sidebar, an unchunked first hop would raise # `OperationalError: too many SQL variables`, get swallowed by the # except below, and silently disable lineage collapse forever. # (Opus pre-release review of v0.50.251, SHOULD-FIX 2.) IN_CHUNK = 500 rows: dict[str, dict] = {} to_fetch = set(wanted) # Cap walk depth to bound worst-case query count. Real lineage # chains seen in production are <10 segments; anything longer is # almost certainly pathological data and not worth chasing. for _hop in range(20): if not to_fetch: break fetch_list = list(to_fetch) to_fetch = set() for i in range(0, len(fetch_list), IN_CHUNK): chunk = fetch_list[i:i + IN_CHUNK] placeholders = ','.join('?' * len(chunk)) cur.execute( f""" SELECT s.id, {source_expr}, {session_source_expr}, {model_config_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason, {message_count_expr} FROM sessions s WHERE s.id IN ({placeholders}) """, chunk, ) for row in cur.fetchall(): rows[row['id']] = dict(row) # Queue up parents we haven't fetched yet. for sid in fetch_list: parent_id = rows.get(sid, {}).get('parent_session_id') if parent_id and parent_id not in rows and parent_id not in to_fetch: to_fetch.add(parent_id) # Fetch descendants from the discovered ancestors using the parent # index. This keeps the sidebar read scoped while still giving the # collapse metadata enough information to choose the active branch. to_expand = set(rows) expanded: set[str] = set() for _hop in range(20): frontier = [sid for sid in to_expand if sid not in expanded] if not frontier: break to_expand = set() for i in range(0, len(frontier), IN_CHUNK): chunk = frontier[i:i + IN_CHUNK] placeholders = ','.join('?' * len(chunk)) cur.execute( f""" SELECT s.id, {source_expr}, {session_source_expr}, {model_config_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason, {message_count_expr} FROM sessions s WHERE s.parent_session_id IN ({placeholders}) """, chunk, ) for row in cur.fetchall(): child = dict(row) rows[child['id']] = child parent_id = child.get('parent_session_id') parent = rows.get(str(parent_id)) if parent_id else None if parent and child['id'] not in expanded and _is_continuation_session(parent, child): to_expand.add(child['id']) expanded.update(frontier) message_stats: dict[str, dict] = {} cur.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'messages'") has_messages_table = cur.fetchone() is not None # Older/minimal state.db schemas can have a `messages` table WITHOUT a # `timestamp` column (or with a non-numeric one). Detect the columns # rather than gating on table existence alone: require `session_id`, # and only select MAX(timestamp) when that column is actually present # so the query can't raise and collapse the whole lineage metadata. messages_has_session_id = False messages_has_timestamp = False if has_messages_table: cur.execute("PRAGMA table_info(messages)") _message_cols = {row[1] for row in cur.fetchall()} messages_has_session_id = 'session_id' in _message_cols messages_has_timestamp = 'timestamp' in _message_cols use_messages_query = has_messages_table and messages_has_session_id row_ids = list(rows) if use_messages_query: last_at_expr = "MAX(timestamp) AS last_message_at" if messages_has_timestamp else "NULL AS last_message_at" for i in range(0, len(row_ids), IN_CHUNK): chunk = row_ids[i:i + IN_CHUNK] placeholders = ','.join('?' * len(chunk)) cur.execute( f""" SELECT session_id, COUNT(*) AS actual_message_count, {last_at_expr} FROM messages WHERE session_id IN ({placeholders}) GROUP BY session_id """, chunk, ) for row in cur.fetchall(): message_stats[row['session_id']] = dict(row) for sid, row in rows.items(): stats = message_stats.get(sid) or {} if use_messages_query: row['actual_message_count'] = int(stats.get('actual_message_count') or 0) else: row['actual_message_count'] = int(row.get('message_count') or 0) row['last_message_at'] = stats.get('last_message_at') except Exception: return {} children_by_parent: dict[str, list[dict]] = {} for row in rows.values(): parent_id = row.get('parent_session_id') if parent_id: children_by_parent.setdefault(parent_id, []).append(row) def continuation_root_and_depth(sid: str) -> tuple[str, int]: root_id = sid current_id = sid depth = 1 seen = {sid} while True: current = rows.get(current_id) raw_parent_id = current.get('parent_session_id') if current else None parent_id = str(raw_parent_id) if raw_parent_id else '' if not parent_id: break parent = rows.get(parent_id) if not parent or parent_id in seen: break if not _is_continuation_session(parent, current): break root_id = parent_id current_id = parent_id seen.add(parent_id) depth += 1 return root_id, depth def freshest_continuation_tip(root_id: str) -> tuple[str, int]: best_id = root_id best_depth = 1 segment_count = 0 best_score = _as_score(rows.get(root_id, {}).get('last_message_at'), rows.get(root_id, {}).get('started_at')) stack: list[tuple[str, int]] = [(root_id, 1)] seen: set[str] = set() while stack: current_id, depth = stack.pop() if current_id in seen: continue seen.add(current_id) current = rows.get(current_id) if not current: continue segment_count += 1 actual_count = int(current.get('actual_message_count') or 0) score = _as_score(current.get('last_message_at'), current.get('started_at')) if actual_count > 0 and (score > best_score or (score == best_score and depth >= best_depth)): best_id = current_id best_depth = depth best_score = score for child in children_by_parent.get(current_id, []): if _is_continuation_session(current, child): stack.append((child['id'], depth + 1)) return best_id, max(segment_count, best_depth) lineage_tip_cache: dict[str, tuple[str, int]] = {} metadata: dict[str, dict] = {} for sid in wanted: row = rows.get(sid) if not row: continue state_title = str(row.get('title') or '').strip() if state_title: metadata.setdefault(sid, {})['_state_db_title'] = state_title state_source = str(row.get('source') or '').strip().lower() if state_source: entry = metadata.setdefault(sid, {}) entry['_state_db_source'] = state_source source_meta = normalize_agent_session_source(state_source) entry['_state_db_source_tag'] = state_source entry['_state_db_raw_source'] = source_meta.get('raw_source') entry['_state_db_session_source'] = source_meta.get('session_source') entry['_state_db_source_label'] = source_meta.get('source_label') parent_id = row.get('parent_session_id') parent_row = rows.get(parent_id) if parent_id else None if parent_id and parent_row: entry = metadata.setdefault(sid, {}) entry['parent_session_id'] = parent_id if not _is_continuation_session(parent_row, row): entry['relationship_type'] = 'child_session' entry['parent_title'] = parent_row.get('title') entry['parent_source'] = parent_row.get('source') parent_source = str(parent_row.get('source') or '').strip().lower() child_source = str(row.get('source') or '').strip().lower() if parent_source and child_source and parent_source != child_source: entry['_cross_surface_child_session'] = True parent_root = _continuation_root_id(rows, parent_id) if parent_root: entry['_parent_lineage_root_id'] = parent_root if parent_root not in lineage_tip_cache: lineage_tip_cache[parent_root] = freshest_continuation_tip(parent_root) entry['_parent_lineage_tip_id'] = lineage_tip_cache[parent_root][0] continue root_id, segment_count = continuation_root_and_depth(sid) if root_id != sid: entry = metadata.setdefault(sid, {}) entry['_lineage_root_id'] = root_id if root_id not in lineage_tip_cache: lineage_tip_cache[root_id] = freshest_continuation_tip(root_id) tip_id, tip_depth = lineage_tip_cache[root_id] entry['_lineage_tip_id'] = tip_id entry['_compression_segment_count'] = max(segment_count, tip_depth) return metadata