fix(session-db): persist post-flush /steer mutation to SQLite

agent/agent_runtime_helpers.py

@@ -33,6 +33,7 @@
 from typing import Any, Dict, List, Optional, Tuple
 
 from hermes_cli.timeouts import get_provider_request_timeout
+from agent.persistence_markers import _DB_CONTENT_UPDATE_PENDING
 from agent.prompt_builder import format_steer_marker
 from agent.tool_dispatch_helpers import _trajectory_normalize_msg, make_tool_result_message
 from agent.trajectory import convert_scratchpad_to_think
@@ -3964,6 +3965,10 @@ def apply_pending_steer_to_tool_results(agent, messages: list, num_tool_msgs: in
             messages[target_idx]["content"] = f"{existing_content}{marker}"
     else:
         messages[target_idx]["content"] = existing_content + marker
+    # The result may already have been incrementally flushed. Mark only this
+    # intentional mutation for an in-place durable update; generic content
+    # drift can also come from sequence repair and must not rewrite other rows.
+    messages[target_idx][_DB_CONTENT_UPDATE_PENDING] = True
     _ra().logger.info(
         "Delivered /steer to agent after tool batch (%d chars): %s",
         len(steer_text),

agent/context_compressor.py

@@ -148,7 +148,11 @@ def _is_summary_access_or_quota_error(exc: Exception) -> bool:
 # micro markers: a batch marker's content is NOT contained in the micro
 # rolling summary, so dropping or rewriting one destroys history.
 MICRO_COMPACT_MARKER_KEY = "_micro_compact_marker"
-_DB_PERSISTED_MARKER = "_db_persisted"
+
+from agent.persistence_markers import (  # noqa: E402
+    _DB_PERSISTED_MARKER,
+    _DB_CONTENT_UPDATE_PENDING,
+)
 
 _NO_USER_TASK_SENTINEL = "None. This session contains no user-authored turns."
 COMPRESSION_CONTINUATION_USER_CONTENT = (
@@ -178,6 +182,7 @@ def _fresh_compaction_message_copy(msg: Dict[str, Any]) -> Dict[str, Any]:
     """
     fresh = msg.copy()
     fresh.pop(_DB_PERSISTED_MARKER, None)
+    fresh.pop(_DB_CONTENT_UPDATE_PENDING, None)
     return fresh
 
 
@@ -227,6 +232,7 @@ def _strip_persistence_markers(messages: List[Dict[str, Any]]) -> None:
     for msg in messages:
         if isinstance(msg, dict):
             msg.pop(_DB_PERSISTED_MARKER, None)
+            msg.pop(_DB_CONTENT_UPDATE_PENDING, None)
 
 
 # Appended to every standalone summary message (and to the merged-into-tail

agent/conversation_loop.py

@@ -73,6 +73,7 @@
     parse_available_output_tokens_from_error,
     save_context_length,
 )
+from agent.persistence_markers import _DB_CONTENT_UPDATE_PENDING
 from agent.process_bootstrap import _install_safe_stdio
 from agent.prompt_caching import (
     build_prompt_cache_plan,
@@ -1514,6 +1515,7 @@ def run_conversation(
                             _sm["content"] = blocks
                         except Exception:
                             pass
+                    _sm[_DB_CONTENT_UPDATE_PENDING] = True
                     _injected = True
                     logger.debug(
                         "Pre-API-call steer drain: injected into tool msg at index %d",

agent/persistence_markers.py

@@ -0,0 +1,20 @@
+"""Private persistence metadata stamped on live message dicts.
+
+The incremental SessionDB flush (``AIAgent._flush_messages_to_session_db``)
+tracks durability directly on the message dicts it writes, so repeated
+flushes stay idempotent without positional slices or ``id()``-keyed sets.
+These keys are private wire metadata: the API payload build strips every
+top-level ``_``-prefixed key before a request leaves the process, and the
+JSON session snapshot / context compressor strip them before reusing a
+message. Define them ONCE here — every producer (steer injection), consumer
+(flush), and stripper (compressor, session log) must agree on the exact
+string or markers silently leak or stop being honoured.
+"""
+
+# Stamped by the flush on each message dict it has written to state.db.
+_DB_PERSISTED_MARKER = "_db_persisted"
+
+# Stamped by intentional post-INSERT content mutations (mid-turn /steer
+# appending its marker to an already-flushed tool result) so the next flush
+# updates the durable row in place instead of skipping the dict.
+_DB_CONTENT_UPDATE_PENDING = "_db_content_update_pending"

gateway/session.py

@@ -3392,7 +3392,9 @@ def load_transcript(self, session_id: str) -> List[Dict[str, Any]]:
             # would otherwise re-trigger the pre-request repair on every
             # request forever — heal it once at the restore boundary.
             return self._db.get_messages_as_conversation(
-                session_id, repair_alternation=True
+                session_id,
+                repair_alternation=True,
+                include_row_ids=True,
             )
         except Exception as e:
             logger.debug("Could not load messages from DB: %s", e)

hermes_cli/cli_agent_setup_mixin.py

@@ -405,7 +405,9 @@ def _init_agent(self, *, model_override: str = None, runtime_override: dict = No
                 if resolved_meta:
                     session_meta = resolved_meta
             restored = self._session_db.get_messages_as_conversation(
-                self.session_id, repair_alternation=True
+                self.session_id,
+                repair_alternation=True,
+                include_row_ids=True,
             )
             if restored:
                 restored = [m for m in restored if m.get("role") != "session_meta"]

hermes_state.py

@@ -6293,7 +6293,8 @@ def append_messages_batch(
         messages: List[Dict[str, Any]],
         compression_lock_holder: Optional[str] = None,
         chunk_rows: Optional[int] = None,
-    ) -> int:
+        return_row_ids: bool = False,
+    ) -> int | List[int]:
         """Append multiple messages atomically in ONE write transaction.
 
         ``messages`` is a list of dicts in the same shape
@@ -6319,12 +6320,28 @@ def append_messages_batch(
         the batch commits in chunks of at most that many rows — same
         recovery semantics as the old per-row loops (a mid-copy failure
         leaves a partial seed), just with bounded lock holds. A turn flush
-        never needs it. Returns the inserted row count.
+        never needs it.
+
+        Returns the inserted row count by default. ``return_row_ids=True``
+        returns the inserted ids in input order for live consumers that must
+        retain exact durable message identity.
         """
         if not messages:
-            return 0
+            return [] if return_row_ids else 0
 
         if chunk_rows is not None and len(messages) > chunk_rows:
+            if return_row_ids:
+                inserted_ids: List[int] = []
+                for start in range(0, len(messages), chunk_rows):
+                    inserted_ids.extend(
+                        self.append_messages_batch(
+                            session_id,
+                            messages[start:start + chunk_rows],
+                            compression_lock_holder=compression_lock_holder,
+                            return_row_ids=True,
+                        )
+                    )
+                return inserted_ids
             inserted_total = 0
             for start in range(0, len(messages), chunk_rows):
                 inserted_total += self.append_messages_batch(
@@ -6338,9 +6355,18 @@ def _do(conn):
             self._check_transcript_write_guards(
                 conn, session_id, compression_lock_holder
             )
-            inserted, tool_calls_total = self._insert_message_rows(
-                conn, session_id, messages
-            )
+            inserted_ids = [] if return_row_ids else None
+            if return_row_ids:
+                inserted, tool_calls_total = self._insert_message_rows(
+                    conn,
+                    session_id,
+                    messages,
+                    inserted_row_ids=inserted_ids,
+                )
+            else:
+                inserted, tool_calls_total = self._insert_message_rows(
+                    conn, session_id, messages
+                )
             # One aggregated counter update for the whole batch.
             if tool_calls_total > 0:
                 conn.execute(
@@ -6353,7 +6379,7 @@ def _do(conn):
                     "UPDATE sessions SET message_count = message_count + ? WHERE id = ?",
                     (inserted, session_id),
                 )
-            return inserted
+            return inserted_ids if return_row_ids else inserted
 
         # Same criticality as append_message: this IS the turn's transcript.
         return self._execute_write(
@@ -6602,7 +6628,80 @@ def get_message_role(self, session_id: str, row_id: int) -> Optional[str]:
 
         return row[0] if row else None
 
-    def _insert_message_rows(self, conn, session_id: str, messages: List[Dict[str, Any]]) -> tuple[int, int]:
+    def update_tool_message_content(
+        self,
+        message_row_id: Optional[int],
+        content: Any,
+        session_id: str,
+        *,
+        tool_call_id: Optional[str] = None,
+        compression_lock_holder: Optional[str] = None,
+    ) -> Optional[int]:
+        """Update an intentionally mutated active tool row and return its id.
+
+        ``message_row_id`` is exact for rows inserted by the live agent or
+        loaded for live replay. ``tool_call_id`` is a fallback for an atomic
+        transcript rewrite that replaced the SQLite row id, and is accepted
+        only when it resolves uniquely. FTS update triggers keep search indexes
+        in sync.
+        """
+        stored_content = self._encode_content(content)
+
+        def _do(conn):
+            self._check_transcript_write_guards(
+                conn, session_id, compression_lock_holder
+            )
+
+            row = None
+            if message_row_id is not None:
+                row = conn.execute(
+                    "SELECT id FROM messages "
+                    "WHERE id = ? AND session_id = ? "
+                    "AND role = 'tool' AND active = 1",
+                    (message_row_id, session_id),
+                ).fetchone()
+            if row is None and tool_call_id:
+                matches = conn.execute(
+                    "SELECT id FROM messages "
+                    "WHERE session_id = ? AND role = 'tool' "
+                    "AND tool_call_id = ? AND active = 1 "
+                    "ORDER BY id LIMIT 2",
+                    (session_id, tool_call_id),
+                ).fetchall()
+                # Durable transcripts may contain retry/crash duplicates that
+                # live replay repairs by keeping only the first. Never guess
+                # which duplicate a metadata-free message represents.
+                if len(matches) == 1:
+                    row = matches[0]
+            if row is None:
+                return None
+            resolved_row_id = int(row["id"])
+            conn.execute(
+                "UPDATE messages SET content = ? WHERE id = ?",
+                (stored_content, resolved_row_id),
+            )
+            return resolved_row_id
+
+        updated_row_id = self._execute_write(
+            _do, patience_s=self._TRANSCRIPT_WRITE_PATIENCE_S
+        )
+        if updated_row_id is None:
+            logger.warning(
+                "update_tool_message_content matched no active row "
+                "(id=%s session_id=%s tool_call_id=%s)",
+                message_row_id,
+                session_id,
+                tool_call_id,
+            )
+        return updated_row_id
+
+    def _insert_message_rows(
+        self,
+        conn,
+        session_id: str,
+        messages: List[Dict[str, Any]],
+        inserted_row_ids: Optional[List[int]] = None,
+    ) -> tuple[int, int]:
         """Insert *messages* as fresh active rows for *session_id*.
 
         Shared by :meth:`replace_messages` (delete-then-insert) and
@@ -6661,7 +6760,7 @@ def _insert_message_rows(self, conn, session_id: str, messages: List[Dict[str, A
 
             api_content = msg.get("api_content")
 
-            conn.execute(
+            cursor = conn.execute(
                 """INSERT INTO messages (session_id, role, content, tool_call_id,
                    tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
                    reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
@@ -6691,6 +6790,8 @@ def _insert_message_rows(self, conn, session_id: str, messages: List[Dict[str, A
                     self._encode_display_metadata(msg.get("display_metadata")),
                 ),
             )
+            if inserted_row_ids is not None:
+                inserted_row_ids.append(int(cursor.lastrowid))
             inserted += 1
             if tool_calls is not None:
                 tool_calls_total += (
@@ -7099,8 +7200,9 @@ def get_messages_as_conversation(
         otherwise re-triggers the pre-request defensive repair on every
         single request for the rest of the session's life — the repair
         mutates only the per-request list, never the stored transcript.
-        Inspection/export consumers keep the default and see the transcript
-        verbatim.
+        ``include_row_ids=True`` adds the private SQLite row id needed by
+        consumers that address or intentionally mutate a durable message.
+        Inspection/export consumers keep the default transcript shape.
         """
         session_ids = [session_id]
         if include_ancestors:
@@ -7165,9 +7267,9 @@ def _rows_to_conversation(
                 content = sanitize_context(content).strip()
             msg = {"role": row["role"], "content": content}
             # Durable per-message identity for surfaces that need to address a
-            # specific row later (desktop reactions). OPT-IN: only the gateway
-            # asks for it — every other consumer (ACP restore, export,
-            # inspection) gets the transcript in its historical shape.
+            # specific row later (desktop reactions and live post-flush
+            # mutations). OPT-IN: other consumers (ACP restore, export,
+            # inspection) keep the transcript's historical shape.
             # Underscore-prefixed so every transport's convert_messages()
             # strips it before the wire.
             if include_row_ids and row["id"] is not None: