hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(state): hand-compact messages mixin docstrings; reuse _json_or/_parse_tool_calls/_read_all; drop identity wrappers

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:15:57 -07:00
parent 1e1134c476
commit 0c1cb793cc
1 changed files with 190 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

441
hermes_state_messages.py
View File

@@ -1,9 +1,5 @@
"""Transcript persistence for SessionDB: message append / replace / rewind, reactions,
resume-conversation assembly and replayed-user-message dedupe.
Mixin bound onto ``SessionDB`` via the MRO, built on its ``_read_ctx`` /
``_execute_write`` / ``_write_rowcount`` / ``_read_one`` / ``_read_all`` primitives.
"""
"""Transcript persistence for SessionDB: message append/replace/rewind, reactions, resume assembly,
replayed-user dedupe. Mixin bound via the MRO, built on SessionDB's _read_ctx/_execute_write/_read_* primitives."""
from __future__ import annotations
@@ -19,8 +15,7 @@ from hermes_state_common import (
_COMPRESSION_LOCK_ROW_SQL, _ENDED_ROW_SQL, _RESET_END_REASONS, _RESET_END_REASONS_SQL, _ended_by_compression,
_legacy_reset_child_sql, _placeholders)
# Log-record parity with the origin module (caplog tests pin "hermes_state").
logger = logging.getLogger("hermes_state")
logger = logging.getLogger("hermes_state") # caplog tests pin the origin module's name
# One INSERT shape for every message writer (append, batch, replace, compact, import).
_INSERT_MESSAGE_SQL = """INSERT INTO messages (session_id, role, content, tool_call_id,
@@ -56,18 +51,6 @@ def _json_or(raw: Any, fallback: Any, warning: str) -> Any:
return fallback
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 (scalar if parsed else 0)
def _coerce_timestamp(value: Any, default: float) -> float:
"""Explicit message timestamp (datetime or number) or *default* when invalid."""
if value is None:
@@ -80,8 +63,7 @@ def _coerce_timestamp(value: Any, default: float) -> float:
def _parse_tool_calls(tool_calls: Any) -> Any:
"""tool_calls may be a list (live agent) or a JSON string (import/export); parse first
so json.dumps never double-encodes."""
"""tool_calls is a list (live agent) or JSON string (import/export); parse so json.dumps never double-encodes."""
if not isinstance(tool_calls, str):
return tool_calls
try:
@@ -94,6 +76,12 @@ def _tool_calls_count(tool_calls: Any) -> int:
return 0 if tool_calls is None else (len(tool_calls) if isinstance(tool_calls, list) else 1)
def _tool_calls_len(raw: Any, scalar: int = 0) -> int:
"""Count of a stored ``tool_calls`` column: list length, *scalar* for a truthy non-list, else 0."""
parsed = _parse_tool_calls(raw)
return len(parsed) if isinstance(parsed, list) else (scalar if parsed else 0)
def _scrub_surrogates(value: Any) -> Any:
"""Lone surrogates make sqlite3 raise UnicodeEncodeError and abort the whole write."""
return _sanitize_surrogates(value) if isinstance(value, str) else value
@@ -109,9 +97,9 @@ 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 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."""
"""Advance the peer's conversation generation past a boundary, in 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()
@@ -124,16 +112,15 @@ class SessionMessagesMixin:
@classmethod
def _encode_content(cls, content: Any) -> Any:
"""Serialize list/dict content (multimodal parts) as a sentinel-prefixed JSON string (sqlite3 binds
only str/bytes/int/float/None). Lone UTF-16 surrogates (web-scraped tool results reach the canonical
history unsanitized) are scrubbed here: left raw, sqlite3 raises UnicodeEncodeError, the flush is
abandoned and the session silently stops persisting. Paired with :meth:`_decode_content`."""
only scalars). Lone UTF-16 surrogates (unsanitized web-scraped tool results) are scrubbed here: left
raw, sqlite3 raises UnicodeEncodeError and the session silently stops persisting. Pairs with
:meth:`_decode_content`."""
if isinstance(content, str):
return _sanitize_surrogates(content)
if content is None or isinstance(content, (bytes, int, float)):
return content
try:
# ensure_ascii=True escapes surrogates as \\udXXX — safe to bind.
return cls._CONTENT_JSON_PREFIX + json.dumps(content)
return cls._CONTENT_JSON_PREFIX + json.dumps(content) # ensure_ascii escapes surrogates: bindable
except (TypeError, ValueError):
return _sanitize_surrogates(str(content))
@@ -148,8 +135,8 @@ class SessionMessagesMixin:
@staticmethod
def _encode_display_metadata(display_metadata: Any) -> Optional[str]:
"""Serialize ``display_metadata`` for its TEXT column without double-encoding an
already-serialized JSON string (import/replace paths hand those in)."""
"""Serialize ``display_metadata`` for its TEXT column; an already-serialized JSON string
(import/replace paths) is not double-encoded."""
if not display_metadata:
return None
if isinstance(display_metadata, str):
@@ -166,16 +153,16 @@ class SessionMessagesMixin:
@staticmethod
def _decode_display_metadata(raw: Any) -> Optional[Dict[str, Any]]:
"""Decode a ``display_metadata`` column into a dict (never the raw TEXT — the desktop does
``'task_count' in meta``). Pre-guard rows are double-encoded: a second string layer is unwrapped."""
"""Decode a ``display_metadata`` column to a dict (never raw TEXT: the desktop does ``'task_count'
in meta``). Pre-guard rows are double-encoded, so a second string layer is unwrapped."""
if raw is None:
return None
try:
meta = json.loads(raw) if isinstance(raw, str) else raw
if isinstance(meta, str):
meta = json.loads(meta)
except (json.JSONDecodeError, TypeError):
logger.warning("Ignoring invalid display metadata on message row")
meta = raw
if isinstance(meta, str):
meta = _json_or(meta, _INVALID, "Ignoring invalid display metadata on message row")
if isinstance(meta, str): # pre-guard rows carry a second string layer
meta = _json_or(meta, _INVALID, "Ignoring invalid display metadata on message row")
if meta is _INVALID:
return None
if not isinstance(meta, dict):
logger.warning("Ignoring non-object display metadata on message row")
@@ -184,9 +171,9 @@ class SessionMessagesMixin:
@staticmethod
def _reasoning_json_text(value: Any) -> Optional[str]:
"""Serialize a structured reasoning field for its TEXT column. Strings are stored as-is:
round-tripping callers (get_messages -> replace_messages) hand back the raw TEXT; re-dumping
would double-encode it and reasoning-replay consumers (``isinstance(..., list)``) would drop it."""
"""Serialize a structured reasoning field for its TEXT column. Strings are stored as-is: round-trips
(get_messages -> replace_messages) hand back raw TEXT; re-dumping would double-encode it and
reasoning-replay consumers (``isinstance(..., list)``) would drop it."""
return None if not value else (value if isinstance(value, str) else json.dumps(value))
def _check_transcript_write_guards(
@@ -195,11 +182,11 @@ 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 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."""
"""Transcript-write admission checks, run INSIDE the write txn 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()
@@ -218,9 +205,8 @@ class SessionMessagesMixin:
raise SessionTurnLeaseLostError(
f"Session turn lease lost; refusing transcript write for {session_id!r}")
if float(lease["expires_at"]) <= now:
# Expiry makes the row reclaimable, it does not prove a takeover; BEGIN
# IMMEDIATE serializes this renewal with acquisition, so a still-matching
# owner recovers from a starved refresher.
# Expiry makes the row reclaimable, not taken over; BEGIN IMMEDIATE serializes this
# renewal with acquisition, so a still-matching owner recovers from a starved refresher.
conn.execute(
"UPDATE session_turn_leases SET expires_at = ? "
"WHERE conversation_id = ? AND holder = ?",
@@ -229,8 +215,7 @@ class SessionMessagesMixin:
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.
# Same reclaim rule as acquisition; deleting 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"]))
@@ -242,9 +227,9 @@ class SessionMessagesMixin:
self, session_id: str, role: str, msg: Dict[str, Any], tool_calls: Any,
message_timestamp: float, *, keep_reasoning: bool,
) -> tuple:
"""Bind values for ``_INSERT_MESSAGE_SQL`` from one message dict. *tool_calls* is the
already-parsed value; *keep_reasoning* False stores NULL for every reasoning column.
``platform_message_id`` falls back to ``message_id`` (yuanbao's message-dict convention)."""
"""Bind values for ``_INSERT_MESSAGE_SQL`` from one message dict (*tool_calls* already parsed;
*keep_reasoning* False NULLs every reasoning column). ``platform_message_id`` falls back to
``message_id`` (yuanbao's message-dict convention)."""
_str_or_none = lambda v: _scrub_surrogates(v) if isinstance(v, str) else None # noqa: E731
_reasoning = lambda key: msg.get(key) if keep_reasoning else None # noqa: E731
return (
@@ -262,8 +247,7 @@ class SessionMessagesMixin:
@staticmethod
def _bump_session_counters(conn, session_id: str, inserted: int, tool_calls: int, *, unit: bool) -> None:
"""Increment sessions.* counters after an insert. *unit* (single append) bakes the
``+ 1`` literal into the SQL instead of binding *inserted*."""
"""Bump sessions.* counters after an insert; *unit* bakes the ``+ 1`` literal into the SQL."""
inc, params = ("1", ()) if unit else ("?", (inserted,))
if tool_calls > 0:
conn.execute(
@@ -285,8 +269,8 @@ class SessionMessagesMixin:
turn_lease_holder: Optional[str] = None, turn_lease_ttl_seconds: float = 300.0,
) -> int:
"""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."""
the platform's own id. ``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).
msg["display_metadata"] = self._encode_display_metadata(display_metadata)
@@ -301,8 +285,8 @@ class SessionMessagesMixin:
msg_id = conn.execute(_INSERT_MESSAGE_SQL, params).lastrowid
self._bump_session_counters(conn, session_id, 1, num_tool_calls, unit=True)
return msg_id
# THE critical write (its failure aborts the turn): long patience so a sibling
# legitimately holding the lock for seconds (VACUUM, checkpoint) can't kill it.
# THE critical write (failure aborts the turn): long patience so a sibling legitimately
# holding the lock for seconds (VACUUM, checkpoint) can't kill it.
return self._execute_write(_do, patience_s=self._TRANSCRIPT_WRITE_PATIENCE_S)
def append_messages_batch(
@@ -310,9 +294,8 @@ class SessionMessagesMixin:
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,
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."""
"""Append *messages* in ONE write txn (all rows land or none, guards run once); returns the inserted
count. ``chunk_rows`` bounds txn size for LARGE copies (branch seeds; FTS triggers run per row)."""
if not messages:
return 0
if chunk_rows is not None and len(messages) > chunk_rows:
@@ -330,8 +313,7 @@ class SessionMessagesMixin:
inserted_rows = resolve_and_repair_transcript_batch(
conn, session_id, messages,
encode_content_fn=self._encode_content, decode_content_fn=self._decode_content)
inserted, tool_calls_total = (
self._insert_message_rows(conn, session_id, inserted_rows) if inserted_rows else (0, 0))
inserted, tool_calls_total = self._insert_message_rows(conn, session_id, inserted_rows)
self._bump_session_counters(conn, session_id, inserted, tool_calls_total, unit=False)
return inserted
return self._execute_write(_do, patience_s=self._TRANSCRIPT_WRITE_PATIENCE_S)
@@ -340,10 +322,9 @@ class SessionMessagesMixin:
self, session_id: str, *, role: str, content: str, display_kind: str,
display_metadata: Optional[Dict[str, Any]] = None,
) -> bool:
"""Stamp presentation metadata on this turn's freshly persisted row (resolved as
newest-active-row-by-content, right after the serial turn has flushed); the model
still receives ``role``/``content`` unchanged, so producer provenance survives
without classifying by content at render time."""
"""Stamp presentation metadata on this turn's freshly persisted row (newest active row by content,
right after the serial turn flushed); the model still sees ``role``/``content`` unchanged, so
producer provenance survives without classifying by content at render time."""
if not session_id or not content or not display_kind:
return False
def _do(conn):
@@ -357,7 +338,7 @@ class SessionMessagesMixin:
"UPDATE messages SET display_kind = ?, display_metadata = ? WHERE id = ?",
(_scrub_surrogates(display_kind), self._encode_display_metadata(display_metadata), row[0]))
return True
return bool(self._execute_write(_do))
return self._execute_write(_do)
def _reaction_list(self, meta: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Well-formed (dict) reactions stored under ``REACTIONS_METADATA_KEY``."""
@@ -367,9 +348,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 (``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."""
"""Set (``emoji=None``: clear) *author*'s reaction. Tapback semantics: one per author per message;
the same emoji again clears, a different one replaces. Returns the list after the write, or
``None`` for a foreign row."""
if not session_id or message_row_id is None:
return None
def _do(conn):
@@ -399,9 +380,8 @@ class SessionMessagesMixin:
return self._reaction_list(self._decode_display_metadata(row[0])) if row is not None else []
def take_unseen_reactions(self, session_id: str, *, author: str = "user") -> List[Dict[str, Any]]:
"""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."""
"""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); ``seen`` makes it exactly once."""
if not session_id:
return []
def _do(conn):
@@ -428,14 +408,13 @@ class SessionMessagesMixin:
if changed:
conn.execute(_SET_DISPLAY_META_SQL, (self._encode_display_metadata(meta), row["id"]))
return pending
return self._execute_write(_do) or []
return self._execute_write(_do)
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 *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."""
"""Row id of the most recent active *role* message, or ``None``. ``offset`` steps back; ``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 ""
@@ -453,9 +432,8 @@ 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, tool_call_count)``; never touches sessions.* counters (callers reconcile
differently). Reasoning columns kept for assistant rows only."""
"""Insert *messages* as fresh active rows in the caller's txn -> ``(inserted, tool_call_count)``.
Never touches sessions.* counters (callers reconcile differently); reasoning kept for assistant rows."""
now_ts = time.time()
inserted = tool_calls_total = 0
for msg in messages:
@@ -475,11 +453,10 @@ 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 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."""
"""Atomically replace a session's messages (/retry, /undo, /compress). DESTRUCTIVE by default (rows
DELETEd, leave FTS). ``active_only`` spares soft-archived rows (needed with in-place compaction).
``archive_dropped`` SOFT-archives live rows rewind-style: what rewind/edit/regenerate must use, since
DELETE leaves nothing to recover. ``reject_active_turn_lease``: in-txn lease check for user rewrites."""
from hermes_state import CompressionSessionClosedError
active_clause = " AND active = 1" if active_only else ""
def _do(conn):
@@ -489,8 +466,7 @@ class SessionMessagesMixin:
elif _ended_by_compression(conn.execute(_ENDED_ROW_SQL, (session_id,)).fetchone()):
raise CompressionSessionClosedError(session_id)
if archive_dropped:
# Content-preserving UPDATE: FTS triggers don't fire on `active`, so the
# replaced turns stay searchable/readable with include_inactive=True.
# FTS triggers don't fire on `active`: replaced turns stay searchable (include_inactive=True).
conn.execute("UPDATE messages SET active = 0 WHERE session_id = ? AND active = 1", (session_id,))
else:
conn.execute(f"DELETE FROM messages WHERE session_id = ?{active_clause}", (session_id,))
@@ -505,23 +481,21 @@ class SessionMessagesMixin:
"SELECT 1 FROM messages WHERE session_id = ? AND active = 0 LIMIT 1", (session_id,)) is not None
def get_active_message_watermark(self, session_id: str) -> int:
"""MAX(id) of the session's active rows — captured at compression START; every active row above it
arrived concurrently and must survive compaction verbatim. 0 for an empty/unknown session."""
"""MAX(id) of the active rows (0 if none), captured at compression START: every active row above it
arrived concurrently and must survive compaction verbatim."""
if not session_id:
return 0
row = self._read_one("SELECT COALESCE(MAX(id), 0) FROM messages WHERE session_id = ? AND active = 1", (session_id,))
return int(row[0]) if row else 0
return int(self._read_one(
"SELECT COALESCE(MAX(id), 0) FROM messages WHERE session_id = ? AND active = 1", (session_id,))[0])
def _tail_rows_after_watermark(self, conn, sql: str, params) -> Tuple[List[int], int]:
"""``(ids, tool_call_count)`` of the concurrent-tail rows selected by *sql*
(``SELECT id, tool_calls ...``)."""
"""``(ids, tool_call_count)`` of the concurrent-tail rows selected by *sql* (``SELECT id, tool_calls``)."""
rows = conn.execute(sql, params).fetchall()
return [int(r["id"]) for r in rows], sum(_tool_calls_len(r["tool_calls"]) for r in rows)
def _clone_message_rows(self, conn, tail_ids: List[int], *, session_id: Optional[str] = None) -> None:
"""Pure-SQL column clone of *tail_ids* as fresh live rows (new id, active=1,
compacted=0, everything else byte-exact; FTS triggers index the clones). With
*session_id* the clones land in that session instead of the originals'."""
"""Pure-SQL clone of *tail_ids* as fresh live rows (new id, active=1, compacted=0, all else byte-exact;
FTS triggers index the clones), into *session_id* when given."""
retarget = session_id is not None
skip = ("id", "active", "compacted") + (("session_id",) if retarget else ())
col_list = ", ".join(c for c in self._message_column_names(conn) if c not in skip)
@@ -536,17 +510,15 @@ 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: soft-archive the active rows
(``active=0, compacted=1`` — "summarized away", still searchable) and insert *compacted_messages* as
fresh active rows, atomically. Returns the new active count (``message_count`` becomes the 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. ``model_config_patch`` merges in the same txn (``None``
removes a key)."""
"""Non-destructive in-place compaction under ONE session id: soft-archive the active rows (``active=0,
compacted=1``: summarized away, still searchable) and insert *compacted_messages* as fresh active
rows, atomically; returns the new ACTIVE count (= ``message_count``). *watermark* (compression
START): rows ``id > watermark`` arrived during the slow summary and are re-sequenced after the
compacted set by a pure-SQL clone (fresh ids); ``None`` archives everything. *lock_holder*: verified
in-txn so a reclaimed lease fails instead of clobbering the winner. *tail_count*: the LAST N compacted
rows are the verbatim carried tail; their originals and the clones' originals are superseded
duplicates and get rewind flags (``active=0, compacted=0``) so search doesn't return each carried
message once per compaction. ``model_config_patch`` merges in the same txn (``None`` removes a key)."""
from hermes_state import SessionCompressionInProgressError
def _do(conn):
if lock_holder is not None:
@@ -555,17 +527,15 @@ class SessionMessagesMixin:
raise SessionCompressionInProgressError(
f"Compression lease for {session_id!r} lost before "
"commit; refusing to publish a stale compaction")
patched_model_config = None
if model_config_patch is not None:
# on_missing="raise": never commit against a vanished session row (the
# compressor's caller turns the error into a keep-the-original no-op).
patched_model_config = self._merge_model_config_json(
conn, session_id, model_config_patch, on_missing="raise")
patch = model_config_patch is not None
# on_missing="raise": never commit against a vanished session row (caller keeps the original).
patched_model_config = self._merge_model_config_json(
conn, session_id, model_config_patch, on_missing="raise") if patch else None
tail_ids, tail_tool_calls = ([], 0) if watermark is None else self._tail_rows_after_watermark(
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 targets sit AT/BELOW the watermark (all the compressor saw); unbounded, a
# concurrent append would steal a LIMIT slot.
rewind_ids: list[int] = []
if tail_count > 0:
bound = watermark is not None
@@ -587,7 +557,6 @@ class SessionMessagesMixin:
self._clone_message_rows(conn, tail_ids)
inserted += len(tail_ids)
tool_calls_total += tail_tool_calls
patch = model_config_patch is not None
conn.execute(
f"{_SET_COUNTERS_SQL}{', model_config = ?' if patch else ''} WHERE id = ?",
(inserted, tool_calls_total, *((patched_model_config,) if patch else ()), session_id))
@@ -601,10 +570,9 @@ class SessionMessagesMixin:
return self._message_columns_cache
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. 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."""
"""Backfill the ``api_content`` sidecar onto the newest ACTIVE user row (0/1 rows). Preflight compaction
inserts that row BEFORE the sidecar exists and the later persist identity-skips compacted dicts;
without this a reload reopens the prompt-cache divergence. ``content`` match guards a racing rewrite."""
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"
@@ -612,10 +580,9 @@ class SessionMessagesMixin:
(_scrub_surrogates(api_content), session_id, self._encode_content(content)))
def _dedupe_display_generations(self, rows):
"""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``."""
"""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 every display projection shares. *rows* must be ordered by ``id``."""
seen: Dict[Tuple[Any, ...], Any] = {}
for row in rows:
dedupe_content = row["content"]
@@ -627,8 +594,8 @@ class SessionMessagesMixin:
"display_metadata": self._decode_display_metadata(row["display_metadata"])})
if handoff is not None and live_view is not None:
dedupe_content = self._encode_content(live_view.get("content"))
# Tool fields are part of the key: identical tool messages across generations
# collapse, distinct tool calls sharing role/content/timestamp never merge.
# Tool fields key too: identical tool messages collapse, distinct calls with equal
# role/content/timestamp never merge.
key = (
row["role"], dedupe_content, row["timestamp"],
row["tool_call_id"], row["tool_calls"], row["tool_name"])
@@ -638,8 +605,8 @@ class SessionMessagesMixin:
return sorted(seen.values(), key=lambda r: r["id"])
def _row_to_message_dict(self, row, *, warn_context: str, summary_flag: bool) -> Dict[str, Any]:
"""``dict(row)`` with content/tool_calls/display_metadata decoded. *summary_flag*
pops ``_compressed_summary`` and keeps it only as ``True``."""
"""``dict(row)`` with content/tool_calls/display_metadata decoded; *summary_flag* keeps
``_compressed_summary`` only as ``True``."""
msg = dict(row)
if summary_flag and msg.pop("_compressed_summary", 0):
msg["_compressed_summary"] = True
@@ -653,26 +620,23 @@ class SessionMessagesMixin:
@staticmethod
def _active_clause(include_inactive: bool, include_compacted: bool) -> str:
"""Audit reads: every row; display reads: active plus compaction-archived (never
Undo/Rewind rows); default: live only."""
"""Audit: every row; display: active plus compaction-archived (never Undo/Rewind rows); default: live."""
return "" if include_inactive else (_DISPLAY_ACTIVE_CLAUSE if include_compacted else " AND active = 1")
def get_messages(
self, session_id: str, include_inactive: bool = False, include_compacted: bool = False,
limit: Optional[int] = None, offset: int = 0, latest: bool = False, after_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
"""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."""
"""Load 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 but returns chronological order; ``after_id``: 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:
raise ValueError("after_id is incompatible with include_compacted (deduped display reads use offset paging)")
active_clause = self._active_clause(include_inactive, include_compacted)
if include_compacted:
# Read the full display set (the UI-level row cap lives in the endpoint),
# dedupe generations, then page (``[:None]`` is a no-op when limit is None).
# Full display set (the UI row cap lives in the endpoint), dedupe, then page ([:None] is a no-op).
rows = self._dedupe_display_generations(self._read_all(
"SELECT * FROM messages WHERE session_id = ?" + active_clause + " ORDER BY id ASC", [session_id]))
rows = rows[::-1][offset:][:limit][::-1] if latest else rows[offset:][:limit]
@@ -692,8 +656,7 @@ class SessionMessagesMixin:
return [self._row_to_message_dict(row, warn_context="get_messages", summary_flag=True) for row in rows]
def find_pr_url_messages(self, session_ids: List[str]) -> List[Dict[str, Any]]:
"""Tool results in these sessions containing ``/pull/`` — a deliberately loose
candidate scan, oldest-first per session so the caller can take the last match."""
"""Tool results containing ``/pull/``: a deliberately loose scan, oldest-first so the caller takes the last."""
found: List[Dict[str, Any]] = []
ids = [s for s in session_ids if s]
for start in range(0, len(ids), 900): # SQLite's bound-variable ceiling.
@@ -707,9 +670,8 @@ class SessionMessagesMixin:
return found
def get_messages_around(self, session_id: str, around_message_id: int, window: int = 5) -> Dict[str, Any]:
"""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*."""
"""Up to *window* messages either side of an anchor id (ascending). ``messages_before``/``_after`` count
strictly around the anchor (fewer than *window* = session boundary). Empty for a foreign anchor."""
window = max(window, 0)
with self._read_ctx() as conn:
anchor = (around_message_id, session_id)
@@ -727,11 +689,10 @@ 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 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."""
"""Redirect a resume target to the descendant holding 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:
@@ -756,30 +717,27 @@ class SessionMessagesMixin:
"ORDER BY child.started_at DESC, child.id DESC LIMIT 1", (current,)).fetchone()
except Exception:
return session_id
child_id = child_row["id"] if child_row is not None else None
if not child_id or child_id in seen:
if child_row is None or not child_row["id"] or child_row["id"] in seen:
break
seen.add(child_id)
current = child_id
current = child_row["id"]
seen.add(current)
return best if best is not None else session_id
def _fetch_conversation_rows(self, session_ids: List[str], active_clause: str, *, with_session_id: bool):
"""``_CONVERSATION_ROW_COLUMNS`` rows for *session_ids*, ORDER BY id (insertion order —
timestamps are not monotonic and would break tool-call adjacency)."""
with self._read_ctx() as conn:
return conn.execute(
f"SELECT {'session_id, ' if with_session_id else ''}{self._CONVERSATION_ROW_COLUMNS} "
f"FROM messages WHERE session_id IN ({_placeholders(session_ids)})"
f"{active_clause} ORDER BY id", tuple(session_ids)).fetchall()
"""``_CONVERSATION_ROW_COLUMNS`` rows for *session_ids* ORDER BY id (timestamps are not monotonic
and would break tool-call adjacency)."""
return self._read_all(
f"SELECT {'session_id, ' if with_session_id else ''}{self._CONVERSATION_ROW_COLUMNS} "
f"FROM messages WHERE session_id IN ({_placeholders(session_ids)})"
f"{active_clause} ORDER BY id", tuple(session_ids))
def get_messages_as_conversation(
self, session_id: str, include_ancestors: bool = False, include_inactive: bool = False,
repair_alternation: bool = False, include_row_ids: bool = False, include_compacted: bool = False,
) -> List[Dict[str, Any]]:
"""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
"""Load messages in OpenAI 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 re-trigger the
per-request repair forever); the stored transcript is never mutated."""
rows = self._fetch_conversation_rows(
self._resume_lineage_ids(session_id) if include_ancestors else [session_id],
@@ -791,10 +749,10 @@ class SessionMessagesMixin:
repair_alternation=repair_alternation, include_row_ids=include_row_ids)
def _dedupe_replayed_user(self, messages, msg, exact_user_clones) -> Tuple[bool, Any]:
"""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."""
"""Ancestor-lineage dedupe of 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 = self._canonical_replayed_user_content(msg)[0]
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
@@ -816,12 +774,10 @@ class SessionMessagesMixin:
self, rows, *, session_id: str, include_ancestors: bool, repair_alternation: bool,
include_row_ids: bool = False, include_summary_markers: bool = False,
) -> List[Dict[str, Any]]:
"""Decode fetched message rows (ordered by id, pre-filtered) into OpenAI format. 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.
``api_content`` is returned VERBATIM (no sanitize/strip): the replay path substitutes it to keep
the provider prompt cache byte-stable."""
"""Decode fetched rows (ordered by id, pre-filtered) into OpenAI format, stable key order. Every dict is
stamped ``_DB_PERSISTED_MARKER_KEY`` (born durable) so an identity-losing handoff never re-appends the
transcript on flush. ``_row_id`` is opt-in (gateway reactions); reasoning restored on assistant rows
only; ``api_content`` VERBATIM (no sanitize/strip) so replay keeps the provider prompt cache byte-stable."""
from hermes_state import _strip_background_review_harness, _strip_stale_tool_call_markers
messages = []
exact_user_clones: Dict[Tuple[Any, str], Dict[str, Any]] = {}
@@ -829,9 +785,8 @@ class SessionMessagesMixin:
content = self._decode_content(row["content"])
if row["role"] in {"user", "assistant"} and isinstance(content, str):
content = sanitize_context(content).strip()
# The persisted marker is underscore-prefixed like ``_row_id``: every transport
# strips it before the wire, and compression's assembly copies deliberately
# strip it so rotated child handoffs still flush (see _fresh_compaction_message_copy).
# Underscore-prefixed like ``_row_id``: transports strip it before the wire; compression's
# assembly copies strip it so rotated child handoffs still flush (_fresh_compaction_message_copy).
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"]
@@ -865,9 +820,8 @@ class SessionMessagesMixin:
messages.append(msg)
if include_ancestors and exact_clone_key is not None:
exact_user_clones[exact_clone_key] = msg
# Defense-in-depth: strip a background-review harness turn (older builds shared
# the parent's session_id) plus its curator reply, and bare tool-call marker
# content ("[memory]") persisted as an answer before the loop fix.
# Defense-in-depth: strip a background-review harness turn (older builds shared the parent's
# session_id) plus its curator reply, and bare tool-call marker content ("[memory]") persisted as an answer.
messages = _strip_stale_tool_call_markers(_strip_background_review_harness(messages))
if repair_alternation and messages:
from agent.agent_runtime_helpers import repair_message_sequence
@@ -880,10 +834,9 @@ 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 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."""
"""``(model_history, display_history)`` for a resume from ONE SELECT; byte-identical to the separate
reads. model: the tip's active rows, alternation-repaired, summary marker kept for pre-compress
checkpointing. display: the full lineage (``/branch`` stands alone), compaction-archived rows deduped."""
rows = self._fetch_conversation_rows(
self._resume_lineage_ids(session_id), _DISPLAY_ACTIVE_CLAUSE, with_session_id=True)
# The model projection stays active-only: it is the compressed working context.
@@ -897,14 +850,12 @@ class SessionMessagesMixin:
return model_history, display_history
def _resume_lineage_ids(self, session_id: str) -> List[str]:
"""Session ids a full (display) resume materializes: the compression lineage, or the
session alone for an explicit ``/branch`` copy. Shared by the resume readers and the
resume guard so the guard counts exactly what a resume loads."""
"""Session ids a display resume materializes: the compression lineage, or the session alone for an
explicit ``/branch`` copy. Shared with the resume guard so it counts exactly what a resume loads."""
return [session_id] if self._is_explicit_branch_session(session_id) else self._session_lineage_root_to_tip(session_id)
def _resume_count_scope(self, session_id: str, tip_only: bool) -> Tuple[List[str], str]:
"""``tip_only``: the tip's ACTIVE rows (model restore); else the full-lineage DISPLAY
set (active + compaction-archived) that get_resume_conversations loads."""
"""``tip_only``: the tip's ACTIVE rows (model restore); else the full-lineage DISPLAY set."""
if tip_only:
return [session_id], "active = 1"
return self._resume_lineage_ids(session_id), "(active = 1 OR compacted = 1)"
@@ -912,16 +863,14 @@ class SessionMessagesMixin:
def get_resume_message_count(self, session_id: str, *, tip_only: bool = False) -> int:
"""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(
return int(self._read_one(
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)
tuple(session_ids))[0])
def assert_resume_safe(self, session_id: str, max_messages: Optional[int] = None, *, tip_only: bool = False) -> int:
"""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."""
"""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 a success, not a rejection."""
from hermes_state import SessionResumeTooLargeError, resolved_max_resume_messages
if max_messages is None:
max_messages = resolved_max_resume_messages()
@@ -930,21 +879,20 @@ class SessionMessagesMixin:
if max_messages == 0:
return 0
session_ids, active_clause = self._resume_count_scope(session_id, tip_only)
row = self._read_one(
message_count = int(self._read_one(
"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))
message_count = int(row[0] if row else 0)
")", (*session_ids, max_messages + 1))[0])
if message_count > max_messages:
raise SessionResumeTooLargeError(
message_count, max_messages, scope="in its tip segment" if tip_only else "across its lineage")
return message_count
def get_ancestor_display_prefix(self, session_id: str) -> List[Dict[str, Any]]:
"""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."""
"""Ancestor-only display messages of a lineage (row ``session_id != tip``) that ``session.resume``
prepends. 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 []
@@ -960,8 +908,8 @@ class SessionMessagesMixin:
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 compression segments and
delegate subagents (Nous Portal usage tagging). Unchanged when there is no recorded parent."""
"""ROOT id of the lineage: the stable conversation id across compression segments and delegate
subagents (Nous Portal usage tagging). Unchanged when there is no recorded parent."""
chain = self._session_lineage_root_to_tip(session_id)
return chain[0] if chain and chain[0] else session_id
@@ -988,10 +936,9 @@ class SessionMessagesMixin:
@staticmethod
def _find_duplicate_replayed_user_message(
messages: List[Dict[str, Any]], msg: Dict[str, Any]) -> Optional[Tuple[int, bool]]:
"""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)."""
"""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 scaffold)."""
if msg.get("role") != "user":
return None
canonical = SessionMessagesMixin._canonical_replayed_user_content
@@ -1009,8 +956,7 @@ class SessionMessagesMixin:
return None
def get_active_message_ids(self, session_id: str) -> List[int]:
"""Ordered physical active ids pinned by rewind CAS checks (includes legacy harness
rows that conversation projections omit)."""
"""Ordered physical active ids for rewind CAS checks (includes legacy harness rows projections omit)."""
return [int(row[0]) for row in self._read_all(_ACTIVE_IDS_SQL, (session_id,))]
@staticmethod
@@ -1020,16 +966,14 @@ 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; 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*."""
"""Validate an active rewind target; return its handoff scaffold (or None). ``ValueError``: inactive /
non-user-originated / missing composite carrier; ``RuntimeError``: live payload changed."""
if not target_row.get("active"):
raise ValueError("rewind target is not active")
from agent.context_compressor import split_user_originated_turn
split_target = target_row.copy()
split_target["content"] = self._decode_content(split_target.get("content"))
split_target["display_metadata"] = self._decode_display_metadata(split_target.get("display_metadata"))
handoff, live_view = split_user_originated_turn(split_target)
handoff, live_view = split_user_originated_turn({
**target_row, "content": self._decode_content(target_row.get("content")),
"display_metadata": self._decode_display_metadata(target_row.get("display_metadata"))})
if live_view is None:
raise ValueError("rewind target is not a user-originated turn")
live_content = live_view.get("content")
@@ -1045,14 +989,13 @@ 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 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."""
"""Soft-delete (``active=0``) every message with id >= *target_message_id*, 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``: target missing or not ``user``.
``expected_active_ids`` / ``expected_target_content`` pin the active set and canonical live payload
in-txn before any mutation (presentation-only metadata changes don't invalidate a rewind). A live turn
lease refuses; expired/dead holders are reclaimed. ``rewind_count`` always increments."""
def _do(conn):
self._check_transcript_write_guards(
conn, session_id, None, reject_active_turn_lease=True, reject_active_compression_lock=True)
@@ -1084,8 +1027,9 @@ class SessionMessagesMixin:
"UPDATE sessions SET rewind_count = COALESCE(rewind_count, 0) + 1 WHERE id = ?", (session_id,))
message_count, tool_call_count = self._active_transcript_counts(conn, session_id)
conn.execute(f"{_SET_COUNTERS_SQL} WHERE id = ?", (message_count, tool_call_count, session_id))
head_row = conn.execute("SELECT MAX(id) FROM messages WHERE session_id = ? AND active = 1", (session_id,)).fetchone()
return target_row, ids, head_row[0] if head_row else None, replacement_message_id
head_id = conn.execute(
"SELECT MAX(id) FROM messages WHERE session_id = ? AND active = 1", (session_id,)).fetchone()[0]
return target_row, ids, head_id, 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.
target_row["content"] = self._decode_content(target_row.get("content"))
@@ -1100,17 +1044,15 @@ class SessionMessagesMixin:
return self._read_one(sql, (session_id,) if session_id else ())[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 probe; the
gateway's transient-failure dedupe guard)."""
"""True when *platform_message_id* exists (partial-index probe; the gateway's transient-failure dedupe)."""
return self._read_one(
"SELECT 1 FROM messages WHERE session_id = ? AND platform_message_id = ? LIMIT 1",
(session_id, platform_message_id)) 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 presence-only matching would misclassify a delegate's
continuation (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 it (same binding as ``_NON_CONTINUATION_CHILD_FILTER_SQL``)."""
if session.get("source") == "tool":
return True
cfg = session.get("model_config")
@@ -1126,22 +1068,20 @@ class SessionMessagesMixin:
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 is not a fork
(``agent/prompt_cache_scope.py`` keeps a declared conversation key from crossing the fork boundary)."""
"""Read-only :meth:`_is_explicit_fork_child_row`; a missing row is not a fork (prompt_cache_scope keeps
a declared conversation key from crossing the fork boundary)."""
session = self.get_session(session_id)
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). Wall-clock-free, so a backwards NTP correction cannot reorder it. DBs upgraded
mid-conversation start at no generation and take their first from the next boundary written (a
pre-upgrade reset shares its predecessor's scope once — costs a warm prompt-cache bucket, never
crosses an identity)."""
"""Conversation boundaries (``_RESET_END_REASONS`` ends) this peer has crossed, or ``None`` if never
reset. The peer is ``(session_key, source)``, never the key alone (an API caller may legally reuse a
Telegram row's key). Read from ``conversation_generations`` (advanced in each boundary's txn), not an
aggregate over session rows: deletes/prunes would re-emit a retired pair. Rows are never GC'd
(dropping one re-issues generation 1: the ABA this prevents). Wall-clock-free, so a backwards NTP
correction cannot reorder it. DBs upgraded mid-conversation take their first generation from the
next boundary written (a pre-upgrade reset shares its predecessor's scope once: costs a warm
prompt-cache bucket, never crosses an identity)."""
if not session_key or not source:
return None
row = self._read_one(
@@ -1158,10 +1098,9 @@ 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`` 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"}``."""
"""Permanently clear bare tool-call marker content ("[memory]") left by pre-fix sessions
(``_rows_to_conversation`` repairs it in memory; this stops the re-scan). Only ``content`` is touched.
``backup``: ``VACUUM INTO`` snapshot first (none when nothing changes)."""
from hermes_state import _STALE_TOOL_CALL_MARKER_RE
def _find_affected(conn) -> List[int]:
cursor = conn.execute(