From b8182791dc28b7b5e4969bde70eb2c229fe0e965 Mon Sep 17 00:00:00 2001 From: Vesna-9 <290871358+Vesna-9@users.noreply.github.com> Date: Mon, 8 Jun 2026 10:58:32 +0300 Subject: [PATCH] fix(hindsight): flush only the un-retained delta on session switch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In append-capable mode (Hindsight >= 0.5.0) `sync_turn` ships only the turns past `_last_retained_turn_count` on each retain and advances that watermark, so already-appended turns are never re-sent. The flush-on-switch path in `on_session_switch`, however, snapshotted the full `_session_turns` (`old_turns = list(self._session_turns)`) and re-appended every turn under `update_mode='append'`, ignoring the watermark. With the default `retain_every_n_turns=1` every turn has already been appended by the time the session rotates, so each `/resume`, `/branch`, `/new`, `/reset`, and context compression appended a duplicate copy of the entire session to the stored document — silent memory-store corruption that skews recall and reflect. The flush now mirrors `sync_turn`: in append mode it sends only `self._session_turns[self._last_retained_turn_count:]` and skips the retain entirely when that delta is empty, while the legacy/overwrite path still resends the whole session (each retain replaces the document). `message_count` metadata is derived from the trimmed delta. ## What does this PR do? Fixes duplicate-turn accumulation in the Hindsight memory provider. The flush that runs when the agent rotates its `session_id` was re-appending the whole buffered session even though `sync_turn` had already appended each turn incrementally, so the backing document grew a fresh duplicate of the conversation on every session switch. Before: switching sessions in append mode re-appended every turn already written by `sync_turn`; with `retain_every_n_turns=1` that meant a full duplicate of the session per `/resume`, `/branch`, `/new`, `/reset`, and compression. After: the flush sends only the turns past `_last_retained_turn_count`, and fires no retain at all when the watermark has caught up to the buffer. The legacy/overwrite path is unchanged. ## Related Issue N/A ## Type of Change - [x] 🐛 Bug fix (non-breaking change that fixes an issue) ## Changes Made - `plugins/memory/hindsight/__init__.py`: in `on_session_switch`, resolve the old session's `update_mode` before computing the flush payload; in append mode slice `_session_turns[self._last_retained_turn_count:]` instead of copying the full buffer, skip the retain when the delta is empty, and size `message_count` from the trimmed delta. The legacy/overwrite branch still flushes the entire session. - `tests/plugins/memory/test_hindsight_provider.py`: add `test_session_switch_does_not_reflush_already_appended_turns` (every turn already retained -> switch fires no flush) and `test_session_switch_flush_ships_only_unretained_delta_in_append_mode` (partially-buffered -> flush carries only the un-retained tail, never the already-appended turns). ## How to Test 1. Run the targeted regression tests: `scripts/run_tests.sh tests/plugins/memory/test_hindsight_provider.py` 2. Both new tests pass against the fix; reverting the slice back to `old_turns = list(self._session_turns)` makes them fail because the flush re-ships `turn1`/`turn2` under `update_mode='append'`. 3. Full file is green: 104 passed. ## Checklist ### Code - [x] I've read the [Contributing Guide](https://github.com/NousResearch/hermes-agent/blob/main/CONTRIBUTING.md) - [x] My commit messages follow [Conventional Commits](https://www.conventionalcommits.org/) (`fix(scope):`, `feat(scope):`, etc.) - [x] I searched for [existing PRs](https://github.com/NousResearch/hermes-agent/pulls) to make sure this isn't a duplicate - [x] My PR contains **only** changes related to this fix/feature (no unrelated commits) - [x] I've run `pytest tests/ -q` and all tests pass - [x] I've added tests for my changes (required for bug fixes, strongly encouraged for features) - [x] I've tested on my platform: macOS 15 (Darwin 25.5) ### Documentation & Housekeeping - [x] I've updated relevant documentation (README, `docs/`, docstrings) — or N/A - [x] I've updated `cli-config.yaml.example` if I added/changed config keys — or N/A - [x] I've updated `CONTRIBUTING.md` or `AGENTS.md` if I changed architecture or workflows — or N/A - [x] I've considered cross-platform impact (Windows, macOS) per the [compatibility guide](https://github.com/NousResearch/hermes-agent/blob/main/CONTRIBUTING.md#cross-platform-compatibility) — or N/A - [x] I've updated tool descriptions/schemas if I changed tool behavior — or N/A --- plugins/memory/hindsight/__init__.py | 117 ++++++++++-------- .../plugins/memory/test_hindsight_provider.py | 74 +++++++++++ 2 files changed, 140 insertions(+), 51 deletions(-) diff --git a/plugins/memory/hindsight/__init__.py b/plugins/memory/hindsight/__init__.py index 53f422b2d7c6c..562b4de3c64cb 100644 --- a/plugins/memory/hindsight/__init__.py +++ b/plugins/memory/hindsight/__init__.py @@ -1653,20 +1653,6 @@ def on_session_switch( # everything before mutating self._* so metadata + tags + doc_id # all reference the old session consistently. if self._session_turns: - old_turns = list(self._session_turns) - old_session_id = self._session_id - old_parent_session_id = self._parent_session_id - old_turn_index = self._turn_index - old_metadata = self._build_metadata( - message_count=len(old_turns) * 2, - turn_index=old_turn_index, - ) - old_lineage_tags: list[str] = [] - if old_session_id: - old_lineage_tags.append(f"session:{old_session_id}") - if old_parent_session_id: - old_lineage_tags.append(f"parent:{old_parent_session_id}") - old_content = "[" + ",".join(old_turns) + "]" # Resolve doc_id + update_mode against the OLD session BEFORE # we rotate _session_id, so the flush lands in the old # session's document either way (legacy: per-process unique; @@ -1674,44 +1660,73 @@ def on_session_switch( old_document_id, old_update_mode = self._resolve_retain_target( self._document_id ) - - def _flush(): - try: - item = self._build_retain_kwargs( - old_content, - context=self._retain_context, - metadata=old_metadata, - tags=old_lineage_tags or None, - ) - item.pop("bank_id", None) - item.pop("retain_async", None) - if old_update_mode is not None: - item["update_mode"] = old_update_mode - logger.debug( - "Hindsight flush-on-switch: bank=%s, doc=%s, mode=%s, num_turns=%d", - self._bank_id, old_document_id, old_update_mode, len(old_turns), - ) - self._run_hindsight_operation( - lambda client: client.aretain_batch( - bank_id=self._bank_id, - items=[item], - document_id=old_document_id, - retain_async=self._retain_async, + # In append mode each sync_turn already shipped its delta and + # advanced _last_retained_turn_count, so the flush must send only + # the turns past that watermark — re-sending the whole buffer would + # append duplicate copies of already-retained turns to the document + # (the delta invariant sync_turn enforces). On legacy/overwrite the + # document is replaced each retain, so the flush must carry the + # entire session. + if old_update_mode == "append": + old_turns = self._session_turns[self._last_retained_turn_count:] + else: + old_turns = list(self._session_turns) + + # In append mode the whole buffer may already be retained (the + # watermark caught up), leaving nothing to flush. Skip the retain + # entirely rather than ship an empty payload. + if old_turns: + old_session_id = self._session_id + old_parent_session_id = self._parent_session_id + old_turn_index = self._turn_index + old_metadata = self._build_metadata( + message_count=len(old_turns) * 2, + turn_index=old_turn_index, + ) + old_lineage_tags: list[str] = [] + if old_session_id: + old_lineage_tags.append(f"session:{old_session_id}") + if old_parent_session_id: + old_lineage_tags.append(f"parent:{old_parent_session_id}") + old_content = "[" + ",".join(old_turns) + "]" + + def _flush(): + try: + item = self._build_retain_kwargs( + old_content, + context=self._retain_context, + metadata=old_metadata, + tags=old_lineage_tags or None, ) - ) - except Exception as e: - logger.warning("Hindsight flush-on-switch failed: %s", e, exc_info=True) - - # Route the flush through the same writer queue sync_turn - # uses. That serializes it behind any still-queued retains - # from the old session (FIFO by document_id), avoids racing - # two threads on aretain_batch against the same document, and - # keeps shutdown's drain semantics intact. Skip enqueue if - # shutdown has already fired — the writer is draining/gone. - if not self._shutting_down.is_set(): - self._ensure_writer() - self._register_atexit() - self._retain_queue.put(_flush) + item.pop("bank_id", None) + item.pop("retain_async", None) + if old_update_mode is not None: + item["update_mode"] = old_update_mode + logger.debug( + "Hindsight flush-on-switch: bank=%s, doc=%s, mode=%s, num_turns=%d", + self._bank_id, old_document_id, old_update_mode, len(old_turns), + ) + self._run_hindsight_operation( + lambda client: client.aretain_batch( + bank_id=self._bank_id, + items=[item], + document_id=old_document_id, + retain_async=self._retain_async, + ) + ) + except Exception as e: + logger.warning("Hindsight flush-on-switch failed: %s", e, exc_info=True) + + # Route the flush through the same writer queue sync_turn + # uses. That serializes it behind any still-queued retains + # from the old session (FIFO by document_id), avoids racing + # two threads on aretain_batch against the same document, and + # keeps shutdown's drain semantics intact. Skip enqueue if + # shutdown has already fired — the writer is draining/gone. + if not self._shutting_down.is_set(): + self._ensure_writer() + self._register_atexit() + self._retain_queue.put(_flush) # 2. Drain any in-flight prefetch from the old session and drop # its cached result so the new session doesn't see stale recall. diff --git a/tests/plugins/memory/test_hindsight_provider.py b/tests/plugins/memory/test_hindsight_provider.py index a7ca66f73f4d5..ef83a2e528665 100644 --- a/tests/plugins/memory/test_hindsight_provider.py +++ b/tests/plugins/memory/test_hindsight_provider.py @@ -1245,6 +1245,80 @@ def test_session_switch_flush_picks_capability_against_old_session( assert kw["document_id"] == "test-session" assert kw["items"][0]["update_mode"] == "append" + def test_session_switch_does_not_reflush_already_appended_turns( + self, provider_with_config, monkeypatch + ): + """Regression: in append mode sync_turn already ships each turn as a + delta, so on_session_switch must NOT re-append the whole session. + + With the default retain_every_n_turns=1 every turn is retained the + moment it lands and the delta watermark catches up to the buffer. + The flush-on-switch then has nothing past the watermark to send, so + no retain should fire at all. Before the fix it snapshotted the full + _session_turns and re-appended every already-retained turn, silently + duplicating them in the stored document on every /resume, /branch, + /new, /reset and context compression. + """ + 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=1, retain_async=False) + p.sync_turn("turn1-user", "turn1-asst") + p.sync_turn("turn2-user", "turn2-asst") + p._retain_queue.join() + + # Every buffered turn is already appended — watermark == buffer len. + assert p._last_retained_turn_count == len(p._session_turns) == 2 + + p._client.aretain_batch.reset_mock() + + p.on_session_switch("new-sid", parent_session_id="test-session", reset=True) + p._retain_queue.join() + + # No turns past the watermark → no flush retain. Without the fix this + # would re-ship [turn1, turn2] under update_mode='append'. + p._client.aretain_batch.assert_not_called() + self._clear_capability_cache() + + def test_session_switch_flush_ships_only_unretained_delta_in_append_mode( + self, provider_with_config, monkeypatch + ): + """When retain_every_n_turns>1 some turns are retained and some are + still buffered at switch time. The flush must carry only the buffered + delta past the watermark, never the already-appended turns.""" + 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, retain_async=False) + # Turns 1+2 hit the boundary → retained as a delta; watermark = 2. + p.sync_turn("turn1-user", "turn1-asst") + p.sync_turn("turn2-user", "turn2-asst") + # Turn 3 is buffered, below the next boundary, so not yet retained. + p.sync_turn("turn3-user", "turn3-asst") + p._retain_queue.join() + assert p._last_retained_turn_count == 2 + assert len(p._session_turns) == 3 + + p._client.aretain_batch.reset_mock() + + p.on_session_switch("new-sid", parent_session_id="test-session", reset=True) + p._retain_queue.join() + + p._client.aretain_batch.assert_called_once() + item = p._client.aretain_batch.call_args.kwargs["items"][0] + assert item["update_mode"] == "append" + # Only the un-retained turn 3 — turns 1+2 were already appended. + assert "turn3-user" in item["content"] + assert "turn1-user" not in item["content"] + assert "turn2-user" not in item["content"] + # message_count reflects only the flushed delta (1 turn -> 2 messages). + assert item["metadata"]["message_count"] == "2" + self._clear_capability_cache() + # --------------------------------------------------------------------------- # System prompt tests