"""Session listing/rich rows, export, and import (portability) for SessionDB. Mixin contract: this is a plain mixin class consumed by ``hermes_state.SessionDB``. It defines no ``__init__`` and no state of its own; methods access the host's attributes (``self._conn``, ``self.db_path``, ``self._execute_write`` and other SessionDB methods) established by ``SessionDB.__init__``. It must never import hermes_state (cycle) — shared module-level constants live in hermes_state_common. """ import logging import json import time from typing import Any, Dict, List, Optional from agent.skill_commands import SKILL_SCAFFOLD_SQL_LIKE from hermes_state_common import ( SCHEMA_SQL, _PREVIEW_ELIGIBLE_SQL, _PREVIEW_RAW_SELECT, _shape_preview, _sql_session_last_active, ) # Moved methods logged under the "hermes_state" logger before the split; # keep that logger identity so log filtering/capture behavior is unchanged. logger = logging.getLogger("hermes_state") class SessionPortabilityMixin: """See module docstring — mixin for SessionDB (Port cluster).""" @classmethod def _compact_session_cols(cls) -> str: """SELECT list for compact_rows: every ``sessions`` column declared in SCHEMA_SQL except prompt storage internals, aliased with the ``s`` prefix used by list_sessions_rich/_get_session_rich_row queries.""" if cls._session_compact_cols_sql is None: declared = cls._parse_schema_columns(SCHEMA_SQL)["sessions"] cls._session_compact_cols_sql = ", ".join( f"s.{name}" for name in declared if name not in cls._SESSION_COMPACT_EXCLUDED ) return cls._session_compact_cols_sql def distinct_session_cwds(self, include_archived: bool = False) -> List[Dict[str, Any]]: """Distinct non-empty session cwds with usage stats, for repo discovery. Aggregates across ALL session history (not a single page), so the desktop can surface every git repo the user has worked in — not just the repos that happen to be in the currently-loaded recents. Children/branches count: a worktree session is still a real workspace signal. """ where = "cwd IS NOT NULL AND TRIM(cwd) != ''" if not include_archived: where += " AND archived = 0" with self._read_ctx() as conn: rows = conn.execute( "SELECT cwd AS cwd, COUNT(*) AS sessions, " "MAX(COALESCE(ended_at, started_at, 0)) AS last_active " f"FROM sessions WHERE {where} GROUP BY cwd" ).fetchall() return [ { "cwd": r["cwd"], "sessions": int(r["sessions"] or 0), "last_active": float(r["last_active"] or 0), } for r in rows ] def list_cron_job_runs( self, job_id: str, limit: int = 20, offset: int = 0, ) -> List[Dict[str, Any]]: """List the run sessions produced by a single cron job, newest first. Cron runs are flat, independent sessions whose id is ``cron_{job_id}_{timestamp}`` (see ``cron/scheduler.run_job``). They are never compression roots and never branch, so this deliberately skips the ``list_sessions_rich`` recursive compression-chain CTE / leading-wildcard ``id_query`` path — that path seeds from *every* ``source='cron'`` row in the DB and only filters to one job's runs after the scan, so it scales with the whole cron pile (a heavy history makes the desktop run-history endpoint time out before it eventually populates). Instead this binds to one job with a ``[prefix, prefix_hi)`` range over the id (an index range scan, not a ``%...%`` substring), filters ``source='cron'``, and orders by ``started_at DESC``. Work scales with the requested window, not the total cron history. Returns the same enriched row shape as ``list_sessions_rich`` (adds ``preview`` + ``last_active``) so callers can reuse it. """ prefix = f"cron_{job_id}_" # Half-open upper bound for an index range scan: increment the final # byte of the prefix so the range covers exactly the ids that start # with ``prefix`` and nothing else. ``prefix`` always ends in '_', but # compute it generically rather than hardcoding the successor char. prefix_hi = prefix[:-1] + chr(ord(prefix[-1]) + 1) query = f""" SELECT s.*, COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved, 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, {_sql_session_last_active("s")} AS last_active FROM sessions s LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash WHERE s.source = 'cron' AND s.id >= ? AND s.id < ? ORDER BY s.started_at DESC, s.id DESC LIMIT ? OFFSET ? """ with self._read_ctx() as conn: cursor = conn.execute(query, (prefix, prefix_hi, limit, offset)) rows = cursor.fetchall() runs: List[Dict[str, Any]] = [] for row in rows: s = self._session_row_dict(row) s["preview"] = _shape_preview(s.pop("_preview_raw", "")) runs.append(s) return runs def _get_session_rich_row(self, session_id: str, compact_rows: bool = False) -> Optional[Dict[str, Any]]: """Fetch a single session with the same enriched columns as ``list_sessions_rich`` (preview + last_active). Returns None if the session doesn't exist. Pass ``compact_rows=True`` to omit the ``system_prompt`` blob (see ``list_sessions_rich`` for details). Thin wrapper over ``_get_session_rich_rows_batch`` so the enriched SELECT lives in exactly one place. """ return self._get_session_rich_rows_batch( [session_id], compact_rows=compact_rows ).get(session_id) def _get_session_rich_rows_batch( self, session_ids, compact_rows: bool = False ) -> Dict[str, Dict[str, Any]]: """Fetch multiple sessions with the same enriched columns as ``_get_session_rich_row``, in a single query. Used by ``list_sessions_rich``'s compression-tip projection to resolve every tip row for a page in one round trip instead of one query per compression-root row. Returns a dict keyed by session id; ids that don't exist are simply absent from the result (same as ``_get_session_rich_row`` returning ``None`` for them). """ ids = [sid for sid in session_ids if sid] if not ids: return {} # Old SQLite builds cap bound variables at 999 # (SQLITE_MAX_VARIABLE_NUMBER); large pages (limit=10000 callers # exist) could exceed it. Chunk the IN list so the helper is safe at # any page size — this is the single choke point for the enriched # multi-row fetch, so the bound lives here, not at call sites. _CHUNK = 900 if len(ids) > _CHUNK: result: Dict[str, Dict[str, Any]] = {} for start in range(0, len(ids), _CHUNK): result.update( self._get_session_rich_rows_batch( ids[start:start + _CHUNK], compact_rows=compact_rows ) ) return result # Same read-your-writes guarantee as list_sessions_rich. self.flush_token_counts() _sel = self._compact_session_cols() if compact_rows else "s.*" placeholders = ",".join("?" for _ in ids) 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" ) query = f""" SELECT {_sel}{prompt_select}, 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, {_sql_session_last_active("s")} AS last_active FROM sessions s {prompt_join} WHERE s.id IN ({placeholders}) """ with self._read_ctx() as conn: cursor = conn.execute(query, ids) rows = cursor.fetchall() result: Dict[str, Dict[str, Any]] = {} for row in rows: s = self._session_row_dict(row) s["preview"] = _shape_preview(s.pop("_preview_raw", "")) result[s["id"]] = s return result def get_session_rich_row(self, session_id: str, compact_rows: bool = False) -> Optional[Dict[str, Any]]: """Public wrapper for :meth:`_get_session_rich_row`. Exposes the single-session enriched row (same columns as ``list_sessions_rich``: preview + last_active) for callers outside this module, e.g. the web server's session-search hydration. """ return self._get_session_rich_row(session_id, compact_rows=compact_rows) def list_skill_scaffolded_sessions(self, limit: int = 200) -> List[Dict[str, Any]]: """Titled sessions whose first user turn was a ``/skill`` invocation. Those titles were generated from the expanded message, which embeds the whole skill body — so they describe the skill rather than the request. Returns ``id``, ``title``, and the full first-turn ``content`` so a caller can re-derive what the user typed. Newest first. """ with self._read_ctx() as conn: rows = conn.execute( """ SELECT s.id, s.title, m.content FROM sessions s JOIN messages m ON m.id = ( SELECT m2.id FROM messages m2 WHERE m2.session_id = s.id AND m2.role = 'user' AND m2.content IS NOT NULL ORDER BY m2.timestamp, m2.id LIMIT 1 ) WHERE s.title IS NOT NULL AND m.content LIKE ? ORDER BY s.started_at DESC LIMIT ? """, (SKILL_SCAFFOLD_SQL_LIKE, int(limit)), ).fetchall() return [dict(row) for row in rows] def get_first_assistant_text(self, session_id: str) -> str: """The session's first assistant reply as plain text ('' when none). Pairs with :meth:`list_skill_scaffolded_sessions` so a re-title can feed the titler the same (request, reply) shape the live path uses. """ with self._read_ctx() as conn: row = conn.execute( "SELECT content FROM messages " "WHERE session_id = ? AND role = 'assistant' AND content IS NOT NULL " "ORDER BY timestamp, id LIMIT 1", (session_id,), ).fetchone() if not row: return "" decoded = self._decode_content(row["content"]) return decoded if isinstance(decoded, str) else "" def export_session(self, session_id: str) -> Optional[Dict[str, Any]]: """Export a single session with all its messages as a dict.""" session = self.get_session(session_id) if not session: return None messages = self.get_messages(session_id) return {**session, "messages": messages} def export_session_lineage(self, session_id: str) -> Optional[Dict[str, Any]]: """Export a compression lineage as one logical session dict.""" lineage_ids = self.get_compression_lineage(session_id) if not lineage_ids: return None segments = [] for sid in lineage_ids: segment = self.export_session(sid) if segment: segments.append(segment) if not segments: return None base = dict(segments[-1]) total_messages = sum(len(seg.get("messages") or []) for seg in segments) base["segments"] = segments base["lineage_session_ids"] = [seg["id"] for seg in segments] base["message_count"] = total_messages base["messages"] = [msg for seg in segments for msg in (seg.get("messages") or [])] return base def export_all(self, source: str = None) -> List[Dict[str, Any]]: """ Export all sessions (with messages) as a list of dicts. Suitable for writing to a JSONL file for backup/analysis. """ sessions = self.search_sessions(source=source, limit=100000) results = [] for session in sessions: messages = self.get_messages(session["id"]) results.append({**session, "messages": messages}) return results def adopt_session_lineage_from( self, donor_db: Any, # a full SessionDB (mixin cannot import it — cycle) session_id: str, *, retire_donor: bool = True, ) -> Dict[str, Any]: """Adopt *session_id*'s full compression lineage from *donor_db* into this store. The stranded-bot-session heal (#93091 follow-up to #93296): before the desktop routed session RPCs by their target session, a profile bot's turns executed on whichever backend held window focus — usually the default one — so the bot's canonical session rows and messages accumulated in the DEFAULT profile's state.db. Once routing was fixed, the profile backend correctly received the RPCs but had no such session, so the same chat 4001'd for the opposite reason. This method moves the conversation to where routing now looks for it. Composition of existing primitives (no new import/export machinery): ``donor_db.export_session_lineage()`` -> ``self.import_sessions()``. Import semantics apply unchanged: gateway routing, handoff, and live activity fields are reset; already-present ids are skipped (idempotent re-adoption after a partial run). When ``retire_donor`` is True and at least one segment was imported (or every segment already exists here), the donor rows are ARCHIVED — never deleted — with ``end_reason='adopted_by_profile'`` so the default profile's list stops advertising a conversation that now lives elsewhere, while the bytes stay recoverable. The archive is deliberately NOT in the recoverable set (agent_close/ws_orphan_reap): canonical-lookup resurrection must not undo an adoption. Returns the ``import_sessions`` result dict, plus ``adopted`` (bool) and ``donor_retired`` (bool — True only when EVERY segment's retirement actually applied). """ payload = donor_db.export_session_lineage(session_id) if not payload: return { "ok": False, "adopted": False, "donor_retired": False, "error": f"session {session_id!r} not found in donor store", } segments = payload.get("segments") or [payload] # Divergence guard: a segment we are about to SKIP (already present # here) may have kept accumulating messages in the donor store after # a partial earlier adoption. Retiring it would strand those newer # messages behind a non-recoverable archive. Compare counts up front # and refuse to retire (still adopt/import) when the donor is ahead. donor_ahead = False for seg in segments: seg_id = seg.get("id") if not seg_id or self.get_session(seg_id) is None: continue donor_count = len(seg.get("messages") or []) local_count = len(self.get_messages(seg_id)) if donor_count > local_count: donor_ahead = True logger.warning( "adoption divergence: donor segment %s has %d messages, " "local copy has %d — donor will NOT be retired", seg_id, donor_count, local_count, ) result = self.import_sessions([dict(seg) for seg in segments]) imported = int(result.get("imported") or 0) skipped = int(result.get("skipped") or 0) adopted = result.get("ok", False) and (imported + skipped) == len(segments) if not adopted: logger.warning( "adoption of %s did not complete: imported=%s skipped=%s " "of %s segment(s); errors=%s", session_id, imported, skipped, len(segments), result.get("errors"), ) donor_retired = False if adopted and retire_donor and not donor_ahead: retire_ok = True for seg in segments: seg_id = seg.get("id") if not seg_id: continue try: # TOCTOU close-out: the guard above compared EXPORT-TIME # counts, but another backend can append donor messages # between export and this loop. Re-read both stores right # before stamping; a donor-ahead signal here skips the # stamp so growth never lands behind a non-recoverable # archive. (Count comparison cannot see equal-count # CONTENT divergence — e.g. a donor rewind+rewrite; that # residual case is accepted: bytes stay in the donor # store either way, only reachability differs.) donor_now = len(donor_db.get_messages(seg_id)) local_now = len(self.get_messages(seg_id)) if donor_now > local_now: retire_ok = False logger.warning( "adoption divergence at retire time: donor " "segment %s grew to %d messages (local %d) — " "leaving donor unretired", seg_id, donor_now, local_now, ) continue # First end_reason wins in end_session(); reopen first so # the adoption boundary is stamped even on ended segments # (e.g. 'compression' parents). donor_db.reopen_session(seg_id) donor_db.end_session(seg_id, "adopted_by_profile") donor_db.set_session_archived(seg_id, True) except Exception: # Best-effort by design: a retirement failure must not # fail the adoption (the profile copy is already whole; # a later resume retries retirement idempotently). But # never claim success we didn't have. retire_ok = False logger.warning( "failed to retire donor segment %s after adoption", seg_id, exc_info=True, ) donor_retired = retire_ok return {**result, "adopted": adopted, "donor_retired": donor_retired} @staticmethod def _import_text_or_none(value: Any, field: str) -> Optional[str]: if value is None: return None if isinstance(value, str): return value raise ValueError(f"{field} must be a string") @staticmethod def _import_json_object_or_none(value: Any, field: str) -> Optional[str]: if value is None: return None if isinstance(value, str): try: parsed = json.loads(value) except json.JSONDecodeError as exc: raise ValueError(f"{field} must be valid JSON") from exc if not isinstance(parsed, dict): raise ValueError(f"{field} must be a JSON object") return value if not isinstance(value, dict): raise ValueError(f"{field} must be a JSON object") try: return json.dumps(value) except (TypeError, ValueError) as exc: raise ValueError(f"{field} must be JSON serializable") from exc @staticmethod def _float_or_none(value: Any) -> Optional[float]: if value is None: return None try: return float(value) except (TypeError, ValueError): return None @staticmethod def _import_int_or_none(value: Any, field: str) -> Optional[int]: if value is None: return None try: return int(value) except (TypeError, ValueError) as exc: raise ValueError(f"{field} must be an integer") from exc @staticmethod def _int_or_default(value: Any, default: int = 0) -> int: if value is None: return default try: return int(value) except (TypeError, ValueError): return default @staticmethod def _reasoning_json_value(value: Any) -> Any: if not isinstance(value, str): return value try: return json.loads(value) except (json.JSONDecodeError, TypeError): return value @staticmethod def _import_error(index: int, session_id: str, error: str) -> Dict[str, Any]: item: Dict[str, Any] = {"index": index, "error": error} if session_id: item["session_id"] = session_id return item def import_sessions(self, sessions: List[Dict[str, Any]]) -> Dict[str, Any]: """Import sessions exported by :meth:`export_session` or ``export_all``. Existing session IDs are skipped. Imported child sessions keep their parent only when that parent already exists or is included in the same import payload; otherwise the child is detached so partial imports don't fail foreign-key validation. Gateway routing, handoff, rewind, and other live runtime state are intentionally reset: this restores conversation history, not ownership of a live channel or process. Activity contract (#76354 review S4): export INCLUDES the live activity fields (``last_activity_at`` / ``last_activity_description`` / ``last_activity_provenance``) because they are part of the durable row, but import deliberately RESETS them to NULL. Resurrecting a stale "working ..." label on a machine where no agent is running would fabricate activity the watchdog and session listings act on. This asymmetry is intentional and covered by regression (tests/gateway/test_watchdog_review_76354.py::test_s4_export_includes_activity_import_resets_it). """ if not isinstance(sessions, list): raise ValueError("sessions must be a list") if len(sessions) > self._IMPORT_MAX_SESSIONS: raise ValueError( f"sessions must contain at most {self._IMPORT_MAX_SESSIONS} entries" ) normalized: List[Dict[str, Any]] = [] errors: List[Dict[str, Any]] = [] seen_ids: set[str] = set() total_messages = 0 total_bytes = 0 session_text_fields = ( "source", "user_id", "model", "system_prompt", "end_reason", "cwd", "git_branch", "git_repo_root", "billing_provider", "billing_base_url", "billing_mode", "cost_status", "cost_source", "pricing_version", "title", ) message_text_fields = ( "role", "tool_call_id", "tool_name", "effect_disposition", "finish_reason", "reasoning", "reasoning_content", "platform_message_id", "message_id", ) for index, raw in enumerate(sessions): if not isinstance(raw, dict): errors.append(self._import_error(index, "", "session must be an object")) continue session_id = str(raw.get("id") or "").strip() if not session_id: errors.append(self._import_error(index, "", "session id is required")) continue if session_id in seen_ids: errors.append(self._import_error(index, session_id, "duplicate session id")) continue messages = raw.get("messages") or [] if not isinstance(messages, list): errors.append(self._import_error(index, session_id, "messages must be a list")) continue if len(messages) > self._IMPORT_MAX_MESSAGES_PER_SESSION: errors.append( self._import_error( index, session_id, "messages exceeds the per-session import limit", ) ) continue if any(not isinstance(msg, dict) for msg in messages): errors.append( self._import_error( index, session_id, "messages must contain only objects", ) ) continue try: session_bytes = len( json.dumps(raw, ensure_ascii=False, separators=(",", ":")).encode("utf-8") ) except (TypeError, ValueError): errors.append( self._import_error(index, session_id, "session must be JSON serializable") ) continue if session_bytes > self._IMPORT_MAX_SESSION_BYTES: errors.append( self._import_error(index, session_id, "session exceeds the import size limit") ) continue total_bytes += session_bytes if total_bytes > self._IMPORT_MAX_TOTAL_BYTES: errors.append( self._import_error(index, session_id, "import exceeds the total size limit") ) continue try: clean_session = dict(raw) clean_session["id"] = session_id clean_session["model_config"] = self._import_json_object_or_none( clean_session.get("model_config"), "model_config" ) clean_session["parent_session_id"] = self._import_text_or_none( clean_session.get("parent_session_id"), "parent_session_id" ) for field in session_text_fields: clean_session[field] = self._import_text_or_none( clean_session.get(field), field ) clean_messages: List[Dict[str, Any]] = [] for message_index, message in enumerate(messages): clean_message = dict(message) role = clean_message.get("role") if not isinstance(role, str) or not role: raise ValueError(f"messages[{message_index}].role must be a non-empty string") for field in message_text_fields: if field == "role": continue clean_message[field] = self._import_text_or_none( clean_message.get(field), field ) clean_message["token_count"] = self._import_int_or_none( clean_message.get("token_count"), "token_count" ) clean_messages.append(clean_message) except ValueError as exc: errors.append(self._import_error(index, session_id, str(exc))) continue total_messages += len(clean_messages) if total_messages > self._IMPORT_MAX_TOTAL_MESSAGES: errors.append( self._import_error( index, session_id, "messages exceeds the total import limit", ) ) continue seen_ids.add(session_id) normalized.append( {"index": index, "session": clean_session, "messages": clean_messages} ) if errors: return { "ok": False, "imported": 0, "skipped": 0, "detached": 0, "errors": errors, } def _do(conn): imported_ids: List[str] = [] skipped_ids: List[str] = [] parent_updates: List[tuple[str, str]] = [] detached = 0 for item in normalized: raw = item["session"] messages = item["messages"] session_id = str(raw.get("id") or "").strip() exists = conn.execute( "SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,), ).fetchone() if exists: skipped_ids.append(session_id) continue started_at = self._float_or_none(raw.get("started_at")) if started_at is None: started_at = time.time() archived = 1 if raw.get("archived") else 0 system_prompt_hash = self._store_system_prompt( conn, raw.get("system_prompt") ) conn.execute( """INSERT INTO sessions ( id, source, user_id, model, model_config, system_prompt, system_prompt_hash, parent_session_id, started_at, ended_at, end_reason, message_count, tool_call_count, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens, cwd, git_branch, git_repo_root, billing_provider, billing_base_url, billing_mode, estimated_cost_usd, actual_cost_usd, cost_status, cost_source, pricing_version, title, api_call_count, archived ) VALUES ( :id, :source, :user_id, :model, :model_config, NULL, :system_prompt_hash, NULL, :started_at, :ended_at, :end_reason, 0, 0, :input_tokens, :output_tokens, :cache_read_tokens, :cache_write_tokens, :reasoning_tokens, :cwd, :git_branch, :git_repo_root, :billing_provider, :billing_base_url, :billing_mode, :estimated_cost_usd, :actual_cost_usd, :cost_status, :cost_source, :pricing_version, :title, :api_call_count, :archived )""", { "id": session_id, "source": str(raw.get("source") or "import"), "user_id": raw.get("user_id"), "model": raw.get("model"), "model_config": raw.get("model_config"), "system_prompt_hash": system_prompt_hash, "started_at": started_at, "ended_at": self._float_or_none(raw.get("ended_at")), "end_reason": raw.get("end_reason"), "input_tokens": self._int_or_default(raw.get("input_tokens")), "output_tokens": self._int_or_default(raw.get("output_tokens")), "cache_read_tokens": self._int_or_default( raw.get("cache_read_tokens") ), "cache_write_tokens": self._int_or_default( raw.get("cache_write_tokens") ), "reasoning_tokens": self._int_or_default( raw.get("reasoning_tokens") ), "cwd": raw.get("cwd"), "git_branch": raw.get("git_branch"), "git_repo_root": raw.get("git_repo_root"), "billing_provider": raw.get("billing_provider"), "billing_base_url": raw.get("billing_base_url"), "billing_mode": raw.get("billing_mode"), "estimated_cost_usd": self._float_or_none( raw.get("estimated_cost_usd") ), "actual_cost_usd": self._float_or_none( raw.get("actual_cost_usd") ), "cost_status": raw.get("cost_status"), "cost_source": raw.get("cost_source"), "pricing_version": raw.get("pricing_version"), "title": raw.get("title"), "api_call_count": self._int_or_default(raw.get("api_call_count")), "archived": archived, }, ) sanitized_messages: List[Dict[str, Any]] = [] for msg in messages: clean = dict(msg) for key in ( "reasoning_details", "codex_reasoning_items", "codex_message_items", ): clean[key] = self._reasoning_json_value(clean.get(key)) sanitized_messages.append(clean) total_messages, total_tool_calls = self._insert_message_rows( conn, session_id, sanitized_messages, ) conn.execute( "UPDATE sessions SET message_count = ?, tool_call_count = ? WHERE id = ?", (total_messages, total_tool_calls, session_id), ) parent_id = str(raw.get("parent_session_id") or "").strip() if parent_id: parent_updates.append((session_id, parent_id)) imported_ids.append(session_id) parent_by_child = dict(parent_updates) def _would_create_cycle(session_id: str, parent_id: str) -> bool: seen = {session_id} current = parent_id while current: if current in seen: return True seen.add(current) if current in parent_by_child: current = parent_by_child[current] continue row = conn.execute( "SELECT parent_session_id FROM sessions WHERE id = ? LIMIT 1", (current,), ).fetchone() if row is None: return False current = row["parent_session_id"] return False for session_id, parent_id in parent_updates: parent_exists = conn.execute( "SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (parent_id,), ).fetchone() if parent_exists and not _would_create_cycle(session_id, parent_id): conn.execute( "UPDATE sessions SET parent_session_id = ? WHERE id = ?", (parent_id, session_id), ) else: # Drop only the closing edge. Later entries can still attach # to this now-root session, preserving the acyclic portion # of a malformed imported lineage. parent_by_child.pop(session_id, None) detached += 1 return { "ok": True, "imported": len(imported_ids), "skipped": len(skipped_ids), "detached": detached, "imported_ids": imported_ids, "skipped_ids": skipped_ids, "errors": [], } return self._execute_write(_do)