Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion plugins/memory/hindsight/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1699,7 +1699,15 @@ def _do_retain() -> None:
# Advance the append watermark only after the delta is queued, so a
# later retain doesn't re-ship turns we've already handed to the writer.
if update_mode == "append":
self._last_retained_turn_count = len(self._session_turns)
# Every buffered turn has now been shipped (the retain content was
# snapshotted into the closure above). Append retains only ever read
# the un-retained tail — sync_turn slices from the watermark and
# flush-on-switch flushes what's left — so drop the retained turns
# instead of letting the buffer grow for the whole session. Overwrite
# mode is deliberately untouched: it resends the full session each
# retain and must keep every turn.
self._session_turns.clear()
self._last_retained_turn_count = 0

def get_tool_schemas(self) -> List[Dict[str, Any]]:
if self._memory_mode == "context":
Expand Down
57 changes: 57 additions & 0 deletions tests/plugins/memory/test_hindsight_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -1169,3 +1169,60 @@ def test_blocked_upgrade_is_nonfatal_and_surfaces_reason(
assert len(calls) == 1 # attempted exactly once, init still completed
assert any("runtime installs are disabled" in r.getMessage()
for r in caplog.records)


class TestSessionTurnsBufferBounding:
"""`_session_turns` must not grow for the whole session in append mode.

Append retains ship only the delta since the last watermark, so a retained
turn is never read again and the buffer is trimmed to the un-retained tail.
Overwrite mode resends the whole session each retain and must keep every
turn, so it is deliberately left unbounded.
"""

def _drive(self, provider, monkeypatch, *, mode, n_turns, retain_every):
provider._auto_retain = True
provider._retain_every_n_turns = retain_every
monkeypatch.setattr(provider, "_ensure_writer", lambda: None)
monkeypatch.setattr(provider, "_register_atexit", lambda: None)
monkeypatch.setattr(provider, "_run_hindsight_operation", lambda op: None)
monkeypatch.setattr(
provider, "_resolve_retain_target", lambda doc: (doc or "doc", mode)
)
# Run queued retain closures inline so the shipped content is captured
# without spinning up the writer thread.
shipped: list[str] = []
real_build = provider._build_retain_kwargs

def _capture(content, **kw):
shipped.append(content)
return real_build(content, **kw)

monkeypatch.setattr(provider, "_build_retain_kwargs", _capture)
q = MagicMock()
q.put = lambda fn: fn()
provider._retain_queue = q

for i in range(n_turns):
provider.sync_turn(f"user {i}", f"assistant {i}")
return shipped

def test_append_trims_retained_turns_without_dropping_data(self, provider, monkeypatch):
shipped = self._drive(provider, monkeypatch, mode="append", n_turns=6, retain_every=3)
# Buffer drained after the last retain; watermark reset.
assert provider._session_turns == []
assert provider._last_retained_turn_count == 0
# Two retains (turn 3 and 6), each shipping only its 3-turn delta — every
# turn shipped exactly once: none re-sent, none lost.
assert len(shipped) == 2
assert sum(len(json.loads(c)) for c in shipped) == 6

def test_append_buffer_stays_bounded_across_many_turns(self, provider, monkeypatch):
self._drive(provider, monkeypatch, mode="append", n_turns=103, retain_every=5)
# Holds only the un-retained tail (turns 101-103), not all 103.
assert len(provider._session_turns) == 3

def test_overwrite_keeps_full_buffer(self, provider, monkeypatch):
self._drive(provider, monkeypatch, mode=None, n_turns=6, retain_every=3)
# Overwrite resends the whole session each retain, so every turn stays.
assert len(provider._session_turns) == 6
Loading