Python: fix: prevent superlinear history growth by deduplicating messages in save_messages

python/packages/core/agent_framework/_sessions.py

@@ -35,7 +35,7 @@
 from dataclasses import dataclass
 from functools import partial
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeAlias, TypeVar, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeAlias, TypeVar, cast, TypeGuard
 
 import msgspec
 
@@ -68,6 +68,7 @@
 JsonDumps: TypeAlias = Callable[[Any], str | bytes]
 JsonLoads: TypeAlias = Callable[[str | bytes], Any]
 ServiceSessionId: TypeAlias = Mapping[str, Any]
+MessageIdentity: TypeAlias = tuple[str, ...]
 StateT = TypeVar("StateT")
 StateEncoder: TypeAlias = Callable[[Any], Mapping[str, Any]]
 StateDecoder: TypeAlias = Callable[[Mapping[str, Any]], Any]
@@ -187,14 +188,69 @@ def _deduplicate_origin_session_ids(origin_session_ids: Iterable[str]) -> list[s
     return unique_origin_session_ids
 
 
+def get_message_identity(message: Message) -> MessageIdentity:
+    """Return a stable identity for a message for deduplication.
+
+    Uses the message's ID if available, otherwise falls back to a hash of
+    its role and serialized contents to prevent duplicate persistence.
+    """
+    msg_id = getattr(message, "message_id", None)
+    if msg_id is None:
+        msg_id = getattr(message, "id", None)
+    if msg_id is not None:
+        return ("id", str(msg_id))
+
+    try:
+        contents_data = [c.to_dict() for c in message.contents] if message.contents else []
+        serialized = json.dumps(contents_data, sort_keys=True, ensure_ascii=False)
+        return ("content", str(message.role), serialized)
+    except Exception:
+        return ("content", str(message.role), str(message.contents))
+
+
+def _get_message_hash(message: Message) -> MessageIdentity:
+    """Stable hash for sequence matching."""
+    return get_message_identity(message)
+
+
+def filter_new_messages(existing: Sequence[Message], incoming: Sequence[Message]) -> list[Message]:
+    """Filters incoming messages to only those that are truly new.
+
+    Handles both 'append-only' and 'full transcript replay' scenarios.
+    Prevents superlinear growth and preserves legitimate duplicate turns.
+    """
+    if not existing:
+        return list(incoming)
+
+    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) :])
+
+    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
+
+
 @dataclass(frozen=True, slots=True)
 class _StateTypeRegistration:
     cls: type[Any]
     type_id: str
     encoder: StateEncoder
     decoder: StateDecoder
 
-
 _STATE_TYPE_REGISTRY: dict[str, _StateTypeRegistration] = {}
 _STATE_CLASS_REGISTRY: dict[type[Any], _StateTypeRegistration] = {}
 
@@ -2095,10 +2151,13 @@ async def save_messages(
     ) -> None:
         """Persist messages to session state."""
         mark_feature_used(FeatureIndex.CORE_IN_MEMORY_HISTORY_PROVIDER)
-        if state is None:
+        if state is None or not messages:
             return
         existing = state.get("messages", [])
-        state["messages"] = [*existing, *messages]
+        new_messages = filter_new_messages(existing, messages)
+
+        if new_messages:
+            state["messages"] = [*existing, *new_messages]
 
 
 @experimental(feature_id=ExperimentalFeature.FILE_HISTORY)
