From 3e204bd771f2a167b818d33ece62d214b2b4b2a8 Mon Sep 17 00:00:00 2001 From: nankingjing <1079826437@qq.com> Date: Fri, 3 Jul 2026 12:48:37 +0800 Subject: [PATCH] fix(agent): strip _db_persisted when assembling rotation compression transcript (#57491) Shallow messages[i].copy() during context compression propagated the _db_persisted marker from cached gateway incremental flushes into the post-rotation compressed list. _flush_messages_to_session_db then skipped every row when writing to the new child session, so gateway restarts lost the compacted transcript (severe amnesia). Strip the marker in _fresh_compaction_message_copy() and add regression tests for rotation flush + compressor assembly. Fixes #57491 --- agent/context_compressor.py | 19 ++++++++- tests/agent/test_context_compressor.py | 11 +++++ .../run_agent/test_compression_persistence.py | 42 +++++++++++++++++++ 3 files changed, 70 insertions(+), 2 deletions(-) diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 6ee5d6cffea..2805f6de43b 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -84,6 +84,21 @@ LEGACY_SUMMARY_PREFIX = "[CONTEXT SUMMARY]:" # poisoning every subsequent request in the session — a bare key like # "is_compressed_summary" would reach the wire and trip exactly that. COMPRESSED_SUMMARY_METADATA_KEY = "_compressed_summary" +_DB_PERSISTED_MARKER = "_db_persisted" + + +def _fresh_compaction_message_copy(msg: Dict[str, Any]) -> Dict[str, Any]: + """Copy a message for compaction assembly without persistence markers. + + Live cached-gateway transcripts stamp ``_db_persisted`` during incremental + flushes. Shallow ``.copy()`` propagates that marker into the post-rotation + compressed list, so ``_flush_messages_to_session_db`` skips every row when + writing to the new child session (#57491). + """ + fresh = msg.copy() + fresh.pop(_DB_PERSISTED_MARKER, None) + return fresh + # Appended to every standalone summary message (and to the merged-into-tail # prefix) so the model has an unambiguous "summary ends here" boundary. @@ -2834,7 +2849,7 @@ This compaction should PRIORITISE preserving all information related to the focu # Phase 4: Assemble compressed message list compressed = [] for i in range(compress_start): - msg = messages[i].copy() + msg = _fresh_compaction_message_copy(messages[i]) if i == 0 and msg.get("role") == "system": existing = msg.get("content") _compression_note = "[Note: Some earlier conversation turns have been compacted into a handoff summary to preserve context space. The current session state may still reflect earlier work, so build on that summary and state rather than re-doing work. Your persistent memory (MEMORY.md, USER.md) remains fully authoritative regardless of compaction.]" @@ -2907,7 +2922,7 @@ This compaction should PRIORITISE preserving all information related to the focu }) for i in range(compress_end, n_messages): - msg = messages[i].copy() + msg = _fresh_compaction_message_copy(messages[i]) if _merge_summary_into_tail and i == compress_end: # Merge the summary into the first tail message, but place # the END MARKER at the very end so the model sees an diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py index 026ee37d951..19fd418cc11 100644 --- a/tests/agent/test_context_compressor.py +++ b/tests/agent/test_context_compressor.py @@ -341,6 +341,17 @@ class TestCompress: # original content is present in either case. assert msgs[-2]["content"] in result[-2]["content"] + def test_compress_strips_db_persisted_from_assembled_messages(self, compressor): + """Regression for #57491: shallow copies must not carry flush markers.""" + msgs = [ + {"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}", "_db_persisted": True} + for i in range(10) + ] + with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): + result = compressor.compress(msgs) + assert len(result) < len(msgs) + assert all("_db_persisted" not in msg for msg in result) + def test_protect_first_n_decays_after_first_compression(self): """Regression for #11996: protect_first_n must protect early turns on the FIRST compaction but DECAY afterwards, so the same early user diff --git a/tests/run_agent/test_compression_persistence.py b/tests/run_agent/test_compression_persistence.py index d38828ccba2..bbec27c5647 100644 --- a/tests/run_agent/test_compression_persistence.py +++ b/tests/run_agent/test_compression_persistence.py @@ -191,6 +191,48 @@ class TestFlushAfterCompression: "final answer", ] + def test_rotation_child_session_flushes_full_compressed_transcript_with_markers(self): + """Regression for #57491: live cached-agent markers must not block child flush.""" + from agent.conversation_compression import compress_context + from hermes_state import SessionDB + + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "test.db" + db = SessionDB(db_path=db_path) + parent_sid = "20260701_152840_parent" + db.create_session(parent_sid, "gateway", model="test/model") + + agent = self._make_agent(db) + agent.session_id = parent_sid + agent.compression_in_place = False + agent._ensure_db_session() + + messages = [ + { + "role": "user" if i % 2 == 0 else "assistant", + "content": f"message {i}", + "_db_persisted": True, + } + for i in range(12) + ] + + with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): + compressed, _ = compress_context( + agent, messages, approx_tokens=100_000, system_message="sys" + ) + + assert agent.session_id != parent_sid + child_sid = agent.session_id + + agent._flush_messages_to_session_db(compressed, None) + + child_rows = db.get_messages(child_sid) + assert len(child_rows) == len(compressed), ( + f"Expected {len(compressed)} rows in child session, got {len(child_rows)}. " + f"_db_persisted marker propagation bug (#57491)." + ) + db.close() + # --------------------------------------------------------------------------- # Part 2: Gateway-side — history_offset after session split