Stage 399: PR #2705 — fix(api): dedupe replayed context_messages so the agent doesn't see duplicates (follow-up to v0.51.96 #2620)

Co-authored-by: AlexeyDsov <AlexeyDsov@users.noreply.github.com>

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,

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)

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"

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"