"""SessionStore transcript I/O: SQLite append with a per-session retry queue, compression-reroute following, FTS corruption recovery, rewrite/rewind/load. Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.""" from __future__ import annotations import contextlib import logging import threading import time from agent.turn_context import extract_api_content_sidecar from typing import TYPE_CHECKING, Any, Dict, List, Optional if TYPE_CHECKING: from gateway.session import SessionEntry # Log-record parity with the origin module. logger = logging.getLogger("gateway.session") class TranscriptReadError(RuntimeError): """Raised when persisted history cannot be read safely.""" def __init__(self, session_id: str) -> None: self.session_id = session_id super().__init__(f"transcript read failed for session {session_id}") def _spool_dropped(session_id: str, message: Dict[str, Any]): """Spool one evicted/undeliverable message to disk (same machinery as the shutdown flush, so it is replayed after DB recovery); path or None.""" try: from gateway.shutdown_flush import spool_dropped_transcript_message return spool_dropped_transcript_message(session_id, message) except Exception: return None # Message keys persisted only for assistant rows (None otherwise). _ASSISTANT_ONLY_KEYS = ( "reasoning", "reasoning_content", "reasoning_details", "codex_reasoning_items", "codex_message_items", ) class SessionTranscriptMixin: """SessionStore transcript I/O: SQLite append with a per-session retry queue, compression-reroute following, FTS corruption recovery, rewrite/rewind/load.""" _MAX_PENDING_PER_SESSION = 200 # in-memory pending messages per session (DB broken) # Cooldown between FTS5 rebuild attempts (see _rebuild_fts_once); avoids permanently # disabling recovery after one failed attempt while still avoiding a rebuild storm against # a database that is corrupt on every write. _FTS_REBUILD_COOLDOWN_SECONDS = 300 # Consecutive transcript-append failures for one session before escalating from WARNING to # ERROR (see _append_to_transcript_serialized); a session stalled past this many attempts is # no longer a transient blip and needs operator attention. _TRANSCRIPT_APPEND_FAILURE_ESCALATION_THRESHOLD = 3 def _compression_tip_for_session_id(self, session_id: Optional[str]) -> Optional[str]: """Latest compression continuation for *session_id* (heals a mapping left pointing at a compressed parent by a restart or failed send).""" db = self._db_for_session_id(session_id) if session_id else None if db is None: return session_id try: return db.get_compression_tip(session_id) or session_id except Exception: logger.debug("Compression-tip lookup failed for session %s", session_id, exc_info=True) return session_id def _heal_compression_tip_locked( self, entry: "SessionEntry", original_session_id: Optional[str], canonical_session_id: Optional[str]) -> bool: """Rewrite *entry* to the compression continuation if stale. Lock held.""" if not original_session_id or not canonical_session_id: return False if entry.session_id != original_session_id or canonical_session_id == original_session_id: return False logger.info( "SessionStore healed compressed session mapping: %s -> %s", entry.session_id, canonical_session_id) entry.session_id = canonical_session_id return True def advance_compression_session( self, session_key: str, expected_session_id: str, target_session_id: str, ) -> Optional[SessionEntry]: """CAS-advance one route along an already-verified compression lineage. Unlike ``switch_session`` this never ends/reopens SQLite rows (the compression transaction owns that). ``None`` means the route moved after the caller's snapshot (e.g. /new) — caller must fail closed.""" if not session_key or not expected_session_id or not target_session_id: return None with self._lock: entry = self._entry_locked(session_key) if entry is None: return None if entry.session_id == target_session_id: return entry if not self._heal_compression_tip_locked(entry, expected_session_id, target_session_id): return None # route moved (session_id != expected) or nothing to heal self._save() # bookkeeping, not user activity: leave ``updated_at`` alone return entry def _get_transcript_drain_lock(self): """Return the lock that serializes pending-queue drain boundaries.""" return self._lazy("_transcript_drain_lock", threading.RLock) def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db: bool = False) -> None: """Serialize transcript draining across queue migration boundaries. A session with no usable store is NOT skipped: the write is queued and counted like any other failed append, so a dead/unopenable state.db escalates and spools instead of dropping turns silently (#114266).""" if skip_db: return with self._get_transcript_drain_lock(): self._append_to_transcript_serialized(self._follow_reroutes(session_id), message) def _follow_reroutes(self, session_id: str) -> str: """Follow the compression reroute chain (cycle-guarded).""" reroutes = self._lazy("_transcript_reroutes", dict) seen = set() while session_id in reroutes and session_id not in seen: seen.add(session_id) session_id = reroutes[session_id] return session_id def _enqueue_transcript_message(self, session_id: str, message: Dict[str, Any]) -> list: """Queue *message* (retry lock held); evicts + spools the oldest past the cap.""" pending = self._dirty_transcripts.setdefault(session_id, []) pending.append(dict(message)) # Cap pending messages per session to avoid unbounded memory growth when the DB is persistently # broken. Spool the evicted oldest message to the on-disk pending spool (same machinery # flush_pending_to_file uses at shutdown) so a runtime cap rotation does not silently discard it # (#78182); it is replayed on the next successful transcript flush. if len(pending) > self._MAX_PENDING_PER_SESSION: spool_path = _spool_dropped(session_id, pending.pop(0)) if spool_path is not None: self._lazy("_spooled_drop_sessions", set).add(session_id) logger.warning( "Session DB transcript pending queue full for %s (cap=%d); spooled oldest " "message to %s for replay after DB recovery", session_id, self._MAX_PENDING_PER_SESSION, spool_path) else: logger.warning( "Session DB transcript pending queue full for %s (cap=%d); dropping oldest " "message to make room (on-disk spool unavailable)", session_id, self._MAX_PENDING_PER_SESSION) return pending def _divert_transcript_after_db_replaced( self, session_id: str, queue_session_id: str, exc: Exception ) -> None: """Stop SQLite writes on a replaced/quarantined handle and divert the backlog to the on-disk spool + JSONL fallback: retrying cannot succeed and the FTS rebuild must never run here.""" logger.error( "Session DB refused further writes on this handle for %s (%s); stopping SQLite writes " "and diverting pending transcripts to the on-disk fallback: %s", session_id, type(exc).__name__, exc) with self._transcript_retry_lock: remaining = list(self._dirty_transcripts.get(queue_session_id, [])) self._dirty_transcripts.pop(queue_session_id, None) self._transcript_append_failures.pop(session_id, None) for dropped in remaining: if _spool_dropped(session_id, dropped) is None: logger.warning( "pending fallback failed for replaced state.db transcript on %s", session_id, exc_info=True) try: from hermes_state import divert_session_transcript_jsonl divert_session_transcript_jsonl(session_id, remaining) except Exception: logger.warning( "JSONL divert failed for replaced state.db transcript on %s", session_id, exc_info=True) def _live_compression_child(self, session_id: str) -> str: """Transitive compression tip of *session_id* if it is a different, still-live row, else "" (a depth-1 lookup misses multi-hop lineages). Uses the PARENT's proven owner handle: the child's id is unpublished until its write succeeds, so a by-id lookup would hit the ambient store.""" owner_db = self._db_for_session_id(session_id) tip = owner_db.get_compression_tip(session_id) if owner_db is not None else None if tip and tip != session_id: tip_row = owner_db.get_session(tip) if tip_row is not None and tip_row.get("ended_at") is None: return str(tip) return "" def _migrate_transcript_queue_to_child( self, session_id: str, queue_session_id: str, child_id: str, pending: list, msg ) -> list: """Move the retry queue + failure counter from parent to child and record the reroute (retry lock held); returns the child's pending list. Older parent backlog must precede messages already queued directly on the child; routing is published only AFTER the queue moved (caller), so new child writes cannot bypass older parent backlog.""" if pending and pending[0] is msg: pending.pop(0) existing_child_pending = self._dirty_transcripts.get(child_id, []) if pending: pending.extend(existing_child_pending) self._dirty_transcripts[child_id] = pending elif existing_child_pending: pending = existing_child_pending self._dirty_transcripts.pop(queue_session_id, None) previous_failures = self._transcript_append_failures.pop(queue_session_id, 0) if previous_failures: self._transcript_append_failures[child_id] = max( previous_failures, self._transcript_append_failures.get(child_id, 0)) self._transcript_reroutes[session_id] = child_id return pending def _publish_transcript_reroute(self, session_id: str, child_id: str) -> None: """Repoint every route at the compression child and save (index authoritative again).""" with self._lock: for entry in self._entries.values(): if entry.session_id == session_id: entry.session_id = child_id self._save() (getattr(self, "_session_owner_hints", None) or {}).pop(child_id, None) def _append_to_transcript_serialized(self, session_id: str, message: Dict[str, Any]) -> None: """Append a message to a session's transcript (SQLite), draining the per-session retry queue. Args: skip_db: When True, skip the SQLite write. Used when the agent already persisted messages to SQLite via its own _flush_messages_to_session_db(), preventing the duplicate-write bug (#860). """ with self._transcript_retry_lock: pending = self._enqueue_transcript_message(session_id, message) msg = pending[0] queue_session_id = session_id def _ack_head() -> bool: """Pop the acknowledged head (retry lock held). True if queue drained.""" if pending and pending[0] is msg: pending.pop(0) if not pending: self._dirty_transcripts.pop(queue_session_id, None) self._transcript_append_failures.pop(session_id, None) return True return False # DB write outside the retry lock so other sessions can append. while True: # Spooled backlog (cap eviction or a stalled session) is older than ``msg``: replay it # first so recovery keeps transcript order; a still-dead DB just fails both. self._drain_spooled_drops(session_id) try: self._append_transcript_message(session_id, msg) except Exception as exc: from hermes_state import StateDbCorruptError, StateDbReplacedError from hermes_state_errors import CompressionSessionClosedError if isinstance(exc, (StateDbReplacedError, StateDbCorruptError)): self._divert_transcript_after_db_replaced(session_id, queue_session_id, exc) return if isinstance(exc, CompressionSessionClosedError): # Adopt only a different, still-live compression tip, else fail closed. _owner_key = self._owner_key_for_session_id(session_id) child_id = self._live_compression_child(session_id) if child_id: # Record the child's owner BEFORE writing to it (the reroute is published # only after the write succeeds — load-bearing for backlog order). if _owner_key: self._lazy("_session_owner_hints", dict)[child_id] = _owner_key try: self._append_transcript_message(child_id, msg) except Exception as reroute_exc: exc = reroute_exc else: with self._transcript_retry_lock: pending = self._migrate_transcript_queue_to_child( session_id, queue_session_id, child_id, pending, msg) queue_session_id = child_id self._publish_transcript_reroute(session_id, child_id) if not pending: return msg = pending[0] session_id = child_id continue else: # Permanent routing invariant failure, not a transient outage: drop it so it # cannot poison later writes. with self._transcript_retry_lock: _ack_head() logger.error( "Session DB transcript append rejected for compression-ended %s with " "no unique live child; not retrying", session_id) return if self._is_fts_corruption_error(exc) and self._rebuild_fts_once(): try: self._append_transcript_message(session_id, msg) except Exception as retry_exc: exc = retry_exc else: with self._transcript_retry_lock: _ack_head() continue with self._transcript_retry_lock: failures = self._transcript_append_failures.get(session_id, 0) + 1 self._transcript_append_failures[session_id] = failures if failures >= self._TRANSCRIPT_APPEND_FAILURE_ESCALATION_THRESHOLD: spooled = self._spool_stalled_backlog(session_id, queue_session_id) logger.error( "Session DB transcript append failed for %s (failure_count=%d, " "pending=%d, spooled_to_disk=%d); session is stalled and needs operator " "attention: %s", session_id, failures, len(pending), spooled, exc) else: logger.warning( "Session DB transcript append failed for %s (failure_count=%d, pending=%d); " "will retry: %s", session_id, failures, len(pending), exc) return else: with self._transcript_retry_lock: queue_empty = _ack_head() if not queue_empty: msg = pending[0] if queue_empty: return continue def _spool_stalled_backlog(self, session_id: str, queue_session_id: str) -> int: """Move a stalled session's in-memory backlog (oldest first) to the on-disk pending spool so a crash/restart during the outage no longer loses it (#114266): ``recover_pending_to_db`` replays it at boot, ``_drain_spooled_drops`` before the next live write. Stops at the first spool failure so order holds; whatever stays in memory remains under the cap.""" with self._transcript_retry_lock: pending = self._dirty_transcripts.get(queue_session_id, []) backlog = list(pending) spooled = 0 for message in backlog: if _spool_dropped(session_id, message) is None: break spooled += 1 if spooled: self._lazy("_spooled_drop_sessions", set).add(session_id) with self._transcript_retry_lock: del pending[:spooled] if not pending: self._dirty_transcripts.pop(queue_session_id, None) return spooled def _drain_spooled_drops(self, session_id: str) -> None: """Replay cap-dropped spooled transcript messages after DB recovery. Best-effort: replay failures keep the spool files for the next successful flush; nothing here may raise.""" spooled_sessions = getattr(self, "_spooled_drop_sessions", None) if not spooled_sessions or session_id not in spooled_sessions: return try: from gateway.shutdown_flush import drain_transcript_spool # Inside an outage the append that follows logs/escalates the same failure; the # replay attempt is only the order-preserving probe, so its failure stays at DEBUG. with self._transcript_retry_lock: known_failing = bool(self._transcript_append_failures.get(session_id)) _replayed, remaining = drain_transcript_spool( session_id, lambda message: self._append_transcript_message(session_id, message), db_known_failing=known_failing, ) if not remaining: spooled_sessions.discard(session_id) except Exception as exc: logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc) def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None: """Write one transcript row. Caller handles retry queuing.""" _db = self._db_for_session_id(session_id) if _db is None: # Named profile with no resolvable home yet: defer (caller queues) instead of writing # into the ambient store. raise RuntimeError( f"no owning session store for {session_id}; deferring transcript write") is_assistant = message.get("role") == "assistant" _db.append_message( session_id=session_id, role=message.get("role", "unknown"), content=message.get("content"), tool_name=message.get("tool_name"), tool_calls=message.get("tool_calls"), tool_call_id=message.get("tool_call_id"), **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS}, platform_message_id=(message.get("platform_message_id") or message.get("message_id")), observed=bool(message.get("observed")), timestamp=message.get("timestamp"), # Exact bytes sent to the API (prompt-cache-stable replay); must survive every # persistence path or the next replay diverges. api_content=extract_api_content_sidecar(message), # Presentation typing (e.g. "internal_notification"); DB-only. # "internal_notification" for self-injected async-delegation/background notification turns, # #82888). DB-only; stripped from provider-bound payloads. display_kind=message.get("display_kind"), display_metadata=message.get("display_metadata"), ) @staticmethod def _is_fts_corruption_error(exc: Exception) -> bool: """True only when the failure is provably scoped to the FTS index. A bare SQLITE_CORRUPT can mean structural B-tree damage; only errors naming ``messages_fts`` or carrying FTS provenance (``SessionDB._is_fts_write_corruption_error``) may authorize the one-shot rebuild-and-retry; everything else takes the retry path. A generic ``database disk image is malformed`` (bare SQLITE_CORRUPT) can mean structural damage to canonical B-trees, not just the FTS shadow tables — treating it as FTS-only here made the store rebuild the index and retry transcript writes against a structurally corrupt database (#97940). """ if "messages_fts" in str(exc).lower(): return True import sqlite3 from hermes_state import SessionDB return isinstance(exc, sqlite3.DatabaseError) and SessionDB._is_fts_write_corruption_error(exc) def _rebuild_fts_once(self) -> bool: """Attempt FTS5 ``rebuild``, at most once per ``_FTS_REBUILD_COOLDOWN_SECONDS`` window; True if any index was rebuilt. A permanent one-shot flag meant a single rebuild failure (e.g. a transient WAL split-brain guard hit) permanently disabled recovery for the life of the process, even though later corruption on the same store could be fixable. Retrying on a cooldown lets the store try again after the underlying condition (e.g. a foreign holder) has likely cleared, without hammering a database that is corrupt on every write. """ now = time.monotonic() last_attempt = self._fts_rebuild_last_attempt_at if last_attempt is not None and (now - last_attempt) < self._FTS_REBUILD_COOLDOWN_SECONDS: return False db = self._db if db is None or not hasattr(db, "rebuild_fts"): return False self._fts_rebuild_last_attempt_at = now # WAL split-brain guard: skip when a foreign process holds state.db. foreign_holders = None if hasattr(db, "_foreign_state_db_holders"): foreign_holders = db._foreign_state_db_holders() if foreign_holders: logger.warning( "Skipping Session DB FTS rebuild while foreign processes hold the database or " "WAL sidecars (%s); canonical transcript writes remain available.", foreign_holders) return False try: rebuilt = db.rebuild_fts() except Exception as exc: logger.warning("Session DB FTS rebuild failed: %s", exc) return False if rebuilt: logger.warning("Rebuilt %d Session DB FTS index(es) after append corruption", rebuilt) return rebuilt > 0 def _clear_dirty_transcript(self, session_id: str) -> None: """Drop queued pending messages so a rewrite/rewind doesn't re-insert them.""" with self._transcript_retry_lock: self._dirty_transcripts.pop(session_id, None) self._transcript_append_failures.pop(session_id, None) def has_platform_message_id(self, session_id: str, platform_message_id: str) -> bool: """Whether a message with this platform_message_id is persisted (False without a DB). Thin wrapper over SessionDB.has_platform_message_id(). Returns False when no DB is available (in-memory sessions). Used by the gateway's transient-failure dedupe guard (#47237). """ db = self._db_for_session_id(session_id) if not db: return False try: return db.has_platform_message_id(session_id, platform_message_id) except Exception: logger.debug("has_platform_message_id lookup failed", exc_info=True) return False def transcript_tail_role(self, session_id: str) -> Optional[str]: """Role of the newest live conversation row on the route ``load_transcript`` reads (``None`` when empty, no DB, or the read fails — the boundary write would fail the same way).""" session_id = self._compression_tip_for_session_id(self._follow_reroutes(session_id)) db = self._db_for_session_id(session_id) if not db: return None try: return db.latest_conversation_role(session_id) except Exception: logger.debug("transcript tail lookup failed for %s", session_id, exc_info=True) return None def rewrite_transcript( self, session_id: str, messages: List[Dict[str, Any]], active_only: bool = False, reject_active_turn_lease: bool = False) -> bool: """Replace a session's transcript (/retry, /compress). DESTRUCTIVE by default: ``active_only=False`` DELETEs every row incl. soft-archived compaction history (pass ``active_only=True`` for sessions that may carry archived rows). True when the write lands or there is no DB, False on failure — callers committing a destructive change on top (/compress repointing) must check it. ``reject_active_turn_lease`` is for user-initiated rewrites that do not own the cross-process turn lease.""" db = self._db_for_session_id(session_id) if not db: return True with self._get_transcript_drain_lock(): try: # Even when the current agent doesn't "own" persistence, the session on disk may already # carry compaction-archived rows — e.g. after a model switch or a /restore, both of which # mint a fresh agent with _session_db_created=False (so the check above is False) yet leave # the durable archived transcript in place. A full-history replace would DELETE those # archived rows just like the owned-agent case. Guard against it by replacing ONLY the live # (active=1) set unconditionally: on a fresh create/fork every row is active=1, so # active-only replace is behaviorally identical to the full replace — and when archived rows # DO exist they survive. An existence probe here (has_archived_messages) would fail OPEN # into the destructive replace on any DB error and can race a concurrent archive_and_compact # — the same probe failure mode #80216's /retry fix (gateway/slash_commands.py) deliberately # avoids. db.replace_messages( session_id, messages, active_only=active_only, reject_active_turn_lease=reject_active_turn_lease) except Exception as e: logger.debug("Failed to rewrite transcript in DB: %s", e) return False self._clear_dirty_transcript(session_id) return True def has_input_owner(self, session_id: str, owner: str) -> bool: """Find this accepted input on the canonical live continuation and its ancestors. Content and unrelated writers cannot establish ownership. Query only existence; compaction archives can contain many megabytes that replay never needs to load. """ try: current = self._follow_reroutes(session_id) db = self._db_for_session_id(current) current = db.get_compression_tip(current) or current seen = set() while current and current not in seen: seen.add(current) if db.has_gateway_input_owner(current, owner): return True row = db.get_session(current) if not row or not db._is_compression_child_row(row): break current = row["parent_session_id"] return False except Exception as e: raise TranscriptReadError(session_id) from e def load_transcript(self, session_id: str) -> List[Dict[str, Any]]: """Load all messages from a session's transcript (state.db is canonical). Reads follow the same routing writes use — the in-memory reroute map, then the durable compression tip — otherwise the transcript "vanishes" while every message sits under the child.""" if not self._db_for_session_id(session_id): return [] session_id = self._follow_reroutes(session_id) with contextlib.suppress(Exception): # Durable successor survives restart; the reroute map doesn't. db = self._db_for_session_id(session_id) session_id = db.get_compression_tip(session_id) or session_id try: # repair_alternation: this feeds LIVE REPLAY; heal a durable user;user wedge once here. return self._db_for_session_id(session_id).get_messages_as_conversation( session_id, repair_alternation=True) except Exception as e: # Empty history is valid data; a failed canonical read is not — live-replay callers # must fail closed, not start from []. logger.error( "Transcript read failed for session %s; refusing to treat the conversation as " "empty: %s", session_id, e, exc_info=True) raise TranscriptReadError(session_id) from e def rewind_session( self, session_id: str, n: int = 1, *, require_retryable_composite: bool = False, ) -> Optional[Dict[str, Any]]: """Back up ``n`` user turns via soft-delete (``active=0``), mirroring CLI ``/undo [N]``. Returns ``{"rewound_count", "turns_undone", "target_text"}`` or ``None`` (no DB / no rewindable turn / persistence failure); ``n`` clamps to the oldest user turn. ``require_retryable_composite`` is the gateway ``/retry`` guard: the selected turn must be a composite carrier whose live payload is losslessly replayable as text — that replay-policy ``ValueError`` propagates so /retry can explain why the carrier is unsafe.""" db = self._db_for_session_id(session_id) if not db: return None from hermes_state_rewind import RewindTargetUnavailableError with self._get_transcript_drain_lock(): try: outcome = db.rewind_user_turn( session_id, -max(n, 1), require_retryable=require_retryable_composite, require_composite=require_retryable_composite) except RewindTargetUnavailableError as e: logger.debug("rewind_session: %s", e) return None except ValueError: raise except Exception as e: logger.debug("rewind_session: rewind failed: %s", e) return None self._clear_dirty_transcript(session_id) return {"rewound_count": outcome.rewound_count, "turns_undone": outcome.turns_undone, "target_text": outcome.live_text}