From 05495fc3cf93beb58fab290b62c8be0262037ed3 Mon Sep 17 00:00:00 2001 From: dakjdakd <1428823446@qq.com> Date: Mon, 28 Sep 2026 21:36:05 +0800 Subject: [PATCH 1/2] Fix repeated turns in Python history persistence --- .../core/agent_framework/_sessions.py | 31 ++++---- .../tests/core/test_middleware_with_agent.py | 15 ++-- .../packages/core/tests/core/test_sessions.py | 79 +++++++++++++++++++ 3 files changed, 101 insertions(+), 24 deletions(-) diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py index 5f512004fe9..b0843f7bf02 100644 --- a/python/packages/core/agent_framework/_sessions.py +++ b/python/packages/core/agent_framework/_sessions.py @@ -175,7 +175,7 @@ def _get_message_hash(message: Message) -> MessageIdentity: def filter_new_messages(existing: Sequence[Message], incoming: Sequence[Message]) -> list[Message]: - """Filters incoming messages to only those that are truly new. + """Return messages after the ordered overlap with persisted history. Handles both 'append-only' and 'full transcript replay' scenarios. Prevents superlinear growth and preserves legitimate duplicate turns. @@ -186,23 +186,20 @@ def filter_new_messages(existing: Sequence[Message], incoming: Sequence[Message] existing_hashes = [_get_message_hash(m) for m in existing] incoming_hashes = [_get_message_hash(m) for m in incoming] - if len(incoming) >= len(existing) and incoming_hashes[: len(existing_hashes)] == existing_hashes: - return list(incoming[len(existing) :]) + for i in range(len(incoming_hashes) - len(existing_hashes) + 1): + if incoming_hashes[i : i + len(existing_hashes)] == existing_hashes: + if i == 0 and len(existing) == 1 and existing[-1].role == "user" and existing_hashes[-1][0] != "id": + break # A repeated input without an ID is more important to retain than a possible replay. + return list(incoming[i + len(existing_hashes) :]) - try: - for i in range(len(incoming_hashes) - len(existing_hashes) + 1): - if incoming_hashes[i : i + len(existing_hashes)] == existing_hashes: - return list(incoming[i + len(existing_hashes) :]) - except Exception: - logger.debug("sequence alignment check failed, falling back to set-based deduplication") - - existing_set = set(existing_hashes) - new_msgs: list[Message] = [] - for m, h in zip(incoming, incoming_hashes): - if h not in existing_set: - new_msgs.append(m) - existing_set.add(h) - return new_msgs + for overlap in range(min(len(existing_hashes), len(incoming_hashes)), 0, -1): + if existing_hashes[-overlap:] != incoming_hashes[:overlap]: + continue + if overlap == 1 and existing[-1].role == "user" and existing_hashes[-1][0] != "id": + continue + return list(incoming[overlap:]) + + return list(incoming) @dataclass(frozen=True, slots=True) diff --git a/python/packages/core/tests/core/test_middleware_with_agent.py b/python/packages/core/tests/core/test_middleware_with_agent.py index 014b9adc810..1bc77997417 100644 --- a/python/packages/core/tests/core/test_middleware_with_agent.py +++ b/python/packages/core/tests/core/test_middleware_with_agent.py @@ -1772,14 +1772,15 @@ async def process(self, context: AgentContext, call_next: Callable[[], Awaitable second_after = thread_states[3] assert second_after["before_next"] is False assert second_after["messages_count"] == 1 # Input messages unchanged - assert second_after["thread_count"] == 3 # Previous history (2) + current input (1) + assert second_after["thread_count"] == 4 # Both runs persist their input and response. assert second_after["messages_text"] == ["second message"] - # Thread should contain: first input + first response + second input - assert "first message" in second_after["thread_messages_text"] - assert "second message" in second_after["thread_messages_text"] - # "test response" should only appear once since the duplicate was correctly filtered - response_count = sum(1 for text in second_after["thread_messages_text"] if "test response" in text) - assert response_count == 1 + # Repeated response text remains attached to each run, in conversation order. + assert second_after["thread_messages_text"] == [ + "first message", + "test response", + "second message", + "test response", + ] class TestChatAgentChatMiddleware: diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py index 545d1e83650..1cf5d5adcfc 100644 --- a/python/packages/core/tests/core/test_sessions.py +++ b/python/packages/core/tests/core/test_sessions.py @@ -43,6 +43,7 @@ _run_identity_scope, _RunPersistenceGate, _suspend_run_persistence_gate, + filter_new_messages, is_local_history_conversation_id, ) from agent_framework._telemetry import FeatureIndex @@ -53,6 +54,51 @@ if TYPE_CHECKING: from agent_framework._agents import SupportsAgentRun + +def test_filter_new_messages_preserves_repeated_user_turn() -> None: + existing = [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["first reply"])] + incoming = [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["second reply"])] + + assert filter_new_messages(existing, incoming) == incoming + + +def test_filter_new_messages_preserves_repeated_user_after_unanswered_input() -> None: + previous_input = Message(role="user", contents=["yes"]) + repeated_input = Message(role="user", contents=["yes"]) + reply = Message(role="assistant", contents=["second reply"]) + + assert filter_new_messages([previous_input], [repeated_input]) == [repeated_input] + assert filter_new_messages([previous_input], [repeated_input, reply]) == [repeated_input, reply] + + identified_input = Message(role="user", contents=["yes"], message_id="turn-1") + assert filter_new_messages([identified_input], [identified_input]) == [] + + +def test_filter_new_messages_aligns_partial_replay_with_repeated_content() -> None: + existing = [ + Message(role="user", contents=["yes"]), + Message(role="assistant", contents=["first reply"]), + Message(role="user", contents=["yes"]), + Message(role="assistant", contents=["second reply"]), + ] + new_messages = [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["third reply"])] + incoming = [*existing[-2:], *new_messages] + + assert filter_new_messages(existing, incoming) == new_messages + + +def test_filter_new_messages_keeps_tool_call_result_pairs_on_replay() -> None: + call = Message( + role="assistant", contents=[Content.from_function_call(call_id="call-1", name="lookup", arguments="{}")] + ) + result = Message(role="tool", contents=[Content.from_function_result(call_id="call-1", result="found")]) + existing = [Message(role="user", contents=["lookup"]), call, result] + new_messages = [Message(role="user", contents=["lookup"]), Message(role="assistant", contents=["again"])] + + assert filter_new_messages(existing, [*existing, *new_messages]) == new_messages + assert filter_new_messages(existing, [result, *new_messages]) == new_messages + + # --------------------------------------------------------------------------- # SessionContext tests # --------------------------------------------------------------------------- @@ -1624,6 +1670,23 @@ async def test_save_messages_preserves_duplicate_content(self) -> None: assert state["messages"][0].text == "yes" assert state["messages"][1].text == "yes" + async def test_save_messages_preserves_repeated_user_turn_across_saves(self) -> None: + provider = InMemoryHistoryProvider() + state: dict[str, Any] = {} + + await provider.save_messages( + "s1", + [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["first reply"])], + state=state, + ) + await provider.save_messages( + "s1", + [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["second reply"])], + state=state, + ) + + assert [message.text for message in state["messages"]] == ["yes", "first reply", "yes", "second reply"] + async def test_save_messages_handles_replayed_transcript_with_duplicates(self) -> None: provider = InMemoryHistoryProvider() state: dict[str, Any] = {} @@ -2111,6 +2174,22 @@ async def test_save_messages_preserves_duplicate_content( assert loaded[0].text == "yes" assert loaded[1].text == "yes" + @pytest.mark.parametrize("serialization_format", ["json", "msgpack"]) + async def test_save_messages_preserves_repeated_user_turn_across_saves( + self, tmp_path: Path, serialization_format: Literal["json", "msgpack"] + ) -> None: + provider = FileHistoryProvider(tmp_path, serialization_format=serialization_format) + + await provider.save_messages( + "s1", [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["first reply"])] + ) + await provider.save_messages( + "s1", [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["second reply"])] + ) + + loaded = await provider.get_messages("s1") + assert [message.text for message in loaded] == ["yes", "first reply", "yes", "second reply"] + # --------------------------------------------------------------------------- # Run-persistence gate tests From cc009b9ac60ed386589b71ffaa292a91465e10ad Mon Sep 17 00:00:00 2001 From: dakjdakd <1428823446@qq.com> Date: Tue, 29 Sep 2026 19:43:35 +0800 Subject: [PATCH 2/2] Test Redis history replay and repeated turns separately --- python/packages/redis/tests/test_providers.py | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/python/packages/redis/tests/test_providers.py b/python/packages/redis/tests/test_providers.py index 9cf10c07bca..bef3283c3e1 100644 --- a/python/packages/redis/tests/test_providers.py +++ b/python/packages/redis/tests/test_providers.py @@ -833,7 +833,7 @@ async def test_only_appends_new_messages(self, mock_redis_client: MagicMock): assert pushed_msg_dict["contents"][0]["text"] == "how are you?" async def test_different_roles_same_text_not_deduplicated(self, mock_redis_client: MagicMock): - msg1 = Message(role="user", contents=["ping"]) + msg1 = Message(role="user", contents=["ping"], message_id="original-ping") mock_redis_client.lrange = AsyncMock(return_value=[json.dumps(msg1.to_dict())]) @@ -846,6 +846,26 @@ async def test_different_roles_same_text_not_deduplicated(self, mock_redis_clien pipeline = mock_redis_client.pipeline.return_value.__aenter__.return_value assert pipeline.rpush.call_count == 1 + pushed_msg_dict = json.loads(pipeline.rpush.call_args.args[1]) + assert pushed_msg_dict["role"] == "assistant" + + async def test_repeated_user_turn_is_not_deduplicated(self, mock_redis_client: MagicMock): + previous = [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["first reply"])] + mock_redis_client.lrange = AsyncMock(return_value=[json.dumps(msg.to_dict()) for msg in previous]) + + with patch("agent_framework_redis._history_provider.redis.from_url") as mock_from_url: + mock_from_url.return_value = mock_redis_client + provider = RedisHistoryProvider("mem", redis_url="redis://localhost:6379", application_id="test-app") + + await provider.save_messages( + "s1", [Message(role="user", contents=["yes"]), Message(role="assistant", contents=["second reply"])] + ) + + pipeline = mock_redis_client.pipeline.return_value.__aenter__.return_value + assert [json.loads(call.args[1])["contents"][0]["text"] for call in pipeline.rpush.await_args_list] == [ + "yes", + "second reply", + ] async def test_trimmed_messages_not_reappended(self, mock_redis_client: MagicMock): """Messages trimmed by max_messages should not be re-appended