import logging import time from datetime import datetime, timezone from types import SimpleNamespace from unittest.mock import MagicMock, Mock import pytest from mem0.exceptions import LLMError from mem0.memory.main import AsyncMemory, Memory def _setup_mocks(mocker): """Helper to setup common mocks for both sync and async fixtures""" mock_embedder = mocker.MagicMock() mock_embedder.return_value.embed.return_value = [0.1, 0.2, 0.3] mocker.patch("mem0.utils.factory.EmbedderFactory.create", mock_embedder) mock_vector_store = mocker.MagicMock() mock_vector_store.return_value.search.return_value = [] mocker.patch( "mem0.utils.factory.VectorStoreFactory.create", side_effect=[mock_vector_store.return_value, mocker.MagicMock()] ) mock_llm = mocker.MagicMock() mocker.patch("mem0.utils.factory.LlmFactory.create", mock_llm) mocker.patch("mem0.memory.storage.SQLiteManager", mocker.MagicMock()) return mock_llm, mock_vector_store class TestAddToVectorStoreErrors: @pytest.fixture def mock_memory(self, mocker): """Fixture that returns a Memory instance with mocker-based mocks""" mock_llm, _ = _setup_mocks(mocker) memory = Memory() memory.config = mocker.MagicMock() memory.config.custom_instructions = None memory.config.custom_update_memory_prompt = None memory.custom_instructions = None memory.api_version = "v1.1" # v3 pipeline needs db.get_last_messages to return a list memory.db.get_last_messages = MagicMock(return_value=[]) memory.db.save_messages = MagicMock() return memory def test_empty_llm_response_fact_extraction(self, mocker, mock_memory, caplog): """Test invalid JSON response from LLM during extraction""" # Setup mock_memory.llm.generate_response.return_value = "invalid json" mocker.patch("mem0.memory.main.capture_event") # Execute with caplog.at_level(logging.ERROR): result = mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, filters={}, infer=True ) # Verify — v3 single-pass pipeline makes 1 LLM call, returns [] on parse error assert mock_memory.llm.generate_response.call_count == 1 assert result == [] assert any("Error parsing extraction response" in record.message for record in caplog.records), "Expected error message not found in logs" def test_empty_llm_response_memory_actions(self, mock_memory, caplog): """Test empty response from LLM during memory actions (v3: single-pass, 1 LLM call)""" # Setup — v3 pipeline does a single LLM call that returns empty/invalid response mock_memory.llm.generate_response.return_value = "" # Execute with caplog.at_level(logging.WARNING): result = mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, filters={}, infer=True ) # Verify — v3 only makes 1 LLM call (no separate merge step) assert mock_memory.llm.generate_response.call_count == 1 assert result == [] # Should return empty list when no memories processed def test_llm_extraction_exception_is_reraised(self, mocker, mock_memory): """A provider error during fact extraction must propagate, not be swallowed. Regression guard for the silent ``return []`` that made it impossible for callers to tell "LLM unavailable" (429/5xx/timeout) from "no facts found". Without the fix this raises AssertionError because the call returns []. """ class _ProviderError(Exception): pass mock_memory.llm.generate_response.side_effect = _ProviderError("429 rate limit") mocker.patch("mem0.memory.main.capture_event") with pytest.raises(LLMError) as exc_info: mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, filters={}, infer=True ) # The documented LLMError contract is honoured, and the original # provider exception is preserved as the cause for debugging. assert isinstance(exc_info.value.__cause__, _ProviderError) class TestPromptOverridesCustomInstructions: @pytest.fixture def mock_memory(self, mocker): mock_llm, _ = _setup_mocks(mocker) mock_llm.return_value.generate_response.return_value = '{"memory": []}' memory = Memory() memory.custom_instructions = "config-level instructions" memory.db.get_last_messages = MagicMock(return_value=[]) memory.db.save_messages = MagicMock() return memory def test_prompt_overrides_custom_instructions(self, mock_memory): mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "hello"}], metadata={}, filters={}, infer=True, prompt="per-call override", ) user_prompt = mock_memory.llm.generate_response.call_args[1]["messages"][1]["content"] assert "per-call override" in user_prompt assert "config-level instructions" not in user_prompt def test_falls_back_to_custom_instructions_when_no_prompt(self, mock_memory): mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "hello"}], metadata={}, filters={}, infer=True, ) user_prompt = mock_memory.llm.generate_response.call_args[1]["messages"][1]["content"] assert "config-level instructions" in user_prompt class TestAsyncUpdate: @pytest.fixture def mock_async_memory(self, mocker): """Fixture for AsyncMemory with mocker-based mocks""" _setup_mocks(mocker) memory = AsyncMemory() return memory @pytest.mark.asyncio async def test_async_update_without_metadata(self, mock_async_memory, mocker): """Test async update passes None metadata by default""" mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed = Mock(return_value=[0.1, 0.2, 0.3]) mock_async_memory._update_memory = mocker.AsyncMock() result = await mock_async_memory.update("test_id", "Updated memory") mock_async_memory._update_memory.assert_called_once_with( "test_id", "Updated memory", {"Updated memory": [0.1, 0.2, 0.3]}, None ) assert result["message"] == "Memory updated successfully!" @pytest.mark.asyncio async def test_async_update_with_metadata(self, mock_async_memory, mocker): """Test async update correctly forwards metadata""" mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed = Mock(return_value=[0.1, 0.2, 0.3]) mock_async_memory._update_memory = mocker.AsyncMock() metadata = {"category": "sports", "priority": "high"} result = await mock_async_memory.update("test_id", "Updated memory", metadata=metadata) mock_async_memory._update_memory.assert_called_once_with( "test_id", "Updated memory", {"Updated memory": [0.1, 0.2, 0.3]}, metadata ) assert result["message"] == "Memory updated successfully!" @pytest.mark.asyncio async def test_async_update_with_empty_metadata(self, mock_async_memory, mocker): """Test async update with empty metadata dict""" mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed = Mock(return_value=[0.1, 0.2, 0.3]) mock_async_memory._update_memory = mocker.AsyncMock() await mock_async_memory.update("test_id", "Updated memory", metadata={}) mock_async_memory._update_memory.assert_called_once_with( "test_id", "Updated memory", {"Updated memory": [0.1, 0.2, 0.3]}, {} ) @pytest.mark.asyncio async def test_async_update_data_is_deprecated_alias_for_text(self, mock_async_memory, mocker, caplog): mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed = Mock(return_value=[0.1, 0.2, 0.3]) mock_async_memory._update_memory = mocker.AsyncMock() # `data=` still works but emits a deprecation warning with caplog.at_level(logging.WARNING): await mock_async_memory.update("test_id", data="via data") assert any("deprecated" in record.message for record in caplog.records) mock_async_memory._update_memory.assert_called_once_with( "test_id", "via data", {"via data": [0.1, 0.2, 0.3]}, None ) # `text` takes precedence when both are passed mock_async_memory._update_memory.reset_mock() await mock_async_memory.update("test_id", text="preferred", data="ignored") mock_async_memory._update_memory.assert_called_once_with( "test_id", "preferred", {"preferred": [0.1, 0.2, 0.3]}, None ) @pytest.mark.asyncio async def test_async_update_can_change_expiration_date_without_changing_text(self, mock_async_memory, mocker): mock_async_memory.embedding_model.embed = Mock(return_value=[0.1, 0.2, 0.3]) mock_async_memory.vector_store.get = Mock( return_value=Mock( payload={ "data": "Existing memory", "user_id": "test_user", "created_at": "2026-01-01T00:00:00+00:00", "expiration_date": "2026-12-31", } ) ) mock_async_memory.vector_store.update = Mock() mock_async_memory.db.add_history = Mock() mock_async_memory._remove_memory_from_entity_store = mocker.AsyncMock() mock_async_memory._link_entities_for_memory = mocker.AsyncMock() result = await mock_async_memory.update("test_id", expiration_date="2999-01-01") assert result["message"] == "Memory updated successfully!" payload = mock_async_memory.vector_store.update.call_args.kwargs["payload"] assert payload["data"] == "Existing memory" assert payload["expiration_date"] == "2999-01-01" mock_async_memory._remove_memory_from_entity_store.assert_not_called() mock_async_memory._link_entities_for_memory.assert_not_called() @pytest.mark.asyncio class TestAsyncAddToVectorStoreErrors: @pytest.fixture def mock_async_memory(self, mocker): """Fixture for AsyncMemory with mocker-based mocks""" mock_llm, _ = _setup_mocks(mocker) memory = AsyncMemory() memory.config = mocker.MagicMock() memory.config.custom_instructions = None memory.config.custom_update_memory_prompt = None memory.custom_instructions = None memory.api_version = "v1.1" # v3 pipeline needs db.get_last_messages to return a list memory.db.get_last_messages = MagicMock(return_value=[]) memory.db.save_messages = MagicMock() return memory @pytest.mark.asyncio async def test_async_empty_llm_response_fact_extraction(self, mock_async_memory, caplog, mocker): """Test invalid JSON response from LLM during extraction (async)""" mocker.patch("mem0.utils.factory.EmbedderFactory.create", return_value=MagicMock()) mock_async_memory.llm.generate_response.return_value = "invalid json" mocker.patch("mem0.memory.main.capture_event") with caplog.at_level(logging.ERROR): result = await mock_async_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, effective_filters={}, infer=True ) assert mock_async_memory.llm.generate_response.call_count == 1 assert result == [] assert any("Error parsing extraction response" in record.message for record in caplog.records), "Expected error message not found in logs" @pytest.mark.asyncio async def test_async_empty_llm_response_memory_actions(self, mock_async_memory, caplog, mocker): """Test empty response in AsyncMemory._add_to_vector_store (v3: single-pass, 1 LLM call)""" mocker.patch("mem0.utils.factory.EmbedderFactory.create", return_value=MagicMock()) mock_async_memory.llm.generate_response.return_value = "" mock_capture_event = mocker.MagicMock() mocker.patch("mem0.memory.main.capture_event", mock_capture_event) with caplog.at_level(logging.WARNING): result = await mock_async_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, effective_filters={}, infer=True ) assert result == [] assert mock_async_memory.llm.generate_response.call_count == 1 @pytest.mark.asyncio async def test_async_llm_extraction_exception_is_reraised(self, mock_async_memory, mocker): """Async counterpart of the sync re-raise guard. A provider error during fact extraction must propagate as ``LLMError`` (with the original exception preserved as the cause), not be swallowed into ``return []``. Without the fix a future revert of the async ``raise`` back to ``return []`` would pass the suite silently. """ mocker.patch("mem0.utils.factory.EmbedderFactory.create", return_value=MagicMock()) class _ProviderError(Exception): pass mock_async_memory.llm.generate_response.side_effect = _ProviderError("429 rate limit") mocker.patch("mem0.memory.main.capture_event") with pytest.raises(LLMError) as exc_info: await mock_async_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, effective_filters={}, infer=True ) assert isinstance(exc_info.value.__cause__, _ProviderError) def _build_memory_instance(mocker, memory_cls): _setup_mocks(mocker) mocker.patch("mem0.memory.main.SQLiteManager", mocker.MagicMock()) mocker.patch("mem0.memory.main.MEM0_TELEMETRY", False) memory = memory_cls() memory.config = mocker.MagicMock() memory.config.custom_instructions = None memory.config.custom_update_memory_prompt = None memory.api_version = "v1.1" memory.vector_store = mocker.MagicMock() memory.db = mocker.MagicMock() return memory def _assert_utc_timestamp(timestamp: str): parsed = datetime.fromisoformat(timestamp) assert parsed.tzinfo == timezone.utc assert parsed.utcoffset().total_seconds() == 0 def test_create_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, Memory) memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] _assert_utc_timestamp(payload["created_at"]) def test_create_memory_sets_updated_at(mocker): memory = _build_memory_instance(mocker, Memory) memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert "updated_at" in payload assert payload["updated_at"] == payload["created_at"] _assert_utc_timestamp(payload["updated_at"]) # History should also receive updated_at history_kwargs = memory.db.add_history.call_args assert history_kwargs.kwargs["updated_at"] == payload["updated_at"] def test_create_memory_preserves_existing_created_at(mocker): memory = _build_memory_instance(mocker, Memory) custom_ts = "2023-05-06T09:19:20+00:00" memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={"created_at": custom_ts}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["created_at"] == custom_ts assert payload["updated_at"] == custom_ts def test_update_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, Memory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old memory", "created_at": "2026-03-17T17:00:00-07:00"} ) memory._update_memory("memory-id", "new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.update.call_args.kwargs["payload"] assert payload["created_at"] == "2026-03-17T17:00:00-07:00" assert payload["updated_at"] is not None @pytest.mark.asyncio async def test_async_create_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, AsyncMemory) await memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] _assert_utc_timestamp(payload["created_at"]) @pytest.mark.asyncio async def test_async_create_memory_sets_updated_at(mocker): memory = _build_memory_instance(mocker, AsyncMemory) await memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert "updated_at" in payload assert payload["updated_at"] == payload["created_at"] _assert_utc_timestamp(payload["updated_at"]) # History should also receive updated_at history_kwargs = memory.db.add_history.call_args assert history_kwargs.kwargs["updated_at"] == payload["updated_at"] @pytest.mark.asyncio async def test_async_create_memory_preserves_existing_created_at(mocker): memory = _build_memory_instance(mocker, AsyncMemory) custom_ts = "2023-05-06T09:19:20+00:00" await memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={"created_at": custom_ts}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["created_at"] == custom_ts assert payload["updated_at"] == custom_ts @pytest.mark.asyncio async def test_async_update_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, AsyncMemory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old memory", "created_at": "2026-03-17T17:00:00-07:00"} ) await memory._update_memory("memory-id", "new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.update.call_args.kwargs["payload"] assert payload["created_at"] == "2026-03-17T17:00:00-07:00" assert payload["updated_at"] is not None def test_create_then_search_and_get_all_return_same_timestamps(mocker): """Reproduces issue #3720: created_at must be identical in search() and get_all().""" memory = _build_memory_instance(mocker, Memory) # Step 1: Create a memory — capture the payload stored in the vector store memory._create_memory("Likes pizza", {"Likes pizza": [0.1, 0.2, 0.3]}, metadata={"user_id": "alice"}) stored_payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] stored_id = memory.vector_store.insert.call_args.kwargs["ids"][0] # Verify both timestamps were stored assert stored_payload["created_at"] is not None assert stored_payload["updated_at"] is not None assert stored_payload["updated_at"] == stored_payload["created_at"] # Step 2: Simulate the vector store returning this memory for both search and get_all mem_result = MagicMock() mem_result.id = stored_id mem_result.payload = stored_payload mem_result.score = 0.95 memory.vector_store.search.return_value = [mem_result] memory.vector_store.list.return_value = [[mem_result]] # Step 3: Call search and get_all, compare timestamps search_results = memory._search_vector_store("pizza", filters={"user_id": "alice"}, limit=10) get_all_results = memory._get_all_from_vector_store(filters={"user_id": "alice"}, limit=100) search_item = search_results[0] get_all_item = get_all_results[0] # The core assertion from issue #3720: created_at must be the same assert search_item["created_at"] == get_all_item["created_at"], ( f"created_at mismatch: search={search_item['created_at']}, get_all={get_all_item['created_at']}" ) assert search_item["updated_at"] == get_all_item["updated_at"], ( f"updated_at mismatch: search={search_item['updated_at']}, get_all={get_all_item['updated_at']}" ) # Neither should be None assert search_item["created_at"] is not None assert search_item["updated_at"] is not None assert get_all_item["created_at"] is not None assert get_all_item["updated_at"] is not None def test_update_preserves_created_at_and_updates_updated_at(mocker): """After an update, created_at must stay the same and updated_at must change.""" memory = _build_memory_instance(mocker, Memory) # Create a memory memory._create_memory("Likes pizza", {"Likes pizza": [0.1, 0.2, 0.3]}, metadata={"user_id": "alice"}) created_payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] created_id = memory.vector_store.insert.call_args.kwargs["ids"][0] original_created_at = created_payload["created_at"] # Update the memory — simulate existing memory in vector store memory.vector_store.get.return_value = MagicMock( id=created_id, payload=created_payload, ) memory._update_memory(created_id, "Loves pizza", {"Loves pizza": [0.2, 0.3, 0.4]}, metadata={}) updated_payload = memory.vector_store.update.call_args.kwargs["payload"] # created_at must be preserved assert updated_payload["created_at"] == original_created_at # updated_at must be set and different from creation time (or at least present) assert updated_payload["updated_at"] is not None _assert_utc_timestamp(updated_payload["updated_at"]) def test_search_and_get_all_consistent_after_update(mocker): """After update, search and get_all must still return the same timestamps.""" memory = _build_memory_instance(mocker, Memory) # Simulate a memory that was created then updated updated_payload = { "data": "Loves pizza", "hash": "abc123", "user_id": "alice", "created_at": "2023-05-06T09:19:20+00:00", "updated_at": "2026-03-23T10:00:00+00:00", } mem_result = MagicMock() mem_result.id = "mem-1" mem_result.payload = updated_payload mem_result.score = 0.9 memory.vector_store.search.return_value = [mem_result] memory.vector_store.list.return_value = [[mem_result]] search_results = memory._search_vector_store("pizza", filters={"user_id": "alice"}, limit=10) get_all_results = memory._get_all_from_vector_store(filters={"user_id": "alice"}, limit=100) assert search_results[0]["created_at"] == get_all_results[0]["created_at"] assert search_results[0]["updated_at"] == get_all_results[0]["updated_at"] # created_at should be the original, not the updated time assert search_results[0]["created_at"] == "2023-05-06T09:19:20+00:00" assert search_results[0]["updated_at"] == "2026-03-23T10:00:00+00:00" class TestMetadataNotMutated: """Tests that metadata dicts passed to memory methods are not mutated in-place (issue #2648).""" def test_create_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, Memory) original_metadata = {"user_id": "test_user", "category": "sports"} metadata_copy = original_metadata.copy() memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"_create_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) def test_create_memory_stores_correct_payload(self, mocker): memory = _build_memory_instance(mocker, Memory) metadata = {"user_id": "test_user", "category": "sports"} memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=metadata) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["data"] == "test data" assert payload["user_id"] == "test_user" assert payload["category"] == "sports" assert "hash" in payload assert "created_at" in payload def test_create_memory_with_none_metadata(self, mocker): memory = _build_memory_instance(mocker, Memory) memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=None) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["data"] == "test data" assert "hash" in payload def test_create_memory_shared_metadata_across_calls(self, mocker): """Verify that sharing a metadata dict between multiple _create_memory calls is safe.""" memory = _build_memory_instance(mocker, Memory) shared_metadata = {"user_id": "test_user"} memory._create_memory("first memory", {"first memory": [0.1, 0.2, 0.3]}, metadata=shared_metadata) memory._create_memory("second memory", {"second memory": [0.4, 0.5, 0.6]}, metadata=shared_metadata) assert shared_metadata == {"user_id": "test_user"}, "shared metadata was mutated across calls" # Verify each call got the correct data first_payload = memory.vector_store.insert.call_args_list[0].kwargs["payloads"][0] second_payload = memory.vector_store.insert.call_args_list[1].kwargs["payloads"][0] assert first_payload["data"] == "first memory" assert second_payload["data"] == "second memory" def test_create_memory_preserves_role_and_actor_id_in_history(self, mocker): """Verify that role and actor_id from metadata flow through to add_history after deepcopy.""" memory = _build_memory_instance(mocker, Memory) metadata = {"user_id": "test_user", "role": "assistant", "actor_id": "bot-1"} memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=metadata) # Verify the payload stored in vector store has all fields payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["role"] == "assistant" assert payload["actor_id"] == "bot-1" assert payload["user_id"] == "test_user" assert payload["data"] == "test data" # Verify add_history received the correct role and actor_id history_call = memory.db.add_history.call_args assert history_call.kwargs["role"] == "assistant" assert history_call.kwargs["actor_id"] == "bot-1" # And the original metadata is still untouched assert metadata == {"user_id": "test_user", "role": "assistant", "actor_id": "bot-1"} def test_create_memory_with_nested_metadata_not_mutated(self, mocker): """Verify deepcopy protects nested structures in metadata.""" memory = _build_memory_instance(mocker, Memory) metadata = {"user_id": "test_user", "tags": ["important", "urgent"], "config": {"key": "val"}} import copy metadata_snapshot = copy.deepcopy(metadata) memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=metadata) assert metadata == metadata_snapshot, "Nested metadata structures were mutated" def test_update_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, Memory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old data", "user_id": "test_user", "created_at": "2026-01-01T00:00:00+00:00"} ) original_metadata = {"category": "updated"} metadata_copy = original_metadata.copy() memory._update_memory("mem-id", "new data", {"new data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"_update_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) def test_add_to_vector_store_no_infer_does_not_mutate_metadata(self, mocker): """Verify _add_to_vector_store with infer=False doesn't leak metadata between messages.""" memory = _build_memory_instance(mocker, Memory) memory.embedding_model.embed.return_value = [0.1, 0.2, 0.3] original_metadata = {"user_id": "test_user"} metadata_copy = original_metadata.copy() messages = [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there", "name": "bot-1"}, ] result = memory._add_to_vector_store(messages, original_metadata, filters={}, infer=False) # Metadata should not be mutated assert original_metadata == metadata_copy, ( f"_add_to_vector_store mutated the caller's metadata: {original_metadata}" ) # Should have created 2 memories assert len(result) == 2 assert result[0]["role"] == "user" assert result[1]["role"] == "assistant" assert result[1]["actor_id"] == "bot-1" # Verify each insert got distinct payloads with correct roles insert_calls = memory.vector_store.insert.call_args_list first_payload = insert_calls[0].kwargs["payloads"][0] second_payload = insert_calls[1].kwargs["payloads"][0] assert first_payload["role"] == "user" assert "actor_id" not in first_payload # user message has no name assert second_payload["role"] == "assistant" assert second_payload["actor_id"] == "bot-1" @pytest.mark.asyncio async def test_async_create_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, AsyncMemory) original_metadata = {"user_id": "test_user", "category": "sports"} metadata_copy = original_metadata.copy() await memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"async _create_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) @pytest.mark.asyncio async def test_async_create_memory_shared_metadata_across_calls(self, mocker): memory = _build_memory_instance(mocker, AsyncMemory) shared_metadata = {"user_id": "test_user"} await memory._create_memory("first memory", {"first memory": [0.1, 0.2, 0.3]}, metadata=shared_metadata) await memory._create_memory("second memory", {"second memory": [0.4, 0.5, 0.6]}, metadata=shared_metadata) assert shared_metadata == {"user_id": "test_user"}, "shared metadata was mutated across async calls" @pytest.mark.asyncio async def test_async_add_to_vector_store_no_infer_does_not_mutate_metadata(self, mocker): """Verify async _add_to_vector_store with infer=False doesn't leak metadata between messages.""" memory = _build_memory_instance(mocker, AsyncMemory) memory.embedding_model.embed.return_value = [0.1, 0.2, 0.3] original_metadata = {"user_id": "test_user"} metadata_copy = original_metadata.copy() messages = [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there", "name": "bot-1"}, ] result = await memory._add_to_vector_store(messages, original_metadata, effective_filters={}, infer=False) assert original_metadata == metadata_copy, ( f"async _add_to_vector_store mutated the caller's metadata: {original_metadata}" ) assert len(result) == 2 assert result[0]["role"] == "user" assert result[1]["role"] == "assistant" @pytest.mark.asyncio async def test_async_update_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, AsyncMemory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old data", "user_id": "test_user", "created_at": "2026-01-01T00:00:00+00:00"} ) original_metadata = {"category": "updated"} metadata_copy = original_metadata.copy() await memory._update_memory("mem-id", "new data", {"new data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"async _update_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) def test_update_preserves_actor_id_when_different_actor_updates(mocker): """actor_id must be preserved from the original memory even when the updating caller passes a different actor_id in metadata (issue #4490).""" memory = _build_memory_instance(mocker, Memory) memory.vector_store.get.return_value = MagicMock( payload={ "data": "I am player #1", "user_id": "team", "actor_id": "Alice", "created_at": "2026-01-01T00:00:00+00:00", } ) memory._update_memory( "mem-id", "Player #1 is a good person", {"Player #1 is a good person": [0.1, 0.2, 0.3]}, metadata={"user_id": "team", "actor_id": "Bob"}, ) stored = memory.vector_store.update.call_args.kwargs["payload"] assert stored["actor_id"] == "Alice" @pytest.mark.asyncio async def test_async_update_preserves_actor_id_when_different_actor_updates(mocker): """Async variant: actor_id must be preserved from the original memory (issue #4490).""" memory = _build_memory_instance(mocker, AsyncMemory) memory.vector_store.get.return_value = MagicMock( payload={ "data": "I am player #1", "user_id": "team", "actor_id": "Alice", "created_at": "2026-01-01T00:00:00+00:00", } ) await memory._update_memory( "mem-id", "Player #1 is a good person", {"Player #1 is a good person": [0.1, 0.2, 0.3]}, metadata={"user_id": "team", "actor_id": "Bob"}, ) stored = memory.vector_store.update.call_args.kwargs["payload"] assert stored["actor_id"] == "Alice" def _make_match(score, linked_memory_ids): return SimpleNamespace(score=score, payload={"linked_memory_ids": linked_memory_ids}) class TestEntityBoostParallelism: """Tests for parallelized entity boost searches (#5214).""" @pytest.fixture def mock_memory(self, mocker): _setup_mocks(mocker) return Memory() @pytest.fixture def mock_async_memory(self, mocker): _setup_mocks(mocker) return AsyncMemory() def test_sync_boosts_preserve_scoring(self, mock_memory): from mem0.utils.scoring import ENTITY_BOOST_WEIGHT mock_memory.embedding_model = Mock() mock_memory.embedding_model.embed_batch = Mock(return_value=[[0.1, 0.2, 0.3], [0.1, 0.2, 0.3]]) results_by_query = { "alice": [_make_match(0.9, ["mem-1"])], "bob": [_make_match(0.6, ["mem-1", "mem-2"])], } def fake_search(query, vectors, top_k, filters): return results_by_query[query] mock_memory._entity_store = Mock() mock_memory._entity_store.search = Mock(side_effect=fake_search) boosts = mock_memory._compute_entity_boosts( [("person", "alice"), ("person", "bob")], {"user_id": "u1"}, ) boost_alice = 0.9 * ENTITY_BOOST_WEIGHT * (1.0 / (1.0 + 0.001 * (0**2))) boost_bob = 0.6 * ENTITY_BOOST_WEIGHT * (1.0 / (1.0 + 0.001 * (1**2))) assert boosts["mem-1"] == pytest.approx(max(boost_alice, boost_bob)) assert boosts["mem-2"] == pytest.approx(boost_bob) def test_sync_embed_batch_called_once(self, mock_memory): mock_memory.embedding_model = Mock() mock_memory.embedding_model.embed_batch = Mock(return_value=[[0.1], [0.1], [0.1]]) mock_memory._entity_store = Mock() mock_memory._entity_store.search = Mock(return_value=[_make_match(0.7, ["mem-1"])]) mock_memory._compute_entity_boosts( [("person", "alice"), ("person", "bob"), ("person", "carol")], {"user_id": "u1"}, ) mock_memory.embedding_model.embed_batch.assert_called_once_with(["alice", "bob", "carol"], "search") @pytest.mark.asyncio async def test_async_boosts_preserve_scoring(self, mock_async_memory): from mem0.utils.scoring import ENTITY_BOOST_WEIGHT mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed_batch = Mock(return_value=[[0.1, 0.2, 0.3], [0.1, 0.2, 0.3]]) results_by_query = { "alice": [_make_match(0.9, ["mem-1"])], "bob": [_make_match(0.6, ["mem-1", "mem-2"])], } def fake_search(query, vectors, top_k, filters): return results_by_query[query] mock_async_memory._entity_store = Mock() mock_async_memory._entity_store.search = Mock(side_effect=fake_search) boosts = await mock_async_memory._compute_entity_boosts_async( [("person", "alice"), ("person", "bob")], {"user_id": "u1"}, ) boost_alice = 0.9 * ENTITY_BOOST_WEIGHT * (1.0 / (1.0 + 0.001 * (0**2))) boost_bob = 0.6 * ENTITY_BOOST_WEIGHT * (1.0 / (1.0 + 0.001 * (1**2))) assert boosts["mem-1"] == pytest.approx(max(boost_alice, boost_bob)) assert boosts["mem-2"] == pytest.approx(boost_bob) @pytest.mark.asyncio async def test_async_embed_batch_called_once(self, mock_async_memory): mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed_batch = Mock(return_value=[[0.1], [0.1], [0.1]]) mock_async_memory._entity_store = Mock() mock_async_memory._entity_store.search = Mock(return_value=[_make_match(0.7, ["mem-1"])]) await mock_async_memory._compute_entity_boosts_async( [("person", "alice"), ("person", "bob"), ("person", "carol")], {"user_id": "u1"}, ) mock_async_memory.embedding_model.embed_batch.assert_called_once_with(["alice", "bob", "carol"], "search") def test_sync_one_entity_failure_does_not_abort_others(self, mock_memory, caplog): mock_memory.embedding_model = Mock() mock_memory.embedding_model.embed_batch = Mock(return_value=[[0.1, 0.2, 0.3], [0.1, 0.2, 0.3]]) def fake_search(query, vectors, top_k, filters): if query == "boom": raise RuntimeError("provider timeout") return [_make_match(0.8, ["mem-9"])] mock_memory._entity_store = Mock() mock_memory._entity_store.search = Mock(side_effect=fake_search) with caplog.at_level(logging.WARNING): boosts = mock_memory._compute_entity_boosts( [("person", "boom"), ("person", "ok")], {"user_id": "u1"}, ) assert "mem-9" in boosts assert any("Entity boost search failed" in r.message for r in caplog.records) @pytest.mark.asyncio async def test_async_one_entity_failure_does_not_abort_others(self, mock_async_memory, caplog): mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed_batch = Mock(return_value=[[0.1, 0.2, 0.3], [0.1, 0.2, 0.3]]) def fake_search(query, vectors, top_k, filters): if query == "boom": raise RuntimeError("provider timeout") return [_make_match(0.8, ["mem-9"])] mock_async_memory._entity_store = Mock() mock_async_memory._entity_store.search = Mock(side_effect=fake_search) with caplog.at_level(logging.WARNING): boosts = await mock_async_memory._compute_entity_boosts_async( [("person", "boom"), ("person", "ok")], {"user_id": "u1"}, ) assert "mem-9" in boosts assert any("Entity boost search failed" in r.message for r in caplog.records) def test_sync_searches_run_concurrently(self, mock_memory): mock_memory.embedding_model = Mock() mock_memory.embedding_model.embed_batch = Mock(return_value=[[0.1]] * 4) concurrent_count = {"current": 0, "peak": 0} def blocking_search(query, vectors, top_k, filters): concurrent_count["current"] += 1 concurrent_count["peak"] = max(concurrent_count["peak"], concurrent_count["current"]) time.sleep(0.2) concurrent_count["current"] -= 1 return [_make_match(0.7, [f"mem-{query}"])] mock_memory._entity_store = Mock() mock_memory._entity_store.search = Mock(side_effect=blocking_search) entities = [("person", f"e{i}") for i in range(4)] start = time.perf_counter() boosts = mock_memory._compute_entity_boosts(entities, {"user_id": "u1"}) elapsed = time.perf_counter() - start assert elapsed < 0.75, f"searches did not run concurrently (took {elapsed:.2f}s)" assert concurrent_count["peak"] >= 2, "no overlap observed between entity searches" assert len(boosts) == 4 @pytest.mark.asyncio async def test_async_searches_run_concurrently(self, mock_async_memory): mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed_batch = Mock(return_value=[[0.1]] * 4) concurrent_count = {"current": 0, "peak": 0} def blocking_search(query, vectors, top_k, filters): concurrent_count["current"] += 1 concurrent_count["peak"] = max(concurrent_count["peak"], concurrent_count["current"]) time.sleep(0.2) concurrent_count["current"] -= 1 return [_make_match(0.7, [f"mem-{query}"])] mock_async_memory._entity_store = Mock() mock_async_memory._entity_store.search = Mock(side_effect=blocking_search) entities = [("person", f"e{i}") for i in range(4)] start = time.perf_counter() boosts = await mock_async_memory._compute_entity_boosts_async(entities, {"user_id": "u1"}) elapsed = time.perf_counter() - start assert elapsed < 0.75, f"searches did not run concurrently (took {elapsed:.2f}s)" assert concurrent_count["peak"] >= 2, "no overlap observed between entity searches" assert len(boosts) == 4 class TestAddPipelineEntityEmbeddingCountGuard: """A misbehaving embedder returning fewer (or more) vectors than entity texts must not silently drop ALL entity links via a swallowed IndexError. Before the fix, `entity_embeddings[i]` (indexed by ordered_keys length) raised IndexError on a short return; the outer `except Exception` in the Phase 7 entity-linking block swallowed it, so no entity was ever searched/inserted and no error surfaced. The search path (_compute_entity_boosts) already guarded this; the add path did not. """ @pytest.fixture def mock_memory(self, mocker): mock_llm, _ = _setup_mocks(mocker) memory = Memory() memory.config = mocker.MagicMock() memory.config.custom_instructions = None memory.config.custom_update_memory_prompt = None memory.custom_instructions = None memory.api_version = "v1.1" memory.db.get_last_messages = MagicMock(return_value=[]) memory.db.save_messages = MagicMock() memory.db.batch_add_history = MagicMock() return memory @pytest.fixture def mock_async_memory(self, mocker): mock_llm, _ = _setup_mocks(mocker) memory = AsyncMemory() memory.config = mocker.MagicMock() memory.config.custom_instructions = None memory.config.custom_update_memory_prompt = None memory.custom_instructions = None memory.api_version = "v1.1" memory.db.get_last_messages = MagicMock(return_value=[]) memory.db.save_messages = MagicMock() memory.db.batch_add_history = MagicMock() return memory @staticmethod def _short_entity_embed_batch(texts, memory_action="add"): # Memory-text embeddings are well-behaved; entity embeddings come back short. if memory_action == "add" and any(t in ("Alice", "Bob") for t in texts): return [[0.1] * 10] # 1 vector for 2 entity texts return [[0.1] * 10 for _ in texts] def test_sync_short_entity_embeddings_still_link_valid_entity(self, mock_memory, mocker, caplog): mock_memory.llm.generate_response.return_value = ( '{"memory": [{"text": "Alice met Bob"}, {"text": "Bob called Alice"}]}' ) mock_memory.embedding_model = Mock() mock_memory.embedding_model.embed_batch = Mock(side_effect=self._short_entity_embed_batch) mock_memory.embedding_model.embed = Mock(return_value=[0.1] * 10) mock_memory._entity_store = Mock() mock_memory._entity_store.search_batch = Mock(return_value=[[]]) mock_memory._entity_store.insert = Mock() mock_memory._entity_store.update = Mock() mocker.patch( "mem0.memory.main.extract_entities_batch", return_value=[ [("person", "Alice"), ("person", "Bob")], [("person", "Bob"), ("person", "Alice")], ], ) mocker.patch("mem0.memory.main.capture_event") with caplog.at_level(logging.WARNING): result = mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "Alice met Bob; Bob called Alice"}], metadata={}, filters={"user_id": "u1"}, infer=True, ) # Both memories persist regardless. assert len(result) == 2 # The entity block did NOT abort: it searched + inserted the one valid entity # instead of swallowing an IndexError and linking nothing. assert mock_memory._entity_store.search_batch.call_count == 1 assert mock_memory._entity_store.insert.call_count == 1 assert not any("Batch entity linking failed" in r.message for r in caplog.records), ( "entity linking aborted on a swallowed IndexError" ) assert any("padding/truncating" in r.message for r in caplog.records), ( "expected count-mismatch warning was not emitted" ) @pytest.mark.asyncio async def test_async_short_entity_embeddings_still_link_valid_entity(self, mock_async_memory, mocker, caplog): mock_async_memory.llm.generate_response.return_value = ( '{"memory": [{"text": "Alice met Bob"}, {"text": "Bob called Alice"}]}' ) mock_async_memory.embedding_model = Mock() mock_async_memory.embedding_model.embed_batch = Mock(side_effect=self._short_entity_embed_batch) mock_async_memory.embedding_model.embed = Mock(return_value=[0.1] * 10) mock_async_memory._entity_store = Mock() mock_async_memory._entity_store.search_batch = Mock(return_value=[[]]) mock_async_memory._entity_store.insert = Mock() mock_async_memory._entity_store.update = Mock() mocker.patch( "mem0.memory.main.extract_entities_batch", return_value=[ [("person", "Alice"), ("person", "Bob")], [("person", "Bob"), ("person", "Alice")], ], ) mocker.patch("mem0.memory.main.capture_event") with caplog.at_level(logging.WARNING): result = await mock_async_memory._add_to_vector_store( messages=[{"role": "user", "content": "Alice met Bob; Bob called Alice"}], metadata={}, effective_filters={"user_id": "u1"}, infer=True, ) assert len(result) == 2 assert mock_async_memory._entity_store.search_batch.call_count == 1 assert mock_async_memory._entity_store.insert.call_count == 1 assert not any("Batch entity linking failed" in r.message for r in caplog.records), ( "async entity linking aborted on a swallowed IndexError" ) assert any("padding/truncating" in r.message for r in caplog.records), ( "expected count-mismatch warning was not emitted" )