diff --git a/gateway/run.py b/gateway/run.py index 26368cd394e3..45946b1e03b4 100644 --- a/gateway/run.py +++ b/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. diff --git a/gateway/session.py b/gateway/session.py index 4d1d26b6467a..f4a424d3261c 100644 --- a/gateway/session.py +++ b/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, diff --git a/hermes_state.py b/hermes_state.py index d83580beeec6..6d09aca7e99c 100644 --- a/hermes_state.py +++ b/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.