From bd1e767b74ca2c4a49f070d854909124c44bde23 Mon Sep 17 00:00:00 2001 From: vivekvar-dl Date: Sat, 8 Aug 2026 12:56:34 +0530 Subject: [PATCH] fix(python-sdks): handle pydantic search results in memory dedup helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pipecat, cartesia, and agent-framework packages passed search results into dedup/format helpers written for plain dicts, but typed SDK results are pydantic models (snake_case attributes, no dict interface). pipecat and cartesia crashed with AttributeError — swallowed upstream, so no memories were ever injected for users with search results — while agent-framework silently dropped every search-result memory. Extract memory fields through a tolerant accessor that accepts both the camelCase dicts returned by the profile endpoint and SDK result models, so the helpers work regardless of installed SDK version. The pipecat/cartesia dependency stubs move from test_empty_profile.py into a shared tests/conftest.py so the new utils tests can reuse them. Fixes #1266 --- .../src/supermemory_agent_framework/utils.py | 15 +-- .../tests/test_utils.py | 26 +++++ .../src/supermemory_cartesia/utils.py | 53 ++++++--- .../cartesia-sdk-python/tests/conftest.py | 46 ++++++++ .../tests/test_empty_profile.py | 36 +------ .../cartesia-sdk-python/tests/test_utils.py | 92 ++++++++++++++++ .../src/supermemory_pipecat/utils.py | 53 ++++++--- packages/pipecat-sdk-python/tests/conftest.py | 101 ++++++++++++++++++ .../tests/test_empty_profile.py | 91 +--------------- .../pipecat-sdk-python/tests/test_utils.py | 92 ++++++++++++++++ 10 files changed, 441 insertions(+), 164 deletions(-) create mode 100644 packages/cartesia-sdk-python/tests/conftest.py create mode 100644 packages/cartesia-sdk-python/tests/test_utils.py create mode 100644 packages/pipecat-sdk-python/tests/conftest.py create mode 100644 packages/pipecat-sdk-python/tests/test_utils.py diff --git a/packages/agent-framework-python/src/supermemory_agent_framework/utils.py b/packages/agent-framework-python/src/supermemory_agent_framework/utils.py index 8b8c9be0..5225895b 100644 --- a/packages/agent-framework-python/src/supermemory_agent_framework/utils.py +++ b/packages/agent-framework-python/src/supermemory_agent_framework/utils.py @@ -92,14 +92,15 @@ def deduplicate_memories( def extract_memory_text(item: Any) -> Optional[str]: if item is None: return None - if isinstance(item, dict): - memory = item.get("memory") - if isinstance(memory, str): - trimmed = memory.strip() - return trimmed if trimmed else None - return None if isinstance(item, str): - trimmed = item.strip() + memory: Any = item + elif isinstance(item, dict): + memory = item.get("memory") + else: + # SDK search results are pydantic models with a `memory` attribute + memory = getattr(item, "memory", None) + if isinstance(memory, str): + trimmed = memory.strip() return trimmed if trimmed else None return None diff --git a/packages/agent-framework-python/tests/test_utils.py b/packages/agent-framework-python/tests/test_utils.py index 6b9362bb..25635cef 100644 --- a/packages/agent-framework-python/tests/test_utils.py +++ b/packages/agent-framework-python/tests/test_utils.py @@ -1,5 +1,7 @@ """Tests for utility functions.""" +from types import SimpleNamespace + import pytest from supermemory_agent_framework.utils import ( @@ -56,6 +58,30 @@ class TestDeduplicateMemories: ) assert result.static == ["valid"] + def test_model_items(self) -> None: + # SDK search results are pydantic models with a `memory` attribute, + # not dicts — they must not be silently dropped (#1266). + result = deduplicate_memories( + search_results=[SimpleNamespace(memory="User prefers async")], + ) + assert result.search_results == ["User prefers async"] + + def test_model_items_deduplicate_against_static(self) -> None: + result = deduplicate_memories( + static=[{"memory": "User likes Python"}], + search_results=[ + SimpleNamespace(memory="User likes Python"), + SimpleNamespace(memory="User prefers async"), + ], + ) + assert result.search_results == ["User prefers async"] + + def test_model_items_without_memory_text_filtered(self) -> None: + result = deduplicate_memories( + search_results=[SimpleNamespace(memory=None), SimpleNamespace(memory=" ")], + ) + assert result.search_results == [] + class TestConvertProfileToMarkdown: def test_empty_profile(self) -> None: diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py index eb366426..8fb35b41 100644 --- a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py +++ b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py @@ -1,7 +1,27 @@ """Utility functions for Supermemory Cartesia integration.""" from datetime import datetime, timezone -from typing import Any, Dict, List, Union +from typing import Any, Dict, List + + +def _get_result_field(result: Any, *keys: str) -> Any: + """Read a field from a search result that may be a dict or an SDK model. + + The profile endpoint returns search results as plain dicts with camelCase + keys, while the typed SDK models expose the same data as snake_case + attributes. Accept both shapes so callers don't depend on the SDK version. + """ + if isinstance(result, dict): + for key in keys: + value = result.get(key) + if value is not None: + return value + return None + for key in keys: + value = getattr(result, key, None) + if value is not None: + return value + return None def get_last_user_message(messages: List[Dict[str, str]]) -> str | None: @@ -52,14 +72,15 @@ def format_relative_time(iso_timestamp: str) -> str: def deduplicate_memories( static: List[str], dynamic: List[str], - search_results: List[Dict[str, Any]], -) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]: + search_results: List[Any], +) -> Dict[str, List[Any]]: """Deduplicate memories. Priority: static > dynamic > search. Args: static: List of static memory strings. dynamic: List of dynamic memory strings. - search_results: List of search result dicts with 'memory' and 'updatedAt'. + search_results: List of search results with 'memory' and 'updatedAt', + either as dicts or as SDK result models. """ seen = set() @@ -71,10 +92,10 @@ def deduplicate_memories( out.append(m) return out - def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + def unique_search(results: List[Any]) -> List[Any]: out = [] for r in results: - memory = r.get("memory", "") + memory = _get_result_field(r, "memory") if memory and memory not in seen: seen.add(memory) out.append(r) @@ -88,7 +109,7 @@ def deduplicate_memories( def format_memories_to_text( - memories: Dict[str, Union[List[str], List[Dict[str, Any]]]], + memories: Dict[str, List[Any]], system_prompt: str = "Based on previous conversations, I recall:\n\n", include_static: bool = True, include_dynamic: bool = True, @@ -116,16 +137,16 @@ def format_memories_to_text( sections.append("## Relevant Memories") lines = [] for item in search_results: - if isinstance(item, dict): - memory = item.get("memory", "") - updated_at = item.get("updatedAt", "") - time_str = format_relative_time(updated_at) if updated_at else "" - if time_str: - lines.append(f"- [{time_str}] {memory}") - else: - lines.append(f"- {memory}") - else: + if isinstance(item, str): lines.append(f"- {item}") + continue + memory = _get_result_field(item, "memory") or "" + updated_at = _get_result_field(item, "updatedAt", "updated_at") or "" + time_str = format_relative_time(updated_at) if updated_at else "" + if time_str: + lines.append(f"- [{time_str}] {memory}") + else: + lines.append(f"- {memory}") sections.append("\n".join(lines)) if not sections: diff --git a/packages/cartesia-sdk-python/tests/conftest.py b/packages/cartesia-sdk-python/tests/conftest.py new file mode 100644 index 00000000..febe082f --- /dev/null +++ b/packages/cartesia-sdk-python/tests/conftest.py @@ -0,0 +1,46 @@ +"""Shared pytest fixtures: stub out heavy optional dependencies. + +The stubs are installed at collection time so test modules can import the +package without pipecat/cartesia, loguru, or pydantic present. +""" + +from __future__ import annotations + +import sys +import types + + +def _install_test_stubs() -> None: + if "loguru" not in sys.modules: + loguru_module = types.ModuleType("loguru") + + class _Logger: + def info(self, *_args, **_kwargs): + return None + + def warning(self, *_args, **_kwargs): + return None + + def error(self, *_args, **_kwargs): + return None + + loguru_module.logger = _Logger() + sys.modules["loguru"] = loguru_module + + if "pydantic" not in sys.modules: + pydantic_module = types.ModuleType("pydantic") + + class BaseModel: + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def Field(*, default=None, **_kwargs): + return default + + pydantic_module.BaseModel = BaseModel + pydantic_module.Field = Field + sys.modules["pydantic"] = pydantic_module + + +_install_test_stubs() diff --git a/packages/cartesia-sdk-python/tests/test_empty_profile.py b/packages/cartesia-sdk-python/tests/test_empty_profile.py index 382e5e5f..c559d26a 100644 --- a/packages/cartesia-sdk-python/tests/test_empty_profile.py +++ b/packages/cartesia-sdk-python/tests/test_empty_profile.py @@ -1,44 +1,10 @@ from __future__ import annotations -import sys -import types import unittest from types import SimpleNamespace from unittest.mock import AsyncMock - -def _install_test_stubs() -> None: - if "loguru" not in sys.modules: - loguru_module = types.ModuleType("loguru") - - class _Logger: - def info(self, *_args, **_kwargs): - return None - - def warning(self, *_args, **_kwargs): - return None - - def error(self, *_args, **_kwargs): - return None - - loguru_module.logger = _Logger() - sys.modules["loguru"] = loguru_module - - if "pydantic" not in sys.modules: - pydantic_module = types.ModuleType("pydantic") - - class BaseModel: - def __init__(self, **kwargs): - for key, value in kwargs.items(): - setattr(self, key, value) - - def Field(*, default=None, **_kwargs): - return default - - pydantic_module.BaseModel = BaseModel - pydantic_module.Field = Field - sys.modules["pydantic"] = pydantic_module - +from .conftest import _install_test_stubs _install_test_stubs() diff --git a/packages/cartesia-sdk-python/tests/test_utils.py b/packages/cartesia-sdk-python/tests/test_utils.py new file mode 100644 index 00000000..da242573 --- /dev/null +++ b/packages/cartesia-sdk-python/tests/test_utils.py @@ -0,0 +1,92 @@ +"""Tests for the memory dedup/format helpers. + +Search results reach these helpers either as plain camelCase dicts (profile +endpoint) or as typed SDK result models (snake_case attributes, no dict +interface). Both shapes must be handled — see issue #1266. +""" + +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from .conftest import _install_test_stubs + +_install_test_stubs() + +from supermemory_cartesia.utils import deduplicate_memories, format_memories_to_text + + +def _model_result(memory, updated_at=None): + """Stand-in for an SDK result model: attribute access only, no .get().""" + return SimpleNamespace(memory=memory, updated_at=updated_at) + + +class TestDeduplicateMemories(unittest.TestCase): + def test_dict_results_with_camel_case_keys(self) -> None: + results = [{"memory": "User prefers async", "updatedAt": "2026-01-01T00:00:00Z"}] + + deduplicated = deduplicate_memories(static=[], dynamic=[], search_results=results) + + self.assertEqual(deduplicated["search_results"], results) + + def test_model_results_use_attribute_access(self) -> None: + result = _model_result("User prefers async") + + deduplicated = deduplicate_memories(static=[], dynamic=[], search_results=[result]) + + self.assertEqual(deduplicated["search_results"], [result]) + + def test_model_results_deduplicate_against_profile(self) -> None: + deduplicated = deduplicate_memories( + static=["User prefers async"], + dynamic=[], + search_results=[ + _model_result("User prefers async"), + _model_result("User works remotely"), + ], + ) + + self.assertEqual(deduplicated["static"], ["User prefers async"]) + self.assertEqual( + [r.memory for r in deduplicated["search_results"]], + ["User works remotely"], + ) + + def test_results_without_memory_are_dropped(self) -> None: + deduplicated = deduplicate_memories( + static=[], + dynamic=[], + search_results=[_model_result(None), {"updatedAt": "2026-01-01T00:00:00Z"}], + ) + + self.assertEqual(deduplicated["search_results"], []) + + +class TestFormatMemoriesToText(unittest.TestCase): + def _format(self, search_results) -> str: + return format_memories_to_text( + {"static": [], "dynamic": [], "search_results": search_results} + ) + + def test_dict_results_render_memory_and_relative_time(self) -> None: + text = self._format( + [{"memory": "User prefers async", "updatedAt": "2020-01-05T00:00:00Z"}] + ) + + self.assertIn("- [5 Jan, 2020] User prefers async", text) + + def test_model_results_render_memory_and_relative_time(self) -> None: + text = self._format([_model_result("User prefers async", "2020-01-05T00:00:00Z")]) + + self.assertIn("- [5 Jan, 2020] User prefers async", text) + + def test_model_results_without_timestamp_render_memory_only(self) -> None: + text = self._format([_model_result("User prefers async")]) + + self.assertIn("- User prefers async", text) + + def test_string_results_render_verbatim(self) -> None: + text = self._format(["User prefers async"]) + + self.assertIn("- User prefers async", text) diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py index a27da256..eb9cc420 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py @@ -1,7 +1,27 @@ """Utility functions for Supermemory Pipecat integration.""" from datetime import datetime, timezone -from typing import Any, Dict, List, Union +from typing import Any, Dict, List + + +def _get_result_field(result: Any, *keys: str) -> Any: + """Read a field from a search result that may be a dict or an SDK model. + + The profile endpoint returns search results as plain dicts with camelCase + keys, while the typed SDK models expose the same data as snake_case + attributes. Accept both shapes so callers don't depend on the SDK version. + """ + if isinstance(result, dict): + for key in keys: + value = result.get(key) + if value is not None: + return value + return None + for key in keys: + value = getattr(result, key, None) + if value is not None: + return value + return None def get_last_user_message(messages: List[Dict[str, str]]) -> str | None: @@ -52,14 +72,15 @@ def format_relative_time(iso_timestamp: str) -> str: def deduplicate_memories( static: List[str], dynamic: List[str], - search_results: List[Dict[str, Any]], -) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]: + search_results: List[Any], +) -> Dict[str, List[Any]]: """Deduplicate memories. Priority: static > dynamic > search. Args: static: List of static memory strings. dynamic: List of dynamic memory strings. - search_results: List of search result dicts with 'memory' and 'updatedAt'. + search_results: List of search results with 'memory' and 'updatedAt', + either as dicts or as SDK result models. """ seen = set() @@ -71,10 +92,10 @@ def deduplicate_memories( out.append(m) return out - def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + def unique_search(results: List[Any]) -> List[Any]: out = [] for r in results: - memory = r.get("memory", "") + memory = _get_result_field(r, "memory") if memory and memory not in seen: seen.add(memory) out.append(r) @@ -88,7 +109,7 @@ def deduplicate_memories( def format_memories_to_text( - memories: Dict[str, Union[List[str], List[Dict[str, Any]]]], + memories: Dict[str, List[Any]], system_prompt: str = "Based on previous conversations, I recall:\n\n", include_static: bool = True, include_dynamic: bool = True, @@ -116,16 +137,16 @@ def format_memories_to_text( sections.append("## Relevant Memories") lines = [] for item in search_results: - if isinstance(item, dict): - memory = item.get("memory", "") - updated_at = item.get("updatedAt", "") - time_str = format_relative_time(updated_at) if updated_at else "" - if time_str: - lines.append(f"- [{time_str}] {memory}") - else: - lines.append(f"- {memory}") - else: + if isinstance(item, str): lines.append(f"- {item}") + continue + memory = _get_result_field(item, "memory") or "" + updated_at = _get_result_field(item, "updatedAt", "updated_at") or "" + time_str = format_relative_time(updated_at) if updated_at else "" + if time_str: + lines.append(f"- [{time_str}] {memory}") + else: + lines.append(f"- {memory}") sections.append("\n".join(lines)) if not sections: diff --git a/packages/pipecat-sdk-python/tests/conftest.py b/packages/pipecat-sdk-python/tests/conftest.py new file mode 100644 index 00000000..08300af4 --- /dev/null +++ b/packages/pipecat-sdk-python/tests/conftest.py @@ -0,0 +1,101 @@ +"""Shared pytest fixtures: stub out heavy optional dependencies. + +The stubs are installed at collection time so test modules can import the +package without pipecat/cartesia, loguru, or pydantic present. +""" + +from __future__ import annotations + +import sys +import types + + +def _install_test_stubs() -> None: + if "loguru" not in sys.modules: + loguru_module = types.ModuleType("loguru") + + class _Logger: + def warning(self, *_args, **_kwargs): + return None + + def error(self, *_args, **_kwargs): + return None + + loguru_module.logger = _Logger() + sys.modules["loguru"] = loguru_module + + if "pydantic" not in sys.modules: + pydantic_module = types.ModuleType("pydantic") + + class BaseModel: + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def Field(*, default=None, **_kwargs): + return default + + pydantic_module.BaseModel = BaseModel + pydantic_module.Field = Field + sys.modules["pydantic"] = pydantic_module + + if "pipecat" not in sys.modules: + pipecat_module = types.ModuleType("pipecat") + sys.modules["pipecat"] = pipecat_module + + frames_module = types.ModuleType("pipecat.frames.frames") + + class Frame: # pragma: no cover - import stub + pass + + class InputAudioRawFrame: # pragma: no cover - import stub + pass + + class LLMContextFrame: # pragma: no cover - import stub + pass + + class LLMMessagesFrame: # pragma: no cover - import stub + pass + + frames_module.Frame = Frame + frames_module.InputAudioRawFrame = InputAudioRawFrame + frames_module.LLMContextFrame = LLMContextFrame + frames_module.LLMMessagesFrame = LLMMessagesFrame + + llm_context_module = types.ModuleType("pipecat.processors.aggregators.llm_context") + + class LLMContext: # pragma: no cover - import stub + pass + + llm_context_module.LLMContext = LLMContext + + openai_context_module = types.ModuleType( + "pipecat.processors.aggregators.openai_llm_context" + ) + + class OpenAILLMContextFrame: # pragma: no cover - import stub + pass + + openai_context_module.OpenAILLMContextFrame = OpenAILLMContextFrame + + frame_processor_module = types.ModuleType("pipecat.processors.frame_processor") + + class FrameDirection: # pragma: no cover - import stub + pass + + class FrameProcessor: + def __init__(self, *args, **kwargs): + return None + + frame_processor_module.FrameDirection = FrameDirection + frame_processor_module.FrameProcessor = FrameProcessor + + sys.modules["pipecat.frames.frames"] = frames_module + sys.modules["pipecat.processors.aggregators.llm_context"] = llm_context_module + sys.modules[ + "pipecat.processors.aggregators.openai_llm_context" + ] = openai_context_module + sys.modules["pipecat.processors.frame_processor"] = frame_processor_module + + +_install_test_stubs() diff --git a/packages/pipecat-sdk-python/tests/test_empty_profile.py b/packages/pipecat-sdk-python/tests/test_empty_profile.py index ec3ccd26..b5c39d79 100644 --- a/packages/pipecat-sdk-python/tests/test_empty_profile.py +++ b/packages/pipecat-sdk-python/tests/test_empty_profile.py @@ -1,99 +1,10 @@ from __future__ import annotations -import sys -import types import unittest from types import SimpleNamespace from unittest.mock import AsyncMock - -def _install_test_stubs() -> None: - if "loguru" not in sys.modules: - loguru_module = types.ModuleType("loguru") - - class _Logger: - def warning(self, *_args, **_kwargs): - return None - - def error(self, *_args, **_kwargs): - return None - - loguru_module.logger = _Logger() - sys.modules["loguru"] = loguru_module - - if "pydantic" not in sys.modules: - pydantic_module = types.ModuleType("pydantic") - - class BaseModel: - def __init__(self, **kwargs): - for key, value in kwargs.items(): - setattr(self, key, value) - - def Field(*, default=None, **_kwargs): - return default - - pydantic_module.BaseModel = BaseModel - pydantic_module.Field = Field - sys.modules["pydantic"] = pydantic_module - - if "pipecat" not in sys.modules: - pipecat_module = types.ModuleType("pipecat") - sys.modules["pipecat"] = pipecat_module - - frames_module = types.ModuleType("pipecat.frames.frames") - - class Frame: # pragma: no cover - import stub - pass - - class InputAudioRawFrame: # pragma: no cover - import stub - pass - - class LLMContextFrame: # pragma: no cover - import stub - pass - - class LLMMessagesFrame: # pragma: no cover - import stub - pass - - frames_module.Frame = Frame - frames_module.InputAudioRawFrame = InputAudioRawFrame - frames_module.LLMContextFrame = LLMContextFrame - frames_module.LLMMessagesFrame = LLMMessagesFrame - - llm_context_module = types.ModuleType("pipecat.processors.aggregators.llm_context") - - class LLMContext: # pragma: no cover - import stub - pass - - llm_context_module.LLMContext = LLMContext - - openai_context_module = types.ModuleType( - "pipecat.processors.aggregators.openai_llm_context" - ) - - class OpenAILLMContextFrame: # pragma: no cover - import stub - pass - - openai_context_module.OpenAILLMContextFrame = OpenAILLMContextFrame - - frame_processor_module = types.ModuleType("pipecat.processors.frame_processor") - - class FrameDirection: # pragma: no cover - import stub - pass - - class FrameProcessor: - def __init__(self, *args, **kwargs): - return None - - frame_processor_module.FrameDirection = FrameDirection - frame_processor_module.FrameProcessor = FrameProcessor - - sys.modules["pipecat.frames.frames"] = frames_module - sys.modules["pipecat.processors.aggregators.llm_context"] = llm_context_module - sys.modules[ - "pipecat.processors.aggregators.openai_llm_context" - ] = openai_context_module - sys.modules["pipecat.processors.frame_processor"] = frame_processor_module - +from .conftest import _install_test_stubs _install_test_stubs() diff --git a/packages/pipecat-sdk-python/tests/test_utils.py b/packages/pipecat-sdk-python/tests/test_utils.py new file mode 100644 index 00000000..8b78def1 --- /dev/null +++ b/packages/pipecat-sdk-python/tests/test_utils.py @@ -0,0 +1,92 @@ +"""Tests for the memory dedup/format helpers. + +Search results reach these helpers either as plain camelCase dicts (profile +endpoint) or as typed SDK result models (snake_case attributes, no dict +interface). Both shapes must be handled — see issue #1266. +""" + +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from .conftest import _install_test_stubs + +_install_test_stubs() + +from supermemory_pipecat.utils import deduplicate_memories, format_memories_to_text + + +def _model_result(memory, updated_at=None): + """Stand-in for an SDK result model: attribute access only, no .get().""" + return SimpleNamespace(memory=memory, updated_at=updated_at) + + +class TestDeduplicateMemories(unittest.TestCase): + def test_dict_results_with_camel_case_keys(self) -> None: + results = [{"memory": "User prefers async", "updatedAt": "2026-01-01T00:00:00Z"}] + + deduplicated = deduplicate_memories(static=[], dynamic=[], search_results=results) + + self.assertEqual(deduplicated["search_results"], results) + + def test_model_results_use_attribute_access(self) -> None: + result = _model_result("User prefers async") + + deduplicated = deduplicate_memories(static=[], dynamic=[], search_results=[result]) + + self.assertEqual(deduplicated["search_results"], [result]) + + def test_model_results_deduplicate_against_profile(self) -> None: + deduplicated = deduplicate_memories( + static=["User prefers async"], + dynamic=[], + search_results=[ + _model_result("User prefers async"), + _model_result("User works remotely"), + ], + ) + + self.assertEqual(deduplicated["static"], ["User prefers async"]) + self.assertEqual( + [r.memory for r in deduplicated["search_results"]], + ["User works remotely"], + ) + + def test_results_without_memory_are_dropped(self) -> None: + deduplicated = deduplicate_memories( + static=[], + dynamic=[], + search_results=[_model_result(None), {"updatedAt": "2026-01-01T00:00:00Z"}], + ) + + self.assertEqual(deduplicated["search_results"], []) + + +class TestFormatMemoriesToText(unittest.TestCase): + def _format(self, search_results) -> str: + return format_memories_to_text( + {"static": [], "dynamic": [], "search_results": search_results} + ) + + def test_dict_results_render_memory_and_relative_time(self) -> None: + text = self._format( + [{"memory": "User prefers async", "updatedAt": "2020-01-05T00:00:00Z"}] + ) + + self.assertIn("- [5 Jan, 2020] User prefers async", text) + + def test_model_results_render_memory_and_relative_time(self) -> None: + text = self._format([_model_result("User prefers async", "2020-01-05T00:00:00Z")]) + + self.assertIn("- [5 Jan, 2020] User prefers async", text) + + def test_model_results_without_timestamp_render_memory_only(self) -> None: + text = self._format([_model_result("User prefers async")]) + + self.assertIn("- User prefers async", text) + + def test_string_results_render_verbatim(self) -> None: + text = self._format(["User prefers async"]) + + self.assertIn("- User prefers async", text)