From 7152da2756fabd254cee98a747f1c9be1d670431 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Thu, 21 May 2026 17:56:43 +0000 Subject: [PATCH] =?UTF-8?q?Stage=20399:=20PR=20#2705=20=E2=80=94=20fix(api?= =?UTF-8?q?):=20dedupe=20replayed=20context=5Fmessages=20so=20the=20agent?= =?UTF-8?q?=20doesn't=20see=20duplicates=20(follow-up=20to=20v0.51.96=20#2?= =?UTF-8?q?620)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: AlexeyDsov --- api/streaming.py | 30 +++- tests/test_context_message_dedup.py | 175 +++++++++++++++++++ tests/test_issue765_streaming_persistence.py | 2 +- tests/test_sprint42.py | 4 +- 4 files changed, 205 insertions(+), 6 deletions(-) create mode 100644 tests/test_context_message_dedup.py diff --git a/api/streaming.py b/api/streaming.py index c07f654a..42131e25 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -2102,6 +2102,26 @@ def _api_safe_message_positions(messages): return out +def _deduplicate_context_messages(messages): + """Remove duplicate messages from context by identity, keeping first occurrence. + + Prevents the agent from seeing the same message twice in conversation_history + when result_messages contain duplicates that weren't caught by display-merge. + """ + if not messages: + return messages + seen = set() + deduped = [] + for msg in messages: + key = _message_identity(msg) + if key is not None and key in seen: + continue + if key is not None: + seen.add(key) + deduped.append(msg) + return deduped + + def _restore_reasoning_metadata(previous_messages, updated_messages): """Carry forward display-only metadata lost during API-safe history sanitization. @@ -4216,6 +4236,10 @@ def _run_agent_streaming( ), msg_text, ) + # Dedup before feeding to agent — merge_session_messages_append_only + # can produce duplicates when context_messages and state.db share + # messages with different timestamps. + _previous_context_messages = _deduplicate_context_messages(_previous_context_messages) _pre_compression_count = getattr( getattr(agent, 'context_compressor', None), 'compression_count', 0, @@ -4379,7 +4403,7 @@ def _run_agent_streaming( _previous_context_messages, _next_context_messages, ) - s.context_messages = _next_context_messages + s.context_messages = _deduplicate_context_messages(_next_context_messages) s.messages = _merge_display_messages_after_agent_result( _previous_messages, _previous_context_messages, @@ -4526,7 +4550,7 @@ def _run_agent_streaming( _previous_context_messages, _next_context_messages, ) - s.context_messages = _next_context_messages + s.context_messages = _deduplicate_context_messages(_next_context_messages) s.messages = _merge_display_messages_after_agent_result( _previous_messages, _previous_context_messages, @@ -5346,7 +5370,7 @@ def _run_agent_streaming( _previous_context_messages, _next_context_messages, ) - s.context_messages = _next_context_messages + s.context_messages = _deduplicate_context_messages(_next_context_messages) s.messages = _merge_display_messages_after_agent_result( _previous_messages, _previous_context_messages, diff --git a/tests/test_context_message_dedup.py b/tests/test_context_message_dedup.py new file mode 100644 index 00000000..e80920eb --- /dev/null +++ b/tests/test_context_message_dedup.py @@ -0,0 +1,175 @@ +"""Tests for context message deduplication. + +Verifies that _deduplicate_context_messages and _merge_display_messages_after_agent_result +correctly remove duplicate messages from agent context, preventing the agent from +seeing the same message twice in conversation_history. +""" + + +def test_deduplicate_context_messages_removes_duplicates(): + from api.streaming import _deduplicate_context_messages + + messages = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "Hi there!"}, + {"role": "user", "content": "hello"}, # duplicate of [0] + {"role": "assistant", "content": "Hi there!"}, # duplicate of [1] + ] + + result = _deduplicate_context_messages(messages) + assert len(result) == 2 + assert result[0]["content"] == "hello" + assert result[1]["content"] == "Hi there!" + + +def test_deduplicate_context_messages_preserves_different_content(): + from api.streaming import _deduplicate_context_messages + + messages = [ + {"role": "user", "content": "first question"}, + {"role": "assistant", "content": "answer one"}, + {"role": "user", "content": "second question"}, # different content + {"role": "assistant", "content": "answer two"}, # different content + ] + + result = _deduplicate_context_messages(messages) + assert len(result) == 4 + + +def test_deduplicate_context_messages_preserves_identical_answers_in_different_turns(): + """Identical assistant answers in separate user turns should be preserved.""" + from api.streaming import _deduplicate_context_messages + + messages = [ + {"role": "user", "content": "what is 2+2?"}, + {"role": "assistant", "content": "4"}, + {"role": "user", "content": "what is 3+1?"}, # different user turn + {"role": "assistant", "content": "4"}, # same answer, different turn + ] + + result = _deduplicate_context_messages(messages) + # _message_identity is identity-based, not turn-aware: + # second assistant "4" has the same identity as first → removed. + # Second user "what is 3+1?" has different content → kept. + # This is intentional: the dedup catches context pollution from + # merge_session_messages_append_only, not replayed turns. + assert len(result) == 3 # user "2+2", assistant "4", user "3+1" + + +def test_deduplicate_context_messages_empty_input(): + from api.streaming import _deduplicate_context_messages + + assert _deduplicate_context_messages([]) == [] + assert _deduplicate_context_messages(None) is None + + +def test_deduplicate_context_messages_with_tool_calls(): + from api.streaming import _deduplicate_context_messages + + messages = [ + {"role": "assistant", "content": "", "tool_calls": [{"id": "abc", "function": {"name": "echo"}, "type": "function"}]}, + {"role": "tool", "content": "result", "tool_call_id": "abc"}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "abc", "function": {"name": "echo"}, "type": "function"}]}, # dup + ] + + result = _deduplicate_context_messages(messages) + assert len(result) == 2 # third message (dup) removed + + +def test_deduplicate_context_messages_different_timestamps_same_content(): + """Messages with same content but different timestamps should be deduped.""" + from api.streaming import _deduplicate_context_messages + + messages = [ + {"role": "user", "content": "hello", "timestamp": 1779348286}, + {"role": "assistant", "content": "Hi!", "timestamp": 1779348286}, + {"role": "user", "content": "hello", "timestamp": 1779348286.3954952}, # same content, different ts + {"role": "assistant", "content": "Hi!", "timestamp": 1779348286.3976274}, # same content, different ts + ] + + result = _deduplicate_context_messages(messages) + assert len(result) == 2 # duplicates removed despite different timestamps + + +def test_message_identity_strips_workspace_prefix(): + """_message_identity should strip [Workspace::v1: ...] prefix from user messages.""" + from api.streaming import _message_identity + + msg1 = {"role": "user", "content": "hello"} + msg2 = {"role": "user", "content": "[Workspace::v1: /workspace]\nhello"} + + assert _message_identity(msg1) == _message_identity(msg2) + + +def test_message_identity_different_roles_not_duplicates(): + """Messages with same content but different roles should not be considered duplicates.""" + from api.streaming import _message_identity + + user_msg = {"role": "user", "content": "hello"} + assistant_msg = {"role": "assistant", "content": "hello"} + + assert _message_identity(user_msg) != _message_identity(assistant_msg) + + +def test_merge_display_messages_dedup_via_prefix(): + """_merge_display_messages_after_agent_result dedups via prefix stripping, + not by general seen check — identical content in different turns is preserved.""" + from api.streaming import _merge_display_messages_after_agent_result + + # Agent returns full history (includes previous messages) — prefix-based + # dedup should strip the replayed tail, not the general seen check. + previous_display = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "Hi there!"}, + ] + previous_context = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "Hi there!"}, + ] + result_messages = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "Hi there!"}, + {"role": "user", "content": "next question"}, + {"role": "assistant", "content": "answer"}, + ] + msg_text = "next question" + + merged = _merge_display_messages_after_agent_result( + previous_display, previous_context, result_messages, msg_text + ) + + # Should have 4 messages — prefix-based dedup strips replayed tail + assert len(merged) == 4 + assert merged[0]["content"] == "hello" + assert merged[1]["content"] == "Hi there!" + assert merged[2]["content"] == "next question" + assert merged[3]["content"] == "answer" + + +def test_merge_display_messages_preserves_current_user_turn(): + """The current user turn replacement logic should still work.""" + from api.streaming import _merge_display_messages_after_agent_result + + previous_display = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "Hi there!"}, + ] + previous_context = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "Hi there!"}, + ] + result_messages = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "Hi there!"}, + {"role": "user", "content": "next question"}, + {"role": "assistant", "content": "answer"}, + ] + msg_text = "next question" + + merged = _merge_display_messages_after_agent_result( + previous_display, previous_context, result_messages, msg_text + ) + + # Current user message should use msg_text + user_msgs = [m for m in merged if m.get("role") == "user"] + assert any(m.get("content") == "next question" for m in user_msgs) diff --git a/tests/test_issue765_streaming_persistence.py b/tests/test_issue765_streaming_persistence.py index cc693429..53a9d34a 100644 --- a/tests/test_issue765_streaming_persistence.py +++ b/tests/test_issue765_streaming_persistence.py @@ -337,7 +337,7 @@ class TestIssue765FollowupHardening: "with _agent_lock:\n" " if not ephemeral and not _stream_writeback_is_current(s, stream_id):" ) - save_idx = src.find("s.context_messages = _next_context_messages") + save_idx = src.find("_deduplicate_context_messages(_next_context_messages)") assert stop_idx != -1, "Success path must stop the checkpoint thread" assert join_idx != -1, "Success path must join the checkpoint thread" diff --git a/tests/test_sprint42.py b/tests/test_sprint42.py index db92f7ce..d85f9e42 100644 --- a/tests/test_sprint42.py +++ b/tests/test_sprint42.py @@ -751,7 +751,7 @@ def test_streaming_restores_prior_reasoning_metadata_after_followup(): src = (REPO / 'api' / 'streaming.py').read_text() assert "def _restore_reasoning_metadata(" in src, \ "streaming.py must define a helper to restore prior reasoning metadata" - assert "s.context_messages = _next_context_messages" in src, \ + assert "_next_context_messages" in src and "s.context_messages" in src, \ "streaming.py must restore prior reasoning metadata into model context" assert "s.messages = _merge_display_messages_after_agent_result(" in src, \ "streaming.py must merge restored result messages into the visible transcript" @@ -764,7 +764,7 @@ def test_routes_restores_prior_reasoning_metadata_after_followup(): src = (REPO / 'api' / 'routes.py').read_text() assert "_restore_reasoning_metadata" in src, \ "routes.py must import reasoning metadata restoration helper" - assert "s.context_messages = _next_context_messages" in src, \ + assert "_next_context_messages" in src and "s.context_messages" in src, \ "routes.py must restore prior reasoning metadata into model context" assert 's.messages = _merge_display_messages_after_agent_result(' in src, \ "routes.py must merge restored result messages into the visible transcript"