From 8e8ebacb0344e120e736f7d62940aa148be3ba53 Mon Sep 17 00:00:00 2001 From: bunnysayzz Date: Sat, 5 Sep 2026 20:53:43 +0000 Subject: [PATCH 1/3] fix(memory): drain pending saves before forget() deletes records `forget()` called `self._storage.delete()` without first draining the pending-save queue, so a `remember_many()` background save submitted before `forget()` could land after the delete and resurrect the forgotten content. `recall()` already has the correct write barrier (`self.drain_writes()` at line 713); this adds the same barrier to `forget()` before the scope resolution. Adds three regression tests in `tests/memory/test_memory_forget_drain.py` that verify `drain_writes()` is called before `_storage.delete()` in `forget()`, and that the ordering matches `recall()`. Fixes #7290 --- .../src/crewai/memory/unified_memory.py | 3 + .../tests/memory/test_memory_forget_drain.py | 102 ++++++++++++++++++ 2 files changed, 105 insertions(+) create mode 100644 lib/crewai/tests/memory/test_memory_forget_drain.py diff --git a/lib/crewai/src/crewai/memory/unified_memory.py b/lib/crewai/src/crewai/memory/unified_memory.py index dcd5383ceb..94fbd76df7 100644 --- a/lib/crewai/src/crewai/memory/unified_memory.py +++ b/lib/crewai/src/crewai/memory/unified_memory.py @@ -836,6 +836,9 @@ def forget( Returns: Number of records deleted. """ + # Write barrier: drain pending background saves before deleting, + # so a save submitted before forget() cannot resurrect deleted content. + self.drain_writes() effective_scope = scope if effective_scope is None and self.root_scope: effective_scope = self.root_scope diff --git a/lib/crewai/tests/memory/test_memory_forget_drain.py b/lib/crewai/tests/memory/test_memory_forget_drain.py new file mode 100644 index 0000000000..1d1fdab664 --- /dev/null +++ b/lib/crewai/tests/memory/test_memory_forget_drain.py @@ -0,0 +1,102 @@ +""" +Regression test for crewAI issue #7290: +Memory.forget() must drain pending background saves before deleting, +otherwise a save submitted before forget() can resurrect deleted content. +""" +import threading +import time +from unittest.mock import MagicMock, patch +import pytest + + +def test_forget_drains_pending_saves_before_delete(): + """ + forget() must call drain_writes() before self._storage.delete(). + Without the fix, a background save submitted before forget() can land + after the delete and resurrect the forgotten content. + """ + from crewai.memory.unified_memory import Memory + + # Track call order + call_order = [] + + # Create a Memory instance with a mock storage + mem = Memory.__new__(Memory) + mem.__pydantic_private__ = {} + + # Patch drain_writes and _storage.delete to record call order + original_forget = Memory.forget + + drain_called = [] + delete_called = [] + + def mock_drain(self): + drain_called.append("drain") + call_order.append("drain_writes") + + def mock_delete(self, **kwargs): + delete_called.append("delete") + call_order.append("storage.delete") + return 1 + + with patch.object(Memory, "drain_writes", mock_drain), \ + patch.object(Memory, "_storage", create=True) as mock_storage: + mock_storage.delete = lambda **kwargs: (call_order.append("storage.delete"), 1)[1] + + # We need a real Memory instance - use a simpler approach + # Just verify the source code has drain_writes() before _storage.delete() + import inspect + source = inspect.getsource(Memory.forget) + lines = source.split("\n") + + drain_line = None + delete_line = None + for i, line in enumerate(lines): + if "self.drain_writes()" in line and drain_line is None: + drain_line = i + if "self._storage.delete(" in line and delete_line is None: + delete_line = i + + assert drain_line is not None, "forget() must call self.drain_writes()" + assert delete_line is not None, "forget() must call self._storage.delete()" + assert drain_line < delete_line, ( + f"drain_writes() (line {drain_line}) must come before " + f"_storage.delete() (line {delete_line}) in forget()" + ) + + +def test_recall_already_has_drain_writes(): + """Confirm recall() has the read barrier (regression guard).""" + from crewai.memory.unified_memory import Memory + import inspect + + source = inspect.getsource(Memory.recall) + assert "self.drain_writes()" in source, "recall() must call self.drain_writes()" + + +def test_forget_drain_ordering_matches_recall(): + """ + Both forget() and recall() should call drain_writes() before + touching storage, ensuring consistent write-barrier semantics. + """ + from crewai.memory.unified_memory import Memory + import inspect + + for method_name in ("forget", "recall"): + method = getattr(Memory, method_name) + source = inspect.getsource(method) + lines = source.split("\n") + + drain_line = next( + (i for i, l in enumerate(lines) if "self.drain_writes()" in l), None + ) + storage_line = next( + (i for i, l in enumerate(lines) + if "self._storage." in l and "drain" not in l), None + ) + + assert drain_line is not None, f"{method_name}() must call drain_writes()" + assert storage_line is not None, f"{method_name}() must access _storage" + assert drain_line < storage_line, ( + f"{method_name}(): drain_writes() must precede _storage access" + ) From 812a57e43eb46e92efd7b9444263be2dc66ee946 Mon Sep 17 00:00:00 2001 From: bunnysayzz Date: Sun, 6 Sep 2026 18:30:47 +0530 Subject: [PATCH 2/3] fix(memory): serialize forget() delete with save submission Hold _reset_lock across drain_writes() + _storage.delete(), mirroring reset(). _submit_save() registers under the same lock, so a save could otherwise register after the drain snapshot but before the delete, land after it, and resurrect forgotten content. Adds a deterministic test probing the lock from a helper thread (fails pre-fix, passes post-fix). --- .../src/crewai/memory/unified_memory.py | 32 ++++++++------- .../tests/memory/test_memory_forget_drain.py | 39 +++++++++++++++++++ 2 files changed, 58 insertions(+), 13 deletions(-) diff --git a/lib/crewai/src/crewai/memory/unified_memory.py b/lib/crewai/src/crewai/memory/unified_memory.py index 94fbd76df7..2047ed8ada 100644 --- a/lib/crewai/src/crewai/memory/unified_memory.py +++ b/lib/crewai/src/crewai/memory/unified_memory.py @@ -838,19 +838,25 @@ def forget( """ # Write barrier: drain pending background saves before deleting, # so a save submitted before forget() cannot resurrect deleted content. - self.drain_writes() - effective_scope = scope - if effective_scope is None and self.root_scope: - effective_scope = self.root_scope - elif effective_scope is not None and self.root_scope: - effective_scope = join_scope_paths(self.root_scope, effective_scope) - return self._storage.delete( - scope_prefix=effective_scope, - categories=categories, - record_ids=record_ids, - older_than=older_than, - metadata_filter=metadata_filter, - ) + # The _reset_lock must be held across the drain AND the delete: + # _submit_save() registers under the same lock, so without it a save + # could register after the drain snapshot but before the delete, land + # after the deletion, and resurrect the forgotten content. reset() + # already serializes the same way. + with self._reset_lock: + self.drain_writes() + effective_scope = scope + if effective_scope is None and self.root_scope: + effective_scope = self.root_scope + elif effective_scope is not None and self.root_scope: + effective_scope = join_scope_paths(self.root_scope, effective_scope) + return self._storage.delete( + scope_prefix=effective_scope, + categories=categories, + record_ids=record_ids, + older_than=older_than, + metadata_filter=metadata_filter, + ) def update( self, diff --git a/lib/crewai/tests/memory/test_memory_forget_drain.py b/lib/crewai/tests/memory/test_memory_forget_drain.py index 1d1fdab664..76aa353731 100644 --- a/lib/crewai/tests/memory/test_memory_forget_drain.py +++ b/lib/crewai/tests/memory/test_memory_forget_drain.py @@ -65,6 +65,45 @@ def mock_delete(self, **kwargs): ) +def test_forget_holds_reset_lock_across_drain_and_delete(): + """forget() must hold _reset_lock across the drain AND the delete. + + _submit_save() registers under _reset_lock. If forget() released the + lock between the drain snapshot and the delete, a concurrent save could + register in between, land after the deletion, and resurrect the + forgotten content. Probed deterministically: a helper thread attempts a + non-blocking acquire of _reset_lock while the delete runs; it must fail. + (A same-thread probe would succeed trivially: the lock is re-entrant.) + """ + from crewai.memory.unified_memory import Memory + + mem = Memory.model_construct() + mem._pending_saves = [] + mem.root_scope = None + + probe = {} + + def fake_delete(**kwargs): + def try_acquire(): + got = mem._reset_lock.acquire(blocking=False) + probe["acquired"] = got + if got: + mem._reset_lock.release() + + t = threading.Thread(target=try_acquire) + t.start() + t.join() + return 1 + + mem._storage = MagicMock() + mem._storage.delete.side_effect = fake_delete + + assert Memory.forget(mem) == 1 + assert probe.get("acquired") is False, ( + "forget() must hold _reset_lock while deleting, otherwise a " + "concurrent _submit_save() can slip between drain and delete and " + "resurrect forgotten content" + ) def test_recall_already_has_drain_writes(): """Confirm recall() has the read barrier (regression guard).""" from crewai.memory.unified_memory import Memory From 5a9fb222c30167b1c48c29ff9d1137929a8c7673 Mon Sep 17 00:00:00 2001 From: bunnysayzz Date: Sun, 6 Sep 2026 20:41:11 +0530 Subject: [PATCH 3/3] test(memory): probe _reset_lock during drain as well as delete CodeRabbit follow-up: the lock probe only covered the delete, so a regression draining outside the lock would still pass. Wrap drain_writes with the same helper-thread probe; assert both fail. --- .../tests/memory/test_memory_forget_drain.py | 32 ++++++++++++++----- 1 file changed, 24 insertions(+), 8 deletions(-) diff --git a/lib/crewai/tests/memory/test_memory_forget_drain.py b/lib/crewai/tests/memory/test_memory_forget_drain.py index 76aa353731..a704257f6f 100644 --- a/lib/crewai/tests/memory/test_memory_forget_drain.py +++ b/lib/crewai/tests/memory/test_memory_forget_drain.py @@ -72,8 +72,9 @@ def test_forget_holds_reset_lock_across_drain_and_delete(): lock between the drain snapshot and the delete, a concurrent save could register in between, land after the deletion, and resurrect the forgotten content. Probed deterministically: a helper thread attempts a - non-blocking acquire of _reset_lock while the delete runs; it must fail. - (A same-thread probe would succeed trivially: the lock is re-entrant.) + non-blocking acquire of _reset_lock while the drain runs and again while + the delete runs; both must fail. (A same-thread probe would succeed + trivially: the lock is re-entrant.) """ from crewai.memory.unified_memory import Memory @@ -83,23 +84,38 @@ def test_forget_holds_reset_lock_across_drain_and_delete(): probe = {} - def fake_delete(**kwargs): - def try_acquire(): + def try_acquire(key): + def _probe(): got = mem._reset_lock.acquire(blocking=False) - probe["acquired"] = got + probe[key] = got if got: mem._reset_lock.release() - t = threading.Thread(target=try_acquire) + t = threading.Thread(target=_probe) t.start() t.join() + + real_drain = Memory.drain_writes + + def probed_drain(self): + try_acquire("drain") + return real_drain(self) + + def fake_delete(**kwargs): + try_acquire("delete") return 1 mem._storage = MagicMock() mem._storage.delete.side_effect = fake_delete - assert Memory.forget(mem) == 1 - assert probe.get("acquired") is False, ( + with patch.object(Memory, "drain_writes", probed_drain): + assert Memory.forget(mem) == 1 + assert probe.get("drain") is False, ( + "forget() must hold _reset_lock while draining, otherwise a " + "concurrent _submit_save() can register after the snapshot and " + "resurrect forgotten content" + ) + assert probe.get("delete") is False, ( "forget() must hold _reset_lock while deleting, otherwise a " "concurrent _submit_save() can slip between drain and delete and " "resurrect forgotten content"