From 52d982281ce02e843223e437a590f495ecbd9068 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:44:38 -0700 Subject: [PATCH 1/4] refactor(state): _stale_holder guard helper; join implicit-concat SQL fragments (AST-identical) --- hermes_state_compression.py | 61 +++++++++++------------------------ hermes_state_messages.py | 63 +++++++++++++++---------------------- hermes_state_titles.py | 9 ++---- 3 files changed, 47 insertions(+), 86 deletions(-) diff --git a/hermes_state_compression.py b/hermes_state_compression.py index eb9dfb5f49..9786f97a28 100644 --- a/hermes_state_compression.py +++ b/hermes_state_compression.py @@ -102,8 +102,7 @@ class SessionCompressionMixin: return False updated = conn.execute( "UPDATE sessions SET ended_at = NULL, end_reason = NULL " - "WHERE id = ? AND ended_at IS NOT NULL " - "AND end_reason = 'compression'", + "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 @@ -115,7 +114,10 @@ class SessionCompressionMixin: 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*.""" + """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) conn.execute( """INSERT INTO sessions ( @@ -126,27 +128,12 @@ class SessionCompressionMixin: thread_id, display_name, origin_json, started_at ) VALUES (?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( - child_session_id, - source, - model, - json.dumps(model_config) if model_config else None, - system_prompt_hash, - parent_session_id, - cwd or parent["cwd"], - parent["git_branch"], + child_session_id, source, model, json.dumps(model_config) if model_config else None, + system_prompt_hash, parent_session_id, cwd or parent["cwd"], parent["git_branch"], parent["git_repo_root"], - # 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. 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(), + parent["user_id"], parent["session_key"], parent["chat_id"], parent["chat_type"], + parent["thread_id"], parent["display_name"], parent["origin_json"], time.time(), ), ) @@ -209,8 +196,7 @@ class SessionCompressionMixin: # Deliberate boundaries still fail closed. if is_automatic_end_reason(parent["end_reason"]): conn.execute( - "UPDATE sessions SET ended_at = NULL, end_reason = NULL " - "WHERE id = ?", + "UPDATE sessions SET ended_at = NULL, end_reason = NULL WHERE id = ?", (parent_session_id,), ) else: @@ -419,8 +405,7 @@ class SessionCompressionMixin: expires_at = time.time() + ttl_seconds try: return self._write_rowcount( - "UPDATE compression_locks SET expires_at = ? " - "WHERE session_id = ? AND holder = ?", + "UPDATE compression_locks SET expires_at = ? WHERE session_id = ? AND holder = ?", (expires_at, session_id, holder), ) > 0 except sqlite3.Error as exc: @@ -448,15 +433,13 @@ class SessionCompressionMixin: current_holder, current_expires_at = row[0], row[1] if current_expires_at < now or _compression_lock_holder_process_is_dead(current_holder): conn.execute( - "DELETE FROM compression_locks " - "WHERE session_id = ? AND holder = ?", + "DELETE FROM compression_locks WHERE session_id = ? AND holder = ?", (session_id, current_holder), ) reclaimed_holder = current_holder conn.execute( "INSERT OR IGNORE INTO compression_locks " - "(session_id, holder, acquired_at, expires_at) " - "VALUES (?, ?, ?, ?)", + "(session_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)", (session_id, holder, now, expires_at), ) row = conn.execute("SELECT holder FROM compression_locks WHERE session_id = ?", (session_id,)).fetchone() @@ -480,8 +463,7 @@ class SessionCompressionMixin: return self._write_sql_logged( "release_compression_lock", session_id, - "DELETE FROM compression_locks " - "WHERE session_id = ? AND holder = ?", + "DELETE FROM compression_locks WHERE session_id = ? AND holder = ?", (session_id, holder), ) @@ -538,22 +520,19 @@ class SessionCompressionMixin: def _do(conn): conversation_id = self._session_turn_lease_key_on_conn(conn, session_id) row = conn.execute( - "SELECT holder, expires_at FROM session_turn_leases " - "WHERE conversation_id = ?", + "SELECT holder, expires_at FROM session_turn_leases WHERE conversation_id = ?", (conversation_id,), ).fetchone() if row is not None: current_holder = row["holder"] if float(row["expires_at"]) <= now or _compression_lock_holder_process_is_dead(current_holder): conn.execute( - "DELETE FROM session_turn_leases " - "WHERE conversation_id = ? AND holder = ?", + "DELETE FROM session_turn_leases WHERE conversation_id = ? AND holder = ?", (conversation_id, current_holder), ) conn.execute( "INSERT OR IGNORE INTO session_turn_leases " - "(conversation_id, holder, acquired_at, expires_at) " - "VALUES (?, ?, ?, ?)", + "(conversation_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)", (conversation_id, holder, now, expires_at), ) owner = conn.execute( @@ -637,8 +616,7 @@ class SessionCompressionMixin: 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 = ?", + "DELETE FROM session_turn_leases WHERE conversation_id = ? AND holder = ?", (conversation_id, holder), ) @@ -649,8 +627,7 @@ class SessionCompressionMixin: if not session_id: return None row = self._read_one( - "SELECT holder FROM compression_locks " - "WHERE session_id = ? AND expires_at >= ?", + "SELECT holder FROM compression_locks WHERE session_id = ? AND expires_at >= ?", (session_id, time.time()), ) return None if row is None else row[0] diff --git a/hermes_state_messages.py b/hermes_state_messages.py index 3dbe8f3555..a318c4487a 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -91,6 +91,12 @@ def _ended_by_compression(row) -> bool: return row is not None and row["ended_at"] is not None and row["end_reason"] == "compression" +def _stale_holder(row, now: float) -> bool: + """A lock/lease row whose holder is expired or a provably dead local process.""" + from hermes_state import _compression_lock_holder_process_is_dead + return float(row["expires_at"]) <= now or _compression_lock_holder_process_is_dead(row["holder"]) + + class SessionMessagesMixin: """Message append/replace/rewind, reactions, resume conversations, replay dedupe.""" @@ -216,17 +222,13 @@ class SessionMessagesMixin: ``reject_active_compression_lock`` / ``reject_active_turn_lease`` so a compressor that captured its watermark cannot resurrect the removed turn. """ - from hermes_state import CompressionSessionClosedError, SessionCompressionInProgressError, SessionTurnLeaseLostError, _compression_lock_holder_process_is_dead + from hermes_state import CompressionSessionClosedError, SessionCompressionInProgressError, SessionTurnLeaseLostError if reject_active_compression_lock: active_lock = conn.execute(_COMPRESSION_LOCK_ROW_SQL, (session_id,)).fetchone() if active_lock is not None: - current_holder = active_lock["holder"] - if ( - float(active_lock["expires_at"]) <= time.time() - or _compression_lock_holder_process_is_dead(current_holder) - ): - conn.execute(_DELETE_COMPRESSION_LOCK_SQL, (session_id, current_holder)) - elif current_holder != compression_lock_holder: + if _stale_holder(active_lock, time.time()): + conn.execute(_DELETE_COMPRESSION_LOCK_SQL, (session_id, active_lock["holder"])) + elif active_lock["holder"] != compression_lock_holder: raise SessionCompressionInProgressError( f"Session {session_id!r} is being compressed by another writer" ) @@ -249,21 +251,16 @@ class SessionMessagesMixin: (now + max(0.1, float(turn_lease_ttl_seconds)), conversation_id, turn_lease_holder), ) elif lease is not None: - current_holder = lease["holder"] - if ( - float(lease["expires_at"]) <= now - or _compression_lock_holder_process_is_dead(current_holder) - ): - # Same reclaim rule as acquisition (expired or provably dead owner); - # deleting here also fences a stale late flush after the mutation. - conn.execute( - "DELETE FROM session_turn_leases WHERE conversation_id = ? AND holder = ?", - (conversation_id, current_holder), - ) - else: + if not _stale_holder(lease, now): raise SessionTurnLeaseLostError( f"Session has an active turn lease; refusing transcript mutation for {session_id!r}" ) + # Same reclaim rule as acquisition (expired or provably dead owner); + # deleting here also fences a stale late flush after the mutation. + conn.execute( + "DELETE FROM session_turn_leases WHERE conversation_id = ? AND holder = ?", + (conversation_id, lease["holder"]), + ) session = conn.execute(_ENDED_BY_COMPRESSION_SQL, (session_id,)).fetchone() if _ended_by_compression(session) and not allow_closed_compression_parent: raise CompressionSessionClosedError(session_id) @@ -519,8 +516,7 @@ class SessionMessagesMixin: def _do(conn): rows = conn.execute( "SELECT id, role, content, display_metadata FROM messages " - "WHERE session_id = ? AND active = 1 AND display_metadata IS NOT NULL " - "ORDER BY id", + "WHERE session_id = ? AND active = 1 AND display_metadata IS NOT NULL ORDER BY id", (session_id,), ).fetchall() pending = [] @@ -758,8 +754,7 @@ class SessionMessagesMixin: 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 > ? " - "ORDER BY id", + "WHERE session_id = ? AND active = 1 AND id > ? ORDER BY id", (session_id, int(watermark)), ) # Rewind targets sit AT/BELOW the watermark (the compressor only saw rows up @@ -768,8 +763,7 @@ class SessionMessagesMixin: if tail_count > 0: if watermark is not None: tail_rows = conn.execute( - "SELECT id FROM messages " - "WHERE session_id = ? AND active = 1 AND id <= ? " + "SELECT id FROM messages WHERE session_id = ? AND active = 1 AND id <= ? " "ORDER BY id DESC LIMIT ?", (session_id, int(watermark), int(tail_count)), ).fetchall() @@ -840,10 +834,8 @@ class SessionMessagesMixin: """ from hermes_state import _scrub_surrogates return self._write_rowcount( - "UPDATE messages SET api_content = ? WHERE id = (" - "SELECT id FROM messages " - "WHERE session_id = ? AND role = 'user' AND active = 1 " - "ORDER BY id DESC LIMIT 1" + "UPDATE messages SET api_content = ? WHERE id = (SELECT id FROM messages " + "WHERE session_id = ? AND role = 'user' AND active = 1 ORDER BY id DESC LIMIT 1" ") AND content IS ?", (_scrub_surrogates(api_content), session_id, self._encode_content(content)), ) @@ -991,15 +983,11 @@ class SessionMessagesMixin: if not anchor_exists: return {"window": [], "messages_before": 0, "messages_after": 0} before_rows = conn.execute( - "SELECT * FROM messages " - "WHERE session_id = ? AND id <= ? " - "ORDER BY id DESC LIMIT ?", + "SELECT * FROM messages WHERE session_id = ? AND id <= ? ORDER BY id DESC LIMIT ?", (session_id, around_message_id, window + 1), ).fetchall() after_rows = conn.execute( - "SELECT * FROM messages " - "WHERE session_id = ? AND id > ? " - "ORDER BY id ASC LIMIT ?", + "SELECT * FROM messages WHERE session_id = ? AND id > ? ORDER BY id ASC LIMIT ?", (session_id, around_message_id, window), ).fetchall() rows = list(reversed(before_rows)) + list(after_rows) @@ -1046,8 +1034,7 @@ class SessionMessagesMixin: best = current try: child_row = conn.execute( - "SELECT id FROM sessions AS child " - "WHERE child.parent_session_id = ? " + "SELECT id FROM sessions AS child WHERE child.parent_session_id = ? " " AND json_extract(COALESCE(child.model_config, '{}'), '$._branched_from') IS NULL " " AND json_extract(COALESCE(child.model_config, '{}'), '$._delegate_from') IS NULL " " AND json_extract(COALESCE(child.model_config, '{}'), '$._reset_from') IS NULL " diff --git a/hermes_state_titles.py b/hermes_state_titles.py index 9404fa74ad..c6ac1dfdc6 100644 --- a/hermes_state_titles.py +++ b/hermes_state_titles.py @@ -166,18 +166,15 @@ class SessionTitlesMixin: if source not in self._TITLE_SOURCE_RANK: raise ValueError(f"invalid title source: {source!r}") return self._write_rowcount( - "UPDATE sessions SET title_source = ? " - "WHERE id = ? AND title IS NOT NULL", + "UPDATE sessions SET title_source = ? WHERE id = ? AND title IS NOT NULL", (source, session_id), ) > 0 def get_session_by_title(self, title: str) -> Optional[Dict[str, Any]]: """Look up a session by exact title. Returns session dict or None.""" row = self._read_one( - "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 " + "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.title = ?", (title,), ) From f54075e0bb6bd3660f3b40eefd157f70c075b1c3 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:52:21 -0700 Subject: [PATCH 2/4] refactor(state): unify lock/lease claim (_claim_lease_row); finalize_orphaned via _write_rowcount --- hermes_state_compression.py | 75 ++++++++++++++----------------------- 1 file changed, 29 insertions(+), 46 deletions(-) diff --git a/hermes_state_compression.py b/hermes_state_compression.py index 9786f97a28..124ffc7015 100644 --- a/hermes_state_compression.py +++ b/hermes_state_compression.py @@ -9,7 +9,7 @@ import json import logging import sqlite3 import time -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from hermes_state_common import _sql_session_last_active, is_automatic_end_reason @@ -27,6 +27,24 @@ def _ended_by_compression(row) -> bool: return row is not None and row["ended_at"] is not None and row["end_reason"] == "compression" +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 + + class SessionCompressionMixin: """Compression lineage, cooldown/streak counters, locks and turn leases.""" @@ -427,23 +445,10 @@ class SessionCompressionMixin: expires_at = now + ttl_seconds def _do(conn): - reclaimed_holder = None - row = conn.execute(_LOCK_ROW_SQL, (session_id,)).fetchone() - if row is not None: - current_holder, current_expires_at = row[0], row[1] - if current_expires_at < now or _compression_lock_holder_process_is_dead(current_holder): - conn.execute( - "DELETE FROM compression_locks WHERE session_id = ? AND holder = ?", - (session_id, current_holder), - ) - reclaimed_holder = current_holder - conn.execute( - "INSERT OR IGNORE INTO compression_locks " - "(session_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)", - (session_id, holder, now, expires_at), + 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), ) - row = conn.execute("SELECT holder FROM compression_locks WHERE session_id = ?", (session_id,)).fetchone() - return row is not None and row[0] == holder, reclaimed_holder try: acquired, reclaimed_holder = self._execute_write(_do) @@ -519,26 +524,10 @@ class SessionCompressionMixin: def _do(conn): conversation_id = self._session_turn_lease_key_on_conn(conn, session_id) - row = conn.execute( - "SELECT holder, expires_at FROM session_turn_leases WHERE conversation_id = ?", - (conversation_id,), - ).fetchone() - if row is not None: - current_holder = row["holder"] - if float(row["expires_at"]) <= now or _compression_lock_holder_process_is_dead(current_holder): - conn.execute( - "DELETE FROM session_turn_leases WHERE conversation_id = ? AND holder = ?", - (conversation_id, current_holder), - ) - conn.execute( - "INSERT OR IGNORE INTO session_turn_leases " - "(conversation_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)", - (conversation_id, holder, now, expires_at), - ) - owner = conn.execute( - "SELECT holder FROM session_turn_leases WHERE conversation_id = ?", (conversation_id,), - ).fetchone() - return owner is not None and owner["holder"] == holder + 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)) @@ -637,10 +626,7 @@ class SessionCompressionMixin: has messages, no end_reason/ended_at, api_call_count=0, older than 7 days) as ``orphaned_compression``. Non-destructive.""" cutoff = time.time() - 604800 # 7 days - - def _do(conn): - now = time.time() - result = conn.execute( + return self._write_rowcount( """ UPDATE sessions SET ended_at = ?, @@ -661,11 +647,8 @@ class SessionCompressionMixin: WHERE m.session_id = sessions.id ) """, - (now, cutoff), - ) - return result.rowcount - - return self._execute_write(_do) or 0 + (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 From 8656cc31f2d7ec9140d07ad2e46671f754074898 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:59:58 -0700 Subject: [PATCH 3/4] refactor(state): table-driven _rows_to_conversation column copy; unify tool-call counting; collapse defensive locals --- hermes_state_compression.py | 21 ++-- hermes_state_messages.py | 191 ++++++++++++++---------------------- hermes_state_titles.py | 3 +- hermes_state_usage.py | 6 +- 4 files changed, 82 insertions(+), 139 deletions(-) diff --git a/hermes_state_compression.py b/hermes_state_compression.py index 124ffc7015..d6eae97865 100644 --- a/hermes_state_compression.py +++ b/hermes_state_compression.py @@ -236,11 +236,9 @@ class SessionCompressionMixin: _ceiling_clause = " AND id <= ?" _params.append(int(watermark_ceiling)) tail_ids, tail_tool_calls = self._tail_rows_after_watermark( - conn, - "SELECT id, tool_calls FROM messages " + conn, "SELECT id, tool_calls FROM messages " "WHERE session_id = ? AND active = 1 AND id > ?" - f"{_ceiling_clause} ORDER BY id", - _params, + f"{_ceiling_clause} ORDER BY id", _params, ) if tail_ids: self._clone_message_rows(conn, tail_ids, session_id=child_session_id) @@ -252,8 +250,7 @@ class SessionCompressionMixin: ) updated = conn.execute( "UPDATE sessions SET ended_at = ?, end_reason = 'compression' " - "WHERE id = ? AND ended_at IS NULL", - (time.time(), parent_session_id), + "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}") @@ -322,8 +319,7 @@ class SessionCompressionMixin: def _do(conn): cursor = conn.execute( "UPDATE sessions SET compression_failure_cooldown_until = ?, " - "compression_failure_error = ? WHERE id = ?", - (deadline, error, session_id), + "compression_failure_error = ? WHERE id = ?", (deadline, error, session_id), ) if cursor.rowcount != 1: raise RuntimeError(f"compression cooldown rollback session missing: {session_id}") @@ -348,8 +344,7 @@ class SessionCompressionMixin: 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,), + "compression_failure_error = NULL WHERE id = ?", (session_id,), ) def _read_session_number(self, column: str, session_id: str, cast: type, zero: Any) -> Any: @@ -484,8 +479,7 @@ class SessionCompressionMixin: def _row(sid: str): row = conn.execute( "SELECT id, parent_session_id, source, model_config, end_reason " - "FROM sessions WHERE id = ?", - (sid,), + "FROM sessions WHERE id = ?", (sid,), ).fetchone() return dict(row) if row else None @@ -590,8 +584,7 @@ class SessionCompressionMixin: conversation_id = self._session_turn_lease_key_on_conn(conn, session_id) cursor = conn.execute( "UPDATE session_turn_leases SET expires_at = ? " - "WHERE conversation_id = ? AND holder = ?", - (expires_at, conversation_id, holder), + "WHERE conversation_id = ? AND holder = ?", (expires_at, conversation_id, holder), ) return cursor.rowcount > 0 diff --git a/hermes_state_messages.py b/hermes_state_messages.py index a318c4487a..0d801db935 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -48,15 +48,18 @@ def _json_or(raw: Any, fallback: Any, warning: str) -> Any: return fallback -def _tool_calls_len(raw: Any) -> int: - """Tool-call count of a stored ``tool_calls`` column (0 unless a JSON list).""" +def _tool_calls_len(raw: Any, scalar: int = 0) -> int: + """Tool-call count of a stored ``tool_calls`` column: list length, else *scalar* + for a truthy non-list value, 0 for empty/undecodable.""" if not raw: return 0 try: parsed = json.loads(raw) if isinstance(raw, str) else raw except (TypeError, ValueError): return 0 - return len(parsed) if isinstance(parsed, list) else 0 + if isinstance(parsed, list): + return len(parsed) + return scalar if parsed else 0 def _coerce_timestamp(value: Any, default: float) -> float: @@ -91,6 +94,14 @@ def _ended_by_compression(row) -> bool: return row is not None and row["ended_at"] is not None and row["end_reason"] == "compression" +# _rows_to_conversation copies these row columns verbatim when truthy, in this order. +# ``api_content`` is returned VERBATIM (no sanitize/strip): the replay path substitutes +# it to keep the provider prompt cache byte-stable. +_VERBATIM_COLS = ("api_content", "display_kind") +_META_COLS = ("timestamp", "tool_call_id", "tool_name", "effect_disposition") +_ASSISTANT_JSON_COLS = ("reasoning_details", "codex_reasoning_items", "codex_message_items") + + def _stale_holder(row, now: float) -> bool: """A lock/lease row whose holder is expired or a provably dead local process.""" from hermes_state import _compression_lock_holder_process_is_dead @@ -162,20 +173,19 @@ class SessionMessagesMixin: return None if isinstance(display_metadata, str): try: - parsed = json.loads(display_metadata) + display_metadata = json.loads(display_metadata) except (json.JSONDecodeError, TypeError): logger.warning("Ignoring non-JSON display metadata on write") return None - if not isinstance(parsed, dict): + if not isinstance(display_metadata, dict): logger.warning("Ignoring non-object display metadata on write") return None - return json.dumps(parsed) - if isinstance(display_metadata, dict): - return json.dumps(display_metadata) - logger.warning( - "Ignoring unexpected display metadata type on write: %s", type(display_metadata).__name__, - ) - return None + elif not isinstance(display_metadata, dict): + logger.warning( + "Ignoring unexpected display metadata type on write: %s", type(display_metadata).__name__, + ) + return None + return json.dumps(display_metadata) @staticmethod def _decode_display_metadata(raw: Any) -> Optional[Dict[str, Any]]: @@ -689,15 +699,13 @@ class SessionMessagesMixin: conn.execute( f"INSERT INTO messages ({col_list}, active, compacted) " f"SELECT {col_list}, 1, 0 FROM messages " - f"WHERE id IN ({placeholders}) ORDER BY id", - tail_ids, + f"WHERE id IN ({placeholders}) ORDER BY id", tail_ids, ) else: conn.execute( f"INSERT INTO messages ({col_list}, session_id, active, compacted) " f"SELECT {col_list}, ?, 1, 0 FROM messages " - f"WHERE id IN ({placeholders}) ORDER BY id", - [session_id, *tail_ids], + f"WHERE id IN ({placeholders}) ORDER BY id", [session_id, *tail_ids], ) def archive_and_compact( @@ -752,20 +760,18 @@ class SessionMessagesMixin: tail_tool_calls = 0 if watermark is not None: tail_ids, tail_tool_calls = self._tail_rows_after_watermark( - conn, - "SELECT id, tool_calls FROM messages " + conn, "SELECT id, tool_calls FROM messages " "WHERE session_id = ? AND active = 1 AND id > ? ORDER BY id", (session_id, int(watermark)), ) # Rewind targets sit AT/BELOW the watermark (the compressor only saw rows up # to it); without the bound a concurrent append would steal a LIMIT slot. - rewind_tail_ids: Optional[list[int]] = None + rewind_ids: list[int] = [] if tail_count > 0: if watermark is not None: tail_rows = conn.execute( "SELECT id FROM messages WHERE session_id = ? AND active = 1 AND id <= ? " - "ORDER BY id DESC LIMIT ?", - (session_id, int(watermark), int(tail_count)), + "ORDER BY id DESC LIMIT ?", (session_id, int(watermark), int(tail_count)), ).fetchall() else: tail_rows = conn.execute( @@ -773,26 +779,23 @@ class SessionMessagesMixin: "WHERE session_id = ? AND active = 1 ORDER BY id DESC LIMIT ?", (session_id, int(tail_count)), ).fetchall() - rewind_tail_ids = [int(row["id"]) for row in tail_rows] - rewind_ids = [*(rewind_tail_ids or []), *tail_ids] + rewind_ids = [int(row["id"]) for row in tail_rows] + rewind_ids += tail_ids if rewind_ids: placeholders = _placeholders(rewind_ids) conn.execute( "UPDATE messages SET active = 0, compacted = 0 " - f"WHERE session_id = ? AND id IN ({placeholders})", - [session_id, *rewind_ids], + f"WHERE session_id = ? AND id IN ({placeholders})", [session_id, *rewind_ids], ) conn.execute( "UPDATE messages SET active = 0, compacted = 1 " "WHERE session_id = ? AND active = 1 " - f"AND id NOT IN ({placeholders})", - [session_id, *rewind_ids], + f"AND id NOT IN ({placeholders})", [session_id, *rewind_ids], ) else: conn.execute( "UPDATE messages SET active = 0, compacted = 1 " - "WHERE session_id = ? AND active = 1", - (session_id,), + "WHERE session_id = ? AND active = 1", (session_id,), ) inserted, tool_calls_total = self._insert_message_rows(conn, session_id, compacted_messages) if tail_ids: @@ -959,7 +962,7 @@ class SessionMessagesMixin: chunk = ids[start : start + 900] rows = self._read_all( f"""SELECT session_id, content FROM messages - WHERE session_id IN ({",".join("?" * len(chunk))}) + WHERE session_id IN ({_placeholders(chunk)}) AND role = 'tool' AND content LIKE '%/pull/%' ORDER BY id ASC""", chunk, @@ -991,14 +994,9 @@ class SessionMessagesMixin: (session_id, around_message_id, window), ).fetchall() rows = list(reversed(before_rows)) + list(after_rows) - return { - "window": [ - self._row_to_message_dict(row, warn_context="get_messages_around", summary_flag=False) - for row in rows - ], - "messages_before": max(0, len(before_rows) - 1), # before_rows includes the anchor - "messages_after": len(after_rows), - } + window_msgs = [self._row_to_message_dict(r, warn_context="get_messages_around", summary_flag=False) for r in rows] + # before_rows includes the anchor itself. + return {"window": window_msgs, "messages_before": max(0, len(before_rows) - 1), "messages_after": len(after_rows)} def resolve_resume_session_id(self, session_id: str) -> str: """Redirect a resume target to the descendant session that holds the messages. @@ -1025,14 +1023,10 @@ class SessionMessagesMixin: best = None # deepest node with messages for _ in range(32): try: - row = conn.execute( + if conn.execute( "SELECT 1 FROM messages WHERE session_id = ? LIMIT 1", (current,), - ).fetchone() - except Exception: - return session_id - if row is not None: - best = current - try: + ).fetchone() is not None: + best = current child_row = conn.execute( "SELECT id FROM sessions AS child WHERE child.parent_session_id = ? " " AND json_extract(COALESCE(child.model_config, '{}'), '$._branched_from') IS NULL " @@ -1040,8 +1034,7 @@ class SessionMessagesMixin: " AND json_extract(COALESCE(child.model_config, '{}'), '$._reset_from') IS NULL " f" AND NOT {_legacy_reset_child_sql('child', _RESET_END_REASONS_SQL)} " " AND COALESCE(child.source, '') != 'tool' " - "ORDER BY child.started_at DESC, child.id DESC LIMIT 1", - (current,), + "ORDER BY child.started_at DESC, child.id DESC LIMIT 1", (current,), ).fetchone() except Exception: return session_id @@ -1062,8 +1055,7 @@ class SessionMessagesMixin: return conn.execute( f"{prefix}{self._CONVERSATION_ROW_COLUMNS} " f"FROM messages WHERE session_id IN ({_placeholders(session_ids)})" - f"{active_clause} ORDER BY id", - tuple(session_ids), + f"{active_clause} ORDER BY id", tuple(session_ids), ).fetchall() def get_messages_as_conversation( @@ -1129,9 +1121,8 @@ class SessionMessagesMixin: Every dict is stamped ``_DB_PERSISTED_MARKER_KEY`` at the source (born durable) so an identity-losing handoff never re-appends the whole transcript on flush. - ``_row_id`` is opt-in (gateway reactions). ``api_content`` is returned VERBATIM - (no sanitize/strip): the replay path substitutes it to keep the provider prompt - cache byte-stable. Reasoning fields are restored on assistant rows only. + ``_row_id`` is opt-in (gateway reactions). Reasoning fields are restored on + assistant rows only. Key order of each dict is stable (see the column tables). """ from hermes_state import _strip_background_review_harness, _strip_stale_tool_call_markers messages = [] @@ -1140,22 +1131,17 @@ class SessionMessagesMixin: content = self._decode_content(row["content"]) if row["role"] in {"user", "assistant"} and isinstance(content, str): content = sanitize_context(content).strip() - msg = {"role": row["role"], "content": content} - msg[_DB_PERSISTED_MARKER_KEY] = True + msg = {"role": row["role"], "content": content, _DB_PERSISTED_MARKER_KEY: True} if include_row_ids and row["id"] is not None: msg["_row_id"] = row["id"] - for col in ("api_content", "display_kind"): - if row[col]: - msg[col] = row[col] + msg.update((col, row[col]) for col in _VERBATIM_COLS if row[col]) if row["display_metadata"]: decoded = self._decode_display_metadata(row["display_metadata"]) if decoded is not None: msg["display_metadata"] = decoded if include_summary_markers and row["_compressed_summary"]: msg["_compressed_summary"] = True - for col in ("timestamp", "tool_call_id", "tool_name", "effect_disposition"): - if row[col]: - msg[col] = row[col] + msg.update((col, row[col]) for col in _META_COLS if row[col]) if row["tool_calls"]: msg["tool_calls"] = _json_or( row["tool_calls"], [], @@ -1167,16 +1153,13 @@ class SessionMessagesMixin: if row["observed"]: msg["observed"] = True if row["role"] == "assistant": - for col in ("finish_reason", "reasoning"): - if row[col]: - msg[col] = row[col] + msg.update((col, row[col]) for col in ("finish_reason", "reasoning") if row[col]) if row["reasoning_content"] is not None: msg["reasoning_content"] = row["reasoning_content"] - for col in ("reasoning_details", "codex_reasoning_items", "codex_message_items"): - if row[col]: - msg[col] = _json_or( - row[col], None, f"Failed to deserialize {col}, falling back to None", - ) + msg.update( + (col, _json_or(row[col], None, f"Failed to deserialize {col}, falling back to None")) + for col in _ASSISTANT_JSON_COLS if row[col] + ) exact_clone_key = None if include_ancestors: skip, exact_clone_key = self._dedupe_replayed_user(messages, msg, exact_user_clones) @@ -1198,9 +1181,7 @@ class SessionMessagesMixin: logger.info( "Repaired %d message-alternation violation(s) while " "restoring session %s — durable transcript kept them, " - "see repair_message_sequence", - repaired, - session_id, + "see repair_message_sequence", repaired, session_id, ) return messages @@ -1246,8 +1227,7 @@ class SessionMessagesMixin: """Count the rows a resume would materialize (see ``_resume_count_scope``).""" session_ids, active_clause = self._resume_count_scope(session_id, tip_only) row = self._read_one( - f"SELECT COUNT(*) FROM messages " - f"WHERE session_id IN ({_placeholders(session_ids)}) AND {active_clause}", + f"SELECT COUNT(*) FROM messages WHERE session_id IN ({_placeholders(session_ids)}) AND {active_clause}", tuple(session_ids), ) return int(row[0] if row else 0) @@ -1273,8 +1253,7 @@ class SessionMessagesMixin: "SELECT COUNT(*) FROM (" f"SELECT 1 FROM messages WHERE session_id IN ({_placeholders(session_ids)}) " f"AND {active_clause} LIMIT ?" - ")", - (*session_ids, max_messages + 1), + ")", (*session_ids, max_messages + 1), ) message_count = int(row[0] if row else 0) if message_count > max_messages: @@ -1303,13 +1282,10 @@ class SessionMessagesMixin: lineage = self._rows_to_conversation( rows, session_id=session_id, include_ancestors=True, repair_alternation=False, include_row_ids=True, ) - prefix: List[Dict[str, Any]] = [] - for message in lineage: - if message.get("_row_id") in ancestor_ids: - projected = message.copy() - projected.pop("_row_id", None) - prefix.append(projected) - return prefix + return [ + {k: v for k, v in message.items() if k != "_row_id"} + for message in lineage if message.get("_row_id") in ancestor_ids + ] def get_conversation_root(self, session_id: str) -> str: """ROOT id of *session_id*'s lineage — the stable conversation id across @@ -1384,20 +1360,7 @@ class SessionMessagesMixin: rows = conn.execute( "SELECT tool_calls FROM messages WHERE session_id = ? AND active = 1", (session_id,), ).fetchall() - tool_call_count = 0 - for row in rows: - raw = row[0] - if not raw: - continue - try: - decoded = json.loads(raw) if isinstance(raw, str) else raw - except (json.JSONDecodeError, TypeError): - continue - if isinstance(decoded, list): - tool_call_count += len(decoded) - elif decoded: - tool_call_count += 1 - return len(rows), tool_call_count + return len(rows), sum(_tool_calls_len(row[0], scalar=1) for row in rows) def _split_rewind_target(self, target_row: Dict[str, Any], expected_target_content: Any, preserve_compaction_handoff: bool): """Validate an active rewind target and return its handoff scaffold (or None). @@ -1453,7 +1416,7 @@ class SessionMessagesMixin: active_rows = conn.execute( "SELECT id FROM messages WHERE session_id = ? AND active = 1 ORDER BY id", (session_id,), ).fetchall() - if [int(active_row[0]) for active_row in active_rows] != expected_active_ids: + if [int(r[0]) for r in active_rows] != expected_active_ids: raise RuntimeError("active transcript changed before the rewind could be persisted") row = conn.execute( "SELECT * FROM messages WHERE id = ? AND session_id = ?", (target_message_id, session_id), @@ -1466,15 +1429,13 @@ class SessionMessagesMixin: f"rewind target must be a 'user' message (got role=" f"{target_row.get('role')!r}, id={target_message_id})" ) - replacement_message_id: Optional[int] = None - replacement: Optional[Dict[str, Any]] = None + replacement_message_id = replacement = None if preserve_compaction_handoff or expected_target_content is not None: replacement = self._split_rewind_target(target_row, expected_target_content, preserve_compaction_handoff) - cursor = conn.execute( + ids = [r[0] for r in conn.execute( "SELECT id FROM messages WHERE session_id = ? AND id >= ? AND active = 1", (session_id, target_message_id), - ) - ids = [r[0] for r in cursor.fetchall()] + ).fetchall()] if ids: conn.execute(f"UPDATE messages SET active = 0 WHERE id IN ({_placeholders(ids)})", ids) if replacement is not None: @@ -1491,8 +1452,7 @@ class SessionMessagesMixin: head_row = conn.execute( "SELECT MAX(id) FROM messages WHERE session_id = ? AND active = 1", (session_id,), ).fetchone() - new_head_id = head_row[0] if head_row and head_row[0] is not None else None - return target_row, ids, new_head_id, replacement_message_id + return target_row, ids, head_row[0] if head_row else None, replacement_message_id target_row, rewound, new_head_id, replacement_message_id = self._execute_write(_do) # Decode for the prompt-buffer prefill without a second fallible DB operation. @@ -1504,12 +1464,9 @@ class SessionMessagesMixin: def message_count(self, session_id: str = None) -> int: """Count messages, optionally for a specific session.""" - with self._read_ctx() as conn: - if session_id: - cursor = conn.execute("SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,)) - else: - cursor = conn.execute("SELECT COUNT(*) FROM messages") - return cursor.fetchone()[0] + if session_id: + return self._read_one("SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,))[0] + return self._read_one("SELECT COUNT(*) FROM messages")[0] def has_platform_message_id(self, session_id: str, platform_message_id: str) -> bool: """True when a message with *platform_message_id* exists (partial index lookup; @@ -1538,12 +1495,9 @@ class SessionMessagesMixin: return False if not isinstance(cfg, dict): return False + markers = (cfg.get("_branched_from"), cfg.get("_delegate_from")) parent_id = session.get("parent_session_id") - branched = cfg.get("_branched_from") - delegated = cfg.get("_delegate_from") - if parent_id: - return branched == parent_id or delegated == parent_id - return branched is not None or delegated is not None + return parent_id in markers if parent_id else any(m is not None for m in markers) def is_explicit_fork_child(self, session_id: str) -> bool: """Public read-only view of :meth:`_is_explicit_fork_child_row`; a missing row @@ -1568,10 +1522,9 @@ class SessionMessagesMixin: "SELECT generation FROM conversation_generations WHERE source = ? AND session_key = ?", (source, session_key), ) - if row is None or row["generation"] is None: + if row is None or row["generation"] is None or int(row["generation"]) <= 0: return None - generation = int(row["generation"]) - return generation if generation > 0 else None + return int(row["generation"]) def clear_messages(self, session_id: str) -> None: """Delete all messages for a session and reset its counters.""" @@ -1625,7 +1578,7 @@ class SessionMessagesMixin: def _do(conn): ids = _find_affected(conn) if ids: - conn.execute(f"UPDATE messages SET content = '' WHERE id IN ({','.join('?' * len(ids))})", ids) + conn.execute(f"UPDATE messages SET content = '' WHERE id IN ({_placeholders(ids)})", ids) return ids affected_ids = self._execute_write(_do) diff --git a/hermes_state_titles.py b/hermes_state_titles.py index c6ac1dfdc6..64272588bf 100644 --- a/hermes_state_titles.py +++ b/hermes_state_titles.py @@ -175,8 +175,7 @@ class SessionTitlesMixin: row = self._read_one( "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.title = ?", - (title,), + "WHERE s.title = ?", (title,), ) return self._session_row_dict(row) if row else None diff --git a/hermes_state_usage.py b/hermes_state_usage.py index f84455053e..0b99762758 100644 --- a/hermes_state_usage.py +++ b/hermes_state_usage.py @@ -245,8 +245,7 @@ class SessionUsageMixin: # Writer stuck mid-apply: leave deltas unapplied rather than race it. logger.warning( "async token accounting: writer did not stop within %.0fs; " - "%d queued delta(s) not persisted", - join_timeout, len(self._token_queue), + "%d queued delta(s) not persisted", join_timeout, len(self._token_queue), ) return # Writer gone: apply leftovers synchronously under the same busy protocol. Wait @@ -366,8 +365,7 @@ class SessionUsageMixin: """ row = conn.execute( "SELECT model, billing_provider, billing_base_url, billing_mode " - "FROM sessions WHERE id = ?", - (session_id,), + "FROM sessions WHERE id = ?", (session_id,), ).fetchone() sess = dict(row) if (row is not None and not task) else {} eff_model = model or sess.get("model") or "unknown" From 388fe39f5535a2f0a04dfaded089ca51224f8487 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 17:06:57 -0700 Subject: [PATCH 4/4] refactor(state): compact messages docstrings (WHY kept); append_message builds msg from its own kwargs --- hermes_state_messages.py | 362 ++++++++++++++------------------------- hermes_state_titles.py | 3 +- hermes_state_usage.py | 3 +- 3 files changed, 126 insertions(+), 242 deletions(-) diff --git a/hermes_state_messages.py b/hermes_state_messages.py index 0d801db935..8e6d75f9ba 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -112,14 +112,10 @@ class SessionMessagesMixin: """Message append/replace/rewind, reactions, resume conversations, replay dedupe.""" def _bump_conversation_generation(self, conn, session_id: str, end_reason: str) -> None: - """Advance this peer's conversation generation past a boundary, inside the - transaction that writes the boundary. - - Only ``_RESET_END_REASONS`` count (``compression`` continues one conversation). - The counter never reads the session rows: an aggregate over them could re-emit a - pair once ``delete_session()``/pruning removes an ended row and hand a new - conversation a retired affinity identity. It only ever increments. - """ + """Advance this peer's conversation generation past a boundary, inside the txn that + writes it. Only ``_RESET_END_REASONS`` count (``compression`` continues one + conversation). Never derived from session rows (deletes/prunes could re-emit a + retired affinity identity); it only ever increments.""" if end_reason not in _RESET_END_REASONS: return row = conn.execute("SELECT source, session_key FROM sessions WHERE id = ?", (session_id,)).fetchone() @@ -143,8 +139,7 @@ class SessionMessagesMixin: def _encode_content(cls, content: Any) -> Any: """Serialize list/dict content (multimodal parts) as a sentinel-prefixed JSON string; sqlite3 can only bind str/bytes/int/float/None. Lone surrogates are - scrubbed from text so persistence never fails. Paired with :meth:`_decode_content`. - """ + scrubbed from text so persistence never fails. Paired with :meth:`_decode_content`.""" if isinstance(content, str): return _sanitize_surrogates(content) if content is None or isinstance(content, (bytes, int, float)): @@ -222,16 +217,12 @@ class SessionMessagesMixin: reject_active_turn_lease: bool = False, reject_active_compression_lock: bool = False, allow_closed_compression_parent: bool = False, ) -> None: - """Transcript-write admission checks, run INSIDE the write txn. - - Shared by every transcript writer so they cannot diverge. Ordinary appends do - NOT check compression_locks: the lock only stops two COMPRESSIONS colliding, and - archive_and_compact() commits against a watermark and clones later rows, so - concurrent appends are safe by construction (blocking them killed turns while a - slow summary held the lease). Destructive user mutations opt in via - ``reject_active_compression_lock`` / ``reject_active_turn_lease`` so a compressor - that captured its watermark cannot resurrect the removed turn. - """ + """Transcript-write admission checks, run INSIDE the write txn (shared by every + writer). Ordinary appends do NOT check compression_locks: the lock only stops two + COMPRESSIONS colliding and archive_and_compact() commits against a watermark, so + concurrent appends are safe (blocking them killed turns during slow summaries). + Destructive user mutations opt in via ``reject_active_*`` so a compressor that + captured its watermark cannot resurrect the removed turn.""" from hermes_state import CompressionSessionClosedError, SessionCompressionInProgressError, SessionTurnLeaseLostError if reject_active_compression_lock: active_lock = conn.execute(_COMPRESSION_LOCK_ROW_SQL, (session_id,)).fetchone() @@ -282,8 +273,7 @@ class SessionMessagesMixin: """Bind values for ``_INSERT_MESSAGE_SQL`` from one message dict. *tool_calls* is the already-parsed value (see ``_parse_tool_calls``). - *keep_reasoning* False stores NULL for every reasoning/codex column. - """ + *keep_reasoning* False stores NULL for every reasoning/codex column.""" from hermes_state import _scrub_surrogates def _str_or_none(value): @@ -330,27 +320,13 @@ class SessionMessagesMixin: compression_lock_holder: Optional[str] = None, turn_lease_holder: Optional[str] = None, turn_lease_ttl_seconds: float = 300.0, ) -> int: - """Append one message; returns the row id. Bumps ``message_count`` (and - ``tool_call_count`` when tool_calls are present). - - ``platform_message_id`` is the platform's own id (Telegram update_id, Yuanbao - msg_id) used by recall-style flows. ``api_content`` is the byte-fidelity - sidecar — the exact string sent to the API when it differed from ``content`` — - stored as sent except lone surrogates (which the loop scrubs anyway). - """ - msg = { - "content": content, "tool_name": tool_name, "tool_call_id": tool_call_id, - "token_count": token_count, "finish_reason": finish_reason, "reasoning": reasoning, - "reasoning_content": reasoning_content, "reasoning_details": reasoning_details, - "codex_reasoning_items": codex_reasoning_items, "codex_message_items": codex_message_items, - "platform_message_id": platform_message_id, "message_id": platform_message_id, - "observed": observed, "effect_disposition": effect_disposition, - "_compressed_summary": _compressed_summary, "api_content": api_content, - "display_kind": display_kind, "display_metadata": display_metadata, - } + """Append one message; returns the row id and bumps the session counters. + ``platform_message_id``: the platform's own id (recall-style flows). + ``api_content``: byte-fidelity sidecar — the exact string sent to the API when it + differed from ``content`` — stored as sent except lone surrogates.""" + msg = dict(locals()) # every keyword above is a message-dict field of the same name # Encode outside the write txn (display metadata first: log-order parity). - display_metadata_json = self._encode_display_metadata(display_metadata) - msg["display_metadata"] = display_metadata_json + msg["display_metadata"] = self._encode_display_metadata(display_metadata) tool_calls = _parse_tool_calls(tool_calls) message_timestamp = _coerce_timestamp(timestamp, time.time()) num_tool_calls = _tool_calls_count(tool_calls) @@ -385,13 +361,10 @@ class SessionMessagesMixin: compression_lock_holder: Optional[str] = None, turn_lease_holder: Optional[str] = None, chunk_rows: Optional[int] = None, turn_lease_ttl_seconds: float = 300.0, ) -> int: - """Append *messages* (``_insert_message_rows`` dict shape) in ONE write txn. - - All rows land or none do; the admission guards run once for the batch. - ``chunk_rows`` bounds transaction size for LARGE copies (branch seeds: FTS - triggers run per row, 10k rows ≈ 2.4s under one lock) — commits in chunks with - the old per-row-loop recovery semantics. Returns the inserted row count. - """ + """Append *messages* (``_insert_message_rows`` dict shape) in ONE write txn: all + rows land or none, guards run once. ``chunk_rows`` bounds txn size for LARGE + copies (branch seeds; FTS triggers run per row) by committing in chunks. + Returns the inserted row count.""" if not messages: return 0 if chunk_rows is not None and len(messages) > chunk_rows: @@ -468,12 +441,9 @@ class SessionMessagesMixin: def set_message_reaction( self, session_id: str, message_row_id: int, emoji: Optional[str], *, author: str = "user", ) -> Optional[List[Dict[str, Any]]]: - """Set (or with ``emoji=None`` clear) *author*'s reaction on one message. - - Tapback semantics: one reaction per author per message; the same emoji again - clears it, a different one replaces it. Returns the message's reaction list - after the write, or ``None`` when the row isn't part of *session_id*. - """ + """Set (``emoji=None``: clear) *author*'s reaction. Tapback semantics: one per + author per message; the same emoji again clears it, a different one replaces it. + Returns the reaction list after the write, or ``None`` for a foreign row.""" from hermes_state import _scrub_surrogates if not session_id or message_row_id is None: return None @@ -518,8 +488,7 @@ class SessionMessagesMixin: """Return *author*'s not-yet-surfaced reactions and mark them seen. Reactions are announced on the NEXT user turn (never by rewriting the reacted - message — cache-safe); the ``seen`` stamp makes each announcement exactly once. - """ + message — cache-safe); the ``seen`` stamp makes each announcement exactly once.""" if not session_id: return [] @@ -562,12 +531,9 @@ class SessionMessagesMixin: def latest_message_row_id( self, session_id: str, *, role: str = "user", offset: int = 0, require_text: bool = True ) -> Optional[int]: - """Row id of the most recent active message with *role*, or ``None``. - - ``offset`` steps to earlier turns (1 = the one before the latest). ``require_text`` - skips rows without plain-text content (tool-call-only turns, attachment stubs) - so "the latest message" never resolves to an invisible bubble. - """ + """Row id of the most recent active *role* message, or ``None``. ``offset`` steps + to earlier turns; ``require_text`` skips rows without plain-text content so "the + latest message" never resolves to an invisible bubble.""" if not session_id or role not in {"user", "assistant"} or offset < 0: return None text_filter = "AND content IS NOT NULL AND TRIM(content) != '' " if require_text else "" @@ -594,12 +560,9 @@ class SessionMessagesMixin: return row[0] if row else None def _insert_message_rows(self, conn, session_id: str, messages: List[Dict[str, Any]]) -> tuple[int, int]: - """Insert *messages* as fresh active rows inside the caller's write txn. - - Returns ``(inserted_count, tool_call_count)``; does NOT touch sessions.* - counters (callers reconcile them differently). Reasoning columns are kept for - assistant rows only. Stamps ``msg["_row_id"]``. - """ + """Insert *messages* as fresh active rows inside the caller's write txn. Returns + ``(inserted, tool_call_count)``; never touches sessions.* counters (callers + reconcile differently). Reasoning columns kept for assistant rows only.""" now_ts = time.time() inserted = tool_calls_total = 0 for msg in messages: @@ -623,18 +586,12 @@ class SessionMessagesMixin: self, session_id: str, messages: List[Dict[str, Any]], active_only: bool = False, archive_dropped: bool = False, reject_active_turn_lease: bool = False, ) -> None: - """Atomically replace the stored messages for a session (/retry, /undo, /compress). - - DESTRUCTIVE by default: every row is DELETEd (and leaves the FTS index). - ``active_only=True`` replaces only ``active = 1`` rows, leaving soft-archived - rows (compacted turns, rewind rows) untouched — required when sharing a session - id with an agent doing in-place compaction. ``archive_dropped=True`` SOFT-archives - the live rows (``active = 0, compacted = 0``, rewind-style) instead of deleting: - the mode rewind/edit/regenerate must use, since a DELETE leaves nothing to - recover from; it implies active-only handling. ``reject_active_turn_lease=True`` - runs the lease check in the same write txn for user-initiated rewrites that do - not own the cross-process lease. - """ + """Atomically replace a session's stored messages (/retry, /undo, /compress). + DESTRUCTIVE by default (rows DELETEd, leave FTS). ``active_only`` spares + soft-archived rows (needed alongside in-place compaction). ``archive_dropped`` + SOFT-archives the live rows rewind-style instead of deleting — what rewind/edit/ + regenerate must use, since a DELETE leaves nothing to recover. ``reject_active_ + turn_lease`` runs the lease check in-txn for user rewrites that don't own it.""" from hermes_state import CompressionSessionClosedError active_clause = " AND active = 1" if active_only else "" @@ -713,28 +670,20 @@ class SessionMessagesMixin: model_config_patch: Optional[Dict[str, Any]] = None, watermark: Optional[int] = None, lock_holder: Optional[str] = None, tail_count: int = 0, ) -> int: - """Non-destructive in-place compaction under ONE durable session id. + """Non-destructive in-place compaction under ONE durable session id: soft-archive + the active rows (``active=0, compacted=1`` — "summarized away", still searchable) + and insert *compacted_messages* as fresh active rows, atomically. - Soft-archives the active rows (``active=0, compacted=1`` — "summarized away", - still found by search_messages and readable with include_inactive) and inserts - *compacted_messages* as fresh active rows, atomically. Live-context loads - filter ``active = 1`` so the model reloads only the compacted set. - - *watermark* (``get_active_message_watermark`` at compression START): rows with - ``id > watermark`` arrived during the slow summary and are re-sequenced after - the compacted set by a pure-SQL column clone (fresh ids — consumers re-resolve - by content); ``None`` archives everything. *lock_holder*: the commit verifies - inside the txn that the compression lock is still held and unexpired, so a - reclaimed lease fails instead of clobbering the winner. *tail_count*: the LAST - N rows of *compacted_messages* are the verbatim carried-forward tail; their - originals (at/below the watermark) and the watermark clones' originals are - superseded duplicates and get rewind-style flags (``active=0, compacted=0``) so - session_search doesn't return each carried message once per compaction. - - ``message_count`` becomes the ACTIVE count; ``model_config_patch`` merges into - the session JSON in the same txn (``None`` value removes a key). Returns the new - active count. - """ + *watermark* (captured at compression START): rows with ``id > watermark`` arrived + during the slow summary and are re-sequenced after the compacted set by a pure- + SQL column clone (fresh ids); ``None`` archives everything. *lock_holder*: the + commit verifies in-txn that the lease is still held, so a reclaimed lease fails + instead of clobbering the winner. *tail_count*: the LAST N compacted rows are the + verbatim carried-forward tail; their originals and the watermark clones' originals + are superseded duplicates and get rewind-style flags (``active=0, compacted=0``) + so search doesn't return each carried message once per compaction. + ``message_count`` becomes the ACTIVE count; ``model_config_patch`` merges in the + same txn (``None`` removes a key). Returns the new active count.""" from hermes_state import SessionCompressionInProgressError def _do(conn): @@ -827,14 +776,10 @@ class SessionMessagesMixin: return cols def set_latest_user_api_content(self, session_id: str, content: Any, api_content: str) -> int: - """Backfill the ``api_content`` sidecar onto the newest ACTIVE user row. - - In-place preflight compaction inserts the current user row BEFORE the turn - prologue composes the sidecar, and the later persist identity-skips compacted - dicts; without this the reload would replay clean content and reopen the - prompt-cache divergence. The ``content`` match guards against a racing rewrite. - Returns rows updated (0 or 1). - """ + """Backfill the ``api_content`` sidecar onto the newest ACTIVE user row. Preflight + compaction inserts that row BEFORE the sidecar is composed and the later persist + identity-skips compacted dicts; without this a reload would reopen the prompt- + cache divergence. The ``content`` match guards a racing rewrite. Returns 0/1.""" from hermes_state import _scrub_surrogates return self._write_rowcount( "UPDATE messages SET api_content = ? WHERE id = (SELECT id FROM messages " @@ -844,14 +789,10 @@ class SessionMessagesMixin: ) def _dedupe_display_generations(self, rows): - """Collapse compaction generations so each logical message appears once. - - Compaction copies the protected tail into each generation (same - role/content/timestamp, different ``active``/id); prefer the live row, then the - newest generation. The ONE definition shared by every display projection - (get_messages, get_resume_conversations, get_ancestor_display_prefix, - get_messages_as_conversation). *rows* must be ordered by ``id``; order is kept. - """ + """Collapse compaction generations so each logical message appears once: the + protected tail is copied into each generation (same role/content/timestamp, + different ``active``/id); prefer the live row, then the newest. The ONE definition + shared by every display projection. *rows* must be ordered by ``id``.""" seen: Dict[Tuple[Any, ...], Any] = {} for row in rows: dedupe_content = row["content"] @@ -907,15 +848,10 @@ class SessionMessagesMixin: limit: Optional[int] = None, offset: int = 0, latest: bool = False, after_id: Optional[int] = None, ) -> List[Dict[str, Any]]: - """Load messages for a session in insertion order (AUTOINCREMENT id, never - timestamp — clocks regress on WSL2/NTP steps). - - ``include_inactive`` loads soft-deleted rewind rows; ``include_compacted`` adds - rows preserved by in-place compaction (durable display history a transcript - read must not drop) but not rewind rows. ``limit``/``offset`` page; ``latest`` - measures the offset back from the newest row and still returns chronological - order. ``after_id`` is keyset paging (``id > after_id``), ascending only. - """ + """Load a session's messages in insertion order (id, never timestamp — clocks + regress). ``include_inactive``: rewind rows too; ``include_compacted``: compaction- + archived display history (not rewind rows). ``latest`` pages back from the newest + row but still returns chronological order; ``after_id`` is keyset paging.""" if after_id is not None and (latest or offset): raise ValueError("after_id is incompatible with latest/offset paging") if after_id is not None and include_compacted: @@ -971,12 +907,9 @@ class SessionMessagesMixin: return found def get_messages_around(self, session_id: str, around_message_id: int, window: int = 5) -> Dict[str, Any]: - """Window of up to *window* messages either side of an anchor id (id ascending). - - ``messages_before``/``messages_after`` count rows in the returned slice strictly - before/after the anchor; less than *window* means a session boundary. Empty - window when the anchor is not a row of *session_id*. - """ + """Up to *window* messages either side of an anchor id (ascending). ``messages_ + before``/``_after`` count the slice strictly around the anchor (fewer than + *window* = session boundary). Empty when the anchor is not in *session_id*.""" window = max(window, 0) with self._read_ctx() as conn: anchor_exists = conn.execute( @@ -999,16 +932,11 @@ class SessionMessagesMixin: return {"window": window_msgs, "messages_before": max(0, len(before_rows) - 1), "messages_after": len(after_rows)} def resolve_resume_session_id(self, session_id: str) -> str: - """Redirect a resume target to the descendant session that holds the messages. - - Follows the compression chain to the live tip first (``get_compression_tip`` is - lineage-aware: only children of compression-ended parents, so delegation/branch - children never hijack the resume), then walks ``parent_session_id`` forward, - returning the deepest node with messages — never short-circuiting on the start - node, since a continuation may hold the newer turns. Branch, delegate, reset - and tool children are skipped (they carry ``parent_session_id`` too). Returns - *session_id* unchanged when nothing has messages. Depth cap 32. - """ + """Redirect a resume target to the descendant that holds the messages: follow the + compression chain to the live tip (lineage-aware, so delegate/branch children never + hijack it), then walk ``parent_session_id`` forward to the DEEPEST node with + messages (a continuation may hold newer turns), skipping branch/delegate/reset/ + tool children. Unchanged when nothing has messages. Depth cap 32.""" if not session_id: return session_id try: @@ -1063,15 +991,11 @@ class SessionMessagesMixin: repair_alternation: bool = False, include_row_ids: bool = False, include_compacted: bool = False, ) -> List[Dict[str, Any]]: - """Load messages in OpenAI conversation format (gateway history restore). - - ``include_compacted`` adds compaction-archived rows deduped by - :meth:`_dedupe_display_generations` — DISPLAY reads only; the model-fed restore - must not pass it or a resume regrows the history compaction summarized away. - ``repair_alternation`` runs ``repair_message_sequence`` on the loaded list - (LIVE REPLAY callers) so a durable ``user;user`` pair doesn't re-trigger the - per-request repair forever; the stored transcript is never mutated. - """ + """Load messages in OpenAI conversation format. ``include_compacted`` (deduped + display history) is for DISPLAY reads only — the model-fed restore must not regrow + what compaction summarized away. ``repair_alternation`` repairs the loaded list for + LIVE REPLAY callers (a durable ``user;user`` pair would otherwise re-trigger the + per-request repair forever); the stored transcript is never mutated.""" session_ids = [session_id] if include_ancestors and not self._is_explicit_branch_session(session_id): session_ids = self._session_lineage_root_to_tip(session_id) @@ -1086,14 +1010,10 @@ class SessionMessagesMixin: ) def _dedupe_replayed_user(self, messages, msg, exact_user_clones) -> Tuple[bool, Any]: - """Ancestor-lineage dedupe for one decoded user *msg*. - - Returns ``(skip, exact_clone_key)``. Watermark rotation column-clones the - concurrent tail into the child after the summary, so the copies need not be - adjacent: an exact ``(timestamp, canonical content)`` clone index is checked - first, then the adjacent-duplicate heuristic. A rotated child carrier wins over - the simpler ancestor copy (it owns the durable row id and the summary scaffold). - """ + """Ancestor-lineage dedupe for one decoded user *msg* -> ``(skip, exact_clone_key)``. + Rotation column-clones the concurrent tail into the child, so copies need not be + adjacent: the exact ``(timestamp, canonical content)`` clone index is checked first, + then the adjacent heuristic. A rotated child carrier wins over the ancestor copy.""" canonical_content, _is_composite = self._canonical_replayed_user_content(msg) exact_clone_key = self._exact_replayed_user_clone_key(msg.get("timestamp"), canonical_content) previous_exact = exact_user_clones.get(exact_clone_key) if exact_clone_key is not None else None @@ -1122,8 +1042,7 @@ class SessionMessagesMixin: Every dict is stamped ``_DB_PERSISTED_MARKER_KEY`` at the source (born durable) so an identity-losing handoff never re-appends the whole transcript on flush. ``_row_id`` is opt-in (gateway reactions). Reasoning fields are restored on - assistant rows only. Key order of each dict is stable (see the column tables). - """ + assistant rows only. Key order of each dict is stable (see the column tables).""" from hermes_state import _strip_background_review_harness, _strip_stale_tool_call_markers messages = [] exact_user_clones: Dict[Tuple[Any, str], Dict[str, Any]] = {} @@ -1186,14 +1105,10 @@ class SessionMessagesMixin: return messages def get_resume_conversations(self, session_id: str) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: - """``(model_history, display_history)`` for a session resume from ONE SELECT. - - ``model_history``: the tip's active rows, alternation-repaired, with the summary - marker kept for pre-compress checkpointing. ``display_history``: the full - compression lineage (``/branch`` sessions are their own lineage) verbatim, with - compaction-archived rows included and deduped, plus replayed-user dedup. Byte- - identical to the separate reads (test_get_resume_conversations_matches_separate_reads). - """ + """``(model_history, display_history)`` for a resume from ONE SELECT. model: the + tip's active rows, alternation-repaired, summary marker kept for pre-compress + checkpointing. display: the full lineage (``/branch`` sessions stand alone) with + compaction-archived rows deduped. Byte-identical to the separate reads.""" session_ids = self._resume_lineage_ids(session_id) rows = self._fetch_conversation_rows(session_ids, _DISPLAY_ACTIVE_CLAUSE, with_session_id=True) # The model projection stays active-only: it is the compressed working context. @@ -1233,14 +1148,10 @@ class SessionMessagesMixin: return int(row[0] if row else 0) def assert_resume_safe(self, session_id: str, max_messages: Optional[int] = None, *, tip_only: bool = False) -> int: - """Return the resume row count or raise ``SessionResumeTooLargeError``. - - ``max_messages=None`` reads ``sessions.max_resume_messages``; 0 disables the - guard and returns 0 without counting. ``tip_only`` bounds only the tip's active - rows, for callers that never materialize the lineage in memory — a heavily - compressed conversation (~29k lineage rows behind a ~700-row tip) is exactly - what compression should produce and must not be rejected. - """ + """Resume row count, or raise ``SessionResumeTooLargeError``. ``max_messages=None`` + reads config; 0 disables the guard without counting. ``tip_only`` bounds only the + tip's active rows for callers that never materialize the lineage — a heavily + compressed conversation is what compression should produce, not a rejection.""" from hermes_state import SessionResumeTooLargeError, resolved_max_resume_messages if max_messages is None: max_messages = resolved_max_resume_messages() @@ -1264,12 +1175,9 @@ class SessionMessagesMixin: return message_count def get_ancestor_display_prefix(self, session_id: str) -> List[Dict[str, Any]]: - """Ancestor-only display messages of a lineage (rows with ``session_id !=`` tip). - - ``session.resume`` prepends this to the live model history. Identifying - ancestors by row origin (not ``display[:len(display) - len(model)]``) avoids - overcounting when alternation repair removes tip messages from the middle. - """ + """Ancestor-only display messages of a lineage (row ``session_id != tip``), which + ``session.resume`` prepends to the model history. Identified by row origin, not + ``display[:len(display) - len(model)]``, so alternation repair cannot overcount.""" session_ids = self._resume_lineage_ids(session_id) if len(session_ids) <= 1: return [] @@ -1323,13 +1231,10 @@ class SessionMessagesMixin: def _find_duplicate_replayed_user_message( messages: List[Dict[str, Any]], msg: Dict[str, Any] ) -> Optional[Tuple[int, bool]]: - """Return an adjacent replay duplicate and whether *msg* must win. - - Rotation may persist the current ask once in the parent and again inside a - composite child carrier; compare the canonical live payload for carriers while - keeping the exact-string dedupe for ordinary replayed users. The child carrier - wins (it owns the durable row id and the retained scaffold). - """ + """Adjacent replay duplicate ``(index, prefer_current)`` or None. Rotation may + persist the current ask in the parent and again inside a composite child carrier: + carriers compare by canonical live payload, ordinary users by exact string. The + child carrier wins (it owns the durable row id and the retained scaffold).""" from hermes_state import SessionDB if msg.get("role") != "user": return None @@ -1363,12 +1268,9 @@ class SessionMessagesMixin: return len(rows), sum(_tool_calls_len(row[0], scalar=1) for row in rows) def _split_rewind_target(self, target_row: Dict[str, Any], expected_target_content: Any, preserve_compaction_handoff: bool): - """Validate an active rewind target and return its handoff scaffold (or None). - - Raises ``ValueError`` for an inactive / non-user-originated target or a missing - composite carrier, ``RuntimeError`` when the canonical live payload no longer - matches *expected_target_content*. - """ + """Validate an active rewind target; return its handoff scaffold (or None). + ``ValueError``: inactive / non-user-originated target or missing composite + carrier; ``RuntimeError``: live payload no longer matches *expected_target_content*.""" if not target_row.get("active"): raise ValueError("rewind target is not active") from agent.context_compressor import split_user_originated_turn @@ -1392,21 +1294,16 @@ class SessionMessagesMixin: self, session_id: str, target_message_id: int, *, preserve_compaction_handoff: bool = False, expected_active_ids: Optional[List[int]] = None, expected_target_content: Any = None, ) -> Dict[str, Any]: - """Soft-delete (``active=0``) every message with id >= *target_message_id*. - - The target itself goes inactive so the caller can pre-fill it as the next - prompt. Returns ``{"rewound_count", "target_message", "new_head_id"}`` (plus - ``replacement_message_id`` with ``preserve_compaction_handoff``, which archives a - composite summary carrier and inserts its hidden handoff scaffold as the new - head in the same txn). Raises ``ValueError`` when the target is missing or not a - ``user`` row. - - ``expected_active_ids`` / ``expected_target_content`` pin the active row set and - the canonical live payload inside the txn before any mutation (presentation-only - metadata changes do not invalidate a rewind). A live cross-process turn lease - refuses the rewind; expired/dead holders are reclaimed. ``rewind_count`` always - increments, even when the target was already inactive. - """ + """Soft-delete (``active=0``) every message with id >= *target_message_id*, the + target included (the caller pre-fills it as the next prompt). Returns + ``{"rewound_count", "target_message", "new_head_id"}``, plus + ``replacement_message_id`` with ``preserve_compaction_handoff`` (archives a + composite summary carrier, inserts its hidden handoff scaffold as the new head). + ``ValueError`` when the target is missing or not a ``user`` row. + ``expected_active_ids`` / ``expected_target_content`` pin the active set and the + canonical live payload in-txn before any mutation (presentation-only metadata + changes don't invalidate a rewind). A live turn lease refuses the rewind; + expired/dead holders are reclaimed. ``rewind_count`` always increments.""" def _do(conn): self._check_transcript_write_guards( @@ -1477,13 +1374,10 @@ class SessionMessagesMixin: ) is not None def _is_explicit_fork_child_row(self, session: Dict[str, Any]) -> bool: - """True when *session* is a branch, delegate, or tool child of its parent. - - Markers only count when they point at ``parent_session_id``: compression copies - ``model_config`` onto the continuation, so a delegate's continuation carries - ``_delegate_from=`` and presence-only matching would - misclassify it (same binding as ``_NON_CONTINUATION_CHILD_FILTER_SQL``). - """ + """True when *session* is a branch, delegate, or tool child of its parent. Markers + only count when they point at ``parent_session_id``: compression copies + ``model_config`` onto the continuation, so presence-only matching would misclassify + a delegate's continuation (same binding as ``_NON_CONTINUATION_CHILD_FILTER_SQL``).""" if session.get("source") == "tool": return True raw = session.get("model_config") @@ -1506,16 +1400,12 @@ class SessionMessagesMixin: return bool(session and self._is_explicit_fork_child_row(session)) def latest_conversation_boundary(self, session_key: str, source: str) -> Optional[int]: - """How many conversation boundaries (``_RESET_END_REASONS`` ends) this routing - peer has crossed, or ``None`` when never reset. - - The peer is ``(session_key, source)`` — the identity recovery uses — never the - key alone (an API caller may legally reuse a Telegram row's key). Read from - ``conversation_generations`` (advanced inside each boundary's txn), not an - aggregate over session rows: deletes/prunes would let an aggregate re-emit a - retired pair. Rows are never garbage-collected, by design (dropping one would - re-issue generation 1 — the ABA this counter prevents). - """ + """Conversation boundaries (``_RESET_END_REASONS`` ends) this routing peer has + crossed, or ``None`` when never reset. The peer is ``(session_key, source)`` — never + the key alone (an API caller may reuse a Telegram row's key). Read from + ``conversation_generations``, not an aggregate over session rows (deletes/prunes + would re-emit a retired pair); rows are never GC'd (dropping one would re-issue + generation 1 — the ABA this counter prevents).""" if not session_key or not source: return None row = self._read_one( @@ -1536,15 +1426,11 @@ class SessionMessagesMixin: self._execute_write(_do) def purge_stale_tool_call_markers(self, *, dry_run: bool = False, backup: bool = True) -> Dict[str, Any]: - """Permanently clear bare tool-call marker content (e.g. "[memory]") left by - pre-fix sessions. ``_rows_to_conversation`` repairs this in memory on every load, - so this is optional; it just stops the re-scan and removes the bytes. - - Only ``content`` is touched (tool_call pairing unaffected). With ``backup`` a - ``VACUUM INTO`` snapshot (safe against a live connection) is taken first; none - when nothing changes. ``dry_run`` reports without writing or backing up. - Returns ``{"dry_run", "rows_affected", "row_ids", "backup_path"}``. - """ + """Permanently clear bare tool-call marker content (e.g. "[memory]") left by pre-fix + sessions (``_rows_to_conversation`` already repairs it in memory; this stops the + re-scan). Only ``content`` is touched. ``backup``: ``VACUUM INTO`` snapshot first + (none when nothing changes). Returns ``{"dry_run", "rows_affected", "row_ids", + "backup_path"}``.""" from hermes_state import _STALE_TOOL_CALL_MARKER_RE def _find_affected(conn) -> List[int]: diff --git a/hermes_state_titles.py b/hermes_state_titles.py index 64272588bf..9aad634895 100644 --- a/hermes_state_titles.py +++ b/hermes_state_titles.py @@ -79,8 +79,7 @@ class SessionTitlesMixin: nothing overwrites a user name, re-running the titler on an llm row is a no-op). No writer may move a hidden canonical Bot Chat off its title. Read and write are one compare-and-swap transaction, so a manual ``/title`` racing an in-flight - generation is not clobbered. - """ + generation is not clobbered.""" title = self.sanitize_title(title) is_user = source == self.TITLE_SOURCE_USER new_rank = self._title_rank(source) if not is_user else None diff --git a/hermes_state_usage.py b/hermes_state_usage.py index 0b99762758..90f32aded3 100644 --- a/hermes_state_usage.py +++ b/hermes_state_usage.py @@ -361,8 +361,7 @@ class SessionUsageMixin: caller's write txn after the ``sessions`` UPDATE. A missing model/provider falls back to the session row (same COALESCE behaviour as the summary update) — except for aux rows (``task`` set), which must NOT inherit the main-loop route (vision - on gemini while the main loop runs anthropic): missing info stays 'unknown'/empty. - """ + on gemini while the main loop runs anthropic): missing info stays 'unknown'/empty.""" row = conn.execute( "SELECT model, billing_provider, billing_base_url, billing_mode " "FROM sessions WHERE id = ?", (session_id,),