From b26062c9e1fa6563503d886e7881a964176ae440 Mon Sep 17 00:00:00 2001 From: draix Date: Mon, 11 May 2026 09:15:42 -0300 Subject: [PATCH] fix(hindsight): clear _session_turns after sync_turn on append mode (#23724) On the modern Hindsight API (>=0.5.0) sync_turn() uses update_mode='append' against a stable session-scoped document_id, so the server preserves prior content across retains. However, _session_turns was never cleared after a successful enqueue, so every subsequent retain re-shipped the full growing transcript. With retain_every_n_turns=N the server ended up with each turn appended ceil(total_turns/N) times, producing duplicate chunks and ~80% unnecessary extraction tokens. Clear _session_turns after building content only when update_mode='append'. On the legacy path (update_mode is None, per-process document_id, replace semantics) the full session must keep being resent, so the buffer stays intact in that case. The session-switch flush path (on_session_switch) already snapshots and clears explicitly, so it is unaffected. --- plugins/memory/hindsight/__init__.py | 13 ++++ .../plugins/memory/test_hindsight_provider.py | 68 +++++++++++++++++++ 2 files changed, 81 insertions(+) diff --git a/plugins/memory/hindsight/__init__.py b/plugins/memory/hindsight/__init__.py index 20772844f16e6..fac188e1dfc59 100644 --- a/plugins/memory/hindsight/__init__.py +++ b/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, diff --git a/tests/plugins/memory/test_hindsight_provider.py b/tests/plugins/memory/test_hindsight_provider.py index fcda46e56b091..6bb5aac4a6051 100644 --- a/tests/plugins/memory/test_hindsight_provider.py +++ b/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()