fix(hindsight): clear _session_turns after sync_turn to send only deltas on append mode

plugins/memory/hindsight/__init__.py

@@ -1453,6 +1453,19 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
         retain_async_flag = self._retain_async
         retain_context = self._retain_context
 
+        # On the modern API (update_mode='append'), the server preserves
+        # prior document content across retains, so the next retain only
+        # needs to ship NEW turns. Without this clear, every retain would
+        # re-append the full growing transcript and the document would
+        # accumulate duplicates of every earlier turn (#23724).
+        #
+        # On the legacy path (update_mode is None, per-process document_id,
+        # replace semantics), each retain overwrites the document, so we
+        # MUST keep the full session buffer to avoid losing earlier turns
+        # on the next retain. Leave _session_turns intact in that case.
+        if update_mode == "append":
+            self._session_turns = []
+
         def _do_retain() -> None:
             item = self._build_retain_kwargs(
                 content,

tests/plugins/memory/test_hindsight_provider.py

@@ -1117,6 +1117,74 @@ def test_modern_api_uses_stable_doc_id_with_append(self, provider, monkeypatch):
         item = kw["items"][0]
         assert item["update_mode"] == "append"
 
+    def test_modern_api_sends_only_new_turns_on_subsequent_retain(
+        self, provider_with_config, monkeypatch
+    ):
+        """With update_mode='append' the server preserves prior content,
+        so each retain must ship ONLY the turns accumulated since the last
+        retain. Resending the full growing transcript would duplicate
+        every earlier turn on the server (#23724)."""
+        self._clear_capability_cache()
+        monkeypatch.setattr(
+            "plugins.memory.hindsight._fetch_hindsight_api_version",
+            lambda *a, **kw: "0.5.6",
+        )
+        p = provider_with_config(retain_every_n_turns=2)
+
+        # First retain batch: turns 1-2 should be shipped.
+        p.sync_turn("turn1-user", "turn1-asst")
+        p.sync_turn("turn2-user", "turn2-asst")
+        p._retain_queue.join()
+        first_content = p._client.aretain_batch.call_args.kwargs["items"][0]["content"]
+        assert "turn1-user" in first_content
+        assert "turn2-user" in first_content
+
+        p._client.aretain_batch.reset_mock()
+
+        # Second retain batch: only turns 3-4 should be shipped — the
+        # server already has turns 1-2 from the previous append.
+        p.sync_turn("turn3-user", "turn3-asst")
+        p.sync_turn("turn4-user", "turn4-asst")
+        p._retain_queue.join()
+        second_content = p._client.aretain_batch.call_args.kwargs["items"][0]["content"]
+        assert "turn3-user" in second_content
+        assert "turn4-user" in second_content
+        assert "turn1-user" not in second_content
+        assert "turn2-user" not in second_content
+
+    def test_legacy_api_still_sends_full_session_on_each_retain(
+        self, provider_with_config, monkeypatch
+    ):
+        """On legacy servers (no update_mode support) each retain replaces
+        the per-process document, so we MUST keep resending the full
+        accumulated session. Otherwise the second retain would overwrite
+        the document with only the new turns and lose earlier content."""
+        self._clear_capability_cache()
+        monkeypatch.setattr(
+            "plugins.memory.hindsight._fetch_hindsight_api_version",
+            lambda *a, **kw: None,
+        )
+        p = provider_with_config(retain_every_n_turns=2)
+
+        p.sync_turn("turn1-user", "turn1-asst")
+        p.sync_turn("turn2-user", "turn2-asst")
+        p._retain_queue.join()
+
+        p._client.aretain_batch.reset_mock()
+
+        p.sync_turn("turn3-user", "turn3-asst")
+        p.sync_turn("turn4-user", "turn4-asst")
+        p._retain_queue.join()
+
+        kw = p._client.aretain_batch.call_args.kwargs
+        content = kw["items"][0]["content"]
+        # Legacy path: no update_mode, replace-on-write, full session each time.
+        assert "update_mode" not in kw["items"][0]
+        assert "turn1-user" in content
+        assert "turn2-user" in content
+        assert "turn3-user" in content
+        assert "turn4-user" in content
+
     def test_capability_cached_per_url(self, provider, monkeypatch):
         """The /version probe must run at most once per (process, api_url)."""
         self._clear_capability_cache()