"""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, _non_continuation_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 # model_config keys that belong to the whole conversation, not one compression segment: a rotation # carries them onto the child (per-segment keys such as usage anchors stay behind). # ``prompt_refresh_epoch`` too: without it a rotated child looks older than any past # ``hermes sessions refresh-prompts`` request and rebuilds its prompt once more after eviction. _CONVERSATION_MODEL_CONFIG_KEYS = ("onboarding_run", "prompt_refresh_epoch") def _walk_compression_chain(session_id: str, step) -> list[str]: """Root-first ids from *session_id* through its tip; ``step(id)`` returns the ``_CHAIN_STEP_SQL`` row.""" 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): row = step(current) 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 # created_source values that are not creation provenance: unset, the accounting placeholder # ``unknown`` and the routing label ``local`` (the store-side copy of session_policy._NOT_A_SURFACE). # The creator upsert, the managed-worker heal and the ancestor stamp may replace them, nothing else. _UNSET_CREATED_SOURCES = frozenset({None, "", "unknown", "local"}) _UNSET_CREATED_SQL = "COALESCE({col}, '') IN ('', 'unknown', 'local')" _CREATED_ROW_SQL = "SELECT parent_session_id, created_source FROM sessions WHERE id = ?" # Is any ancestor unset, walking up parent_session_id through compressed parents on the same # surface? One query, so a fully stamped chain is skipped without the per-hop validated walk below. # A parent that did not end by compression (a reset or branch source) is not a chain edge, so an # unset row behind one does not send every write down that walk. # CROSS JOIN keeps ``up`` the outer loop: without ANALYZE stats SQLite otherwise scans sessions # and probes ``up`` for each row, a full-table read on every stamped write. _UNSET_ANCESTOR_SQL = """ WITH RECURSIVE up(pid, n) AS ( SELECT parent_session_id, 1 FROM sessions WHERE id = ? UNION ALL SELECT s.parent_session_id, up.n + 1 FROM up JOIN sessions s ON s.id = up.pid WHERE s.end_reason = 'compression' AND s.created_source = ? AND up.n < ? ) SELECT 1 FROM up CROSS JOIN sessions s ON s.id = up.pid WHERE s.end_reason = 'compression' AND """ + _UNSET_CREATED_SQL.format(col="s.created_source") + """ LIMIT 1 """ def stamp_compression_ancestors(conn, session_id: str) -> None: """Copy *session_id*'s ``created_source`` onto compression ancestors that have none. A list page shows a compressed conversation as its root, painted as the tip, and filters on the root's ``created_source``. Automatic compression publishes the child before any writer knows the surface, so the turn's usage write stamps only the tip. Stamping the chain here, in that same transaction, keeps the root matchable without a list-time chain walk. Each hop must be the parent's preferred continuation (``_CHAIN_STEP_SQL``); ``unknown`` and ``local`` count as unset, as in the creator upsert. An ancestor already on the same surface is walked through (a NULL root behind a stamped mid segment still gets stamped); another surface stops it. A chain with no unset ancestor returns after one query (``_UNSET_ANCESTOR_SQL``). """ row = conn.execute(_CREATED_ROW_SQL, (session_id,)).fetchone() surface = row["created_source"] if row else None if surface in _UNSET_CREATED_SOURCES: return if conn.execute(_UNSET_ANCESTOR_SQL, (session_id, surface, _CHAIN_CAP)).fetchone() is None: return current, parent_id = session_id, row["parent_session_id"] for _ in range(_CHAIN_CAP): parent = parent_id and conn.execute(_CREATED_ROW_SQL, (parent_id,)).fetchone() if not parent or parent["created_source"] not in (*_UNSET_CREATED_SOURCES, surface): return step = conn.execute(_CHAIN_STEP_SQL, (parent_id,)).fetchone() if step is None or step["id"] != current: return if parent["created_source"] != surface: conn.execute("UPDATE sessions SET created_source = ? WHERE id = ?", (surface, parent_id)) current, parent_id = parent_id, parent["parent_session_id"] # Compression edges whose parent is unset and whose continuation child carries a real surface: what # a store from before write-time stamping still hides from ``sessions list --source``. A child # forked from that parent (branch, delegate or reset marker naming it) or a tool child is no # continuation (as in prune) and never matches. A stamped store, or one whose unset chains have no # stamped segment, answers with no rows after one scan of sessions. A stamped child the stamp walk # does not follow (a non-preferred sibling, or one with a fork marker inherited from an earlier row) # still matches; the walk stops at its first hop. _UNSTAMPED_EDGE_CHILD_SQL = f""" SELECT child.id FROM sessions parent JOIN sessions child ON child.parent_session_id = parent.id WHERE parent.end_reason = 'compression' AND {_UNSET_CREATED_SQL.format(col="parent.created_source")} AND NOT {_UNSET_CREATED_SQL.format(col="child.created_source")} {_non_continuation_child_sql("child.", "parent.id")}""" 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 stamp_legacy_compression_chains(self) -> int: """Stamp the unset ancestors of chains compressed before write-time stamping from their stamped segments, through ``stamp_compression_ancestors`` (same edge checks; a known value is never replaced). One write transaction and idempotent, so the owner runs it on every start: a store with nothing to repair pays one scan of sessions under the write lock, plus a one-hop walk per stamped child of an unset compression parent that the walk does not follow (a non-preferred sibling). Returns the rows stamped.""" def _do(conn): before = conn.total_changes for row in conn.execute(_UNSTAMPED_EDGE_CHILD_SQL).fetchall(): stamp_compression_ancestors(conn, row["id"]) return conn.total_changes - before return self._execute_write(_do) 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.""" from hermes_state_sessions import _parse_model_config # Pin/archive and the conversation keys are compression-lineage state; canonical branch/reset # verbs also publish through here, and their child is its own conversation. continuation = not any(k in (model_config or {}) for k in ("_branched_from", "_reset_from")) carried = {key: value for key, value in _parse_model_config(parent["model_config"]).items() if continuation and key in _CONVERSATION_MODEL_CONFIG_KEYS} model_config = {**carried, **(model_config or {})} system_prompt_hash = self._store_system_prompt(conn, system_prompt) # Copy a real creating surface. Automatic compression passes the parent's live # label (``local`` after the peer refresh); reset, branch and the compress # mutation pass the policy label. ``local`` and ``unknown`` are not surfaces, # so those stay NULL. Compression sets ``_session_db_created`` and does not # create the child again. The turn's ``update_token_counts(source=...)`` fills # that NULL; auxiliary usage must not write ``unknown`` into it first. from gateway.session_policy import created_surface created_source = created_surface(parent["created_source"]) or created_surface(source) # 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, created_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, pinned, started_at, archived, auto_archived ) VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( child_session_id, source, created_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"], # The pin is lineage-wide (set_session_pinned); a segment published after it joins it. int(parent["pinned"] or 0) if continuation else 0, 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) if continuation else 0, (parent["auto_archived"] or 0) if continuation else 0), ) def publish_compression_child( self, *, parent_session_id: str, child_session_id: str, source: str, messages: list[dict[str, Any]], model: str | None = None, model_config: dict[str, Any] | None = None, system_prompt: str | None = None, cwd: str | None = None, profile_name: str | None = None, compression_lock_holder: str | None = 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* (the parent's highest row already represented in the handoff: its ``get_active_message_watermark`` at compression start, or the newest row of an adopted durable snapshot): 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. """ self._execute_transcript_write(lambda conn: self._publish_compression_child_on_conn( conn, parent_session_id=parent_session_id, child_session_id=child_session_id, source=source, messages=messages, model=model, model_config=model_config, system_prompt=system_prompt, cwd=cwd, profile_name=profile_name, compression_lock_holder=compression_lock_holder, require_compression_lease=require_compression_lease, require_lease_refresh=require_lease_refresh, lease_ttl_seconds=lease_ttl_seconds, watermark=watermark, watermark_ceiling=watermark_ceiling), messages) def _publish_compression_child_on_conn( self, conn, *, parent_session_id: str, child_session_id: str, source: str, messages: list[dict[str, Any]], model: str | None = None, model_config: dict[str, Any] | None = None, system_prompt: str | None = None, cwd: str | None = None, profile_name: str | None = None, compression_lock_holder: str | None = 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: """:meth:`publish_compression_child` on the caller's write transaction (never commits). The owner and the worker/canonical-mutation verbs share this one body so they cannot drift.""" from hermes_state_errors import CompressionSessionBusyError 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, pinned, model_config, created_source 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}") # A child on a real surface stamps the chain's unset ancestors, so compressing a conversation # whose root predates write-time stamping lists it again under its surface. stamp_compression_ancestors(conn, child_session_id) from hermes_state_local_lineage import advance_local_target advance_local_target(conn, parent_session_id, child_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) 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 session_turn_lease_expires_at(self, session_id: str, holder: str) -> Optional[float]: """Committed ``expires_at`` of ``holder``'s turn lease, or None when it does not hold it.""" if not session_id or not holder: return None with self._read_ctx() as conn: row = conn.execute( "SELECT expires_at FROM session_turn_leases WHERE conversation_id = ? AND holder = ?", (self._session_turn_lease_key_on_conn(conn, session_id), holder), ).fetchone() return float(row[0]) if row else None def refresh_session_turn_lease( self, session_id: str, holder: str, *, ttl_seconds: float = 300.0, patience_s: Optional[float] = None, ) -> bool: """Extend a turn lease only while ``holder`` still owns it. ``expires_at`` is stamped before the write-lock wait, so a caller timestamp taken before this call never exceeds 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, patience_s=patience_s)) 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``.""" def step(current: str): with self._read_ctx() as conn: return conn.execute(_CHAIN_STEP_SQL, (current,)).fetchone() return _walk_compression_chain(session_id, step) 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 _compression_tip_on(self, conn, session_id: str) -> str: """``get_compression_tip`` read on a writer's own connection: inside its transaction no rotation can publish between finding the tip and writing it.""" return _walk_compression_chain(session_id, lambda current: conn.execute(_CHAIN_STEP_SQL, (current,)).fetchone())[-1] 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]