diff --git a/gateway/session_lifecycle.py b/gateway/session_lifecycle.py index 4eb6ab5cbb..7738ddba46 100644 --- a/gateway/session_lifecycle.py +++ b/gateway/session_lifecycle.py @@ -1,9 +1,6 @@ -"""SessionStore reset/expiry policy and crash-recovery markers: idle/daily reset -evaluation, expiry finalization, active-turn tokens, resume_pending, -suspension and pruning. Also home of the shared clock/id helpers. - -Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO. -""" +"""SessionStore reset/expiry policy and crash-recovery markers (idle/daily reset, expiry +finalization, active-turn tokens, resume_pending, suspension, pruning) plus the shared clock/id +helpers. Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.""" from __future__ import annotations @@ -43,19 +40,14 @@ def _parse_iso(value) -> Optional[datetime]: return None -# Default auto-continue freshness window (1 hour): a restart-interrupted session is only -# auto-resumed while within this window of when ``resume_pending`` was marked. ``gateway/run.py`` +# Auto-continue freshness window (1 hour) after the ``resume_pending`` mark; ``gateway/run.py`` # bridges config.yaml ``agent.gateway_auto_continue_freshness`` into the env var at startup. _AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT = 60 * 60 def auto_continue_freshness_window() -> float: - """Auto-continue freshness window in seconds (single source of truth for - the resume scheduler and the routing-time zombie gate). - - Reads ``HERMES_AUTO_CONTINUE_FRESHNESS``; falls back to the default when - unset or malformed. Non-positive disables the gate. - """ + """Auto-continue freshness window in seconds (one source of truth for the resume scheduler and + the routing-time zombie gate); env var, default when unset/malformed; non-positive disables.""" raw = os.environ.get("HERMES_AUTO_CONTINUE_FRESHNESS") try: return float(raw) if raw else float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT) @@ -69,20 +61,15 @@ class SessionLifecycleMixin: def set_expiry_finalized( self, entry: SessionEntry, *, clear_model_override: bool = True ) -> None: - """Mark a session entry expiry-finalized in memory, sessions.json, AND state.db. - - Single write-path for the expiry watcher so the durable flag survives - sessions.json loss. ``clear_model_override=False`` = flag only. - """ + """Mark a session entry expiry-finalized in memory, sessions.json, AND state.db (single + write-path for the expiry watcher). ``clear_model_override=False`` = flag only.""" with self._lock: entry.expiry_finalized = True if clear_model_override: - # Finalization is a conversation boundary: drop the persisted - # /model override so a later message cannot rehydrate it. + # Finalization is a conversation boundary: a later message must not rehydrate it. entry.model_override = None self._save() - # Background caller never entered ``_profile_runtime_scope``: resolve - # the store from the key, not the ambient scope. + # Background caller never entered ``_profile_runtime_scope``: resolve the store by key. _db = self._db_for_key(entry.session_key) if not _db: return @@ -94,8 +81,7 @@ class SessionLifecycleMixin: logger.debug("Session DB expiry_finalized write failed for %s: %s", entry.session_id, exc) try: # Without a durable ``session_reset`` end_reason, later agent cleanup ends the row as - # ``agent_close``, which stale-route recovery treats as resumable. Promotion only - # upgrades live/ agent_close rows; explicit boundaries are preserved. + # ``agent_close``, which stale-route recovery treats as resumable. _db.promote_to_session_reset(entry.session_id) except Exception as exc: logger.debug("Session DB promote_to_session_reset failed for %s: %s", entry.session_id, exc) @@ -117,11 +103,8 @@ class SessionLifecycleMixin: return None def _is_session_expired(self, entry: SessionEntry) -> bool: - """Whether the entry's reset policy has expired it (entry alone, no source). - - Used by the background expiry watcher. Sessions with active - background processes are never considered expired. - """ + """Whether the reset policy has expired *entry* (expiry watcher); sessions with active + background processes never expire.""" if self._has_active_processes_safe(entry.session_key, context="expiry"): logger.debug("Session %s not expired — active background processes", entry.session_key) return False @@ -129,13 +112,8 @@ class SessionLifecycleMixin: return self._policy_reset_reason(policy, entry.updated_at) is not None def is_session_finalizable(self, entry: SessionEntry) -> bool: - """True if the expiry watcher will *ever* finalize this session. - - A ``mode == "none"`` session never expires, so the agent-cache idle - sweep must reap its agent itself instead of deferring to the watcher - (deferring would pin the agent for the gateway's lifetime). Policy - resolution errors count as "not finalizable" (sweep reaps — safe). - """ + """True if the expiry watcher will *ever* finalize this session; ``mode == "none"`` never + expires, so the agent-cache sweep must reap its agent itself. Policy errors -> False.""" try: policy = self.config.get_reset_policy(platform=entry.platform, session_type=entry.chat_type) return policy.mode != "none" @@ -143,14 +121,9 @@ class SessionLifecycleMixin: return False def _is_session_ended_in_db(self, session_id: str) -> bool: - """True iff state.db has this session with a non-null end_reason. - - Same staleness test as ``_prune_stale_sessions_locked`` (no DB, no - row, or DB error -> False, keep). Lets ``get_or_create_session`` - self-heal at routing time, since the startup prune cannot see a - session ended while the gateway stays alive. Store resolved from the - row's owning profile, not the ambient scope. - """ + """True iff state.db has this session with a non-null end_reason (same staleness test as + ``_prune_stale_sessions_locked``; no DB/row or DB error -> False). Lets routing self-heal a + session ended while the gateway stays alive. Store resolved from the owning profile.""" db = self._db_for_session_id(session_id) if not db or not session_id: return False @@ -161,8 +134,8 @@ class SessionLifecycleMixin: return bool(row is not None and row.get("end_reason") is not None) def _should_reset(self, entry: SessionEntry, source: SessionSource) -> Optional[str]: - """Reset reason ("idle"/"daily") if policy says reset, else None. - Sessions with active background processes are never reset.""" + """Reset reason ("idle"/"daily") if policy says reset, else None; sessions with active + background processes are never reset.""" session_key = self._generate_session_key(source) if self._has_active_processes_safe(session_key, context="reset"): logger.debug("Session reset skipped for %s — active background processes", session_key) @@ -173,14 +146,11 @@ class SessionLifecycleMixin: def _route_reset_reason( self, entry: SessionEntry, source: SessionSource, now: datetime ) -> Optional[str]: - """Reset decision for an existing route (no lock; DB/config I/O). - - ``suspended`` always resets. Otherwise the reset policy decides; a - still-pending resume marker is additionally freshness-gated — but - ``session_reset.mode: none`` (user opted out of ALL automatic resets) - makes an expired marker fall through to a normal resume, never a - silent fresh session. - """ + """Reset decision for an existing route (no lock; DB/config I/O). ``suspended`` always + resets; otherwise the reset policy decides, and a still-pending resume marker is also + freshness-gated — but ``session_reset.mode: none`` (user opted out of ALL automatic + resets) makes an expired marker fall through to a normal resume, never a silent fresh + session.""" if entry.suspended: return "suspended" reason = self._should_reset(entry, source) @@ -196,8 +166,8 @@ class SessionLifecycleMixin: return None def _update_entry(self, session_key: str, mutate) -> bool: - """Apply ``mutate(entry)`` under ``_lock`` and full-save; False when the - entry is missing or *mutate* returned False (nothing to persist).""" + """Apply ``mutate(entry)`` under ``_lock`` and full-save; False when the entry is missing + or *mutate* returned False (nothing to persist).""" with self._lock: entry = self._entry_locked(session_key) if entry is None or mutate(entry) is False: @@ -206,8 +176,8 @@ class SessionLifecycleMixin: return True def _update_all_entries_locked(self, mutate) -> int: - """Apply ``mutate(entry) -> bool`` to every entry under ``_lock``; save once - if any returned True. Returns the count that did.""" + """Apply ``mutate(entry) -> bool`` to every entry under ``_lock``; save once if any + returned True. Returns the count that did.""" with self._lock: self._ensure_loaded_locked() changed = sum(1 for entry in self._entries.values() if mutate(entry)) @@ -216,19 +186,18 @@ class SessionLifecycleMixin: return changed def suspend_session(self, session_key: str) -> bool: - """Mark a session suspended so it auto-resets on next access (/stop). - Returns True if the session existed.""" + """Mark a session suspended so it auto-resets on next access (/stop). True if it existed.""" return self._update_entry(session_key, lambda e: setattr(e, "suspended", True)) def _set_turn_marker_locked(self, session_key: str, entry: SessionEntry, token, started_at) -> None: - """Persist the active-turn pair BEFORE publishing it in memory, so a failed - write can neither leak an unowned token nor drop a live one. Lock held.""" + """Persist the active-turn pair BEFORE publishing it in memory, so a failed write can + neither leak an unowned token nor drop a live one. Lock held.""" candidate = entry.to_dict() candidate["active_turn_token"] = token candidate["active_turn_started_at"] = _iso(started_at) if started_at is not None: - # Keeps the legacy 120s startup heuristic working for an older - # binary during a rolling downgrade/upgrade window. + # Keeps the legacy 120s startup heuristic working for an older binary during a rolling + # downgrade/upgrade window. candidate["updated_at"] = started_at.isoformat() self._save_entry(session_key, entry_data=candidate, lock_held=True) entry.active_turn_token = token @@ -237,12 +206,9 @@ class SessionLifecycleMixin: entry.updated_at = started_at def mark_turn_active(self, session_key: str) -> Optional[str]: - """Persist exact ownership of the agent turn running for *session_key*. - - The opaque token is returned to the caller and must be supplied to - :meth:`clear_turn_active`. Re-marking replaces the previous token so - a stale asynchronous unwind cannot clear a newer turn. - """ + """Persist exact ownership of the running agent turn; returns the opaque token for + :meth:`clear_turn_active`. Re-marking replaces the previous token so a stale asynchronous + unwind cannot clear a newer turn.""" token = uuid.uuid4().hex with self._lock: entry = self._entry_locked(session_key) @@ -252,8 +218,8 @@ class SessionLifecycleMixin: return token def clear_turn_active(self, session_key: str, token: str) -> bool: - """Compare-and-swap clear an active-turn marker; ``False`` when the - entry disappeared or a newer turn owns it.""" + """Compare-and-swap clear an active-turn marker; ``False`` when the entry disappeared or a + newer turn owns it.""" with self._lock: entry = self._entry_locked(session_key) if entry is None or entry.active_turn_token != token: @@ -263,10 +229,8 @@ class SessionLifecycleMixin: def recover_interrupted_turns(self, max_age_seconds: int = 60 * 60) -> int: """Promote crash-left turn markers into ``resume_pending`` (unclean startup only). - - Old/invalid markers are cleared without resuming; suspended sessions - are never re-armed. Returns the number of newly promoted sessions. - """ + Old/invalid markers are cleared without resuming; suspended sessions are never re-armed. + Returns the number of newly promoted sessions.""" now = _now() max_age = timedelta(seconds=max(0, max_age_seconds)) promoted = 0 @@ -277,13 +241,12 @@ class SessionLifecycleMixin: return False started_at = entry.active_turn_started_at try: - marker_is_stale = ( - started_at is None or (max_age_seconds > 0 and now - started_at > max_age) + marker_is_stale = started_at is None or ( + max_age_seconds > 0 and now - started_at > max_age ) except TypeError: # Mixed aware/naive timestamps: clear rather than risk an unsafe old resume. marker_is_stale = True - if not marker_is_stale and not entry.suspended: if entry.resume_pending: # A drain-timeout marker is more specific; keep it. @@ -292,10 +255,8 @@ class SessionLifecycleMixin: else: entry.resume_pending = True entry.resume_reason = "restart_interrupted" - # Freshness starts at discovery, not turn start. - entry.last_resume_marked_at = now + entry.last_resume_marked_at = now # freshness starts at discovery promoted += 1 - entry.active_turn_token = None entry.active_turn_started_at = None return True @@ -311,45 +272,36 @@ class SessionLifecycleMixin: entry.active_turn_token = None entry.active_turn_started_at = None return True - return self._update_all_entries_locked(_discard) def mark_resume_pending(self, session_key: str, reason: str = "restart_timeout") -> bool: - """Mark a session resumable after a restart interruption (keeps the - session_id/transcript, unlike ``suspend_session``). True if marked.""" + """Mark a session resumable after a restart interruption (keeps the session_id/transcript, + unlike ``suspend_session``). True if marked.""" def _apply(entry: SessionEntry): - # Never override an explicit ``suspended`` (hard forced-wipe). - if entry.suspended: + if entry.suspended: # never override an explicit ``suspended`` (hard forced-wipe) return False entry.resume_pending = True entry.resume_reason = reason entry.last_resume_marked_at = _now() - return self._update_entry(session_key, _apply) def clear_resume_pending(self, session_key: str) -> bool: - """Clear the resume-pending flag after a successful resumed turn. - Returns True if a flag was cleared.""" + """Clear the resume-pending flag after a successful resumed turn; True if cleared.""" def _apply(entry: SessionEntry): if not entry.resume_pending: return False entry.resume_pending = False entry.resume_reason = None entry.last_resume_marked_at = None - return self._update_entry(session_key, _apply) def prune_old_entries(self, max_age_days: int) -> int: - """Drop routing entries idle (by ``updated_at``) for more than max_age_days. - - Suspended entries and entries with active background processes are - kept. The SQLite transcript stays; only the key -> session_id mapping - is dropped. ``max_age_days <= 0`` disables. Returns the count removed. - """ + """Drop routing entries idle (by ``updated_at``) for more than max_age_days; suspended + entries and entries with active background processes are kept. Only the key -> session_id + mapping is dropped (the transcript stays). ``max_age_days <= 0`` disables. Returns count.""" if max_age_days is None or max_age_days <= 0: return 0 cutoff = _now() - timedelta(days=max_age_days) - with self._lock: self._ensure_loaded_locked() removed_keys = [ @@ -363,18 +315,14 @@ class SessionLifecycleMixin: self._entries.pop(key, None) if removed_keys: self._save() - if removed_keys: - logger.info( - "SessionStore pruned %d entries older than %d days", - len(removed_keys), max_age_days, - ) + logger.info("SessionStore pruned %d entries older than %d days", + len(removed_keys), max_age_days) return len(removed_keys) def suspend_recently_active(self, max_age_seconds: int = 120) -> int: - """Mark sessions active within *max_age_seconds* as ``resume_pending`` - after a crash/fast restart (already-pending and suspended entries are - skipped). Returns the number marked.""" + """Mark sessions active within *max_age_seconds* as ``resume_pending`` after a crash/fast + restart (already-pending and suspended entries are skipped). Returns the number marked.""" cutoff = _now() - timedelta(seconds=max_age_seconds) def _mark(entry: SessionEntry) -> bool: @@ -384,5 +332,4 @@ class SessionLifecycleMixin: entry.resume_reason = "restart_interrupted" entry.last_resume_marked_at = _now() return True - return self._update_all_entries_locked(_mark) diff --git a/gateway/session_persistence.py b/gateway/session_persistence.py index 392f1b160e..b0635d8abc 100644 --- a/gateway/session_persistence.py +++ b/gateway/session_persistence.py @@ -1,8 +1,6 @@ """SessionStore storage plumbing: per-profile SessionDB handle resolution and the routing-index -load/save paths (state.db gateway_routing primary, sessions.json legacy mirror). - -Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO. -""" +load/save paths (state.db gateway_routing primary, sessions.json legacy mirror). Mixin split out of +``gateway/session.py``; bound onto ``SessionStore`` via the MRO.""" from __future__ import annotations @@ -22,20 +20,17 @@ if TYPE_CHECKING: # Log-record parity with the origin module. logger = logging.getLogger("gateway.session") -# "No SessionDB pinned" sentinel: lets ``_db`` distinguish "resolve from the -# active scope" from a deliberate ``store._db = None`` (JSONL fallback). +# "No SessionDB pinned" sentinel: lets ``_db`` distinguish "resolve from the active scope" from a +# deliberate ``store._db = None`` (JSONL fallback). _DB_UNPINNED = object() # Self-documenting sentinel written first into sessions.json; "_" keys are skipped on load. _SESSIONS_JSON_README = ( - "LEGACY MIRROR of the gateway routing index (the primary copy " - "lives in the gateway_routing table in ~/.hermes/state.db). " - "Maps messaging session keys (agent:main::...) to " - "active session IDs. This is NOT the session list. ALL " - "sessions (CLI, TUI, and gateway) live in ~/.hermes/state.db " - "and are shown by `hermes sessions list` and `/sessions`. " - "Disable this file with `gateway.write_sessions_json: false` " - "in config.yaml." + "LEGACY MIRROR of the gateway routing index (the primary copy lives in the gateway_routing " + "table in ~/.hermes/state.db). Maps messaging session keys (agent:main::...) to " + "active session IDs. This is NOT the session list. ALL sessions (CLI, TUI, and gateway) live " + "in ~/.hermes/state.db and are shown by `hermes sessions list` and `/sessions`. Disable this " + "file with `gateway.write_sessions_json: false` in config.yaml." ) @@ -45,26 +40,20 @@ def _is_live_system_guard(exc: BaseException) -> bool: class SessionPersistenceMixin: - """SessionStore storage plumbing: SessionDB handle resolution and the - routing-index load/save paths.""" + """SessionStore storage plumbing: SessionDB handle resolution and routing-index load/save.""" def _open_session_db_for_active_scope(self, db_path: Optional[Path] = None): - """SessionDB for the profile scope active on this task. - - ``db_path`` pins the store; otherwise ``_default_db_path()`` follows the - context-local HERMES_HOME from ``_profile_runtime_scope`` (resolved per - call so multiplexed profiles reach their own store). Handles are cached - per path; failed opens enter a bounded backoff during which callers keep - using the JSONL fallback. - """ + """SessionDB for the profile scope active on this task. ``db_path`` pins the store; + otherwise ``_default_db_path()`` follows the context-local HERMES_HOME (resolved per call so + multiplexed profiles reach their own store). Handles are cached per path; failed opens enter + a bounded backoff during which callers keep using the JSONL fallback.""" from hermes_state import _default_db_path, get_shared_session_db path = Path(db_path) if db_path is not None else Path(_default_db_path()) def _open(): try: - # Process-wide shared registry: one writer connection per path. - return get_shared_session_db(path) + return get_shared_session_db(path) # process-wide registry: one writer per path except Exception as e: if not _is_live_system_guard(e): print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}") @@ -78,13 +67,10 @@ class SessionPersistenceMixin: @property def _db(self): - """The SessionDB for the active profile scope, or a pinned override. - - Assigning ``store._db`` pins that value for every subsequent read - (tests install a fake or disable the DB with ``store._db = None``). - Unpinned, each read resolves the scope so a multiplexed profile's - writes reach its own store. - """ + """The SessionDB for the active profile scope, or a pinned override. Assigning ``store._db`` + pins that value for every subsequent read (tests install a fake or disable the DB with + ``store._db = None``); unpinned, each read resolves the scope so a multiplexed profile's + writes reach its own store.""" pinned = self._pinned_db() if pinned is not _DB_UNPINNED: return pinned @@ -96,15 +82,11 @@ class SessionPersistenceMixin: @property def _routing_db(self): - """The one store that owns the routing index, whatever scope is active. - - ``_entries`` is one flat dict holding every profile's keys, so it must - persist to ONE file (``_routing_home``), not whichever profile is - scoped — otherwise a mid-turn rewrite and the unscoped startup load see - different copies and crash markers under a secondary profile go - unrecovered. A pinned handle still wins; bare test instances lacking - the handle cache report no DB. - """ + """The one store that owns the routing index, whatever scope is active. ``_entries`` is one + flat dict holding every profile's keys, so it must persist to ONE file (``_routing_home``), + not whichever profile is scoped — otherwise a mid-turn rewrite and the unscoped startup + load see different copies and crash markers under a secondary profile go unrecovered. A + pinned handle still wins; bare test instances lacking the handle cache report no DB.""" pinned = self._pinned_db() if pinned is not _DB_UNPINNED: return pinned @@ -117,12 +99,9 @@ class SessionPersistenceMixin: return None def _named_profile_for_key(self, session_key: Optional[str]) -> Optional[str]: - """The non-default profile that owns *session_key*, or None. - - None means the ambient store is authoritative (multiplexing off, or legacy ``agent:main`` - namespace). It deliberately does NOT cover "that profile has no directory" — ownership and - resolvability are separate questions that ``_db_for_key`` answers separately. - """ + """The non-default profile that owns *session_key*, or None (ambient store is authoritative: + multiplexing off, or legacy ``agent:main``). Deliberately does NOT cover "that profile has + no directory" — ownership and resolvability are separate questions for ``_db_for_key``.""" if not getattr(self.config, "multiplex_profiles", False): return None profile = self._profile_from_session_key(session_key) @@ -131,8 +110,8 @@ class SessionPersistenceMixin: return profile def _profile_home_for_key(self, session_key: Optional[str]) -> Optional[Path]: - """HERMES_HOME of the profile that owns *session_key*, or None (no named - owner, or the owner's directory could not be resolved).""" + """HERMES_HOME of the profile owning *session_key*, or None (no named owner or + unresolvable).""" profile = self._named_profile_for_key(session_key) if profile is None: return None @@ -148,21 +127,18 @@ class SessionPersistenceMixin: except Exception as exc: logger.debug("Could not resolve profile home for %r: %s", session_key, exc) home = None - # Only hits are memoized: a profile directory can be provisioned - # *after* startup (enrollment bridge), and a cached miss would pin - # that profile's rows to the ambient store for the process lifetime. + # Only hits are memoized: a profile directory can be provisioned *after* startup (enrollment + # bridge), and a cached miss would pin that profile's rows to the ambient store for life. if home is not None: cache[profile] = home return home def _db_for_key(self, session_key: Optional[str]): - """The SessionDB holding *session_key*'s rows, whatever scope is active. - - ``_db`` follows the ambient HERMES_HOME that only the inbound message - path installs; background work (expiry watcher) runs unscoped and would - write profile rows into the ROOT store until the stale-route self-heal - drops a live conversation. The owning profile is encoded in the key. - """ + """The SessionDB holding *session_key*'s rows, whatever scope is active. ``_db`` follows the + ambient HERMES_HOME that only the inbound message path installs; background work (expiry + watcher) runs unscoped and would write profile rows into the ROOT store until the + stale-route self-heal drops a live conversation. The owning profile is encoded in the + key.""" pinned = self._pinned_db() if pinned is not _DB_UNPINNED: return pinned @@ -171,28 +147,21 @@ class SessionPersistenceMixin: return self._db home = self._profile_home_for_key(session_key) if home is None: - # Named owner we cannot resolve (not provisioned yet, or lookup failed). Falling back to - # the ambient store would split ONE session identity across two physical stores — fail - # closed; callers already handle a missing DB. + # Falling back to the ambient store would split ONE session identity across two + # physical stores — fail closed; callers already handle a missing DB. logger.warning( - "gateway.session: profile %r has no resolvable home (key %r); " - "refusing to fall back to the ambient store", - profile, session_key, - ) + "gateway.session: profile %r has no resolvable home (key %r); refusing to fall " + "back to the ambient store", profile, session_key) return None try: return self._open_session_db_for_active_scope(db_path=home / "state.db") except Exception: - # Same contract as ``_db``: a failed open degrades to JSONL fallback. - return None + return None # same contract as ``_db``: a failed open degrades to JSONL fallback def _owner_key_for_session_id(self, session_id: Optional[str]) -> Optional[str]: - """The routing key that owns *session_id*, or None. - - The published index is authoritative; ``_session_owner_hints`` covers - the window where ownership is proven but routing not yet published. - Deliberately lock-free: several callers already hold ``_lock``. - """ + """The routing key that owns *session_id*, or None. The published index is authoritative; + ``_session_owner_hints`` covers the window where ownership is proven but routing not yet + published. Deliberately lock-free: several callers already hold ``_lock``.""" if not session_id: return None try: @@ -204,22 +173,18 @@ class SessionPersistenceMixin: return (getattr(self, "_session_owner_hints", None) or {}).get(session_id) def _db_for_session_id(self, session_id: Optional[str]): - """The SessionDB holding *session_id*'s row (owner recovered from the - index or a pre-published hint; unknown ids fall back to the ambient store).""" + """The SessionDB holding *session_id*'s row (owner from the index or a pre-published hint; + unknown ids fall back to the ambient store).""" if not session_id: return self._db return self._db_for_key(self._owner_key_for_session_id(session_id)) def close_all_db_handles(self) -> None: - """Close every SessionDB handle this store opened (one per path). - - Closing only ``store._db`` would strand secondary profiles' handles with - their WAL lock held ('database is locked' on restart). Drained under the - lock, closed outside it; a pinned handle is the pinner's to close. - """ + """Close every SessionDB handle this store opened (one per path). Closing only ``store._db`` + would strand secondary profiles' handles with their WAL lock held ('database is locked' on + restart). Drained under the lock, closed outside it; a pinned handle is the pinner's.""" def _close(db) -> None: - # Shared instances no-op on close(); release the refcount instead. - from hermes_state import release_or_close + from hermes_state import release_or_close # shared instances no-op on close() try: release_or_close(db) except Exception as exc: @@ -238,8 +203,8 @@ class SessionPersistenceMixin: return self._entries.get(session_key) def _routing_scope(self) -> str: - """Namespace for this store's gateway_routing rows: the resolved - sessions_dir, so stores with different dirs never share entries.""" + """Namespace for this store's gateway_routing rows: the resolved sessions_dir, so stores + with different dirs never share entries.""" try: return str(Path(self.sessions_dir).resolve()) except Exception: @@ -252,8 +217,8 @@ class SessionPersistenceMixin: return method if callable(method) else None def _load_routing_rows_locked(self) -> bool: - """Load state.db routing entries into ``_entries``; False when there is - no loader or the load failed (warned). Lock held.""" + """Load state.db routing entries into ``_entries``; False when there is no loader or the + load failed (warned). Lock held.""" loader = self._routing_db_method("load_gateway_routing_entries") if loader is None: return False @@ -280,28 +245,19 @@ class SessionPersistenceMixin: return None def _ensure_loaded_locked(self) -> None: - """Load the routing index. Must be called with self._lock held. - - state.db ``gateway_routing`` is primary; sessions.json is the legacy - import for keys the DB lacks (persisted to the DB on the next _save). - """ + """Load the routing index (lock held). state.db ``gateway_routing`` is primary; + sessions.json is the legacy import for keys the DB lacks (persisted on the next _save).""" if self._loaded: self._reconcile_recovered_routing_locked() return - self.sessions_dir.mkdir(parents=True, exist_ok=True) - db_load_succeeded = self._load_routing_rows_locked() db_had_entries = db_load_succeeded and bool(self._entries) - self._import_legacy_sessions_json(db_had_entries) - self._loaded = True self._routing_db_loaded = db_load_succeeded self._routing_fallback_baseline = None if db_load_succeeded else self._entries_as_dicts() - - # A hard crash skips graceful shutdown and leaves sessions.json - # pointing at ended sessions; self-heal before the first message. + # A hard crash skips graceful shutdown and leaves sessions.json pointing at ended sessions. self._prune_stale_sessions_locked() def _import_legacy_sessions_json(self, db_had_entries: bool) -> None: @@ -318,12 +274,10 @@ class SessionPersistenceMixin: # "_"-prefixed keys are sentinels (e.g. "_README"), not entries. if key.startswith("_") or key in self._entries: continue - # A non-dict entry (corrupt file) must not abort the whole load. - if not isinstance(entry_data, dict): + if not isinstance(entry_data, dict): # corrupt file must not abort the whole load logger.warning( - "Skipping invalid session entry %r: expected dict, got %s", - key, type(entry_data).__name__, - ) + "Skipping invalid session entry %r: expected dict, got %s", key, + type(entry_data).__name__) continue try: self._entries[key] = SessionEntry.from_dict(entry_data) @@ -332,28 +286,23 @@ class SessionPersistenceMixin: logger.warning("Skipping invalid session entry %r: %s", key, e) if imported and db_had_entries: logger.info( - "gateway.session: imported %d legacy sessions.json " - "entr%s missing from state.db routing table", - imported, "y" if imported == 1 else "ies", - ) + "gateway.session: imported %d legacy sessions.json entr%s missing from " + "state.db routing table", imported, "y" if imported == 1 else "ies") except Exception as e: print(f"[gateway] Warning: Failed to load sessions: {e}") def _prune_stale_sessions_locked(self) -> None: - """Remove routing entries whose session has ended in state.db (startup, lock held). - - Stale == ``end_reason IS NOT NULL``. Rows absent from the DB are kept; - a ``None`` DB handle is a no-op; DB errors are non-fatal. - """ + """Remove routing entries whose session has ended in state.db (startup, lock held). Stale == + ``end_reason IS NOT NULL``; rows absent from the DB are kept; a ``None`` DB handle is a + no-op; DB errors are non-fatal.""" if not self._entries: return - stale_keys: list = [] recovered_keys = 0 try: for key, entry in self._entries.items(): - # Ask the store that owns the key, not the ambient handle, or a - # live secondary-profile session gets pruned on the root copy. + # Ask the store that owns the key, not the ambient handle, or a live + # secondary-profile session gets pruned on the root copy. db = self._db_for_key(key) if db is None: continue @@ -369,58 +318,44 @@ class SessionPersistenceMixin: except Exception as exc: logger.warning("gateway.session: stale-entry pruning skipped due to DB error: %s", exc) return - for key in stale_keys: del self._entries[key] - if stale_keys or recovered_keys: self._save() def _stale_entry_verdict(self, key: str, entry, row): - """For a routing entry whose row has ended: ``"prune"``, a replacement - entry (repoint), or None (keep as-is).""" + """For a routing entry whose row has ended: ``"prune"``, a replacement entry (repoint), or + None (keep as-is).""" from gateway.session import _now recovered_entry = None if entry.origin is not None: try: recovered_entry = self._recover_session_from_db( - session_key=key, source=entry.origin, now=_now(), raise_on_lookup_error=True, - ) + session_key=key, source=entry.origin, now=_now(), raise_on_lookup_error=True) except Exception as exc: # Indeterminate: keep the only routing handle. logger.debug( - "gateway.session: recovery lookup failed for stale " - "sessions.json entry %r -> %s: %s", - key, entry.session_id, exc, - ) + "gateway.session: recovery lookup failed for stale sessions.json entry %r -> " + "%s: %s", key, entry.session_id, exc) return None - # Compression-ended parent with a newer live child for the same peer: repoint instead of # dropping, or queued/resume-pending work vanishes until the next message. if recovered_entry is not None and recovered_entry.session_id != entry.session_id: logger.warning( - "gateway.session: repointing stale sessions.json entry " - "%r from ended %s (end_reason=%r) to recovered %s", - key, entry.session_id, row["end_reason"], recovered_entry.session_id, - ) + "gateway.session: repointing stale sessions.json entry %r from ended %s " + "(end_reason=%r) to recovered %s", key, entry.session_id, row["end_reason"], + recovered_entry.session_id) return recovered_entry - - # Same-id recovery == successful resume: keep the ORIGINAL entry object - # (the recovered one is rebuilt minimal and would drop counters, - # model_override, resume markers, metadata). Nothing changes, no save. + # Same-id recovery == successful resume: keep the ORIGINAL entry object (the recovered one + # is rebuilt minimal and would drop counters, model_override, resume markers, metadata). if recovered_entry is not None: logger.info( - "gateway.session: reopened ended session %s for " - "sessions.json entry %r (end_reason=%r); keeping route", - entry.session_id, key, row["end_reason"], - ) + "gateway.session: reopened ended session %s for sessions.json entry %r " + "(end_reason=%r); keeping route", entry.session_id, key, row["end_reason"]) return None - logger.warning( - "gateway.session: pruning stale sessions.json entry " - "%r -> %s (end_reason=%r); left by a crashed gateway", - key, entry.session_id, row["end_reason"], - ) + "gateway.session: pruning stale sessions.json entry %r -> %s (end_reason=%r); left by " + "a crashed gateway", key, entry.session_id, row["end_reason"]) return "prune" def _entries_as_dicts(self) -> Dict[str, Any]: @@ -433,12 +368,9 @@ class SessionPersistenceMixin: self._persist_routing_data(data, generation) def _next_routing_generation_locked(self) -> int: - """Bump and return the shared routing counter. Caller holds ``_lock``. - - Full snapshots AND single-entry fast saves MUST allocate from this one - counter: the stale-write protection is a total order over - serialization times and silently breaks otherwise. - """ + """Bump and return the shared routing counter (lock held). Full snapshots AND single-entry + fast saves MUST allocate from this one counter: the stale-write protection is a total order + over serialization times and silently breaks otherwise.""" self._routing_generation = getattr(self, "_routing_generation", 0) + 1 return self._routing_generation @@ -447,7 +379,6 @@ class SessionPersistenceMixin: baseline = getattr(self, "_routing_fallback_baseline", None) if getattr(self, "_routing_db_loaded", False) or baseline is None: return - loader = self._routing_db_method("load_gateway_routing_entries") if loader is None: return @@ -456,24 +387,19 @@ class SessionPersistenceMixin: except Exception as exc: logger.warning("gateway.session: recovered state.db routing load failed: %s", exc) return - current = self._entries_as_dicts() for key, entry_json in durable.items(): durable_entry = self._routing_entry_from_json(key, entry_json) if durable_entry is None: continue - if key not in baseline: - # A key created while on fallback wins over a DB-only key; - # otherwise restore the authoritative row that fallback never saw. + # A key created while on fallback wins over a DB-only key; otherwise restore the + # authoritative row that fallback never saw. self._entries.setdefault(key, durable_entry) elif key not in current: - # The key was loaded from fallback and deliberately removed. - continue + continue # loaded from fallback and deliberately removed elif current[key] == baseline[key]: - # Unchanged fallback data yields to the authoritative DB copy. - self._entries[key] = durable_entry - + self._entries[key] = durable_entry # unchanged fallback data yields to the DB copy self._routing_db_loaded = True self._routing_fallback_baseline = None @@ -487,8 +413,8 @@ class SessionPersistenceMixin: with self._lazy("_save_lock", threading.Lock): if generation <= getattr(self, "_persisted_routing_generation", 0): return - # Fold in fast upserts numbered above this snapshot: they were - # serialized after us and a delayed full rewrite must not regress them. + # Fold in fast upserts numbered above this snapshot: they were serialized after us and + # a delayed full rewrite must not regress them. fast_persisted = getattr(self, "_fast_persisted_entries", None) if fast_persisted: for key, (revision, entry_json) in fast_persisted.items(): @@ -508,16 +434,14 @@ class SessionPersistenceMixin: except Exception as exc: if not db_saved: raise - # state.db is authoritative. A failed legacy mirror must not - # report the already-committed primary write as failed. + # state.db is authoritative: a failed legacy mirror must not report the + # already-committed primary write as failed. logger.warning( - "gateway.session: sessions.json mirror save failed " - "after state.db commit: %s", - exc, - ) + "gateway.session: sessions.json mirror save failed after state.db commit: " + "%s", exc) self._persisted_routing_generation = generation - # This rewrite supersedes fast records at or below its - # generation; newer ones stay for the next delayed full writer. + # This rewrite supersedes fast records at or below its generation; newer ones stay for + # the next delayed full writer. if fast_persisted: for key in [k for k, (rev, _) in fast_persisted.items() if rev <= generation]: del fast_persisted[key] @@ -549,23 +473,17 @@ class SessionPersistenceMixin: def _save_entry( self, session_key: str, *, entry_data: Optional[Dict[str, Any]] = None, - lock_held: bool = False, - ) -> None: - """Persist ONE routing entry via UPSERT — the per-turn fast path - (a full rewrite fsyncs a multi-MB sessions.json, ~50ms at ~1100 keys). - - Invariants: the key -> session_id mapping never changes here — - structural transitions (create/recover/reset/switch/prune/heal) use the - full rewrite, which also refreshes the sessions.json mirror (it may lag - in metadata only; state.db stays primary). The entry is serialized under - ``_lock`` with a revision from the shared routing generation counter - (higher == same-or-newer); under ``_save_lock`` the upsert is skipped if - a full snapshot or a newer fast save of this key already persisted (the - reverse case lives in ``_persist_routing_data``). No DB or a failed - upsert falls back to the full rewrite so DB-less installs stay durable. - ``entry_data`` persists a candidate BEFORE it is published to the live - entry (failure-atomic transitions); the fallback carries the same candidate. - """ + lock_held: bool = False) -> None: + """Persist ONE routing entry via UPSERT — the per-turn fast path (a full rewrite fsyncs a + multi-MB sessions.json, ~50ms at ~1100 keys). The key -> session_id mapping never changes + here: structural transitions use the full rewrite, which also refreshes the sessions.json + mirror (it may lag in metadata only; state.db stays primary). The entry is serialized under + ``_lock`` with a revision from the shared routing generation counter; under ``_save_lock`` + the upsert is skipped if a full snapshot or a newer fast save of this key already persisted + (the reverse case lives in ``_persist_routing_data``). No DB or a failed upsert falls back + to the full rewrite so DB-less installs stay durable. ``entry_data`` persists a candidate + BEFORE it is published to the live entry (failure-atomic transitions); the fallback carries + the same candidate.""" def _capture() -> Optional[tuple[str, int, Optional[Dict[str, Any]]]]: entry = self._entries.get(session_key) if entry is None: @@ -601,10 +519,8 @@ class SessionPersistenceMixin: return except Exception as exc: logger.warning( - "gateway.session: single-entry routing save failed for %r " - "(%s); falling back to full index rewrite", - session_key, exc, - ) + "gateway.session: single-entry routing save failed for %r (%s); falling back " + "to full index rewrite", session_key, exc) if candidate_entry is not None: # Full-snapshot fallback carrying the candidate transition. fallback_data = _locked(self._entries_as_dicts) diff --git a/gateway/session_recovery.py b/gateway/session_recovery.py index 203d4ba65a..a35efa7842 100644 --- a/gateway/session_recovery.py +++ b/gateway/session_recovery.py @@ -1,9 +1,6 @@ -"""SessionStore durable-row recovery: session-key generation, legacy Slack key -migration, rebuilding a routing entry from state.db, and the SQLite side of -routing transitions (promote/reopen/create/peer). - -Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO. -""" +"""SessionStore durable-row recovery: session-key generation, legacy Slack key migration, rebuilding +a routing entry from state.db, and the SQLite side of routing transitions (promote/reopen/create/ +peer). Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.""" from __future__ import annotations @@ -36,8 +33,8 @@ class SessionRecoveryMixin: """SessionStore durable-row recovery and the SQLite side of routing transitions.""" def _resolve_profile_for_key(self, source: Optional[SessionSource] = None) -> Optional[str]: - """Profile namespace for session keys: None when multiplexing is off - (legacy ``agent:main``), else ``source.profile`` or the active profile.""" + """Profile namespace for session keys: None when multiplexing is off (legacy + ``agent:main``), else ``source.profile`` or the active profile.""" if not getattr(self.config, "multiplex_profiles", False): return None if source is not None and source.profile: @@ -68,45 +65,34 @@ class SessionRecoveryMixin: return "default" def _recovered_row_allowed_for_active_profile( - self, *, requested_session_key: str, recovered: Dict[str, Any], + self, *, requested_session_key: str, recovered: Dict[str, Any] ) -> bool: - """Prevent a gateway from reviving another profile's row. - - Single-profile: the row's namespace must match the ACTIVE profile. - Multiplexed: it must match the namespace of the requested key (the - active profile is meaningless there). Keyless rows stay adoptable. - """ + """Prevent a gateway from reviving another profile's row. Single-profile: the row's + namespace must match the ACTIVE profile. Multiplexed: it must match the requested key's + namespace (the active profile is meaningless there). Keyless rows stay adoptable.""" recovered_key = str(recovered.get("session_key") or "") if not recovered_key or recovered_key == requested_session_key: return True - recovered_profile = self._profile_from_session_key(recovered_key) if recovered_profile is None: return True - if getattr(self.config, "multiplex_profiles", False): requested_profile = self._profile_from_session_key(requested_session_key) return requested_profile is None or recovered_profile == requested_profile - return recovered_profile == self._active_profile_name() def _generate_session_key(self, source: SessionSource, key_source: Optional[SessionSource] = None) -> str: - """Session key for *source* (profile resolved from *source*, key built - from *key_source* when given).""" + """Session key for *source* (profile from *source*; key from *key_source* if given).""" from gateway.session import build_session_key return build_session_key( key_source if key_source is not None else source, group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False), - profile=self._resolve_profile_for_key(source), - ) + profile=self._resolve_profile_for_key(source)) def _legacy_slack_session_key(self, source: SessionSource) -> Optional[str]: - """Pre-workspace Slack key for an explicitly scoped source. - - Deliberately Slack-only; an unscoped Slack session may be claimed by - only one workspace because its old key cannot distinguish teams. - """ + """Pre-workspace Slack key for an explicitly scoped source. Deliberately Slack-only: an + unscoped Slack session may be claimed by only one workspace (old key cannot tell teams).""" if source.platform != Platform.SLACK or not source.scope_id: return None return self._generate_session_key(source, replace(source, scope_id=None, guild_id=None)) @@ -126,12 +112,9 @@ class SessionRecoveryMixin: def _recovered_row_matches_source_scope( recovered: Dict[str, Any], source: SessionSource ) -> bool: - """Reject recovered rows whose recorded origin belongs to another workspace. - - A workspace-scoped Slack lookup adopts a row only if its origin_json - names the same scope_id; rows without a parseable origin are rejected - (an unattributable transcript is exactly the ambiguity to avoid). - """ + """Reject recovered rows whose origin belongs to another workspace: a workspace-scoped Slack + lookup adopts a row only if its origin_json names the same scope_id; rows without a + parseable origin are rejected (an unattributable transcript is exactly the ambiguity).""" if source.platform != Platform.SLACK or source.chat_type == "dm" or not source.scope_id: return True try: @@ -165,20 +148,14 @@ class SessionRecoveryMixin: session_key=session_key, session_id=str(row["id"]), created_at=created_at, updated_at=updated_at, origin=source, display_name=source.chat_name, platform=source.platform, chat_type=source.chat_type, - reset_had_activity=bool(had_activity), - ) + reset_had_activity=bool(had_activity)) def _find_gateway_session_row( self, *, session_key: str, source: SessionSource, allow_peer_fallback: bool, - raise_on_lookup_error: bool = False, - ) -> Optional[Dict[str, Any]]: - """Query one durable gateway session row. - - Scoped Slack lookups disable SessionDB's platform/chat/user fallback: - that tuple does not contain a workspace id and could therefore revive - another team's session. The caller performs one explicit exact lookup - of the old unscoped key instead. - """ + raise_on_lookup_error: bool = False) -> Optional[Dict[str, Any]]: + """Query one durable gateway session row. Scoped Slack lookups disable SessionDB's + platform/chat/user fallback: that tuple has no workspace id and could revive another team's + session; the caller performs one explicit exact lookup of the old unscoped key instead.""" db = self._db_for_key(session_key) finder = getattr(db, "find_latest_gateway_session_for_peer", None) if db else None if not callable(finder): @@ -188,8 +165,7 @@ class SessionRecoveryMixin: source=source.platform.value, user_id=source.user_id, session_key=session_key, chat_id=source.chat_id if allow_peer_fallback else None, chat_type=source.chat_type if allow_peer_fallback else None, - thread_id=source.thread_id, - ) + thread_id=source.thread_id) except Exception as exc: logger.debug("Gateway session DB recovery failed for %s: %s", session_key, exc) if raise_on_lookup_error: @@ -198,18 +174,13 @@ class SessionRecoveryMixin: def _recover_session_from_db( self, *, session_key: str, source: SessionSource, now: datetime, - raise_on_lookup_error: bool = False, - ) -> Optional[SessionEntry]: - """Rebuild a missing session-key mapping from durable state.db data. - - Returns ``None`` when no row is recoverable, or when the recovered - session is already overdue under the reset policy — the row is then - durably promoted to a reset boundary instead of resurrected. - """ + raise_on_lookup_error: bool = False) -> Optional[SessionEntry]: + """Rebuild a missing session-key mapping from durable state.db data. ``None`` when no row is + recoverable, or when the recovered session is already overdue under the reset policy — the + row is then durably promoted to a reset boundary instead of resurrected.""" entry, migrated_legacy = self._query_recoverable_row( session_key=session_key, source=source, now=now, - raise_on_lookup_error=raise_on_lookup_error, - ) + raise_on_lookup_error=raise_on_lookup_error) if entry is None: return None reset_reason = self._should_reset(entry, source) @@ -217,83 +188,62 @@ class SessionRecoveryMixin: self._promote_session_reset( session_key, entry.session_id, reset_reason, log=lambda exc: logger.debug( - "Gateway recovered-session reset promotion failed for %s: %s", - session_key, exc, + "Gateway recovered-session reset promotion failed for %s: %s", session_key, exc, ), ) return None self._reopen_session_row(session_key, entry.session_id) if migrated_legacy: self._record_gateway_session_peer( - entry.session_id, session_key, source, display_name=entry.display_name, - ) + entry.session_id, session_key, source, display_name=entry.display_name) return entry def _query_recoverable_session(self, *, session_key, source, now): - """DB-only half of _recover_session_from_db (no lock needed). - - Returns a SessionEntry or None. Caller assigns _entries[key] under - lock. The row is NOT reopened here: the caller evaluates reset policy - first (an agent_close/ws_orphan row may need promotion to a real reset - boundary instead). - """ + """DB-only half of _recover_session_from_db (no lock needed): a SessionEntry or None; the + caller assigns _entries[key] under lock. The row is NOT reopened here: the caller evaluates + reset policy first (an agent_close/ws_orphan row may need promotion to a real reset).""" entry, migrated_legacy = self._query_recoverable_row( - session_key=session_key, source=source, now=now, - ) + session_key=session_key, source=source, now=now) if entry is not None and migrated_legacy: self._record_gateway_session_peer( - entry.session_id, session_key, source, display_name=entry.display_name, - ) + entry.session_id, session_key, source, display_name=entry.display_name) return entry def _query_recoverable_row( self, *, session_key, source, now, raise_on_lookup_error=False, ) -> tuple[Optional[SessionEntry], bool]: - """Find and gate a recoverable row -> (entry or None, migrated_legacy). - - The legacy (pre-workspace) Slack key fallback lives here: exact-key lookup, claimed once per - process; ``migrated_legacy`` tells the caller to rewrite the peer row to the scoped key. - """ + """Find and gate a recoverable row -> (entry or None, migrated_legacy). The legacy + (pre-workspace) Slack key fallback lives here: exact-key lookup, claimed once per process; + ``migrated_legacy`` tells the caller to rewrite the peer row to the scoped key.""" legacy_key = self._legacy_slack_session_key(source) recovered = self._find_gateway_session_row( session_key=session_key, source=source, allow_peer_fallback=legacy_key is None, - raise_on_lookup_error=raise_on_lookup_error, - ) + raise_on_lookup_error=raise_on_lookup_error) migrated_legacy = False if not recovered and legacy_key and self._claim_legacy_slack_key(legacy_key): recovered = self._find_gateway_session_row( session_key=legacy_key, source=source, allow_peer_fallback=False, - raise_on_lookup_error=raise_on_lookup_error, - ) + raise_on_lookup_error=raise_on_lookup_error) migrated_legacy = bool(recovered) if not isinstance(recovered, dict): return None, False if not self._recovered_row_matches_source_scope(recovered, source): return None, False if not self._recovered_row_allowed_for_active_profile( - requested_session_key=session_key, recovered=recovered, - ): + requested_session_key=session_key, recovered=recovered): logger.warning( - "Gateway session DB recovery ignored %s for %s because " - "the row belongs to a different profile", - recovered.get("session_key"), - session_key, - ) + "Gateway session DB recovery ignored %s for %s because the row belongs to a " + "different profile", recovered.get("session_key"), session_key) return None, False entry = self._create_entry_from_recovered_row( - row=recovered, session_key=session_key, source=source, now=now, - ) + row=recovered, session_key=session_key, source=source, now=now) return entry, migrated_legacy def _promote_session_reset(self, session_key: str, session_id: str, reason: str, *, log) -> None: - """End *session_id* with *reason* via ``promote_to_session_reset``. - - Promote (not plain ``end_session``): a row already ended with a - recoverable accidental reason (agent_close / ws_orphan_reap) must be - upgraded to the explicit boundary, or stale-route recovery resurrects - it over the reset. Falls back to ``end_session`` on old SessionDBs. - ``log(exc)`` reports failures (each caller has its own message). - """ + """End *session_id* with *reason* via ``promote_to_session_reset`` (``end_session`` on old + SessionDBs). Promote, not plain end: a row already ended with a recoverable accidental + reason (agent_close / ws_orphan_reap) must be upgraded to the explicit boundary, or + stale-route recovery resurrects it over the reset. ``log(exc)`` reports failures.""" try: db = self._db_for_key(session_key) promote = getattr(db, "promote_to_session_reset", None) @@ -316,8 +266,7 @@ class SessionRecoveryMixin: def _record_gateway_session_peer( self, session_id: str, session_key: str, source: Optional[SessionSource], - display_name: Optional[str] = None, include_compression_ancestors: bool = False, - ) -> None: + display_name: Optional[str] = None, include_compression_ancestors: bool = False) -> None: """Persist the routing peer for an existing gateway session row.""" db = self._db_for_key(session_key) if not db or not source: @@ -327,17 +276,14 @@ class SessionRecoveryMixin: return peer = dict( source=source.platform.value, user_id=source.user_id, session_key=session_key, - chat_id=source.chat_id, chat_type=source.chat_type, thread_id=source.thread_id, - ) + chat_id=source.chat_id, chat_type=source.chat_type, thread_id=source.thread_id) try: recorder( session_id, **peer, display_name=display_name or source.chat_name, origin_json=_origin_json(source), - include_compression_ancestors=include_compression_ancestors, - ) + include_compression_ancestors=include_compression_ancestors) except TypeError: - # Older SessionDB without display_name/origin_json kwargs. - try: + try: # older SessionDB without display_name/origin_json kwargs recorder(session_id, **peer) except Exception as exc: logger.debug("Gateway session peer record failed for %s: %s", session_key, exc) @@ -345,14 +291,10 @@ class SessionRecoveryMixin: logger.debug("Gateway session peer record failed for %s: %s", session_key, exc) def _adopt_legacy_slack_entry(self, source: SessionSource, session_key: str) -> None: - """One-time migration of pre-workspace-scope Slack keys. - - MOVE (not copy) the legacy entry so a second workspace with identical - Slack ids cannot attach to the same transcript. Adopt when the legacy - origin names the same workspace; a scope-less DM is claimed once by - the first workspace; a scope-less channel/group is refused (channel - ids collide across workspaces). - """ + """One-time migration of pre-workspace-scope Slack keys: MOVE (not copy) the legacy entry so + a second workspace with identical Slack ids cannot attach to the same transcript. Adopt when + the legacy origin names the same workspace; a scope-less DM is claimed once by the first + workspace; a scope-less channel/group is refused (channel ids collide across workspaces).""" legacy_key = self._legacy_slack_session_key(source) if not legacy_key: return @@ -376,51 +318,40 @@ class SessionRecoveryMixin: if migrated is not None: self._save_entries() self._record_gateway_session_peer( - migrated.session_id, session_key, source, display_name=migrated.display_name, - ) + migrated.session_id, session_key, source, display_name=migrated.display_name) def _finish_route_transition( self, session_key: str, *, end_session_id: Optional[str], end_reason: str, create_kwargs: Optional[Dict[str, Any]], origin: Optional[SessionSource], - display_name: Optional[str], during: str = "", - ) -> None: - """SQLite side of a routing transition, outside ``_lock``. - - Promotes the predecessor row to an explicit reset boundary (with the - specific reason so state.db is auditable, e.g. ``resume_pending_expired`` - vs a plain ``session_reset``), then INSERTs the new row + routing peer. - Both are best-effort: failures are warned and self-healed by the next - per-turn peer refresh. - """ + display_name: Optional[str], during: str = "") -> None: + """SQLite side of a routing transition, outside ``_lock``: promote the predecessor row to an + explicit reset boundary (with the specific reason so state.db is auditable, e.g. + ``resume_pending_expired`` vs plain ``session_reset``), then INSERT the new row + routing + peer. Both best-effort: failures are warned and self-healed by the next peer refresh.""" if self._db_for_key(session_key) and end_session_id: self._promote_session_reset( session_key, end_session_id, end_reason, log=lambda e: logger.warning( - "Failed to end predecessor session row %s for %s%s: %s — " - "the old row remains open and may win restart recovery " - "until the next successful peer refresh", - end_session_id, session_key, during, e, - ), + "Failed to end predecessor session row %s for %s%s: %s — the old row remains " + "open and may win restart recovery until the next successful peer refresh", + end_session_id, session_key, during, e), ) if self._db_for_key(session_key) and create_kwargs: self._create_session_row( session_key, create_kwargs, origin, display_name, log=lambda e: logger.warning( - "Failed to create session row %s for %s%s: %s — deferring " - "to the self-healing peer refresh on the next turn", - create_kwargs.get("session_id"), session_key, during, e, - ), + "Failed to create session row %s for %s%s: %s — deferring to the " + "self-healing peer refresh on the next turn", + create_kwargs.get("session_id"), session_key, during, e), ) @staticmethod def _session_create_kwargs( *, session_id, session_key, origin, source_value, display_name, parent_session_id, ) -> Dict[str, Any]: - """kwargs for ``SessionDB.create_session``. - - Identity (origin_json) and lineage (parent/_reset_from) land atomically - in the INSERT so a crash right after cannot strand the row unroutable. - """ + """kwargs for ``SessionDB.create_session``. Identity (origin_json) and lineage + (parent/_reset_from) land atomically in the INSERT so a crash right after cannot strand the + row unroutable.""" return { "session_id": session_id, "source": source_value, @@ -437,15 +368,12 @@ class SessionRecoveryMixin: } def _create_session_row(self, session_key, db_create_kwargs, origin, display_name, *, log) -> None: - """INSERT a session row and record its routing peer; ``log(exc)`` on failure. - - A failed create is a routing hazard (visible warning), but the row is - self-healed with full identity by the next per-turn peer refresh. - """ + """INSERT a session row and record its routing peer; ``log(exc)`` on failure. A failed + create is a routing hazard (visible warning), but the row is self-healed with full identity + by the next per-turn peer refresh.""" try: self._db_for_key(session_key).create_session(**db_create_kwargs) self._record_gateway_session_peer( - db_create_kwargs["session_id"], session_key, origin, display_name=display_name, - ) + db_create_kwargs["session_id"], session_key, origin, display_name=display_name) except Exception as e: log(e) diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index 92a0d49a1b..5211ca1c27 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -1,8 +1,6 @@ -"""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. -""" +"""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 @@ -35,11 +33,10 @@ def _plain_text(content) -> str: 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.""" + """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 @@ -52,8 +49,8 @@ class SessionTranscriptMixin: _MAX_PENDING_PER_SESSION = 200 # in-memory pending messages per session (DB broken) 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).""" + """Latest compression continuation for *session_id* (heals a mapping left pointing at a + compressed parent by a restart or failed send).""" if not session_id: return session_id db = self._db_for_session_id(session_id) @@ -67,8 +64,7 @@ class SessionTranscriptMixin: def _heal_compression_tip_locked( self, entry: "SessionEntry", original_session_id: Optional[str], - canonical_session_id: Optional[str], - ) -> bool: + canonical_session_id: Optional[str]) -> bool: """Rewrite *entry* to the compression continuation if stale. Lock held.""" if ( not original_session_id @@ -78,24 +74,20 @@ class SessionTranscriptMixin: ): return False logger.info( - "SessionStore healed compressed session mapping: %s -> %s", - entry.session_id, canonical_session_id, - ) + "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. - """ + """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: @@ -106,8 +98,7 @@ class SessionTranscriptMixin: return None if not self._heal_compression_tip_locked(entry, expected_session_id, target_session_id): return None - # Bookkeeping, not user activity: leave ``updated_at`` alone. - self._save() + self._save() # bookkeeping, not user activity: leave ``updated_at`` alone return entry def _get_transcript_drain_lock(self): @@ -139,34 +130,25 @@ class SessionTranscriptMixin: 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, - ) + "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, - ) + "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. - - Retrying cannot succeed and the FTS rebuild must never run on this - handle; the pending queue goes to the on-disk spool + JSONL fallback. - """ + """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, - ) + "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) @@ -174,30 +156,24 @@ class SessionTranscriptMixin: for dropped in remaining: try: from gateway.shutdown_flush import spool_dropped_transcript_message - spool_dropped_transcript_message(session_id, dropped) except Exception: logger.warning( "pending fallback failed for replaced state.db transcript on %s", session_id, - exc_info=True, - ) + 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, - ) + 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 not published - until after its write succeeds, so a by-id lookup would fall back to - the ambient store. - """ + """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) if owner_db is None: return "" @@ -211,13 +187,10 @@ class SessionTranscriptMixin: 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 publish - 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. - """ + """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, []) @@ -230,8 +203,7 @@ class SessionTranscriptMixin: 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), - ) + previous_failures, self._transcript_append_failures.get(child_id, 0)) self._transcript_reroutes[session_id] = child_id return pending @@ -247,8 +219,8 @@ class SessionTranscriptMixin: _hints.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.""" + """Append a message to a session's transcript (SQLite), draining the per-session retry + queue.""" with self._transcript_retry_lock: pending = self._enqueue_transcript_message(session_id, message) msg = pending[0] @@ -272,11 +244,9 @@ class SessionTranscriptMixin: from hermes_state import ( CompressionSessionClosedError, StateDbCorruptError, StateDbReplacedError, ) - 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) @@ -293,8 +263,7 @@ class SessionTranscriptMixin: else: with self._transcript_retry_lock: pending = self._migrate_transcript_queue_to_child( - session_id, queue_session_id, child_id, pending, msg - ) + 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: @@ -303,15 +272,13 @@ class SessionTranscriptMixin: session_id = child_id continue else: - # Permanent routing invariant failure, not a transient - # outage: drop it so it cannot poison later writes. + # 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, - ) + "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: @@ -326,10 +293,8 @@ class SessionTranscriptMixin: failures = self._transcript_append_failures.get(session_id, 0) + 1 self._transcript_append_failures[session_id] = failures logger.warning( - "Session DB transcript append failed for %s " - "(failure_count=%d, pending=%d); will retry: %s", - session_id, failures, len(pending), exc, - ) + "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: @@ -343,17 +308,13 @@ class SessionTranscriptMixin: continue 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 into the caller. - """ + """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 - _replayed, remaining = drain_transcript_spool( session_id, lambda message: self._append_transcript_message(session_id, message), ) @@ -366,11 +327,10 @@ class SessionTranscriptMixin: """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. + # 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" - ) + f"no owning session store for {session_id}; deferring transcript write") is_assistant = message.get("role") == "assistant" _db.append_message( session_id=session_id, @@ -387,8 +347,8 @@ class SessionTranscriptMixin: 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. + # 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. display_kind=message.get("display_kind"), @@ -397,19 +357,14 @@ class SessionTranscriptMixin: @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 (per - ``SessionDB._is_fts_write_corruption_error``) may authorize the - one-shot rebuild-and-retry. Everything else takes the retry path. - """ + """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.""" 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: @@ -425,11 +380,9 @@ class SessionTranscriptMixin: 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, - ) + "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() @@ -459,17 +412,13 @@ class SessionTranscriptMixin: 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. - """ + 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 @@ -477,8 +426,7 @@ class SessionTranscriptMixin: try: db.replace_messages( session_id, messages, active_only=active_only, - reject_active_turn_lease=reject_active_turn_lease, - ) + reject_active_turn_lease=reject_active_turn_lease) except Exception as e: logger.debug("Failed to rewrite transcript in DB: %s", e) return False @@ -486,12 +434,9 @@ class SessionTranscriptMixin: return True 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 - (compression rotation), then the durable compression tip — otherwise - the transcript "vanishes" while every message sits under the child. - """ + """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) @@ -503,32 +448,25 @@ class SessionTranscriptMixin: except Exception: pass try: - # repair_alternation: this feeds LIVE REPLAY; heal a durable - # user;user wedge once here instead of on every request. + # 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 - ) + 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 []. + # 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, - ) + "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 user turn). ``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 before anything changes. - """ + Returns ``{"rewound_count", "turns_undone", "target_text"}`` or ``None`` (no DB / no user + turn); ``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 before anything changes.""" db = self._db_for_session_id(session_id) if not db: return None @@ -537,13 +475,11 @@ class SessionTranscriptMixin: from agent.context_compressor import ( retryable_user_text, split_user_originated_turn, user_originated_turn_view, ) - try: expected_active_ids = db.get_active_message_ids(session_id) durable = db.get_messages_as_conversation(session_id, include_row_ids=True) user_indices = [ - index - for index, message in enumerate(durable) + index for index, message in enumerate(durable) if user_originated_turn_view(message) is not None ] if not user_indices: @@ -562,15 +498,14 @@ class SessionTranscriptMixin: logger.debug("rewind_session: failed to resolve canonical target: %s", e) return None if require_retryable_composite: - # Keep replay-policy failures distinct from persistence errors - # so /retry can explain why the selected carrier is unsafe. + # Keep replay-policy failures distinct from persistence errors so /retry can explain + # why the selected carrier is unsafe. target_text = retryable_user_text(target_view.get("content")) try: result = db.rewind_to_message( session_id, target_id, preserve_compaction_handoff=handoff is not None, expected_active_ids=expected_active_ids, - expected_target_content=target_view.get("content"), - ) + expected_target_content=target_view.get("content")) except ValueError as e: logger.debug("rewind_session: %s", e) return None @@ -578,8 +513,8 @@ class SessionTranscriptMixin: logger.debug("rewind_session: rewind_to_message failed: %s", e) return None self._clear_dirty_transcript(session_id) - # ``target_view`` is the live projection; a composite carrier's raw - # row holds the summary wrapper and must not be echoed as prompt. + # ``target_view`` is the live projection; a composite carrier's raw row holds the + # summary wrapper and must not be echoed as prompt. if not require_retryable_composite: target_text = _plain_text(target_view.get("content") or "") return {