"""Compression lineage, cooldown/streak counters, locks and turn leases for SessionDB. Mixin bound onto ``SessionDB`` via the MRO, built on its ``_read_ctx`` / ``_execute_write`` / ``_write_sql`` / ``_read_one`` primitives.""" from __future__ import annotations import json import logging import sqlite3 import time from typing import Any, Dict, List, Optional, Tuple from hermes_state_common import ( _BOUNDARY_END_REASONS, _COMPRESSION_LOCK_ROW_SQL as _LOCK_ROW_SQL, _ENDED_ROW_SQL, _ended_by_compression, _RESET_CHILD_SQL, _sql_json_extract, _sql_session_last_active, is_automatic_end_reason) # Log-record parity with the origin module (caplog tests pin "hermes_state"). logger = logging.getLogger("hermes_state") _COOLDOWN_ROW_SQL = ( "SELECT compression_failure_cooldown_until, compression_failure_error FROM sessions WHERE id = ?" ) # One forward step of get_compression_chain: the preferred continuation child of ``?``. A reset # fork is a separate user-visible conversation (_LISTABLE_CHILD_SQL already surfaces it as its # own row), so following it here would hijack the lineage tip projection onto the reset sibling # and make the real continuation invisible (#114271). _CHAIN_STEP_SQL = f""" SELECT child.id FROM sessions parent JOIN sessions child ON child.parent_session_id = parent.id WHERE parent.id = ? AND parent.end_reason = 'compression' AND {_sql_json_extract('child.model_config', '$._branched_from')} IS NULL AND {_sql_json_extract('child.model_config', '$._delegate_from')} IS NULL AND NOT ({_RESET_CHILD_SQL.format(a='child')}) AND COALESCE(child.source, '') != 'tool' ORDER BY CASE WHEN child.end_reason = 'compression' THEN 0 WHEN child.ended_at IS NULL THEN 1 ELSE 2 END, {_sql_session_last_active("child")} DESC, child.started_at DESC, child.id DESC LIMIT 1 """ # Turn-lease rows expired longer than this are swept by the next acquisition of any conversation. _TURN_LEASE_SWEEP_GRACE_S = 86400.0 def _cooldown_row(exists: bool, cooldown_until, error) -> Dict[str, Any]: return {"session_exists": exists, "cooldown_until": float(cooldown_until) if cooldown_until is not None else None, "error": error} def _claim_lease_row(conn, table: str, key_col: str, key: str, holder: str, now: float, expires_at: float, stale) -> Tuple[bool, Optional[str]]: """Single-transaction lease claim: DELETE a stale holder's row (``stale(holder, expires_at)``), INSERT OR IGNORE ours, then SELECT to confirm ownership (INSERT OR IGNORE gives no rowcount signal). Returns ``(acquired, reclaimed_holder)``.""" reclaimed_holder = None row = conn.execute(f"SELECT holder, expires_at FROM {table} WHERE {key_col} = ?", (key,)).fetchone() if row is not None and stale(row["holder"], row["expires_at"]): conn.execute(f"DELETE FROM {table} WHERE {key_col} = ? AND holder = ?", (key, row["holder"])) reclaimed_holder = row["holder"] conn.execute( f"INSERT OR IGNORE INTO {table} ({key_col}, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)", (key, holder, now, expires_at)) owner = conn.execute(f"SELECT holder FROM {table} WHERE {key_col} = ?", (key,)).fetchone() return owner is not None and owner["holder"] == holder, reclaimed_holder # Defensive bound on the forward compression-chain walk; ``seen`` guards cycles. # 100 truncated real compression lineages (~180 deep), stranding root→tip walks # on a stale mid id (#125041). Named so tests can simulate the REAL walk. _CHAIN_CAP = 1000 class SessionCompressionMixin: """Compression lineage, cooldown/streak counters, locks and turn leases.""" def reopen_if_explicitly_closed( self, session_id: str, *, provenance: str, patience_s: Optional[float] = None, ) -> Optional[str]: """Clear an explicit-close stamp (``tui_close``, ``cli_close``, ``webhook_complete``, ...) from a session a HOST has just proven is still routed to it, returning the reason cleared or None (#106459). Narrow twin of ``reopen_session()``: automatic stamps are left to publish (#88197); ``compression``, boundary (reset reasons, CLI ``new_session``) and stamps with a published continuation own lineage elsewhere and are never touched. The UPDATE is conditional on the exact stamp read, so a close landing between read and write survives. Only the routing host can make this call. Publication cannot: ``end_session()`` is first-stamp-wins, so a close made during a turn that began on a stale stamp is a no-op write. Turn-lease admission cannot: the TUI starts its worker before it reaches ``run_conversation()``, so ``session.close`` can stamp ``tui_close`` in between and a late lease would clear a deliberate close. Call it under the lock that makes the host's registry claim atomic with its teardown (#54878 on the routing table).""" if not session_id: return None def _do(conn): row = conn.execute(_ENDED_ROW_SQL, (session_id,)).fetchone() if row is None or row["ended_at"] is None: return None reason = row["end_reason"] if is_automatic_end_reason(reason) or reason == "compression" or reason in _BOUNDARY_END_REASONS: return None superseded = conn.execute( "SELECT 1 FROM sessions WHERE parent_session_id = ?" + self._NON_CONTINUATION_CHILD_FILTER_SQL.format(alias="") + " LIMIT 1", (session_id,) * 4).fetchone() if superseded is not None: return None conn.execute( "UPDATE sessions SET ended_at = NULL, end_reason = NULL " "WHERE id = ? AND ended_at = ? AND end_reason = ?", (session_id, row["ended_at"], reason)) return str(reason) reason = self._execute_write(_do, patience_s=patience_s) if reason is not None: logger.warning( "Session %s carried a stale %r end stamp while %s; cleared so the conversation can " "compress and a later close is recorded (#106459)", session_id, reason, provenance) return reason def find_live_compression_child(self, parent_session_id: str) -> Optional[Dict[str, Any]]: """The unique live direct child of a compression-ended session, else None. A stale agent whose parent was rotated elsewhere may recover only when the lineage names exactly one live continuation; more than one fails closed.""" if not parent_session_id: return None with self._read_ctx() as conn: if not _ended_by_compression(conn.execute(_ENDED_ROW_SQL, (parent_session_id,)).fetchone()): return None rows = conn.execute( """ SELECT s.*, COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved FROM sessions s LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash WHERE s.parent_session_id = ? AND s.ended_at IS NULL """ + self._NON_CONTINUATION_CHILD_FILTER_SQL.format(alias="s.") + """ ORDER BY s.started_at ASC LIMIT 2 """, (parent_session_id,) * 4, ).fetchall() return self._session_row_dict(rows[0]) if len(rows) == 1 else None def reopen_orphaned_compression_session(self, session_id: str) -> bool: """Reopen a compression parent only when no continuation was published (older builds could leave a closed parent after an interrupted handoff). Conservative: an active lease or any canonical child means another path owns the lineage.""" if not session_id: return False def _do(conn): if not _ended_by_compression(conn.execute(_ENDED_ROW_SQL, (session_id,)).fetchone()): return False # Any non-branch/non-delegate/non-reset/non-tool child is a continuation, ended or not. child = conn.execute( """ SELECT 1 FROM sessions WHERE parent_session_id = ? """ + self._NON_CONTINUATION_CHILD_FILTER_SQL.format(alias="") + """ LIMIT 1 """, (session_id,) * 4, ).fetchone() if child is not None: return False # refresh_compression_lock() lets an owner revive its own expired row, so reclaim # it inside this write txn: refresh-first makes the lease active and aborts # recovery; recovery-first deletes the holder so a refresh can't resurrect it. now = time.time() lock_row = conn.execute(_LOCK_ROW_SQL, (session_id,)).fetchone() if lock_row is not None: expires_at = lock_row["expires_at"] if expires_at is None or float(expires_at) >= now: return False deleted = conn.execute( "DELETE FROM compression_locks WHERE session_id = ? AND holder = ? AND expires_at = ?", (session_id, lock_row["holder"], expires_at)) if deleted.rowcount != 1: return False updated = conn.execute( # A parent stamped ended by AUTOMATIC cleanup (tui_shutdown, ws_disconnect, orphan reap, # idle/LRU evict) while a live agent is publishing its rotation is stale by construction — # this writer holds the compression lease and is actively continuing the conversation the # stamp claims is over. Left in place it wedges rotation forever: every attempt aborts here, # nothing clears the stamp, and each attempt's pre-publish flush re-grows the parent until # the provider rejects the request (#88197: 303 unique messages → 2,611 rows → HTTP 400). # Clear it in this same transaction and proceed; the closure UPDATE below re-stamps the # parent with its true boundary (end_reason='compression'). Deliberate boundaries # (compression, session_reset, explicit close) still fail closed — those mean another path # owns lineage. "UPDATE sessions SET ended_at = NULL, end_reason = NULL " "WHERE id = ? AND ended_at IS NOT NULL AND end_reason = 'compression'", (session_id,)) # rowcount==1 is guaranteed by the parent SELECT in this same txn. A False return added past # this point must raise instead: the lease DELETE above commits unless _do raises. return updated.rowcount == 1 return bool(self._execute_write(_do)) def _publish_child_session_row(self, conn, parent, *, parent_session_id, child_session_id, source, model, model_config, system_prompt, cwd, profile_name) -> None: """INSERT the compression child's ``sessions`` row copied from *parent*. Same contract as _insert_session_row's compression-fork backfill: the child stays on the parent's profile and keeps gateway routing/origin columns; no owner on either side -> this store's profile.""" system_prompt_hash = self._store_system_prompt(conn, system_prompt) # The child continues the parent's tools[] pin (the compaction refresh re-pinned it just # before publish), or its first hop to another surface re-derives the array. conn.execute( """INSERT INTO sessions ( id, source, model, model_config, system_prompt, system_prompt_hash, tool_names, parent_session_id, cwd, git_branch, git_repo_root, profile_name, user_id, session_key, chat_id, chat_type, thread_id, display_name, origin_json, started_at, archived, auto_archived ) VALUES (?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( child_session_id, source, model, json.dumps(model_config) if model_config else None, system_prompt_hash, parent["tool_names"], parent_session_id, cwd or parent["cwd"], parent["git_branch"], parent["git_repo_root"], profile_name or parent["profile_name"] or self._own_profile_name(), parent["user_id"], parent["session_key"], parent["chat_id"], parent["chat_type"], parent["thread_id"], parent["display_name"], parent["origin_json"], time.time(), # Inherit the lineage's archive state so a manually archived chat stays uniformly # archived (a mixed lineage let the sweep re-stamp its fresh tip as auto-archived). parent["archived"] or 0, parent["auto_archived"] or 0), ) def publish_compression_child( self, *, parent_session_id: str, child_session_id: str, source: str, messages: List[Dict[str, Any]], model: str = None, model_config: Dict[str, Any] = None, system_prompt: str = None, cwd: str = None, profile_name: str = None, compression_lock_holder: str = None, require_compression_lease: bool = True, require_lease_refresh: bool = False, lease_ttl_seconds: float = 300.0, watermark: Optional[int] = None, watermark_ceiling: Optional[int] = None) -> None: """Atomically close a parent and publish its durable compression child: closure, child row, and handoff commit in one transaction, so readers see the live parent or a complete child, never an ended parent with a missing/empty child. *watermark* (parent's ``get_active_message_watermark`` at compression start): parent rows with ``id > watermark`` — appends landed during the slow summary — are column-cloned into the child AFTER the handoff. *watermark_ceiling* bounds the clone: the rotation path flushes its OWN transcript to the parent just before publishing and those rows are already in the handoff, so only ``(watermark, watermark_ceiling]`` is foreign tail (``None`` = unbounded). *require_lease_refresh* + *compression_lock_holder* refreshes the lease on the same ``conn`` before the expiry check (no TOCTOU window), so a refresher that died on transient DB errors gets one last chance. See #75316. ``None`` = unbounded (no internal flush happened). See #47202. """ from hermes_state_errors import CompressionSessionBusyError def _do(conn): if require_lease_refresh and compression_lock_holder: conn.execute( "UPDATE compression_locks SET expires_at = ? WHERE session_id = ? AND holder = ?", (time.time() + lease_ttl_seconds, parent_session_id, compression_lock_holder)) lock_row = conn.execute(_LOCK_ROW_SQL, (parent_session_id,)).fetchone() if require_compression_lease and ( lock_row is None or not compression_lock_holder or lock_row["holder"] != compression_lock_holder or float(lock_row["expires_at"]) <= time.time() ): raise CompressionSessionBusyError( f"Compression lease lost before publication: {parent_session_id}") parent = conn.execute( """SELECT ended_at, end_reason, cwd, git_branch, git_repo_root, user_id, session_key, chat_id, chat_type, thread_id, display_name, origin_json, profile_name, tool_names, archived, auto_archived FROM sessions WHERE id = ?""", (parent_session_id,), ).fetchone() if parent is None: raise RuntimeError(f"Compression parent not found: {parent_session_id}") if parent["ended_at"] is not None: # An AUTOMATIC end stamp (tui_shutdown, ws_disconnect, orphan reap, idle/LRU # evict) is stale by construction — this lease holder is still continuing the # conversation, and left alone it wedges rotation forever. Clear it; the closure # UPDATE below re-stamps end_reason='compression'. Deliberate boundaries fail closed. if not is_automatic_end_reason(parent["end_reason"]): raise RuntimeError(f"Compression parent already ended: {parent_session_id}") conn.execute( "UPDATE sessions SET ended_at = NULL, end_reason = NULL WHERE id = ?", (parent_session_id,)) if not messages: raise RuntimeError("Compression child handoff must not be empty") self._publish_child_session_row( conn, parent, parent_session_id=parent_session_id, child_session_id=child_session_id, source=source, model=model, model_config=model_config, system_prompt=system_prompt, cwd=cwd, profile_name=profile_name) # Carried handoff tail rows arrive without a timestamp and would otherwise be stamped # `now`, breaking their _display_dedupe_key identity with the parent's durable originals # and duplicating them in the lineage display read (#59661). self._carry_parent_timestamps(conn, parent_session_id, messages) total_messages, total_tool_calls = self._insert_message_rows(conn, child_session_id, messages) if watermark is not None: # Clone the parent's concurrent tail into the child after the handoff; # originals stay in the closed parent for lineage recovery. bounded = watermark_ceiling is not None tail_ids, tail_tool_calls = self._tail_rows_after_watermark( conn, "SELECT id, tool_calls FROM messages " "WHERE session_id = ? AND active = 1 AND id > ?" f"{' AND id <= ?' if bounded else ''} ORDER BY id", [parent_session_id, int(watermark), *([int(watermark_ceiling)] if bounded else [])]) if tail_ids: self._clone_message_rows(conn, tail_ids, session_id=child_session_id) total_messages += len(tail_ids) total_tool_calls += tail_tool_calls conn.execute( "UPDATE sessions SET message_count = ?, tool_call_count = ? WHERE id = ?", (total_messages, total_tool_calls, child_session_id)) updated = conn.execute( "UPDATE sessions SET ended_at = ?, end_reason = 'compression' " "WHERE id = ? AND ended_at IS NULL", (time.time(), parent_session_id)) if updated.rowcount != 1: raise RuntimeError(f"Compression parent changed during publication: {parent_session_id}") if parent["archived"]: # A live continuation under an idle-sweep archive re-activates the chat; after the # closure above the child is linked into the lineage walk (#117713). self._unarchive_auto_archived_lineage(conn, child_session_id) self._execute_transcript_write(_do, messages) def _write_sql_logged(self, op: str, session_id: str, sql: str, params) -> None: """``_write_sql`` that logs (never raises) on ``sqlite3.Error``.""" try: self._write_sql(sql, params) except sqlite3.Error as exc: logger.warning("%s(%s) failed: %s", op, session_id, exc) def record_compression_failure_cooldown( self, session_id: str, cooldown_until: float, error: Optional[str] = None) -> None: """Persist the active compression-failure cooldown. Merge-max with any longer live deadline so a later shorter write can't reopen the thrash window; error always takes the latest diagnostic.""" if not session_id: return self._write_sql_logged( "record_compression_failure_cooldown", session_id, # Merge-max with any longer live deadline so a later shorter write cannot reopen the thrash # window (#96775). The error column always takes the latest diagnostic. "UPDATE sessions SET compression_failure_cooldown_until = CASE " "WHEN compression_failure_cooldown_until IS NOT NULL AND compression_failure_cooldown_until > ? " "THEN compression_failure_cooldown_until ELSE ? END, compression_failure_error = ? WHERE id = ?", (cooldown_until, cooldown_until, error, session_id)) def get_compression_failure_cooldown(self, session_id: str) -> Optional[Dict[str, Any]]: """Return the active (unexpired) compression-failure cooldown, or None.""" now = time.time() row = self._read_one(_COOLDOWN_ROW_SQL, (session_id,)) if session_id else None if row is None or row[0] is None or float(row[0]) <= now: return None return {"cooldown_until": float(row[0]), "remaining_seconds": float(row[0]) - now, "error": row[1]} def get_compression_failure_cooldown_row(self, session_id: str) -> Dict[str, Any]: """Exact stored cooldown columns, no expiry filtering, so compression cancellation can roll back an expired, partially-null, or absent row exactly.""" row = self._read_one(_COOLDOWN_ROW_SQL, (session_id,)) if session_id else None return _cooldown_row(False, None, None) if row is None else _cooldown_row(True, row[0], row[1]) def restore_compression_failure_cooldown_row(self, session_id: str, snapshot: Dict[str, Any]) -> None: """Restore and verify an exact cooldown-row snapshot. Unlike record/clear this rollback API propagates write and verification failures: cancellation must not be reported mutation-free when compensation failed. The tolerated exception is a session row that vanished mid-attempt — before or after the compensating write — since its cooldown died with it and there is nothing left to restore (#106271).""" if not snapshot.get("session_exists", False): if self.get_compression_failure_cooldown_row(session_id).get("session_exists", False): raise RuntimeError("cannot restore absent compression cooldown row: session now exists") return deadline = snapshot.get("cooldown_until") error = snapshot.get("error") def _do(conn): cursor = conn.execute( "UPDATE sessions SET compression_failure_cooldown_until = ?, " "compression_failure_error = ? WHERE id = ?", (deadline, error, session_id)) return cursor.rowcount == 1 if not self._execute_write(_do): logger.warning("compression cooldown rollback session missing: %s", session_id) return actual = self.get_compression_failure_cooldown_row(session_id) expected = _cooldown_row(True, deadline, error) if actual != expected: if not actual.get("session_exists", False): logger.warning("compression cooldown rollback session missing after restore: %s", session_id) return raise RuntimeError( f"compression cooldown rollback verification failed: expected={expected!r}, actual={actual!r}") def clear_compression_failure_cooldown(self, session_id: str) -> None: """Clear any persisted compression-failure cooldown for a session.""" if not session_id: return self._write_sql_logged( "clear_compression_failure_cooldown", session_id, "UPDATE sessions SET compression_failure_cooldown_until = NULL, " "compression_failure_error = NULL WHERE id = ?", (session_id,)) def _read_session_number(self, column: str, session_id: str, cast: type, zero: Any) -> Any: """Read one numeric ``sessions`` column clamped at ``zero``; a missing session, NULL, or unparsable value also reads as ``zero``.""" row = self._read_one(f"SELECT {column} FROM sessions WHERE id = ?", (session_id,)) if session_id else None try: return zero if row is None else max(zero, cast(row[0] or zero)) except (TypeError, ValueError): return zero def _write_session_column(self, column: str, session_id: str, value: Any) -> None: self._write_sql(f"UPDATE sessions SET {column} = ? WHERE id = ?", (value, session_id)) def get_compression_fallback_streak(self, session_id: str) -> int: """Return the persisted deterministic-fallback streak.""" return self._read_session_number("compression_fallback_streak", session_id, int, 0) def set_compression_fallback_streak(self, session_id: str, streak: int) -> None: """Persist the deterministic-fallback streak for one session.""" if session_id: self._write_session_column("compression_fallback_streak", session_id, max(0, int(streak))) def get_compression_ineffective_count(self, session_id: str) -> int: """Persisted ineffective-compaction strike count — the durable half of the built-in compressor's anti-thrash guard, so a fresh compressor bound to a resumed session inherits an armed/tripped guard across restarts.""" return self._read_session_number("compression_ineffective_count", session_id, int, 0) def set_compression_ineffective_count(self, session_id: str, count: int) -> None: """Persist the ineffective-compaction strike count for one session.""" if session_id: self._write_session_column("compression_ineffective_count", session_id, max(0, int(count))) def get_compression_recovery_deadline(self, session_id: str) -> float: """Persisted anti-thrash recovery deadline (epoch; ``0.0`` = not armed). Durable because the gateway rebuilds the compressor every turn / cache eviction. The deadline is the durable half of the 14694 recovery clock: the gateway rebuilds the compressor on every turn / cache eviction, so a process-local deadline restarted the wait on each rebuild and a tripped session never earned its probe (#100185). """ return self._read_session_number("compression_recovery_deadline", session_id, float, 0.0) def set_compression_recovery_deadline(self, session_id: str, deadline: float) -> None: """Persist the anti-thrash recovery deadline; ``0`` / ``None`` disarms it.""" if not session_id: return try: normalized = max(0.0, float(deadline or 0.0)) except (TypeError, ValueError): normalized = 0.0 self._write_session_column("compression_recovery_deadline", session_id, normalized or None) def get_compression_overload_streak(self, session_id: str) -> int: """Return the persisted sustained-overload abort streak (#123167).""" return self._read_session_number("compression_overload_streak", session_id, int, 0) def set_compression_overload_streak(self, session_id: str, streak: int) -> None: """Persist the sustained-overload abort streak for one session.""" if session_id: self._write_session_column("compression_overload_streak", session_id, max(0, int(streak))) def increment_compression_overload_streak(self, session_id: str) -> Optional[int]: """Atomically bump the overload streak and return the new value (None when no row). One UPDATE ... RETURNING, so concurrent agents on one session cannot lose a strike.""" if not session_id: return None def _do(conn): row = conn.execute( "UPDATE sessions SET compression_overload_streak = compression_overload_streak + 1" " WHERE id = ? RETURNING compression_overload_streak", (session_id,)).fetchone() return None if row is None else int(row[0]) return self._execute_write(_do) def refresh_compression_lock(self, session_id: str, holder: str, ttl_seconds: float = 300.0) -> bool: """Extend the compression lock lease if ``holder`` still owns it. Ownership is decided by ``holder`` alone, deliberately NOT ``expires_at``: a live owner whose refresher stalled past its TTL must be able to revive its still-unclaimed row, otherwise it keeps compressing with no lease — the window in which a competing path can fork the lineage. It cannot resurrect a lock someone else took: SQLite serialises writes, so the reclaim (DELETE-expired + INSERT OR IGNORE) never interleaves with this UPDATE.""" if not session_id or not holder: return False expires_at = time.time() + ttl_seconds try: return self._write_rowcount( "UPDATE compression_locks SET expires_at = ? WHERE session_id = ? AND holder = ?", (expires_at, session_id, holder)) > 0 except sqlite3.Error as exc: logger.warning("refresh_compression_lock(%s) failed: %s", session_id, exc) return False def try_acquire_compression_lock(self, session_id: str, holder: str, ttl_seconds: float = 300.0) -> bool: """Try to atomically acquire the compression lock for ``session_id``. ``False``: another holder owns a live lock and the caller MUST NOT compress (its rotation would split the lineage). Expired locks and structured holders whose local ``pid=`` is dead are reclaimed transparently.""" from hermes_state import _compression_lock_holder_process_is_dead if not session_id: return False now = time.time() expires_at = now + ttl_seconds def _do(conn): return _claim_lease_row( conn, "compression_locks", "session_id", session_id, holder, now, expires_at, lambda h, e: e < now or _compression_lock_holder_process_is_dead(h)) try: acquired, reclaimed_holder = self._execute_write(_do) if reclaimed_holder: logger.warning("Reclaimed stale compression lock for session=%s (holder=%s)", session_id, reclaimed_holder) return bool(acquired) except sqlite3.Error as exc: # False makes the caller skip compression — safe when the lock subsystem is broken. logger.warning("try_acquire_compression_lock(%s) failed: %s", session_id, exc) return False def release_compression_lock(self, session_id: str, holder: str) -> None: """Release the compression lock iff we own it; idempotent when gone/reclaimed.""" if not session_id: return self._write_sql_logged( "release_compression_lock", session_id, "DELETE FROM compression_locks WHERE session_id = ? AND holder = ?", (session_id, holder)) def _session_turn_lease_key_on_conn(self, conn, session_id: str) -> str: """Walk compression parents on ``conn`` to the conversation lease key. Must share the connection of the lease INSERT/UPDATE/DELETE: a failed lookup must not yield a child id the write then persists. Markers bind to ``parent_session_id``. Lock errors propagate so ``_execute_write`` can retry.""" if not session_id: return session_id def _row(sid: str): row = conn.execute( "SELECT id, parent_session_id, source, model_config, end_reason FROM sessions WHERE id = ?", (sid,)).fetchone() return dict(row) if row else None current = _row(session_id) seen = {session_id} while current: parent_id = current.get("parent_session_id") if not parent_id or parent_id in seen or self._is_explicit_fork_child_row(current, include_reset=True): break parent = _row(parent_id) if not parent or parent.get("end_reason") != "compression": break seen.add(parent_id) current = parent return str(current.get("id") or session_id) if current else session_id def _session_turn_lease_key(self, session_id: str) -> str: """Stable serialization key for every compression segment (tests/diagnostics; the write paths resolve it inside their own txn). Does not swallow lock errors.""" if not session_id: return session_id with self._read_ctx() as conn: return self._session_turn_lease_key_on_conn(conn, session_id) def try_acquire_session_turn_lease( self, session_id: str, holder: str, *, ttl_seconds: float = 300.0, patience_s: Optional[float] = None, ) -> bool: """Atomically acquire the cross-process turn lease for a conversation (keyed by the lineage root). The walk, the INSERT, and reclaim of expired or dead-local-PID leases share one write transaction.""" from hermes_state import _compression_lock_holder_process_is_dead if not session_id or not holder: return False now = time.time() expires_at = now + max(0.1, float(ttl_seconds)) def _do(conn): # Sweep rows that expired long ago: a holder that died without releasing leaves its row # behind, and nothing else revisits a conversation nobody resumes. The grace keeps the # recent expiries a still-live owner can renew from a starved refresher; it is also the # longest a suspended holder can go unrefreshed and still keep its lease. conn.execute("DELETE FROM session_turn_leases WHERE expires_at < ?", (now - _TURN_LEASE_SWEEP_GRACE_S,)) conversation_id = self._session_turn_lease_key_on_conn(conn, session_id) return _claim_lease_row( conn, "session_turn_leases", "conversation_id", conversation_id, holder, now, expires_at, lambda h, e: float(e) <= now or _compression_lock_holder_process_is_dead(h), )[0] return bool(self._execute_write(_do, patience_s=patience_s)) def acquire_session_turn_lease( self, session_id: str, holder: str, *, ttl_seconds: float = 300.0, wait_seconds: float = 1800.0, poll_interval_seconds: float = 1.0, on_wait=None, wait_notice_interval_seconds: float = 15.0, should_abort=None, acquire_patience_s: float = 0.5, on_contended=None, ) -> bool: """Wait for a cross-process turn lease without holding a SQLite lock. ``on_wait(elapsed)`` is best-effort: called when another holder has the lease and about every ``wait_notice_interval_seconds`` after. A busy database is not a holder: the attempt is retried at once with a longer write patience, and ``on_contended()`` is called instead, since the busy writer may be the holder's last flush. ``should_abort()`` True (e.g. ``/stop``) returns False at once.""" from hermes_state import classify_persistence_error deadline = time.monotonic() + max(0.0, float(wait_seconds)) patience = acquire_patience_s wait_started = None last_notice_at = None notice_every = max(0.0, float(wait_notice_interval_seconds)) while True: if should_abort is not None: try: if should_abort(): return False except Exception: logger.debug("session turn lease should_abort callback failed", exc_info=True) try: if self.try_acquire_session_turn_lease( session_id, holder, ttl_seconds=ttl_seconds, patience_s=patience): return True except sqlite3.Error as exc: # Another writer's transaction outlasted the write patience. That is not a lease # holder, so no poll sleep or "another process" notice; the retry waits on the write # lock itself (the poll interval moves into its patience) and wins it once free. if classify_persistence_error(exc) != "locked": raise if on_contended is not None: try: on_contended() except Exception: logger.debug("session turn lease on_contended callback failed", exc_info=True) patience = min(acquire_patience_s + max(0.0, float(poll_interval_seconds)), max(acquire_patience_s, deadline - time.monotonic())) if not self._sleep_before_write_retry(deadline, 0.0): return False continue patience = acquire_patience_s now = time.monotonic() remaining = deadline - now if remaining <= 0: return False if wait_started is None: wait_started = now if on_wait is not None and ( last_notice_at is None or notice_every == 0.0 or (now - last_notice_at) >= notice_every ): try: on_wait(max(0.0, now - wait_started)) except Exception: logger.debug("session turn lease on_wait callback failed", exc_info=True) last_notice_at = now time.sleep(min(max(0.01, float(poll_interval_seconds)), remaining)) def refresh_session_turn_lease(self, session_id: str, holder: str, *, ttl_seconds: float = 300.0) -> bool: """Extend a turn lease only while ``holder`` still owns it.""" if not session_id or not holder: return False expires_at = time.time() + max(0.1, float(ttl_seconds)) def _do(conn): conversation_id = self._session_turn_lease_key_on_conn(conn, session_id) return conn.execute( "UPDATE session_turn_leases SET expires_at = ? " "WHERE conversation_id = ? AND holder = ?", (expires_at, conversation_id, holder), ).rowcount > 0 return bool(self._execute_write(_do)) def release_session_turn_lease(self, session_id: str, holder: str) -> None: """Release a turn lease iff ``holder`` still owns it; idempotent.""" if not session_id or not holder: return def _do(conn): conversation_id = self._session_turn_lease_key_on_conn(conn, session_id) conn.execute( "DELETE FROM session_turn_leases WHERE conversation_id = ? AND holder = ?", (conversation_id, holder)) self._execute_write(_do) def get_compression_lock_holder(self, session_id: str) -> Optional[str]: """Current (non-expired) holder for ``session_id``, or None. Diagnostic only.""" if not session_id: return None row = self._read_one( "SELECT holder FROM compression_locks WHERE session_id = ? AND expires_at >= ?", (session_id, time.time())) return None if row is None else row[0] def finalize_orphaned_compression_sessions(self) -> int: """Mark orphaned compression continuations (parent ended by compression; child has messages, no end_reason/ended_at, api_call_count=0, older than 7 days) as ``orphaned_compression``. Non-destructive. Fix for #20001. """ cutoff = time.time() - 604800 # 7 days return self._write_rowcount( """ UPDATE sessions SET ended_at = ?, end_reason = 'orphaned_compression' WHERE api_call_count = 0 AND end_reason IS NULL AND ended_at IS NULL AND started_at < ? AND parent_session_id IS NOT NULL AND EXISTS ( SELECT 1 FROM sessions p WHERE p.id = sessions.parent_session_id AND p.end_reason = 'compression' AND p.ended_at IS NOT NULL ) AND EXISTS ( SELECT 1 FROM messages m WHERE m.session_id = sessions.id ) """, (time.time(), cutoff), ) or 0 def get_compression_chain(self, session_id: str) -> List[str]: """Walk the compression-continuation chain forward: root-first through the tip (``[session_id]`` when no continuation); ``get_compression_tip`` is the last element. A continuation is a child of a session with ``end_reason='compression'``. The old ``child.started_at >= parent.ended_at`` test was too brittle (gateway + compression races insert the real continuation before ``ended_at`` is written, while a stale websocket later creates a sibling that passes it). Instead exclude branch/delegate/tool children and prefer children that continue the chain or are still live over stale closed siblings such as ``ws_orphan_reap``.""" current = session_id chain = [current] if current else [] seen = set(chain) # Defensive bound; ``seen`` guards cycles. 100 truncated real # compression lineages (~180 deep), stranding root→tip walks on a # stale mid id (#125041). for _ in range(_CHAIN_CAP): with self._read_ctx() as conn: row = conn.execute(_CHAIN_STEP_SQL, (current,)).fetchone() child_id = row["id"] if row is not None else None if not child_id or child_id in seen: return chain seen.add(child_id) current = child_id chain.append(child_id) return chain def get_compression_tip(self, session_id: str) -> Optional[str]: """Live tip of a compression chain (``get_compression_chain`` semantics); the input id when no continuation exists.""" chain = self.get_compression_chain(session_id) return chain[-1] if chain else session_id def _is_compression_child_row(self, child: Dict[str, Any]) -> bool: parent_id = child.get("parent_session_id") # A reset fork of a compression-ended parent is its own conversation, not the continuation (#114271). if not parent_id or self._is_explicit_fork_child_row(child, include_reset=True): return False parent = self.get_session(parent_id) return bool(parent and parent.get("end_reason") == "compression") def get_compression_lineage(self, session_id: str) -> List[str]: """Return compression ancestors through tip in chronological order.""" session = self.get_session(session_id) if not session or self._is_explicit_fork_child_row(session): return [session_id] if session else [] root = session ancestors = {root["id"]} while self._is_compression_child_row(root): parent = self.get_session(root["parent_session_id"]) if not parent or parent["id"] in ancestors: break root = parent ancestors.add(root["id"]) lineage = [root["id"]] seen = {root["id"]} current = root while current.get("end_reason") == "compression": rows = self._read_all( """ SELECT * FROM sessions WHERE parent_session_id = ? ORDER BY started_at ASC """, (current["id"],)) next_child = next((dict(row) for row in rows if self._is_compression_child_row(dict(row))), None) if not next_child or next_child["id"] in seen: break lineage.append(next_child["id"]) seen.add(next_child["id"]) current = next_child # Later tips are included only when the requested session itself was compacted. return lineage if session_id in lineage else [session_id]