@@ -2257,9 +2316,26 @@ async def save_messages(
         def _append_messages() -> None:
             with file_lock:
                 if self.serialization_format == "json":
-                    with file_path.open("a", encoding="utf-8") as file_handle:
-                        for message in messages:
-                            file_handle.write(f"{self._serialize_json_message(message)}\n")
+                    existing_messages: list[Message] = []
+                    if file_path.exists():
+                        with file_path.open("r", encoding="utf-8") as f:
+                            for line in f:
+                                line = line.strip()
+                                if not line:
+                                    continue
+                                try:
+                                    payload = self.loads(line)
+                                    msg = Message.from_dict(dict(cast(Mapping[str, Any], payload)))
+                                    existing_messages.append(msg)
+                                except Exception:
+                                    logger.debug("failed to parse history line for deduplication")
+                                    continue
+
+                    new_messages = filter_new_messages(existing_messages, messages)
+                    if new_messages:
+                        with file_path.open("a", encoding="utf-8") as file_handle:
+                            for message in new_messages:
+                                file_handle.write(f"{self._serialize_json_message(message)}\n")
                     return
                 with file_path.open("ab") as file_handle:
                     for message in messages:

python/packages/core/tests/core/test_agent_hooks.py

@@ -1893,7 +1893,7 @@ async def test_drained_and_discarded_attempt_flushes_on_allow(streaming: bool) -
     expected_response = "update - hello there" if streaming else "test response - hello there"
     assert final.text == expected_response
     stored = cast("list[Message]", session.state[provider.source_id]["messages"])
-    assert [message.text for message in stored] == ["hello there", expected_response] * 2
+    assert [message.text for message in stored] == ["hello there", expected_response]
 
 
 class _DrainThenTerminateWithoutResultMiddleware(AgentMiddleware):

python/packages/core/tests/core/test_middleware_with_agent.py

@@ -1778,14 +1778,14 @@ 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"] == 4  # Previous history + current input + current response
+        assert second_after["thread_count"] == 3  # Previous history (2) + current input (1)
         assert second_after["messages_text"] == ["second message"]
-        # Thread should contain: first input + first response + second input + second response
+        # 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"]
-        # Should have two "test response" entries (one for each run)
+        # "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 == 2
+        assert response_count == 1
 
 
 class TestChatAgentChatMiddleware:

python/packages/core/tests/core/test_sessions.py

@@ -8,7 +8,7 @@
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any, cast
 from unittest.mock import patch
 
 import msgspec
@@ -47,6 +47,9 @@
 from agent_framework._telemetry import FeatureIndex
 from agent_framework.exceptions import MiddlewareException
 
+if TYPE_CHECKING:
+    from agent_framework._agents import SupportsAgentRun
+
 # ---------------------------------------------------------------------------
 # SessionContext tests
 # ---------------------------------------------------------------------------
@@ -1381,6 +1384,131 @@ async def test_source_id_attribution(self) -> None:
         ctx.extend_messages("custom-source", [Message(role="user", contents=["test"])])
         assert "custom-source" in ctx.context_messages
 
+    async def test_save_messages_deduplicates_identical_messages(self) -> None:
+        """Test that save_messages does not re-append messages already in the store."""
+        provider = InMemoryHistoryProvider()
+        state: dict[str, Any] = {}
+
+        msg1 = Message(role="user", contents=["hello"])
+        msg2 = Message(role="assistant", contents=["hi there"])
+
+        await provider.save_messages("s1", [msg1, msg2], state=state)
+        assert len(state["messages"]) == 2
+
+        await provider.save_messages("s1", [msg1, msg2], state=state)
+        assert len(state["messages"]) == 2
+
+    async def test_save_messages_only_appends_new_messages(self) -> None:
+        """Test that save_messages filters out old messages and only appends new ones"""
+        provider = InMemoryHistoryProvider()
+        state: dict[str, Any] = {}
+
+        msg1 = Message(role="user", contents=["hello"])
+        msg2 = Message(role="assistant", contents=["hi there"])
+        msg3 = Message(role="user", contents=["how are you?"])
+
+        await provider.save_messages("s1", [msg1, msg2], state=state)
+        assert len(state["messages"]) == 2
+
+        await provider.save_messages("s1", [msg1, msg2, msg3], state=state)
+        assert len(state["messages"]) == 3
+        assert state["messages"][2].text == "how are you?"
+
+    async def test_save_messages_different_roles_same_text_not_deduplicated(self) -> None:
+        """Test that messages with the same text but different roles are kept separate."""
+        provider = InMemoryHistoryProvider()
+        state: dict[str, Any] = {}
+
+        msg1 = Message(role="user", contents=["ping"])
+        msg2 = Message(role="assistant", contents=["ping"])
+
+        await provider.save_messages("s1", [msg1, msg2], state=state)
+        assert len(state["messages"]) == 2
+
+    async def test_save_messages_deduplication_with_none_state(self) -> None:
+        """Test that save_messages with None state does not raise."""
+        provider = InMemoryHistoryProvider()
+        msg = Message(role="user", contents=["hello"])
+        await provider.save_messages("s1", [msg], state=None)
+
+    async def test_full_loop_does_not_grow_superlinearly(self) -> None:
+        """Regression test: a multi-round looped run must not re-persist the whole
+        conversation on every round."""
+        from agent_framework import AgentResponse
+
+        provider = InMemoryHistoryProvider()
+        session = AgentSession()
+        provider_state = session.state.setdefault(provider.source_id, {})
+        ctx1 = SessionContext(session_id="s1", input_messages=[Message(role="user", contents=["turn 1"])])
+
+        await provider.before_run(
+            agent=cast("SupportsAgentRun", None),
+            session=session,
+            context=ctx1,
+            state=provider_state,
+        )
+        ctx1._response = AgentResponse(messages=[Message(role="assistant", contents=["reply 1"])])
+        await provider.after_run(
+            agent=cast("SupportsAgentRun", None),
+            session=session,
+            context=ctx1,
+            state=provider_state,
+        )
+
+        ctx2 = SessionContext(session_id="s1", input_messages=[Message(role="user", contents=["turn 2"])])
+        await provider.before_run(
+            agent=cast("SupportsAgentRun", None),
+            session=session,
+            context=ctx2,
+            state=provider_state,
+        )
+        ctx2._response = AgentResponse(messages=[Message(role="assistant", contents=["reply 2"])])
+        await provider.after_run(
+            agent=cast("SupportsAgentRun", None),
+            session=session,
+            context=ctx2,
+            state=provider_state,
+        )
+
+        stored = session.state[provider.source_id]["messages"]
+        assert len(stored) == 4
+        texts = [m.text for m in stored]
+        assert texts == ["turn 1", "reply 1", "turn 2", "reply 2"]
+
+    async def test_save_messages_preserves_duplicate_content(self) -> None:
+        """Two separate user 'yes' replies in the same batch must both be persisted."""
+        provider = InMemoryHistoryProvider()
+        state: dict[str, Any] = {}
+
+        yes_1 = Message(role="user", contents=["yes"])
+        yes_2 = Message(role="user", contents=["yes"])
+
+        await provider.save_messages("s1", [yes_1, yes_2], state=state)
+
+        assert len(state["messages"]) == 2
+        assert state["messages"][0].text == "yes"
+        assert state["messages"][1].text == "yes"
+
+
+    async def test_save_messages_handles_replayed_transcript_with_duplicates(self) -> None:
+        provider = InMemoryHistoryProvider()
+        state: dict[str, Any] = {}
+
+        msg_b = Message (role = "user", contents=["B"])
+        await provider.save_messages("s1", [msg_b], state = state)
+        assert len(state["messages"]) == 1
+
+        msg_a = Message(role="user", contents=["A"])
+        msg_c = Message(role="user", contents=["C"])
+        msg_b2 = Message(role="user", contents=["B"])
+        msg_d = Message(role="user", contents=["D"])
+
+        await provider.save_messages("s1", [msg_a, msg_b, msg_c, msg_b2, msg_d], state = state)
+
+        assert len(state["messages"]) == 4
+        texts = [m.text for m in state["messages"]]
+        assert texts == ["B", "C", "B", "D"]
+
 
 class TestFileHistoryProvider:
     def test_is_marked_experimental(self) -> None:
@@ -1665,6 +1793,81 @@ def tracked_open(path: Path, *args: Any, **kwargs: Any) -> Any:
         loaded = await provider.get_messages(session_id)
         assert [message.text for message in loaded] == ["first", "second"]
 
+    async def test_save_messages_deduplicates_identical_messages(self, tmp_path: Path) -> None:
+        """Test that FileHistoryProvider does not re-append already persisted messages."""
+        provider = FileHistoryProvider(tmp_path)
+
+        msg1 = Message(role="user", contents=["hello"])
+        msg2 = Message(role="assistant", contents=["hi there"])
+
+        await provider.save_messages("s1", [msg1, msg2])
+        loaded = await provider.get_messages("s1")
+        assert len(loaded) == 2
+
+        await provider.save_messages("s1", [msg1, msg2])
+        loaded = await provider.get_messages("s1")
+        assert len(loaded) == 2
+
+    async def test_save_messages_only_appends_new_messages(self, tmp_path: Path) -> None:
+        """Test that FileHistoryProvider filters out old messages and only appends new ones"""
+        provider = FileHistoryProvider(tmp_path)
+
+        msg1 = Message(role="user", contents=["hello"])
+        msg2 = Message(role="assistant", contents=["hi there"])
+        msg3 = Message(role="user", contents=["how are you?"])
+
+        await provider.save_messages("s1", [msg1, msg2])
+        loaded = await provider.get_messages("s1")
+        assert len(loaded) == 2
+
+        await provider.save_messages("s1", [msg1, msg2, msg3])
+        loaded = await provider.get_messages("s1")
+        assert len(loaded) == 3
+        assert loaded[2].text == "how are you?"
+
+    async def test_save_messages_different_roles_same_text_not_deduplicated(self, tmp_path: Path) -> None:
+        """Test that messages with the same text but different roles are kept separate."""
+        provider = FileHistoryProvider(tmp_path)
+
+        msg1 = Message(role="user", contents=["ping"])
+        msg2 = Message(role="assistant", contents=["ping"])
+
+        await provider.save_messages("s1", [msg1, msg2])
+        loaded = await provider.get_messages("s1")
+        assert len(loaded) == 2
+
+    async def test_deduplication_file_integrity(self, tmp_path: Path) -> None:
+        """Test that deduplication writes the correct number of lines to the JSONL file."""
+        provider = FileHistoryProvider(tmp_path)
+
+        msg1 = Message(role="user", contents=["hello"])
+        msg2 = Message(role="assistant", contents=["hi there"])
+        msg3 = Message(role="user", contents=["follow-up"])
+
+        await provider.save_messages("s1", [msg1, msg2])
+
+        session_file = provider._session_file_path("s1")
+        raw_lines = (await asyncio.to_thread(session_file.read_text, encoding="utf-8")).splitlines()
+        assert len(raw_lines) == 2
+
+        await provider.save_messages("s1", [msg1, msg2, msg3])
+        raw_lines = (await asyncio.to_thread(session_file.read_text, encoding="utf-8")).splitlines()
+        assert len(raw_lines) == 3
+
+    async def test_save_messages_preserves_duplicate_content(self, tmp_path: Path) -> None:
+        """Test that two identical user turns in the same batch are both persisted."""
+        provider = FileHistoryProvider(tmp_path)
+
+        yes_1 = Message(role="user", contents=["yes"])
+        yes_2 = Message(role="user", contents=["yes"])
+
+        await provider.save_messages("s1", [yes_1, yes_2])
+        loaded = await provider.get_messages("s1")
+
+        assert len(loaded) == 2
+        assert loaded[0].text == "yes"
+        assert loaded[1].text == "yes"
+
 
 # ---------------------------------------------------------------------------
 # Run-persistence gate tests