From 0fdadbe2554caba8012551bc3fe303b7b6f35cd4 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:17:38 -0700 Subject: [PATCH] refactor(agent/session_persistence): split 292-LOC flush into phase helpers; unify override/durable-content logic --- agent/session_persistence.py | 752 ++++++++++++++++------------------- 1 file changed, 339 insertions(+), 413 deletions(-) diff --git a/agent/session_persistence.py b/agent/session_persistence.py index 30eb76635a..8526d35e62 100644 --- a/agent/session_persistence.py +++ b/agent/session_persistence.py @@ -1,8 +1,7 @@ -"""Durable transcript persistence for ``AIAgent``. +"""Durable transcript persistence for ``AIAgent`` (mixin; MRO-resolved from ``run_agent``). -SQLite session flush (intrinsic ``_DB_PERSISTED_MARKER`` dedup), ephemeral-scaffolding filtering, the JSON -session log and trajectory export. -Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged. +SQLite session flush with intrinsic ``_DB_PERSISTED_MARKER`` dedup, ephemeral-scaffolding +filtering, the optional JSON session log and trajectory export. """ import hashlib import json @@ -28,108 +27,141 @@ from utils import atomic_json_write logger = logging.getLogger("run_agent") -# Flags marking ephemeral empty-response/prefill recovery scaffolding. The loop pops these before -# appending the real response; persistence must skip them or a resumed session replays synthetic turns. +# Flags marking ephemeral recovery scaffolding the loop pops before appending the real response. +# Persistence must skip them or a resumed session replays synthetic turns / breaks prefix-cache reuse. _EPHEMERAL_SCAFFOLDING_FLAGS = ( "_empty_recovery_synthetic", "_empty_terminal_sentinel", "_thinking_prefill", - # verify-on-stop / pre_verify nudges: persisting them poisons the resumed transcript and breaks - # prompt-prefix cache reuse. The assistant candidate is NOT synthetic (#65919). - "_verification_stop_synthetic", + "_verification_stop_synthetic", # verify-on-stop nudge; the assistant candidate itself is NOT synthetic "_pre_verify_synthetic", - # kanban worker stop-guard: narrated exit without kanban_complete/block - "_kanban_stop_synthetic", - # dropped tool-call re-prompt pair: internal retry instruction, must not replay as user context on resume. - "_dropped_toolcall_nudge", + "_kanban_stop_synthetic", # kanban worker stop-guard + "_dropped_toolcall_nudge", # internal retry instruction; must not replay as user context ) +_IMAGE_PART_TYPES = {"image", "image_url", "input_image"} + def _is_ephemeral_scaffolding(msg: Any) -> bool: - """Return True when ``msg`` is internal recovery scaffolding that must never be persisted to the - durable transcript (SQLite session store or JSON log).""" - return isinstance(msg, dict) and any( - msg.get(flag) for flag in _EPHEMERAL_SCAFFOLDING_FLAGS - ) + """True when ``msg`` is internal recovery scaffolding that must never reach the durable transcript.""" + return isinstance(msg, dict) and any(msg.get(flag) for flag in _EPHEMERAL_SCAFFOLDING_FLAGS) -# `_DB_PERSISTED_MARKER` (agent.context_compressor) — intrinsic "already written to SQLite" marker. An id(msg) -# dedup set can alias a freed dict's address onto a new message and silently skip persisting it; a marker on -# the dict cannot. The `_` prefix is mandatory: wire sanitizers strip `_`-prefixed keys. CONTRACT (#92231): -# the marker asserts the dict's CONTENT is durable as written — any in-place mutation that must persist MUST -# pop it (see turn_finalizer, context_compressor). +# `_DB_PERSISTED_MARKER` (agent.context_compressor) is the intrinsic "already written to SQLite" marker: +# an id(msg) set can alias a freed dict's address onto a new message, a key on the dict cannot. The `_` +# prefix is mandatory (wire sanitizers strip `_` keys). CONTRACT: the marker asserts the dict's CONTENT +# is durable as written — any in-place mutation that must persist MUST pop it (turn_finalizer, +# context_compressor). def _safe_session_filename_component(session_id: str) -> str: - """Return a stable, path-safe filename component for a session ID. + """Path-safe filename component for a (possibly untrusted ``X-Hermes-Session-Id``) session ID. - Session IDs may be untrusted (``X-Hermes-Session-Id``) and are interpolated into ``~/.hermes/sessions/`` - filenames. Collapses non ``[A-Za-z0-9_-]`` chars to ``_``, caps length, and appends a short content - hash when sanitization changed the string so distinct IDs cannot collide. + Collapses non ``[A-Za-z0-9_-]`` chars to ``_``, caps length, and appends a short content hash + whenever sanitization changed the string so distinct IDs cannot collide. """ raw = str(session_id or "").strip() sanitized = re.sub(r"[^\w-]", "_", raw).strip("._") sanitized = sanitized[:96] or "session" if raw and sanitized == raw: return sanitized - digest = hashlib.sha256( - raw.encode("utf-8", errors="surrogatepass") - ).hexdigest()[:12] + digest = hashlib.sha256(raw.encode("utf-8", errors="surrogatepass")).hexdigest()[:12] return f"{sanitized}_{digest}" +def _override_replaces_content(msg: Dict, content: Any, override: Any) -> bool: + """Whether the persist user-message override may replace ``content``. + + A plain-text override must not replace native image/audio blocks (a list override is the clean + multimodal payload and does). Preflight compaction may re-anchor the index at a message MERGED + with the compaction summary — overwriting it would drop the summary from the durable transcript. + """ + return ( + override is not None + and not msg.get(COMPRESSED_SUMMARY_METADATA_KEY) + and (not isinstance(content, list) or isinstance(override, list)) + ) + + +def _summary_display_kind(msg: Dict) -> Any: + """Standalone reference handoffs are always hidden so they never occupy the active user slot in + retry/undo dispatch; merge-into-tail carriers keep their prior visibility.""" + if ( + msg.get(COMPRESSED_SUMMARY_METADATA_KEY) + and user_originated_turn_view(msg) is None + and ( + ContextCompressor.classify_summary_content(msg.get("content")) == "standalone" + or not msg.get("_compressed_summary_has_user_turn") + ) + ): + return "hidden" + return msg.get("display_kind") + + +def _durable_content(content: Any) -> Any: + """Text-only projection for the DB: multimodal envelopes become their summary, OpenAI-style part + lists keep text and replace images with ``[screenshot]`` (base64 bloats the DB).""" + if _is_multimodal_tool_result(content): + return _multimodal_text_summary(content) + if isinstance(content, list): + txt = [] + for p in content: + if not isinstance(p, dict): + continue + if p.get("type") == "text": + txt.append(str(p.get("text", ""))) + elif p.get("type") in _IMAGE_PART_TYPES: + txt.append("[screenshot]") + return "\n".join(txt) if txt else None + return content + + +def _tool_calls_data(msg: Dict) -> Any: + if hasattr(msg, "tool_calls") and isinstance(msg.tool_calls, list) and msg.tool_calls: + return [{"name": tc.function.name, "arguments": tc.function.arguments} for tc in msg.tool_calls] + if isinstance(msg.get("tool_calls"), list): + return msg["tool_calls"] + return None + + class SessionPersistenceMixin: """Session DB flush, session log and trajectory persistence (see module docstring).""" def _apply_persist_user_message_override(self, messages: List[Dict]) -> None: - """Rewrite the current-turn user message before persistence/return. + """Rewrite the current-turn user message in place before persistence/return. - Some paths use an API-only user-message variant that must not leak into transcripts or resumed - history; mutate the in-memory list in place so both persistence and returned history stay clean. + Some paths send an API-only user-message variant that must not leak into transcripts or + resumed history; mutating the live list keeps persistence and returned history clean. """ idx = getattr(self, "_persist_user_message_idx", None) override = getattr(self, "_persist_user_message_override", None) timestamp = getattr(self, "_persist_user_message_timestamp", None) platform_id = getattr(self, "_persist_user_message_platform_id", None) - if idx is None or ( - override is None and timestamp is None and platform_id is None - ): + if idx is None or (override is None and timestamp is None and platform_id is None): return - if 0 <= idx < len(messages): - msg = messages[idx] - if isinstance(msg, dict) and msg.get("role") == "user": - # A plain-text override must not replace native image/audio blocks; a list override is the - # clean multimodal payload and does. Preflight compaction may re-anchor this index at a - # message MERGED with the compaction summary — overwriting it would drop the summary (see the - # twin guard in _flush_messages_to_session_db_unlocked). - if ( - override is not None - and not msg.get(COMPRESSED_SUMMARY_METADATA_KEY) - and ( - not isinstance(msg.get("content"), list) - or isinstance(override, list) - ) - ): - msg["content"] = override - if timestamp is not None: - msg["timestamp"] = timestamp - # Platform message id: load-bearing for restart drain-window recovery dedup - # (has_platform_message_id). Stamped here too so it survives the override path. - if platform_id is not None: - msg["platform_message_id"] = platform_id + if not 0 <= idx < len(messages): + return + msg = messages[idx] + if not (isinstance(msg, dict) and msg.get("role") == "user"): + return + if _override_replaces_content(msg, msg.get("content"), override): + msg["content"] = override + if timestamp is not None: + msg["timestamp"] = timestamp + # Load-bearing for restart drain-window recovery dedup (has_platform_message_id). + if platform_id is not None: + msg["platform_message_id"] = platform_id def _persist_session(self, messages: List[Dict], conversation_history: List[Dict] = None): """Save session state to both JSON log and SQLite on any exit path. - Trailing empty-response scaffolding is dropped from the live list. The persist user-message override - is NOT applied here — ``_flush_messages_to_session_db`` writes it to the DB row only. + Trailing empty-response scaffolding is dropped from the live list. The persist user-message + override is NOT applied here — ``_flush_messages_to_session_db`` writes it to the DB row only. """ - # Scaffolding removal mutates the live list on purpose. Close and turn-start persistence can run on - # separate CLI threads, so the marker test-and-append below must be one critical section. + # Close and turn-start persistence can run on separate CLI threads, so scaffolding removal + # plus the marker test-and-append must be one critical section. from agent.agent_runtime_helpers import note_turn_persisted - persist_lock = getattr(self, "_session_persist_lock", None) - def _persist_and_drain() -> None: self._drop_trailing_empty_response_scaffolding(messages) self._session_messages = messages @@ -140,46 +172,32 @@ class SessionPersistenceMixin: self._session_db.flush_token_counts() note_turn_persisted(self) + persist_lock = getattr(self, "_session_persist_lock", None) if persist_lock is None: _persist_and_drain() return - with persist_lock: _persist_and_drain() def _drop_trailing_empty_response_scaffolding(self, messages: List[Dict]) -> None: """Remove private empty-response retry/failure scaffolding from transcript tails. - Also rewinds a trailing tool-result / assistant(tool_calls) pair the failed iteration left hanging; - otherwise the next user turn lands as ``...tool, user`` and providers return empty content forever. + Also rewinds a trailing tool-result / assistant(tool_calls) pair the failed iteration left + hanging; otherwise the next user turn lands as ``...tool, user`` and providers return empty + content forever. The rewind only runs when scaffolding was actually present. """ - # Pass 1: strip the flagged scaffolding messages themselves. + def _tail(*keys: str) -> bool: + return bool(messages) and isinstance(messages[-1], dict) and any(messages[-1].get(k) for k in keys) + dropped_scaffolding = False - while ( - messages - and isinstance(messages[-1], dict) - and ( - messages[-1].get("_empty_recovery_synthetic") - or messages[-1].get("_empty_terminal_sentinel") - ) - ): + while _tail("_empty_recovery_synthetic", "_empty_terminal_sentinel"): messages.pop() dropped_scaffolding = True - - # Pass 2: after stripping scaffolding, rewind trailing tool results and the assistant(tool_calls) - # that produced them, so role alternation holds. Only runs when scaffolding was present. if not dropped_scaffolding: return - - # Drop any trailing tool-result messages - while ( - messages - and isinstance(messages[-1], dict) - and messages[-1].get("role") == "tool" - ): + while messages and isinstance(messages[-1], dict) and messages[-1].get("role") == "tool": messages.pop() - - # Drop the assistant(tool_calls) whose results were just popped — providers reject a dangling one. + # Providers reject a dangling assistant(tool_calls) whose results were just popped. if ( messages and isinstance(messages[-1], dict) @@ -202,6 +220,199 @@ class SessionPersistenceMixin: with persist_lock: return self._flush_messages_to_session_db_unlocked(messages, conversation_history) + # --- flush phases (called only by _flush_messages_to_session_db_unlocked) --- + + def _db_flush_seed_ids(self) -> set: + """One-shot ``_flushed_db_message_ids`` seed, honoured only for the same session after a + non-empty flush; translated to markers by the scan and cleared afterwards.""" + current_session_id = getattr(self, "session_id", None) + flushed_session_id = getattr(self, "_flushed_db_message_session_id", None) + if flushed_session_id != current_session_id or self._last_flushed_db_idx == 0: + seed_ids = set() + else: + seed_ids = getattr(self, "_flushed_db_message_ids", None) + if not isinstance(seed_ids, set): + seed_ids = set() + self._flushed_db_message_session_id = current_session_id + return seed_ids + + def _db_flush_scan_start(self, messages: List[Dict]) -> int: + """Bounded scan: skip the identity-matched, still-marked prefix of the previous flush's + snapshot — every message in it already got its final disposition.""" + scan_start = 0 + prev_prefix = getattr(self, "_db_flush_scan_prefix", None) + if isinstance(prev_prefix, list): + limit = min(len(prev_prefix), len(messages)) + while ( + scan_start < limit + and messages[scan_start] is prev_prefix[scan_start] + and bool(messages[scan_start].get(_DB_PERSISTED_MARKER)) + ): + scan_start += 1 + return scan_start + + def _db_flush_row(self, msg: Dict, is_current_turn_user: bool) -> Dict[str, Any]: + """Build the session-db row for ``msg``, applying the persist override to THIS row only.""" + role = msg.get("role", "unknown") + content = msg.get("content") + # api_content sidecar: exact bytes sent to the API when they differ from clean content, so + # replay reproduces the sent prefix byte-for-byte. + api_content = msg.get("api_content") + if not isinstance(api_content, str): + api_content = None + timestamp = msg.get("timestamp") + if is_current_turn_user and msg.get("role") == "user": + override = getattr(self, "_persist_user_message_override", None) + if _override_replaces_content(msg, content, override): + # Live content is what the wire sent, the override is the clean transcript; keep the + # sent bytes in api_content so replay matches the wire. + if api_content is None and isinstance(content, str) and content != override: + api_content = content + content = override + ov_timestamp = getattr(self, "_persist_user_message_timestamp", None) + if ov_timestamp is not None: + timestamp = ov_timestamp + if api_content == content: + api_content = None + # get_messages_as_conversation replays rows through sanitize_context().strip(); capture the + # sent bytes when they would differ (compared in wire form). + if ( + api_content is None + and role in ("user", "assistant") + and isinstance(content, str) + and content + and sanitize_context(content).strip() != content.strip() + ): + api_content = content + row = { + "role": role, + "content": _durable_content(content), + "tool_name": msg.get("tool_name"), + "tool_calls": _tool_calls_data(msg), + "tool_call_id": msg.get("tool_call_id"), + "finish_reason": msg.get("finish_reason"), + # Reasoning/codex fields are role-gated (assistant-only) inside _insert_message_rows. + "reasoning": msg.get("reasoning"), + "reasoning_content": msg.get("reasoning_content"), + "reasoning_details": msg.get("reasoning_details"), + "codex_reasoning_items": msg.get("codex_reasoning_items"), + "codex_message_items": msg.get("codex_message_items"), + "_compressed_summary": bool(msg.get(COMPRESSED_SUMMARY_METADATA_KEY)), + "timestamp": timestamp, + "api_content": api_content, + "display_kind": _summary_display_kind(msg), + "display_metadata": msg.get("display_metadata"), + # Load-bearing for restart drain-window recovery dedup. + "platform_message_id": msg.get("platform_message_id"), + } + if isinstance(msg.get("_row_id"), int): + row["_row_id"] = msg["_row_id"] + return row + + def _db_flush_collect(self, messages: List[Dict], conversation_history: Optional[List[Dict]]): + """Scan for un-flushed messages; returns ``(rows, msgs)`` to write in one transaction.""" + seed_ids = self._db_flush_seed_ids() + history_ids = {id(item) for item in (conversation_history or []) if isinstance(item, dict)} + ov_idx = getattr(self, "_persist_user_message_idx", None) + # Also match the staged CLI dict by identity — the close safety-net may flush a shortened + # snapshot whose turn index refers to the full history. + pending_cli_message = getattr(self, "_pending_cli_user_message", None) + batch_rows: List[Dict[str, Any]] = [] + batch_msgs: List[Dict] = [] + for msg_idx in range(self._db_flush_scan_start(messages), len(messages)): + msg = messages[msg_idx] + if not isinstance(msg, dict): + continue + # The flush is append-only: a mid-turn persist of scaffolding could commit a synthetic + # turn the end-of-turn drop cannot un-write. Skip regardless of position. + if _is_ephemeral_scaffolding(msg) or msg.get(_DB_PERSISTED_MARKER): + continue + # Already durable (history copy or caller-seeded): stamp so future flushes skip it. + if id(msg) in history_ids or id(msg) in seed_ids: + msg[_DB_PERSISTED_MARKER] = True + continue + batch_rows.append(self._db_flush_row(msg, ov_idx == msg_idx or msg is pending_cli_message)) + batch_msgs.append(msg) + return batch_rows, batch_msgs + + def _db_flush_write(self, batch_rows: List[Dict[str, Any]], batch_msgs: List[Dict]) -> None: + """One transaction for the turn's new rows; on failure no rows land and no markers are + stamped, so the next flush re-writes the tail.""" + if not batch_rows: + return + self._session_db.append_messages_batch( + session_id=self.session_id, + messages=batch_rows, + compression_lock_holder=getattr(self, "_active_compression_lock_holder", None), + turn_lease_holder=getattr(self, "_active_session_turn_lease_holder", None), + turn_lease_ttl_seconds=getattr(self, "_active_session_turn_lease_ttl_seconds", 300.0) or 300.0, + ) + from agent.transcript_repair import sync_flushed_message_markers + + sync_flushed_message_markers(batch_msgs, batch_rows) + + def _db_flush_adopt_compression_tip(self) -> bool: + """Adopt the live continuation of a session closed by compression, if there is one. + + ``get_compression_tip`` returning the same id means no continuation exists; a tip whose row + is missing or already ended is not adopted either. + """ + old_id = self.session_id + tip = None + try: + tip = self._session_db.get_compression_tip(old_id) + except Exception as tip_exc: + logger.warning("compression tip lookup failed for %s: %s", old_id, tip_exc) + if not tip or tip == old_id: + return False + try: + tip_row = self._session_db.get_session(tip) + except Exception: + tip_row = None + if tip_row is None or tip_row.get("ended_at") is not None: + return False + logger.warning("Adopted live compression tip %s for closed session %s; retrying flush once", tip, old_id) + self.session_id = tip + self._flushed_db_message_ids = set() + self._last_flushed_db_idx = 0 + self._compression_adoption_failed = False + return True + + def _db_flush_failed(self, e: Exception, batch_rows: List[Dict[str, Any]], adoption_budget: int) -> bool: + """Classify a failed flush; True when the caller should retry once on an adopted compression tip.""" + # Force a full re-scan next flush: an exception mid-loop leaves mixed dispositions. + self._db_flush_scan_prefix = None + # The only place the SQLite error is visible before it becomes a bare False — classify it so + # the turn-end explanation can distinguish lock contention from disk-full/read-only. + from hermes_state import ( + CompressionSessionClosedError, + StateDbCorruptError, + StateDbReplacedError, + classify_persistence_error, + divert_session_transcript_jsonl, + ) + + self._last_persistence_error_cause = classify_persistence_error(e) + if isinstance(e, (StateDbReplacedError, StateDbCorruptError)): + # A replaced/quarantined handle will not take this batch again — keep it on disk. + try: + divert_session_transcript_jsonl(getattr(self, "session_id", "") or "", batch_rows) + except Exception: + logger.warning( + "JSONL divert failed after state.db %s for %s", + self._last_persistence_error_cause, getattr(self, "session_id", None), exc_info=True, + ) + if isinstance(e, CompressionSessionClosedError): + # Compression race: another path rotated this session mid-write. Retry exactly once on the + # live tip; a second closed-parent write fails closed. + if adoption_budget > 0 and self._db_flush_adopt_compression_tip(): + return True + # The flag lets the turn explanation name compression rotation instead of misleading + # full-disk advice. + self._compression_adoption_failed = True + logger.warning("Session DB append_message failed: %s", e) + return False + def _flush_messages_to_session_db_unlocked( self, messages: List[Dict], @@ -210,312 +421,45 @@ class SessionPersistenceMixin: ): """Persist any un-flushed messages to the SQLite session store. - Dedup is an intrinsic ``_DB_PERSISTED_MARKER`` on each written dict — not positional slices (drift - after sequence repair) nor a retained ``id(msg)`` set (address reuse). ``_flushed_db_message_ids`` is - only a one-shot seed translated to markers and cleared each flush. + Dedup is an intrinsic ``_DB_PERSISTED_MARKER`` on each written dict — not positional slices + (drift after sequence repair) nor a retained ``id(msg)`` set (address reuse). The persist + user-message override is applied ONLY to the written row, never to the live dict. A + compression-closed session adopts its live tip and retries exactly once. """ - # Persistence-isolated agents (background review fork) share the parent's session_id for cache - # warmth; a write here would land the curator's harness turn in the user's real history. Hard-stop. + # Persistence-isolated agents (background review fork) share the parent's session_id for + # cache warmth; a write here would land the curator's turn in the user's real history. if getattr(self, "_persist_disabled", False): return None if not self._session_db: return None - # Persist user-message override (#48677): resolved here and applied ONLY to the written row, never - # to the live dict — the early crash-resilience persist runs before the API call is built. - _ov_idx = getattr(self, "_persist_user_message_idx", None) - _ov_content = getattr(self, "_persist_user_message_override", None) - _ov_timestamp = getattr(self, "_persist_user_message_timestamp", None) + batch_rows: List[Dict[str, Any]] = [] try: # Retry row creation if the earlier attempt failed transiently. if not self._session_db_created: self._ensure_db_session() - # Positional slicing broke when repair_message_sequence shrank the list (#46053). Persistence is - # tracked by an intrinsic per-message marker (see _DB_PERSISTED_MARKER); `_flushed_db_message_ids` - # is honoured only as a one-shot seed translated to markers and then cleared. - current_session_id = getattr(self, "session_id", None) - flushed_session_id = getattr(self, "_flushed_db_message_session_id", None) - if flushed_session_id != current_session_id or self._last_flushed_db_idx == 0: - seed_ids = set() - else: - seed_ids = getattr(self, "_flushed_db_message_ids", None) - if not isinstance(seed_ids, set): - seed_ids = set() - self._flushed_db_message_session_id = current_session_id - history_ids = { - id(item) for item in (conversation_history or []) - if isinstance(item, dict) - } - - # Bounded scan: skip the identity-matched prefix of the previous flush's snapshot. Every message - # in it already got its final disposition, and no live dict has its marker popped in place. - _scan_start = 0 - _prev_prefix = getattr(self, "_db_flush_scan_prefix", None) - if isinstance(_prev_prefix, list): - _limit = min(len(_prev_prefix), len(messages)) - while ( - _scan_start < _limit - and messages[_scan_start] is _prev_prefix[_scan_start] - and bool(messages[_scan_start].get(_DB_PERSISTED_MARKER)) - ): - _scan_start += 1 - - # Collect this flush's new rows and write them in ONE transaction - # at the end of the scan (see append_messages_batch). - _batch_rows: List[Dict[str, Any]] = [] - _batch_msgs: List[Dict] = [] - for _msg_idx in range(_scan_start, len(messages)): - msg = messages[_msg_idx] - if not isinstance(msg, dict): - continue - # Never write ephemeral scaffolding: the flush is append-only, so a mid-turn persist could - # commit a synthetic turn that the end-of-turn drop cannot un-write. Skip regardless of - # position. - if _is_ephemeral_scaffolding(msg): - continue - if msg.get(_DB_PERSISTED_MARKER): - continue - # Already-durable (history copy or caller-seeded): stamp so future flushes skip without id() - # sets. - if id(msg) in history_ids or id(msg) in seed_ids: - msg[_DB_PERSISTED_MARKER] = True - continue - role = msg.get("role", "unknown") - content = msg.get("content") - # api_content sidecar: exact bytes sent to the API when they differ from clean content, so - # replay reproduces the sent prefix byte-for-byte. - _row_api_content = msg.get("api_content") - if not isinstance(_row_api_content, str): - _row_api_content = None - _row_timestamp = msg.get("timestamp") - # Apply the persist override to THIS row only. A list override replaces a noted payload; a - # text override must not erase an image/audio summary. Also match the staged CLI dict by - # identity — the close safety-net may flush a shortened snapshot whose turn index refers to - # the full history. - pending_cli_message = getattr(self, "_pending_cli_user_message", None) - is_current_turn_user = ( - _ov_idx == _msg_idx or msg is pending_cli_message - ) - if is_current_turn_user and msg.get("role") == "user": - # Preflight compaction may have re-anchored the index at a message MERGED with the - # compaction summary; overwriting it with the clean text would drop the summary from the - # durable transcript. - if ( - _ov_content is not None - and (not isinstance(content, list) or isinstance(_ov_content, list)) - and not msg.get(COMPRESSED_SUMMARY_METADATA_KEY) - ): - # Live content is what the wire sent, the override is the clean transcript; keep the - # sent bytes in api_content so replay matches the wire (#48677). - if ( - _row_api_content is None - and isinstance(content, str) - and content != _ov_content - ): - _row_api_content = content - content = _ov_content - if _ov_timestamp is not None: - _row_timestamp = _ov_timestamp - # Store the sidecar only when it actually differs. - if _row_api_content == content: - _row_api_content = None - # Load-time sanitize divergence: get_messages_as_conversation replays rows through - # sanitize_context().strip(); capture the sent bytes when they would differ (compared in wire - # form). - if ( - _row_api_content is None - and role in ("user", "assistant") - and isinstance(content, str) - and content - and sanitize_context(content).strip() != content.strip() - ): - _row_api_content = content - # Persist multimodal tool results as text summary only — base64 images bloat the DB. - if _is_multimodal_tool_result(content): - content = _multimodal_text_summary(content) - elif isinstance(content, list): - # List of OpenAI-style content parts: strip images, keep text. - _txt = [] - for p in content: - if isinstance(p, dict) and p.get("type") == "text": - _txt.append(str(p.get("text", ""))) - elif isinstance(p, dict) and p.get("type") in {"image", "image_url", "input_image"}: - _txt.append("[screenshot]") - content = "\n".join(_txt) if _txt else None - tool_calls_data = None - if hasattr(msg, "tool_calls") and isinstance(msg.tool_calls, list) and msg.tool_calls: - tool_calls_data = [ - {"name": tc.function.name, "arguments": tc.function.arguments} - for tc in msg.tool_calls - ] - elif isinstance(msg.get("tool_calls"), list): - tool_calls_data = msg["tool_calls"] - _row = { - "role": role, - "content": content, - "tool_name": msg.get("tool_name"), - "tool_calls": tool_calls_data, - "tool_call_id": msg.get("tool_call_id"), - "finish_reason": msg.get("finish_reason"), - # Reasoning/codex fields are role-gated (assistant-only) - # inside _insert_message_rows — pass through untouched. - "reasoning": msg.get("reasoning"), - "reasoning_content": msg.get("reasoning_content"), - "reasoning_details": msg.get("reasoning_details"), - "codex_reasoning_items": msg.get("codex_reasoning_items"), - "codex_message_items": msg.get("codex_message_items"), - "_compressed_summary": bool(msg.get(COMPRESSED_SUMMARY_METADATA_KEY)), - "timestamp": _row_timestamp, - "api_content": _row_api_content, - # Standalone reference handoffs are always hidden so they never occupy the active user - # slot in retry/undo dispatch (#80622); merge-into-tail carriers keep prior visibility. - "display_kind": ( - "hidden" - if ( - msg.get(COMPRESSED_SUMMARY_METADATA_KEY) - and user_originated_turn_view(msg) is None - and ( - ContextCompressor.classify_summary_content( - msg.get("content") - ) - == "standalone" - or not msg.get( - "_compressed_summary_has_user_turn" - ) - ) - ) - else msg.get("display_kind") - ), - "display_metadata": msg.get("display_metadata"), - # Platform message id — load-bearing for restart drain-window recovery dedup. - "platform_message_id": msg.get("platform_message_id"), - } - if isinstance(msg.get("_row_id"), int): - _row["_row_id"] = msg["_row_id"] - _batch_rows.append(_row) - _batch_msgs.append(msg) - # One transaction for the turn's new rows. All-or-nothing pairs with the marker stamping below: - # on failure no rows landed and no markers were stamped, so the next flush re-writes the tail. - if _batch_rows: - self._session_db.append_messages_batch( - session_id=self.session_id, - messages=_batch_rows, - compression_lock_holder=getattr( - self, "_active_compression_lock_holder", None - ), - turn_lease_holder=getattr( - self, "_active_session_turn_lease_holder", None - ), - turn_lease_ttl_seconds=getattr( - self, "_active_session_turn_lease_ttl_seconds", 300.0 - ) - or 300.0, - ) - from agent.transcript_repair import sync_flushed_message_markers - - sync_flushed_message_markers(_batch_msgs, _batch_rows) + batch_rows, batch_msgs = self._db_flush_collect(messages, conversation_history) + self._db_flush_write(batch_rows, batch_msgs) # Markers are now the sole truth; reset the one-shot seed so no id() outlives this flush. self._flushed_db_message_ids = set() self._last_flushed_db_idx = len(messages) - # Snapshot for the bounded scan above — only on full success, so - # a partially-processed list can never be treated as settled. + # Snapshot for the bounded scan — only on full success, so a partially-processed list can + # never be treated as settled. self._db_flush_scan_prefix = messages[:] return True except Exception as e: - # Force a full re-scan on the next flush: an exception mid-loop - # leaves messages with mixed dispositions. - self._db_flush_scan_prefix = None - # The only place the SQLite error is visible before it becomes a bare False — classify it so the - # turn-end explanation can distinguish lock contention from disk-full/read-only. - from hermes_state import ( - CompressionSessionClosedError, - StateDbCorruptError, - StateDbReplacedError, - classify_persistence_error, - divert_session_transcript_jsonl, - ) - - self._last_persistence_error_cause = classify_persistence_error(e) - if isinstance(e, (StateDbReplacedError, StateDbCorruptError)): - # Replaced/quarantined handle will not take this batch again — keep it on disk, not only in - # RAM. - try: - divert_session_transcript_jsonl( - getattr(self, "session_id", "") or "", - _batch_rows, - ) - except Exception: - logger.warning( - "JSONL divert failed after state.db %s for %s", - self._last_persistence_error_cause, - getattr(self, "session_id", None), - exc_info=True, - ) - if isinstance(e, CompressionSessionClosedError): - # Compression race: another path rotated this session mid-write. Adopt the continuation tip - # (get_compression_tip) ONLY when it is a different, live row, and retry exactly once; a - # second closed-parent write fails closed. tip == session_id means no continuation exists. - if _adoption_budget > 0: - old_id = self.session_id - tip = None - try: - tip = self._session_db.get_compression_tip(old_id) - except Exception as tip_exc: - logger.warning( - "compression tip lookup failed for %s: %s", - old_id, - tip_exc, - ) - if tip and tip != old_id: - tip_row = None - try: - tip_row = self._session_db.get_session(tip) - except Exception: - tip_row = None - if tip_row is not None and tip_row.get("ended_at") is None: - logger.warning( - "Adopted live compression tip %s for closed " - "session %s; retrying flush once", - tip, - old_id, - ) - self.session_id = tip - self._flushed_db_message_ids = set() - self._last_flushed_db_idx = 0 - self._compression_adoption_failed = False - return self._flush_messages_to_session_db_unlocked( - messages, - conversation_history, - _adoption_budget=0, - ) - # No live tip or budget exhausted: fail closed. The flag lets the turn explanation name - # compression rotation instead of misleading full-disk advice. - self._compression_adoption_failed = True - logger.warning("Session DB append_message failed: %s", e) - return False - logger.warning("Session DB append_message failed: %s", e) + if self._db_flush_failed(e, batch_rows, _adoption_budget): + return self._flush_messages_to_session_db_unlocked( + messages, conversation_history, _adoption_budget=0, + ) return False def _get_messages_up_to_last_assistant(self, messages: List[Dict]) -> List[Dict]: - """Get messages up to (but not including) the last assistant turn. - - The rollback point when the final assistant message is incomplete or malformed. - """ - if not messages: - return [] - - # Find the index of the last assistant message - last_assistant_idx = None + """Messages up to (not including) the last assistant turn — the rollback point when the final + assistant message is incomplete or malformed. All messages when there is none.""" for i in range(len(messages) - 1, -1, -1): if messages[i].get("role") == "assistant": - last_assistant_idx = i - break - - if last_assistant_idx is None: - # No assistant message found, return all messages - return messages.copy() - - # Return everything up to (not including) the last assistant message - return messages[:last_assistant_idx] + return messages[:i] + return messages.copy() _format_tools_for_system_message = _forward("agent.system_prompt", "format_tools_for_system_message") @@ -525,7 +469,6 @@ class SessionPersistenceMixin: """Save conversation trajectory to JSONL file.""" if not self.save_trajectories: return - trajectory = self._convert_to_trajectory_format(messages, user_query, completed) _save_trajectory_to_file(trajectory, self.model, completed) @@ -545,42 +488,35 @@ class SessionPersistenceMixin: @staticmethod def _redact_message_content(content): - """Apply secret redaction to message content (str or list-of-parts). - - Only text fields pass through ``redact_sensitive_text``; image/binary parts are untouched. - No-op when ``HERMES_REDACT_SECRETS`` disables redaction. - """ - if content is None: - return content + """Redact secrets in str or list-of-parts content; only text fields are touched. + No-op when ``HERMES_REDACT_SECRETS`` disables redaction.""" if isinstance(content, str): return redact_sensitive_text(content) - if isinstance(content, list): - redacted = [] - for part in content: - if isinstance(part, dict): - part = dict(part) - if isinstance(part.get("text"), str): - part["text"] = redact_sensitive_text(part["text"]) - if isinstance(part.get("content"), str): - part["content"] = redact_sensitive_text(part["content"]) - redacted.append(part) - return redacted - return content + if not isinstance(content, list): + return content + redacted = [] + for part in content: + if isinstance(part, dict): + part = dict(part) + for key in ("text", "content"): + if isinstance(part.get(key), str): + part[key] = redact_sensitive_text(part[key]) + redacted.append(part) + return redacted def _save_session_log(self, messages: List[Dict[str, Any]] = None): - """Optional per-session JSON snapshot writer (``sessions.write_json_snapshots``, default False). + """Optional per-session JSON snapshot (``sessions.write_json_snapshots``, default False). - state.db is canonical; this exists for external tooling reading ``session_{sid}.json``. Rewrites the - full list after every persistence point, never overwriting a larger log with fewer messages. + state.db is canonical; this exists for external tooling reading ``session_{sid}.json``. + Rewrites the full list after every persistence point, never overwriting a larger log with + fewer messages (resumed agent with partial history). """ if not getattr(self, "_session_json_enabled", False): return messages = messages or self._session_messages if not messages: return - - # Re-derive the path each call so /branch and /compress land in the right file. Session IDs can be - # untrusted (X-Hermes-Session-Id) — sanitize to a single traversal-free segment. + # Re-derive the path each call so /branch and /compress land in the right file. try: safe_sid = _safe_session_filename_component(self.session_id) log_file = self.logs_dir / f"session_{safe_sid}.json" @@ -590,21 +526,18 @@ class SessionPersistenceMixin: try: cleaned = [] for msg in messages: - # Mirror the SQLite flush: ephemeral recovery scaffolding is - # internal retry state, never durable transcript content. + # Mirror the SQLite flush: scaffolding is never durable transcript content. if _is_ephemeral_scaffolding(msg): continue if msg.get("role") == "assistant" and msg.get("content"): msg = dict(msg) msg["content"] = self._clean_session_content(msg["content"]) - # Defence-in-depth: redact credentials from every message before persistence; respects - # HERMES_REDACT_SECRETS via redact_sensitive_text (#19798, #19845). + # Defence-in-depth credential redaction (respects HERMES_REDACT_SECRETS). if "content" in msg: msg = dict(msg) msg["content"] = self._redact_message_content(msg.get("content")) cleaned.append(msg) - # Never overwrite a larger session log with fewer messages (resumed agent with partial history). if log_file.exists(): try: existing = json.loads(log_file.read_text(encoding="utf-8")) @@ -630,14 +563,7 @@ class SessionPersistenceMixin: "message_count": len(cleaned), "messages": cleaned, } - - atomic_json_write( - log_file, - entry, - indent=2, - default=str, - ) - + atomic_json_write(log_file, entry, indent=2, default=str) except Exception as e: if self.verbose_logging: logging.warning(f"Failed to save session log: {e}")