fix: gateway persists transcript to state.db itself instead of relying on agent's internal flush

gateway/run.py

@@ -8625,10 +8625,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                     }
                 )
 
-            # The agent already persisted these messages to SQLite via
-            # _flush_messages_to_session_db(), so skip the DB write here
-            # to prevent the duplicate-write bug (#860 / #42039).
-            agent_persisted = self._session_db is not None
+            # The gateway persists transcript entries to state.db so that
+            # agent messages survive process restarts.  A last-message
+            # dedup in append_to_transcript() prevents the duplicate-row
+            # bug (#860) when the agent has already flushed the same
+            # message via _flush_messages_to_session_db().
 
             # Find only the NEW messages from this turn (skip history we loaded).
             # Use the filtered history length (history_offset) that was actually
@@ -8647,7 +8648,6 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                 self.session_store.append_to_transcript(
                     session_entry.session_id,
                     _user_entry,
-                    skip_db=agent_persisted,
                 )
             else:
                 history_len = agent_result.get("history_offset", len(history))
@@ -8661,13 +8661,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                     self.session_store.append_to_transcript(
                         session_entry.session_id,
                         _user_entry,
-                        skip_db=agent_persisted,
                     )
                     if response:
                         self.session_store.append_to_transcript(
                             session_entry.session_id,
                             {"role": "assistant", "content": response, "timestamp": ts},
-                            skip_db=agent_persisted,
                         )
                 else:
                     # Attach the inbound platform message_id to the first user
@@ -8691,7 +8689,6 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                             _user_msg_id_attached = True
                         self.session_store.append_to_transcript(
                             session_entry.session_id, entry,
-                            skip_db=agent_persisted,
                         )
 
             # Token counts and model are now persisted by the agent directly.

gateway/session.py

@@ -1268,12 +1268,28 @@ def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db
         """Append a message to a session's transcript (SQLite).
 
         Args:
-            skip_db: When True, skip the SQLite write. Used when the agent
-                     already persisted messages to SQLite via its own
-                     _flush_messages_to_session_db(), preventing the
-                     duplicate-write bug (#860).
+            skip_db: When True, skip the SQLite write entirely.  Kept for
+                     backward compatibility with other callers.
         """
         if self._db and not skip_db:
+            role = message.get("role", "unknown")
+            content = message.get("content")
+            # Skip duplicate writes: if the last message in this session
+            # has the same role + content, the agent already persisted it
+            # via _flush_messages_to_session_db() and we should not create
+            # a duplicate row.  This lets the gateway always attempt the
+            # write (surviving process restarts) while avoiding the
+            # double-row bug (#860).
+            try:
+                last = self._db.get_last_message(session_id)
+                if (
+                    last
+                    and last.get("role") == role
+                    and last.get("content") == content
+                ):
+                    return
+            except Exception:
+                pass
             try:
                 self._db.append_message(
                     session_id=session_id,

hermes_state.py

@@ -2090,6 +2090,22 @@ def _do(conn):
 
         return self._execute_write(_do)
 
+    def get_last_message(self, session_id: str) -> Optional[Dict[str, Any]]:
+        """Return the most recent message for a session, or None."""
+        if self._conn is None:
+            return None
+        with self._lock:
+            cursor = self._conn.execute(
+                """SELECT role, content FROM messages
+                   WHERE session_id = ?
+                   ORDER BY id DESC LIMIT 1""",
+                (session_id,),
+            )
+            row = cursor.fetchone()
+        if row:
+            return {"role": row["role"], "content": row["content"]}
+        return None
+
     def replace_messages(self, session_id: str, messages: List[Dict[str, Any]]) -> None:
         """Atomically replace every message for a session.