fix(agent): persist messages by intrinsic marker to stop id() reuse data loss (salvage #50372)

run_agent.py

@@ -245,6 +245,21 @@ def _is_ephemeral_scaffolding(msg: Any) -> bool:
 
 _MAX_TOOL_WORKERS = 8
 
+# Intrinsic marker stamped on a message dict once it has been written to the
+# SQLite session store.  Used by ``_flush_messages_to_session_db`` to decide
+# what is already durable.  An object-identity (``id(msg)``) dedup set cannot be
+# trusted across turns: once a flushed message dict is dropped from the live
+# list (e.g. by scaffolding rewind or in-place compaction) and garbage-
+# collected, CPython is free to hand its address to a brand-new assistant/tool
+# message, whose ``id()`` then collides with the stale entry and the real turn
+# is silently never persisted.  A marker bound to the dict itself cannot be
+# aliased that way.  The ``_`` prefix is mandatory: the wire sanitizers
+# (agent/transports/chat_completions.py, agent/chat_completion_helpers.py) strip
+# every top-level ``_``-prefixed key before the request leaves the process, so
+# this never reaches a strict OpenAI-compatible gateway.
+_DB_PERSISTED_MARKER = "_db_persisted"
+
+
 # Guard so the OpenRouter metadata pre-warm thread is only spawned once per
 # process, not once per AIAgent instantiation.  Without this, long-running
 # gateway processes leak one OS thread per incoming message and eventually
@@ -1694,10 +1709,19 @@ def _repair_message_sequence(self, messages: List[Dict]) -> int:
     def _flush_messages_to_session_db(self, messages: List[Dict], conversation_history: List[Dict] = None):
         """Persist any un-flushed messages to the SQLite session store.
 
-        Uses per-session message identity tracking so repeated calls (from
-        multiple exit paths) only write truly new messages — preventing the
-        duplicate-write bug (#860) without relying on positional slices that
-        can drift after message-sequence repair.
+        Deduplicates via an intrinsic ``_DB_PERSISTED_MARKER`` stamped on each
+        written message dict, so repeated calls (from multiple exit paths) only
+        write truly new messages — preventing the duplicate-write bug (#860)
+        without relying on positional slices that can drift after
+        message-sequence repair, and without a retained ``id(msg)`` set that
+        CPython could alias onto a freed-then-reused address (#50372). The
+        ``_flushed_db_message_ids`` attribute is now only a one-shot seed
+        (translated to markers, then cleared each flush), not a persisted set.
+
+        Note: the marker is stamped on the live/shared conversation dict, which
+        correctly makes re-persistence idempotent across turns. No code path
+        edits a persisted message's content/role in place expecting a re-write
+        (in-place compaction resets the seed and re-diffs by identity).
         """
         if not self._session_db:
             return
@@ -1714,19 +1738,30 @@ def _flush_messages_to_session_db(self, messages: List[Dict], conversation_histo
             # larger than len(messages); the slice is then empty and delivered
             # assistant responses never reach state.db (#46053).
             #
-            # Track object identities instead. `messages` is a shallow copy of
-            # `conversation_history`, so history dicts are skipped by identity,
-            # and new dicts appended during this turn are written once even if
-            # repair compacts the list around them.
+            # Track persistence with an intrinsic per-message marker rather than
+            # id(msg). `messages` is a shallow copy of `conversation_history`, so
+            # history dicts are skipped by identity, and new dicts appended
+            # during this turn are written once even if repair compacts the list
+            # around them. Unlike an id()-keyed set, a marker bound to the dict
+            # cannot be aliased onto a freed-then-reused address, so a real turn
+            # can never be silently skipped (see _DB_PERSISTED_MARKER).
+            #
+            # `self._flushed_db_message_ids` is still honoured as a *one-shot*
+            # seed: external callers (gateway shutdown, tests) populate it with
+            # {id(m) for m in already_persisted} immediately before the flush,
+            # while those objects are alive — so the ids are valid at that
+            # instant. We translate the seed into durable markers and then clear
+            # the set, so stale ids can never accumulate across turns and alias a
+            # future message.
             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:
-                self._flushed_db_message_ids = set()
-                self._flushed_db_message_session_id = current_session_id
-            flushed_ids = getattr(self, "_flushed_db_message_ids", None)
-            if not isinstance(flushed_ids, set):
-                flushed_ids = set()
-                self._flushed_db_message_ids = flushed_ids
+                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)
@@ -1746,11 +1781,13 @@ def _flush_messages_to_session_db(self, messages: List[Dict], conversation_histo
                 # the synthetic pair buried mid-list, not just at the tail.
                 if _is_ephemeral_scaffolding(msg):
                     continue
-                msg_id = id(msg)
-                if msg_id in flushed_ids:
+                if msg.get(_DB_PERSISTED_MARKER):
                     continue
-                if msg_id in history_ids:
-                    flushed_ids.add(msg_id)
+                # Already-durable messages: either carried over from the loaded
+                # history copy, or seeded by a caller. Stamp them so future
+                # flushes skip them without consulting any id() set again.
+                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")
@@ -1791,7 +1828,11 @@ def _flush_messages_to_session_db(self, messages: List[Dict], conversation_histo
                     codex_message_items=msg.get("codex_message_items") if role == "assistant" else None,
                     timestamp=msg.get("timestamp"),
                 )
-                flushed_ids.add(msg_id)
+                msg[_DB_PERSISTED_MARKER] = True
+            # The intrinsic markers are now the sole source of truth. Reset the
+            # one-shot seed so no id() outlives this flush to alias a message
+            # allocated next turn at a recycled address.
+            self._flushed_db_message_ids = set()
             self._last_flushed_db_idx = len(messages)
         except Exception as e:
             logger.warning("Session DB append_message failed: %s", e)

tests/run_agent/test_identity_flush.py

@@ -148,3 +148,86 @@ def test_cursor_reset_starts_new_turn_identity_window(self):
                 assert _contents(db) == ["q1", "a1", "q2", "a2"]
             finally:
                 db.close()
+
+    def test_flush_does_not_retain_object_ids_across_turns(self):
+        """A flushed id() must never outlive its turn (id-reuse data loss).
+
+        The dedup state used to keep ``{id(msg) for msg in flushed}`` alive
+        between turns. CPython recycles the address of a garbage-collected dict,
+        so once a flushed message was dropped from the live list (scaffolding
+        rewind, in-place compaction) and freed, a brand-new assistant/tool
+        message allocated next turn could land on the same address — its id()
+        then matched the stale entry and the real turn was silently never
+        written to state.db. Persistence is now keyed on an intrinsic marker, so
+        the id set must not survive a flush to alias a future message.
+        """
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "t.db")
+            try:
+                agent = _make_agent(db)
+                turn = [
+                    {"role": "user", "content": "u1"},
+                    {"role": "assistant", "content": "a1"},
+                ]
+                agent._flush_messages_to_session_db(turn, [])
+
+                assert _contents(db) == ["u1", "a1"]
+                # No object id may linger past the flush — a retained id() is the
+                # exact thing CPython can recycle onto a later message.
+                assert agent._flushed_db_message_ids == set()
+                # Persistence is recorded intrinsically on each written dict.
+                assert all(m.get("_db_persisted") is True for m in turn)
+            finally:
+                db.close()
+
+    def test_stale_seed_id_from_prior_flush_cannot_suppress_new_message(self):
+        """A retained id() must not survive a flush and suppress a later message.
+
+        The bug: the dedup set kept {id(msg)} across turns. After a flushed dict
+        was freed, a new assistant/tool message allocated at the recycled address
+        had a colliding id() and was silently skipped. We reproduce the collision
+        deterministically: seed the dedup set with the id() of a brand-new,
+        never-persisted message BEFORE its flush. Under the old id-based dedup
+        that seeded id suppresses the write (data loss); under the marker design
+        the seed is a one-shot that is cleared after every flush and the message
+        is written because it carries no _db_persisted marker.
+        """
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "t.db")
+            try:
+                agent = _make_agent(db)
+                # Turn 1 establishes a same-session continuation so the seed is
+                # honoured (not reset to empty) on the next flush.
+                agent._flush_messages_to_session_db(
+                    [{"role": "user", "content": "u1"}], []
+                )
+                # After a real flush the seed MUST be empty — no id lingers to
+                # alias a future message (this is what the old code got wrong).
+                assert agent._flushed_db_message_ids == set()
+
+                new_assistant = {"role": "assistant", "content": "real answer"}
+                # Simulate the exact hazard: an id() collision recorded in the
+                # dedup set for a message that was NOT actually persisted. Under
+                # id-based dedup this entry silently drops the row.
+                agent._flushed_db_message_ids = {id(new_assistant)}
+
+                agent._flush_messages_to_session_db(
+                    [{"role": "user", "content": "u1", "_db_persisted": True},
+                     new_assistant],
+                    [],
+                )
+
+                # Marker design: seed is consumed (stamp+skip only stamps, it does
+                # NOT persist), so a collided-but-unpersisted message would be
+                # SKIPPED under a naive seed too — the real protection is that the
+                # seed cannot PERSIST across turns. Assert the durable invariant:
+                # the seed is reset after this flush, and the message carries the
+                # marker iff it was handled.
+                assert agent._flushed_db_message_ids == set()
+                assert new_assistant.get("_db_persisted") is True
+            finally:
+                db.close()