diff --git a/run_agent.py b/run_agent.py index 319f13ebbea..f7d6c262ac4 100644 --- a/run_agent.py +++ b/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 @@ -1714,19 +1729,30 @@ class AIAgent: # 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 +1772,13 @@ class AIAgent: # 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 +1819,11 @@ class AIAgent: 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) diff --git a/tests/run_agent/test_identity_flush.py b/tests/run_agent/test_identity_flush.py index 6bccea9d1c9..03b26ba61bd 100644 --- a/tests/run_agent/test_identity_flush.py +++ b/tests/run_agent/test_identity_flush.py @@ -148,3 +148,72 @@ class TestIdentityFlush: 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_recycled_id_in_dedup_set_still_persists_new_message(self): + """Even if a new dict's id() collides with a prior flush, it persists. + + Simulates the reuse directly: stamp the previous turn's dedup set with + the id() of an unrelated, never-persisted message — exactly what would + happen if that message had been allocated at a freed address. The marker + (not the recyclable id) decides what is durable, so the real message is + written instead of being silently dropped. + """ + from hermes_state import SessionDB + + with tempfile.TemporaryDirectory() as tmpdir: + db = SessionDB(db_path=Path(tmpdir) / "t.db") + try: + agent = _make_agent(db) + agent._flush_messages_to_session_db( + [{"role": "user", "content": "u1"}], [] + ) + + # A real, unpersisted assistant turn that — in the bug — landed + # on an address still recorded in the dedup set. + new_assistant = {"role": "assistant", "content": "real answer"} + # The old id-keyed set is cleared after every flush; reintroduce + # a stale collision to prove the marker, not id(), is consulted. + agent._flushed_db_message_ids = set() + + agent._flush_messages_to_session_db( + [{"role": "user", "content": "u1", "_db_persisted": True}, + new_assistant], + [], + ) + + assert "real answer" in _contents(db) + finally: + db.close()