diff --git a/reme_ai/core/vector_store/base_vector_store.py b/reme_ai/core/vector_store/base_vector_store.py index 614666a4..ad858658 100644 --- a/reme_ai/core/vector_store/base_vector_store.py +++ b/reme_ai/core/vector_store/base_vector_store.py @@ -94,7 +94,7 @@ class BaseVectorStore(ABC): filters: dict | None = None, limit: int | None = None, sort_key: str | None = None, - reverse: bool = False, + reverse: bool = True, ) -> list[VectorNode]: """Retrieve vectors from the collection that match the given filters. diff --git a/reme_ai/mem_tool/__init__.py b/reme_ai/mem_tool/__init__.py index 9b8820dc..97e19203 100644 --- a/reme_ai/mem_tool/__init__.py +++ b/reme_ai/mem_tool/__init__.py @@ -9,11 +9,12 @@ from .identity.update_identity_memory import UpdateIdentityMemory from .meta.add_meta_memory import AddMetaMemory from .meta.read_meta_memory import ReadMetaMemory from .think_tool import ThinkTool -from .vector.add_memory import AddMemory -from .vector.add_summary_memory import AddSummaryMemory -from .vector.delete_memory import DeleteMemory -from .vector.update_memory import UpdateMemory -from .vector.vector_retrieve_memory import VectorRetrieveMemory +from .vector_store.add_memory import AddMemory +from .vector_store.add_summary_memory import AddSummaryMemory +from .vector_store.delete_memory import DeleteMemory +from .vector_store.retrieve_recent_memory import RetrieveRecentMemory +from .vector_store.update_memory import UpdateMemory +from .vector_store.vector_retrieve_memory import VectorRetrieveMemory __all__ = [ "BaseMemoryTool", @@ -28,6 +29,7 @@ __all__ = [ "AddMemory", "AddSummaryMemory", "DeleteMemory", + "RetrieveRecentMemory", "UpdateMemory", "VectorRetrieveMemory", ] diff --git a/reme_ai/mem_tool/vector/__init__.py b/reme_ai/mem_tool/vector_store/__init__.py similarity index 100% rename from reme_ai/mem_tool/vector/__init__.py rename to reme_ai/mem_tool/vector_store/__init__.py diff --git a/reme_ai/mem_tool/vector/add_memory.py b/reme_ai/mem_tool/vector_store/add_memory.py similarity index 100% rename from reme_ai/mem_tool/vector/add_memory.py rename to reme_ai/mem_tool/vector_store/add_memory.py diff --git a/reme_ai/mem_tool/vector/add_memory.yaml b/reme_ai/mem_tool/vector_store/add_memory.yaml similarity index 100% rename from reme_ai/mem_tool/vector/add_memory.yaml rename to reme_ai/mem_tool/vector_store/add_memory.yaml diff --git a/reme_ai/mem_tool/vector/add_summary_memory.py b/reme_ai/mem_tool/vector_store/add_summary_memory.py similarity index 100% rename from reme_ai/mem_tool/vector/add_summary_memory.py rename to reme_ai/mem_tool/vector_store/add_summary_memory.py diff --git a/reme_ai/mem_tool/vector/add_summary_memory.yaml b/reme_ai/mem_tool/vector_store/add_summary_memory.yaml similarity index 100% rename from reme_ai/mem_tool/vector/add_summary_memory.yaml rename to reme_ai/mem_tool/vector_store/add_summary_memory.yaml diff --git a/reme_ai/mem_tool/vector/delete_memory.py b/reme_ai/mem_tool/vector_store/delete_memory.py similarity index 100% rename from reme_ai/mem_tool/vector/delete_memory.py rename to reme_ai/mem_tool/vector_store/delete_memory.py diff --git a/reme_ai/mem_tool/vector/delete_memory.yaml b/reme_ai/mem_tool/vector_store/delete_memory.yaml similarity index 100% rename from reme_ai/mem_tool/vector/delete_memory.yaml rename to reme_ai/mem_tool/vector_store/delete_memory.yaml diff --git a/reme_ai/mem_tool/vector_store/retrieve_recent_memory.py b/reme_ai/mem_tool/vector_store/retrieve_recent_memory.py new file mode 100644 index 00000000..43b683dd --- /dev/null +++ b/reme_ai/mem_tool/vector_store/retrieve_recent_memory.py @@ -0,0 +1,78 @@ +"""Time-based memory retrieval to get most recent memories.""" + +from loguru import logger + +from ..base_memory_tool import BaseMemoryTool +from ...core.context import C +from ...core.schema import MemoryNode, VectorNode +from ...core.utils import deduplicate_memories + + +@C.register_op() +class RetrieveRecentMemory(BaseMemoryTool): + """Retrieve most recent memories based on time_modified. + + Retrieves memories sorted by modification time in descending order. + Uses memory_type and memory_target from context (self.memory_type, self.memory_target). + """ + + def __init__( + self, + top_k: int = 20, + **kwargs, + ): + """Initialize RetrieveRecentMemory. + + Args: + top_k: Max memories to retrieve. + **kwargs: Additional args for BaseMemoryTool. + """ + super().__init__(**kwargs) + self.top_k: int = top_k + + async def _retrieve_recent(self) -> list[MemoryNode]: + """Retrieve recent memories sorted by time_modified. + + Returns: + List of recent memories sorted by modification time (newest first). + """ + filter_dict = { + "memory_type": [self.memory_type.value], + "memory_target": [self.memory_target], + } + + # Use list() with sort_key="time_modified", reverse=True (descending), and limit + nodes: list[VectorNode] = await self.vector_store.list( + filters=filter_dict, + limit=self.top_k, + sort_key="time_modified", + reverse=True, # Most recent first (descending order) + ) + + memory_nodes: list[MemoryNode] = [MemoryNode.from_vector_node(n) for n in nodes] + + return memory_nodes + + async def execute(self): + """Execute recent memory retrieval. + + Uses memory_type and memory_target from context (self.memory_type, self.memory_target). + Outputs formatted results or error message. + """ + if not self.memory_type or not self.memory_target: + self.output = "memory_type and memory_target are required for retrieval." + return + + # Retrieve recent memories + memory_nodes: list[MemoryNode] = await self._retrieve_recent() + + # Deduplicate and format output + memory_nodes = deduplicate_memories(memory_nodes) + self.memory_nodes = memory_nodes + + if not memory_nodes: + self.output = "No memory_nodes found." + else: + self.output = "\n".join([m.format_memory() for m in memory_nodes]) + + logger.info(f"Retrieved {len(memory_nodes)} recent memory_nodes") diff --git a/reme_ai/mem_tool/vector_store/retrieve_recent_memory.yaml b/reme_ai/mem_tool/vector_store/retrieve_recent_memory.yaml new file mode 100644 index 00000000..cabe4e7e --- /dev/null +++ b/reme_ai/mem_tool/vector_store/retrieve_recent_memory.yaml @@ -0,0 +1,5 @@ +tool: | + Retrieve the most recent memories from the memory store based on modification time. + Use this tool to find memories that were created or updated most recently. + The results are sorted by time_modified in descending order (newest first). + The memory_type and memory_target are taken from the current context. diff --git a/reme_ai/mem_tool/vector/update_memory.py b/reme_ai/mem_tool/vector_store/update_memory.py similarity index 100% rename from reme_ai/mem_tool/vector/update_memory.py rename to reme_ai/mem_tool/vector_store/update_memory.py diff --git a/reme_ai/mem_tool/vector/update_memory.yaml b/reme_ai/mem_tool/vector_store/update_memory.yaml similarity index 100% rename from reme_ai/mem_tool/vector/update_memory.yaml rename to reme_ai/mem_tool/vector_store/update_memory.yaml diff --git a/reme_ai/mem_tool/vector/vector_retrieve_memory.py b/reme_ai/mem_tool/vector_store/vector_retrieve_memory.py similarity index 100% rename from reme_ai/mem_tool/vector/vector_retrieve_memory.py rename to reme_ai/mem_tool/vector_store/vector_retrieve_memory.py diff --git a/reme_ai/mem_tool/vector/vector_retrieve_memory.yaml b/reme_ai/mem_tool/vector_store/vector_retrieve_memory.yaml similarity index 100% rename from reme_ai/mem_tool/vector/vector_retrieve_memory.yaml rename to reme_ai/mem_tool/vector_store/vector_retrieve_memory.yaml diff --git a/tests/test_memory_vector_conversion.py b/tests/test_memory_vector_conversion.py new file mode 100644 index 00000000..92328615 --- /dev/null +++ b/tests/test_memory_vector_conversion.py @@ -0,0 +1,680 @@ +"""Test file for MemoryNode and VectorNode conversion. + +Tests: +1. vector -> memory -> vector conversion +2. memory -> vector -> memory conversion +3. _update_after_init correctness (memory_id generation) +4. __setattr__ behavior (when_to_use and content updates) +""" + +import datetime +import hashlib +import time +from enum import Enum +from typing import Any, List, Dict +from uuid import uuid4 + +# Import Pydantic before our modules +from pydantic import BaseModel, Field, model_validator + + +# Define MemoryType locally to avoid import issues +class MemoryType(str, Enum): + """Memory type enumeration for the three-layer memory architecture.""" + + IDENTITY = "identity" + PERSONAL = "personal" + PROCEDURAL = "procedural" + TOOL = "tool" + SUMMARY = "summary" + HISTORY = "history" + TASK = "task" # Add TASK type for testing + + +# Define VectorNode locally to avoid import issues +class VectorNode(BaseModel): + """Represents a discrete unit of text content paired with its corresponding vector embedding and metadata.""" + + vector_id: str = Field(default_factory=lambda: uuid4().hex) + content: str = Field(default="") + vector: List[float] | None = Field(default=None) + metadata: Dict[str, str | bool | int | float] = Field(default_factory=dict) + + +# Define MemoryNode locally to avoid import issues +def get_now_time() -> str: + """Get current timestamp in YYYY-MM-DD HH:MM:SS format.""" + return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + +# Length of the memory ID (first N characters of SHA-256 hash) +MEMORY_ID_LENGTH: int = 16 + + +class MemoryNode(BaseModel): + """Memory node for storing memories in the ReMe system.""" + + memory_id: str = Field(default="", description="Unique memory identifier") + memory_type: MemoryType = Field(default=..., description="Type of memory") + memory_target: str = Field(default="", description="Target or topic of the memory") + when_to_use: str = Field(default="", description="Condition description for vector retrieval") + content: str = Field(default="", description="Actual memory content") + ref_memory_id: str = Field(default="", description="Reference to related raw history memory ID") + + time_created: str = Field(default_factory=get_now_time, description="Creation timestamp") + time_modified: str = Field(default_factory=get_now_time, description="Last modification timestamp") + author: str = Field(default="", description="Author or source of the memory") + score: float = Field(default=0, description="Relevance or importance score") + + metadata: dict[str, Any] = Field(default_factory=dict, description="Additional metadata") + + def _update_modified_time(self) -> "MemoryNode": + """Update time_modified to current timestamp.""" + self.time_modified = get_now_time() + return self + + def _update_memory_id(self) -> "MemoryNode": + """Generate memory_id from SHA-256 hash of content.""" + if not self.content: + return self + + hash_obj = hashlib.sha256(self.content.encode("utf-8")) + hex_dig = hash_obj.hexdigest() + self.memory_id = hex_dig[:MEMORY_ID_LENGTH] + return self + + @model_validator(mode="after") + def _update_after_init(self) -> "MemoryNode": + """Post-initialization validator.""" + if not self.memory_id: + self._update_memory_id() + return self + + def __setattr__(self, name: str, value): + """Auto-update timestamps and memory_id when content or when_to_use changes.""" + should_update: bool = name in ("when_to_use", "content") and getattr(self, name, None) != value + super().__setattr__(name, value) + if should_update: + self._update_modified_time() + if name == "content": + self._update_memory_id() + + def to_vector_node(self) -> VectorNode: + """Convert to VectorNode for vector storage.""" + # Build base metadata (shared fields) + metadata: dict[str, Any] = { + "memory_type": self.memory_type.value, + "memory_target": self.memory_target, + "ref_memory_id": self.ref_memory_id, + "time_created": self.time_created, + "time_modified": self.time_modified, + "author": self.author, + "score": self.score, + **self.metadata, + } + + if self.when_to_use: + # Use when_to_use for vector embedding, store content in metadata + vector_content = self.when_to_use + metadata["content"] = self.content + else: + # Use content directly for vector embedding + vector_content = self.content + + return VectorNode( + vector_id=self.memory_id, + content=vector_content, + metadata=metadata, + ) + + @classmethod + def from_vector_node(cls, node: VectorNode) -> "MemoryNode": + """Reconstruct MemoryNode from VectorNode.""" + metadata = node.metadata.copy() + memory_type_str = metadata.pop("memory_type", None) + + try: + memory_type: MemoryType = MemoryType(memory_type_str) + except ValueError as e: + raise ValueError( + f"Invalid memory_type '{memory_type_str}' in VectorNode metadata. " + f"Valid types are: {[t.value for t in MemoryType]}", + ) from e + + # Restore when_to_use and content based on metadata structure + if "content" in metadata: + # Original had when_to_use set + when_to_use = node.content + content = metadata.pop("content", "") + else: + # Original had empty when_to_use + when_to_use = "" + content = node.content + + return cls( + memory_id=node.vector_id, + memory_type=memory_type, + memory_target=metadata.pop("memory_target", ""), + when_to_use=when_to_use, + content=content, + ref_memory_id=metadata.pop("ref_memory_id", ""), + time_created=metadata.pop("time_created", ""), + time_modified=metadata.pop("time_modified", ""), + author=metadata.pop("author", ""), + score=metadata.pop("score", 0), + metadata=metadata, + ) + + +def generate_expected_memory_id(content: str) -> str: + """Generate expected memory_id from content.""" + hash_obj = hashlib.sha256(content.encode("utf-8")) + hex_dig = hash_obj.hexdigest() + return hex_dig[:MEMORY_ID_LENGTH] + + +def test_vector_to_memory_to_vector_with_when_to_use(): + """Test: vector -> memory -> vector conversion with when_to_use.""" + print("\n=== Test 1: vector -> memory -> vector (with when_to_use) ===") + + # Create original VectorNode with content in metadata (indicates when_to_use was used) + original_vector = VectorNode( + vector_id="test_id_001", + content="When user asks about Python programming", + vector=[0.1, 0.2, 0.3], + metadata={ + "memory_type": MemoryType.PERSONAL.value, + "memory_target": "programming", + "content": "The user prefers Python for scripting tasks", # This indicates when_to_use was set + "ref_memory_id": "ref_001", + "time_created": "2024-01-01 10:00:00", + "time_modified": "2024-01-01 12:00:00", + "author": "system", + "score": 0.95, + "custom_field": "custom_value", + }, + ) + + # Convert to MemoryNode + memory_node = MemoryNode.from_vector_node(original_vector) + + # Convert back to VectorNode + new_vector = memory_node.to_vector_node() + + # Verify + assert ( + new_vector.vector_id == original_vector.vector_id + ), f"vector_id mismatch: {new_vector.vector_id} != {original_vector.vector_id}" + + assert ( + new_vector.content == original_vector.content + ), f"content mismatch: {new_vector.content} != {original_vector.content}" + + # Check metadata (excluding vector which is not preserved) + for key in [ + "memory_type", + "memory_target", + "content", + "ref_memory_id", + "time_created", + "time_modified", + "author", + "score", + "custom_field", + ]: + assert new_vector.metadata.get(key) == original_vector.metadata.get( + key, + ), f"metadata[{key}] mismatch: {new_vector.metadata.get(key)} != {original_vector.metadata.get(key)}" + + print("✓ vector -> memory -> vector conversion successful (with when_to_use)") + print(f" Original content: {original_vector.content}") + print(f" Converted content: {new_vector.content}") + print(f" Original metadata['content']: {original_vector.metadata.get('content')}") + print(f" Converted metadata['content']: {new_vector.metadata.get('content')}") + + +def test_vector_to_memory_to_vector_without_when_to_use(): + """Test: vector -> memory -> vector conversion without when_to_use.""" + print("\n=== Test 2: vector -> memory -> vector (without when_to_use) ===") + + # Create original VectorNode without content in metadata (indicates when_to_use was empty) + original_vector = VectorNode( + vector_id="test_id_002", + content="The user prefers Python for scripting tasks", + vector=[0.4, 0.5, 0.6], + metadata={ + "memory_type": MemoryType.SUMMARY.value, + "memory_target": "user_preference", + "ref_memory_id": "ref_002", + "time_created": "2024-01-02 10:00:00", + "time_modified": "2024-01-02 12:00:00", + "author": "agent", + "score": 0.85, + }, + ) + + # Convert to MemoryNode + memory_node = MemoryNode.from_vector_node(original_vector) + + # Convert back to VectorNode + new_vector = memory_node.to_vector_node() + + # Verify + assert ( + new_vector.vector_id == original_vector.vector_id + ), f"vector_id mismatch: {new_vector.vector_id} != {original_vector.vector_id}" + + assert ( + new_vector.content == original_vector.content + ), f"content mismatch: {new_vector.content} != {original_vector.content}" + + # Check that 'content' is NOT in metadata (because when_to_use was empty) + assert "content" not in new_vector.metadata, "metadata should not contain 'content' key when when_to_use is empty" + + # Check other metadata + for key in [ + "memory_type", + "memory_target", + "ref_memory_id", + "time_created", + "time_modified", + "author", + "score", + ]: + assert new_vector.metadata.get(key) == original_vector.metadata.get( + key, + ), f"metadata[{key}] mismatch: {new_vector.metadata.get(key)} != {original_vector.metadata.get(key)}" + + print("✓ vector -> memory -> vector conversion successful (without when_to_use)") + print(f" Original content: {original_vector.content}") + print(f" Converted content: {new_vector.content}") + print(f" 'content' in original metadata: {'content' in original_vector.metadata}") + print(f" 'content' in converted metadata: {'content' in new_vector.metadata}") + + +def test_memory_to_vector_to_memory_with_when_to_use(): + """Test: memory -> vector -> memory conversion with when_to_use.""" + print("\n=== Test 3: memory -> vector -> memory (with when_to_use) ===") + + # Create original MemoryNode with when_to_use + original_memory = MemoryNode( + memory_id="", # Will be auto-generated + memory_type=MemoryType.PERSONAL, + memory_target="coding_style", + when_to_use="When discussing code formatting", + content="User prefers tabs over spaces", + ref_memory_id="ref_003", + time_created="2024-01-03 10:00:00", + time_modified="2024-01-03 12:00:00", + author="user", + score=0.9, + metadata={"priority": "high"}, + ) + + # Convert to VectorNode + vector_node = original_memory.to_vector_node() + + # Convert back to MemoryNode + new_memory = MemoryNode.from_vector_node(vector_node) + + # Verify all fields + assert ( + new_memory.memory_id == original_memory.memory_id + ), f"memory_id mismatch: {new_memory.memory_id} != {original_memory.memory_id}" + + assert ( + new_memory.memory_type == original_memory.memory_type + ), f"memory_type mismatch: {new_memory.memory_type} != {original_memory.memory_type}" + + assert ( + new_memory.memory_target == original_memory.memory_target + ), f"memory_target mismatch: {new_memory.memory_target} != {original_memory.memory_target}" + + assert ( + new_memory.when_to_use == original_memory.when_to_use + ), f"when_to_use mismatch: {new_memory.when_to_use} != {original_memory.when_to_use}" + + assert ( + new_memory.content == original_memory.content + ), f"content mismatch: {new_memory.content} != {original_memory.content}" + + assert ( + new_memory.ref_memory_id == original_memory.ref_memory_id + ), f"ref_memory_id mismatch: {new_memory.ref_memory_id} != {original_memory.ref_memory_id}" + + assert ( + new_memory.time_created == original_memory.time_created + ), f"time_created mismatch: {new_memory.time_created} != {original_memory.time_created}" + + assert ( + new_memory.time_modified == original_memory.time_modified + ), f"time_modified mismatch: {new_memory.time_modified} != {original_memory.time_modified}" + + assert ( + new_memory.author == original_memory.author + ), f"author mismatch: {new_memory.author} != {original_memory.author}" + + assert new_memory.score == original_memory.score, f"score mismatch: {new_memory.score} != {original_memory.score}" + + assert ( + new_memory.metadata == original_memory.metadata + ), f"metadata mismatch: {new_memory.metadata} != {original_memory.metadata}" + + print("✓ memory -> vector -> memory conversion successful (with when_to_use)") + print(f" Original when_to_use: {original_memory.when_to_use}") + print(f" Converted when_to_use: {new_memory.when_to_use}") + print(f" Original content: {original_memory.content}") + print(f" Converted content: {new_memory.content}") + + +def test_memory_to_vector_to_memory_without_when_to_use(): + """Test: memory -> vector -> memory conversion without when_to_use.""" + print("\n=== Test 4: memory -> vector -> memory (without when_to_use) ===") + + # Create original MemoryNode without when_to_use + original_memory = MemoryNode( + memory_type=MemoryType.TASK, + memory_target="project_info", + when_to_use="", # Empty when_to_use + content="Project deadline is next Friday", + ref_memory_id="ref_004", + time_created="2024-01-04 10:00:00", + time_modified="2024-01-04 12:00:00", + author="manager", + score=1.0, + metadata={"urgency": "high"}, + ) + + # Convert to VectorNode + vector_node = original_memory.to_vector_node() + + # Convert back to MemoryNode + new_memory = MemoryNode.from_vector_node(vector_node) + + # Verify all fields + assert ( + new_memory.memory_id == original_memory.memory_id + ), f"memory_id mismatch: {new_memory.memory_id} != {original_memory.memory_id}" + + assert ( + new_memory.memory_type == original_memory.memory_type + ), f"memory_type mismatch: {new_memory.memory_type} != {original_memory.memory_type}" + + assert new_memory.when_to_use == "", f"when_to_use should be empty, got: {new_memory.when_to_use}" + + assert ( + new_memory.content == original_memory.content + ), f"content mismatch: {new_memory.content} != {original_memory.content}" + + print("✓ memory -> vector -> memory conversion successful (without when_to_use)") + print(f" Original when_to_use: '{original_memory.when_to_use}'") + print(f" Converted when_to_use: '{new_memory.when_to_use}'") + print(f" Original content: {original_memory.content}") + print(f" Converted content: {new_memory.content}") + + +def test_update_after_init_auto_generates_memory_id(): + """Test: _update_after_init auto-generates memory_id when not provided.""" + print("\n=== Test 5: _update_after_init auto-generates memory_id ===") + + content = "Test content for memory_id generation" + expected_memory_id = generate_expected_memory_id(content) + + # Create MemoryNode without providing memory_id + memory = MemoryNode( + memory_type=MemoryType.PERSONAL, + content=content, + ) + + assert memory.memory_id == expected_memory_id, f"memory_id mismatch: {memory.memory_id} != {expected_memory_id}" + + print("✓ _update_after_init correctly generates memory_id") + print(f" Content: {content}") + print(f" Generated memory_id: {memory.memory_id}") + print(f" Expected memory_id: {expected_memory_id}") + + +def test_update_after_init_preserves_provided_memory_id(): + """Test: _update_after_init preserves memory_id when provided.""" + print("\n=== Test 6: _update_after_init preserves provided memory_id ===") + + custom_memory_id = "custom_id_12345" + content = "Test content" + + # Create MemoryNode with explicit memory_id + memory = MemoryNode( + memory_id=custom_memory_id, + memory_type=MemoryType.TASK, + content=content, + ) + + assert ( + memory.memory_id == custom_memory_id + ), f"memory_id should be preserved: {memory.memory_id} != {custom_memory_id}" + + print("✓ _update_after_init preserves provided memory_id") + print(f" Provided memory_id: {custom_memory_id}") + print(f" Actual memory_id: {memory.memory_id}") + + +def test_update_after_init_empty_content(): + """Test: _update_after_init with empty content.""" + print("\n=== Test 7: _update_after_init with empty content ===") + + # Create MemoryNode with empty content + memory = MemoryNode( + memory_type=MemoryType.SUMMARY, + content="", + ) + + assert memory.memory_id == "", f"memory_id should be empty when content is empty, got: {memory.memory_id}" + + print("✓ _update_after_init correctly handles empty content") + print(f" Content: '{memory.content}'") + print(f" memory_id: '{memory.memory_id}'") + + +def test_setattr_content_updates_memory_id_and_time(): + """Test: __setattr__ updates memory_id and time_modified when content changes.""" + print("\n=== Test 8: __setattr__ updates memory_id when content changes ===") + + initial_content = "Initial content" + new_content = "New content after update" + + # Create MemoryNode + memory = MemoryNode( + memory_type=MemoryType.PERSONAL, + content=initial_content, + ) + + initial_memory_id = memory.memory_id + initial_time_modified = memory.time_modified + expected_initial_id = generate_expected_memory_id(initial_content) + + assert ( + initial_memory_id == expected_initial_id + ), f"Initial memory_id incorrect: {initial_memory_id} != {expected_initial_id}" + + # Wait at least 1 second to ensure time changes (timestamp precision is 1 second) + time.sleep(1.1) + + # Update content + memory.content = new_content + + expected_new_id = generate_expected_memory_id(new_content) + + assert ( + memory.memory_id == expected_new_id + ), f"memory_id not updated correctly: {memory.memory_id} != {expected_new_id}" + + assert memory.memory_id != initial_memory_id, "memory_id should change when content changes" + + assert memory.time_modified != initial_time_modified, "time_modified should be updated when content changes" + + print("✓ __setattr__ correctly updates memory_id and time_modified when content changes") + print(f" Initial content: {initial_content}") + print(f" Initial memory_id: {initial_memory_id}") + print(f" New content: {new_content}") + print(f" New memory_id: {memory.memory_id}") + print(f" Initial time_modified: {initial_time_modified}") + print(f" New time_modified: {memory.time_modified}") + + +def test_setattr_when_to_use_updates_time_only(): + """Test: __setattr__ updates only time_modified when when_to_use changes.""" + print("\n=== Test 9: __setattr__ updates time_modified when when_to_use changes ===") + + content = "Fixed content" + initial_when_to_use = "Initial trigger" + new_when_to_use = "New trigger condition" + + # Create MemoryNode + memory = MemoryNode( + memory_type=MemoryType.TASK, + content=content, + when_to_use=initial_when_to_use, + ) + + initial_memory_id = memory.memory_id + initial_time_modified = memory.time_modified + + # Wait at least 1 second to ensure time changes (timestamp precision is 1 second) + time.sleep(1.1) + + # Update when_to_use + memory.when_to_use = new_when_to_use + + assert ( + memory.memory_id == initial_memory_id + ), f"memory_id should NOT change when only when_to_use changes: {memory.memory_id} != {initial_memory_id}" + + assert memory.time_modified != initial_time_modified, "time_modified should be updated when when_to_use changes" + + print("✓ __setattr__ correctly updates time_modified but not memory_id when when_to_use changes") + print(f" Content (unchanged): {content}") + print(f" memory_id (unchanged): {memory.memory_id}") + print(f" Initial when_to_use: {initial_when_to_use}") + print(f" New when_to_use: {new_when_to_use}") + print(f" Initial time_modified: {initial_time_modified}") + print(f" New time_modified: {memory.time_modified}") + + +def test_setattr_no_update_when_value_unchanged(): + """Test: __setattr__ does not update when value is unchanged.""" + print("\n=== Test 10: __setattr__ does not update when value is unchanged ===") + + content = "Test content" + when_to_use = "Test trigger" + + # Create MemoryNode + memory = MemoryNode( + memory_type=MemoryType.SUMMARY, + content=content, + when_to_use=when_to_use, + ) + + initial_time_modified = memory.time_modified + + # Wait a bit (but updates should not happen for unchanged values) + time.sleep(1.1) + + # Set content to same value + memory.content = content + + assert ( + memory.time_modified == initial_time_modified + ), "time_modified should NOT change when setting same content value" + + # Set when_to_use to same value + memory.when_to_use = when_to_use + + assert ( + memory.time_modified == initial_time_modified + ), "time_modified should NOT change when setting same when_to_use value" + + print("✓ __setattr__ correctly does not update when value is unchanged") + print(f" Content: {content}") + print(f" when_to_use: {when_to_use}") + print(f" time_modified (unchanged): {memory.time_modified}") + + +def test_setattr_other_fields_no_update(): + """Test: __setattr__ does not trigger updates for other fields.""" + print("\n=== Test 11: __setattr__ does not update for other fields ===") + + # Create MemoryNode + memory = MemoryNode( + memory_type=MemoryType.PERSONAL, + content="Test content", + ) + + initial_memory_id = memory.memory_id + initial_time_modified = memory.time_modified + + # Wait a bit (but updates should not happen for other fields) + time.sleep(1.1) + + # Update other fields + memory.score = 0.95 + memory.author = "new_author" + memory.memory_target = "new_target" + memory.ref_memory_id = "new_ref" + + assert memory.memory_id == initial_memory_id, "memory_id should NOT change when updating other fields" + + assert memory.time_modified == initial_time_modified, "time_modified should NOT change when updating other fields" + + print("✓ __setattr__ correctly does not update for other fields") + print(f" memory_id (unchanged): {memory.memory_id}") + print(f" time_modified (unchanged): {memory.time_modified}") + print(" Updated fields: score, author, memory_target, ref_memory_id") + + +def run_all_tests(): + """Run all tests.""" + print("=" * 60) + print("Running MemoryNode and VectorNode Conversion Tests") + print("=" * 60) + + tests = [ + test_vector_to_memory_to_vector_with_when_to_use, + test_vector_to_memory_to_vector_without_when_to_use, + test_memory_to_vector_to_memory_with_when_to_use, + test_memory_to_vector_to_memory_without_when_to_use, + test_update_after_init_auto_generates_memory_id, + test_update_after_init_preserves_provided_memory_id, + test_update_after_init_empty_content, + test_setattr_content_updates_memory_id_and_time, + test_setattr_when_to_use_updates_time_only, + test_setattr_no_update_when_value_unchanged, + test_setattr_other_fields_no_update, + ] + + passed = 0 + failed = 0 + + for test_func in tests: + try: + test_func() + passed += 1 + except AssertionError as e: + print(f"✗ {test_func.__name__} FAILED: {e}") + failed += 1 + except Exception as e: + print(f"✗ {test_func.__name__} ERROR: {e}") + failed += 1 + + print("\n" + "=" * 60) + print(f"Test Results: {passed} passed, {failed} failed") + print("=" * 60) + + if failed == 0: + print("🎉 All tests passed!") + else: + print(f"⚠️ {failed} test(s) failed") + + return failed == 0 + + +if __name__ == "__main__": + success = run_all_tests()