From 36856cc92a7d90a1e8b385e027f6758a55797457 Mon Sep 17 00:00:00 2001 From: Saurabh Meena Date: Wed, 22 Jul 2026 22:56:16 +0530 Subject: [PATCH] Add Markdown round-trip support to AgentMemory --- docs/reference/context.md | 41 ++ semantica/context/agent_memory.py | 620 +++++++++++++++++++- tests/context/test_agent_memory_markdown.py | 406 +++++++++++++ 3 files changed, 1046 insertions(+), 21 deletions(-) create mode 100644 tests/context/test_agent_memory_markdown.py diff --git a/docs/reference/context.md b/docs/reference/context.md index 3a415816..591f11a5 100644 --- a/docs/reference/context.md +++ b/docs/reference/context.md @@ -586,6 +586,47 @@ history = memory.get_conversation_history(conversation_id="conv_001", max_items= | `max_memory_size` | `int` | `10000` | Max items before LRU eviction | | `retention_policy` | `str` | `"unlimited"` | `"N_days"` (e.g. `"30_days"`) or `"unlimited"` | +### Markdown Round Trips + +`AgentMemory` can export human-editable Markdown and import the edited files back. +Each file contains one memory item, with required metadata in YAML frontmatter and +the memory content in the Markdown body: + +```markdown +--- +id: mem_compliance_rule +created_at: '2026-07-22T09:00:00+00:00' +updated_at: '2026-07-22T10:30:00+00:00' +type: compliance +tags: +- trading +- approval +--- + +All trades must be pre-approved. +``` + +```python +from pathlib import Path + +# A single selected memory can be returned as Markdown text. +document = memory.export(format="markdown", type="compliance") + +# Export a memory set as one stable Markdown file per item. +memory.export(format="markdown", destination="memory_export/") + +# New IDs create memories; existing IDs are updated in place. +count = memory.import_data(Path("memory_export/"), format="markdown") +``` + +The required frontmatter fields are `id`, `created_at`, `updated_at`, and either +`type` or `kind`. Optional metadata can be edited at the top level. Imports reject +malformed or duplicate fields before changing memory, and re-importing unchanged +files is idempotent. Memory-local `entities` and `relationships` are preserved as +provenance but are not applied to `ContextGraph` by Markdown import. Use a dedicated +export directory: matching files are overwritten, but unrelated or stale Markdown +files are not deleted automatically. + ## PolicyEngine diff --git a/semantica/context/agent_memory.py b/semantica/context/agent_memory.py index 1811623b..7965be3c 100644 --- a/semantica/context/agent_memory.py +++ b/semantica/context/agent_memory.py @@ -58,18 +58,59 @@ Author: Semantica Contributors License: MIT """ +import copy +import hashlib +import re from collections import deque from dataclasses import dataclass, field -from datetime import datetime, timedelta -from typing import Any, Dict, List, Optional, Union +from datetime import date, datetime, timedelta +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union import numpy as np +import yaml from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker from ..utils.types import EntityDict, RelationshipDict +class _UniqueKeySafeLoader(yaml.SafeLoader): + """Safe YAML loader that rejects ambiguous duplicate mapping keys.""" + + +def _construct_unique_mapping( + loader: _UniqueKeySafeLoader, node: yaml.MappingNode, deep: bool = False +) -> Dict[Any, Any]: + loader.flatten_mapping(node) + mapping = {} + for key_node, value_node in node.value: + key = loader.construct_object(key_node, deep=deep) + try: + duplicate = key in mapping + except TypeError as exc: + raise yaml.constructor.ConstructorError( + "while constructing a mapping", + node.start_mark, + "found an unhashable key", + key_node.start_mark, + ) from exc + if duplicate: + raise yaml.constructor.ConstructorError( + "while constructing a mapping", + node.start_mark, + f"found duplicate key {key!r}", + key_node.start_mark, + ) + mapping[key] = loader.construct_object(value_node, deep=deep) + return mapping + + +_UniqueKeySafeLoader.add_constructor( + yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _construct_unique_mapping +) + + @dataclass class MemoryItem: """Memory item structure.""" @@ -121,6 +162,20 @@ class AgentMemory: • Knowledge Graph: Structured context integration """ + _MARKDOWN_RESERVED_FIELDS = frozenset( + { + "id", + "created_at", + "updated_at", + "type", + "kind", + "entities", + "relationships", + "metadata", + } + ) + _MARKDOWN_EXTENSIONS = frozenset({".md", ".markdown"}) + def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs): """ Initialize agent memory. @@ -206,7 +261,7 @@ class AgentMemory: with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) elif os.path.exists(legacy_path): - # Legacy pickle files: refuse to load them to prevent deserialization attacks. + # Refuse legacy pickle files to prevent deserialization attacks. # Users must re-save memory in the new JSON format. self.logger.warning( f"Legacy pickle file found at {legacy_path}. " @@ -218,9 +273,7 @@ class AgentMemory: return raw_items = data.get("memory_items", {}) - self.memory_items = { - k: MemoryItem.from_dict(v) for k, v in raw_items.items() - } + self.memory_items = {k: MemoryItem.from_dict(v) for k, v in raw_items.items()} raw_index = data.get("memory_index", []) self.memory_index = deque(raw_index, maxlen=self.max_memory_size) self.short_term_memory = [ @@ -253,6 +306,7 @@ class AgentMemory: - memory_id: Custom memory ID - timestamp: Custom timestamp - skip_vector: If True, skip vector store (Short-term only) + - skip_graph: If True, keep entities local to the memory item Returns: Memory ID @@ -313,7 +367,8 @@ class AgentMemory: self.memory_index.append(memory_id) # 3. Update Knowledge Graph - if self.knowledge_graph and entities: + skip_graph = options.get("skip_graph", False) + if self.knowledge_graph and entities and not skip_graph: self.progress_tracker.update_tracking( tracking_id, message="Updating knowledge graph..." ) @@ -516,14 +571,29 @@ class AgentMemory: except Exception as e: self.logger.warning(f"Failed to delete from vector store: {e}") + memory_item = self.memory_items[memory_id] + # Remove from memory del self.memory_items[memory_id] - # Remove from index - if memory_id in self.memory_index: - self.memory_index.remove(memory_id) + # Remove every occurrence in case a custom ID was stored more than once. + self.memory_index = deque( + (item_id for item_id in self.memory_index if item_id != memory_id), + maxlen=self.memory_index.maxlen, + ) + + self.short_term_memory = [ + item for item in self.short_term_memory if item.memory_id != memory_id + ] self.stats["total_items"] = max(0, self.stats["total_items"] - 1) + item_type = memory_item.metadata.get("type", "general") + if item_type in self.stats["items_by_type"]: + self.stats["items_by_type"][item_type] = max( + 0, self.stats["items_by_type"][item_type] - 1 + ) + if self.stats["items_by_type"][item_type] == 0: + del self.stats["items_by_type"][item_type] self.logger.debug(f"Deleted memory item: {memory_id}") return True @@ -628,7 +698,6 @@ class AgentMemory: def _generate_memory_id(self) -> str: """Generate unique memory ID.""" - import hashlib import time timestamp = str(time.time()) @@ -917,6 +986,8 @@ class AgentMemory: memory_id: str, content: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, + entities: Optional[List[EntityDict]] = None, + relationships: Optional[List[RelationshipDict]] = None, **kwargs, ) -> bool: """ @@ -926,6 +997,8 @@ class AgentMemory: memory_id: Memory ID to update content: New content (optional) metadata: New metadata (optional, merged with existing) + entities: Replacement entities (optional) + relationships: Replacement relationships (optional) **kwargs: Additional fields to update Returns: @@ -942,18 +1015,68 @@ class AgentMemory: current_metadata = memory_item.metadata.copy() if metadata: current_metadata.update(metadata) + current_entities = entities if entities is not None else memory_item.entities + current_relationships = ( + relationships if relationships is not None else memory_item.relationships + ) - # Delete old and create new - self.delete_memory(memory_id) - new_id = self.store( + return self._replace_memory_item( + memory_id, current_content, metadata=current_metadata, - entities=memory_item.entities, - relationships=memory_item.relationships, + entities=current_entities, + relationships=current_relationships, + timestamp=kwargs.pop("timestamp", memory_item.timestamp), **kwargs, ) - return new_id is not None + def _replace_memory_item( + self, + memory_id: str, + content: str, + metadata: Dict[str, Any], + entities: List[EntityDict], + relationships: List[RelationshipDict], + timestamp: datetime, + **options, + ) -> bool: + """Replace a memory while retaining its identity and recoverable state.""" + state = self._snapshot_memory_state() + try: + self.delete_memory(memory_id) + new_id = self.store( + content, + metadata=metadata, + entities=entities, + relationships=relationships, + memory_id=memory_id, + timestamp=timestamp, + **options, + ) + except Exception: + self._restore_memory_state(state) + raise + + if new_id != memory_id or not self.exists(memory_id): + self._restore_memory_state(state) + return False + return True + + def _snapshot_memory_state(self) -> Dict[str, Any]: + """Capture mutable in-memory state for an update or import rollback.""" + return { + "memory_items": dict(self.memory_items), + "memory_index": deque(self.memory_index, maxlen=self.memory_index.maxlen), + "short_term_memory": list(self.short_term_memory), + "stats": copy.deepcopy(self.stats), + } + + def _restore_memory_state(self, state: Dict[str, Any]) -> None: + """Restore state captured by :meth:`_snapshot_memory_state`.""" + self.memory_items = state["memory_items"] + self.memory_index = state["memory_index"] + self.short_term_memory = state["short_term_memory"] + self.stats = state["stats"] def delete(self, memory_id: str) -> bool: """ @@ -1332,14 +1455,19 @@ class AgentMemory: # Export/Import def export( - self, conversation_id: Optional[str] = None, format: str = "json", **filters + self, + conversation_id: Optional[str] = None, + format: str = "json", + destination: Optional[Union[str, Path]] = None, + **filters, ) -> Union[str, Dict[str, Any]]: """ Export memories. Args: conversation_id: Export specific conversation (optional) - format: Export format ('json' or 'dict', default: 'json') + format: Export format ('json', 'dict', or 'markdown', default: 'json') + destination: Optional directory for one-file-per-memory Markdown export **filters: Additional filters Returns: @@ -1369,17 +1497,19 @@ class AgentMemory: import json return json.dumps(export_data, indent=2, default=str) + if format == "markdown": + return self._export_markdown(memories, destination=destination) return export_data def import_data( - self, data: Union[str, Dict[str, Any]], format: str = "json" + self, data: Union[str, Path, Dict[str, Any]], format: str = "json" ) -> int: """ Import memories. Args: data: Data to import - format: Data format ('json' or 'dict', default: 'json') + format: Data format ('json', 'dict', or 'markdown', default: 'json') Returns: Number of memories imported @@ -1392,6 +1522,9 @@ class AgentMemory: if isinstance(data, str): data = json.loads(data) + elif format == "markdown": + memories = self._import_markdown_payload(data) + return self._import_markdown_records(memories) if not isinstance(data, dict): raise ValueError("Invalid data format") @@ -1414,6 +1547,451 @@ class AgentMemory: return imported + def _export_markdown( + self, + memories: List[Dict[str, Any]], + destination: Optional[Union[str, Path]] = None, + ) -> str: + documents = [] + filenames = set() + for memory in memories: + filename = self._memory_markdown_filename(memory.get("memory_id")) + normalized_filename = filename.casefold() + if normalized_filename in filenames: + raise ValueError( + f"Cannot export Markdown: duplicate filename {filename!r}." + ) + filenames.add(normalized_filename) + documents.append((filename, self._memory_to_markdown(memory))) + + if destination is None: + if not documents: + return "" + if len(documents) > 1: + raise ValueError( + "Markdown export without a destination supports one memory item. " + "Pass a destination directory to export multiple items." + ) + return documents[0][1] + + destination_path = Path(destination) + if destination_path.exists() and not destination_path.is_dir(): + raise ValueError( + f"Markdown export destination is not a directory: {destination_path}" + ) + destination_path.mkdir(parents=True, exist_ok=True) + + for filename, document in documents: + file_path = destination_path / filename + try: + file_path.write_text(document, encoding="utf-8") + except OSError as exc: + raise OSError( + f"Failed to write Markdown memory to {file_path}" + ) from exc + + return str(destination_path) + + def _memory_to_markdown(self, memory: Dict[str, Any]) -> str: + memory_id = memory.get("memory_id") + source = f"memory {memory_id!r}" + if not isinstance(memory_id, str) or not memory_id.strip(): + raise ValueError(f"Cannot export {source}: 'memory_id' must be a string.") + + raw_metadata = memory.get("metadata") + if raw_metadata is None: + raw_metadata = {} + if not isinstance(raw_metadata, dict): + raise ValueError(f"Cannot export {source}: 'metadata' must be a mapping.") + metadata = dict(raw_metadata) + + created_at = self._parse_markdown_datetime( + memory.get("timestamp"), "created_at", source + ) + updated_at = self._parse_markdown_datetime( + metadata.pop("updated_at", created_at), "updated_at", source + ) + memory_type = metadata.pop("type", "general") + if not isinstance(memory_type, str) or not memory_type.strip(): + raise ValueError(f"Cannot export {source}: 'type' must be a string.") + + frontmatter = { + "id": memory_id, + "created_at": created_at.isoformat(), + "updated_at": updated_at.isoformat(), + "type": memory_type, + } + + nested_metadata = {} + if any(not isinstance(key, str) for key in metadata): + raise ValueError(f"Cannot export {source}: metadata keys must be strings.") + for key in sorted(metadata): + value = metadata[key] + if key in self._MARKDOWN_RESERVED_FIELDS: + nested_metadata[key] = value + else: + frontmatter[key] = value + + if nested_metadata: + frontmatter["metadata"] = nested_metadata + entities = memory.get("entities") + relationships = memory.get("relationships") + if entities is None: + entities = [] + if relationships is None: + relationships = [] + self._validate_markdown_mapping_list(entities, "entities", source) + self._validate_markdown_mapping_list(relationships, "relationships", source) + if entities: + frontmatter["entities"] = entities + if relationships: + frontmatter["relationships"] = relationships + + try: + yaml_text = yaml.safe_dump( + frontmatter, + sort_keys=False, + allow_unicode=True, + default_flow_style=False, + ) + except yaml.YAMLError as exc: + raise ValueError( + f"Cannot export {source}: metadata is not YAML serializable." + ) from exc + + body = memory.get("content", "") + if not isinstance(body, str): + raise ValueError(f"Cannot export {source}: 'content' must be a string.") + return f"---\n{yaml_text}---\n\n{body}" + + def _memory_markdown_filename(self, memory_id: Optional[str]) -> str: + if not isinstance(memory_id, str) or not memory_id.strip(): + raise ValueError( + "Cannot create Markdown filename without a string memory ID." + ) + + slug = re.sub(r"[^A-Za-z0-9._-]+", "-", memory_id) + slug = re.sub(r"-+", "-", slug).strip("._-")[:80].rstrip("._-") + slug = slug or "memory" + digest = hashlib.sha256(memory_id.encode("utf-8")).hexdigest()[:12] + return f"{slug}--{digest}.md" + + def _import_markdown_payload( + self, data: Union[str, Path, Dict[str, Any]] + ) -> List[Tuple[str, Dict[str, Any]]]: + if isinstance(data, Path): + documents = self._read_markdown_path(data) + elif isinstance(data, str): + if not data: + return [] + + documents = None + if "\n" not in data and "\r" not in data: + candidate = Path(data) + try: + if candidate.exists(): + documents = self._read_markdown_path(candidate) + except OSError: + pass + + if documents is None: + documents = [("markdown document", data)] + else: + raise ValueError( + "Invalid Markdown data format. Expected Markdown text or a path." + ) + + memories = [] + memory_sources = {} + for source, document in documents: + memory = self._markdown_to_memory_dict(document, source=source) + memory_id = memory["memory_id"] + if memory_id in memory_sources: + raise ValueError( + f"Duplicate Markdown memory ID {memory_id!r} in {source}; " + f"already defined in {memory_sources[memory_id]}." + ) + memory_sources[memory_id] = source + memories.append((source, memory)) + + return memories + + def _read_markdown_path(self, path: Path) -> List[Tuple[str, str]]: + if not path.exists(): + raise FileNotFoundError(f"Markdown import path does not exist: {path}") + + if path.is_dir(): + file_paths = sorted( + ( + file_path + for file_path in path.iterdir() + if file_path.is_file() + and file_path.suffix.lower() in self._MARKDOWN_EXTENSIONS + ), + key=lambda file_path: (file_path.name.casefold(), file_path.name), + ) + elif path.is_file(): + file_paths = [path] + else: + raise ValueError(f"Markdown import path is not a file or directory: {path}") + + return [ + (str(file_path), file_path.read_text(encoding="utf-8")) + for file_path in file_paths + ] + + def _markdown_to_memory_dict( + self, document: str, source: str = "markdown document" + ) -> Dict[str, Any]: + if not document.startswith("---"): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + "document must start with '---'." + ) + + lines = document.splitlines(keepends=True) + if not lines or lines[0].rstrip("\r\n") != "---": + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + "opening delimiter is malformed." + ) + + closing_index = next( + ( + index + for index, line in enumerate(lines[1:], start=1) + if line.rstrip("\r\n") == "---" + ), + None, + ) + if closing_index is None: + raise ValueError( + f"Invalid Markdown frontmatter in {source}: missing closing '---'." + ) + + yaml_text = "".join(lines[1:closing_index]) + try: + loaded_frontmatter = yaml.load(yaml_text, Loader=_UniqueKeySafeLoader) + frontmatter = {} if loaded_frontmatter is None else loaded_frontmatter + except yaml.YAMLError as exc: + raise ValueError( + f"Invalid Markdown frontmatter in {source}: {exc}" + ) from exc + + if not isinstance(frontmatter, dict): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: expected a YAML mapping." + ) + + if any(not isinstance(key, str) for key in frontmatter): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + "field names must be strings." + ) + + missing_fields = [ + field + for field in ("id", "created_at", "updated_at") + if field not in frontmatter + ] + if "type" not in frontmatter and "kind" not in frontmatter: + missing_fields.append("type or kind") + if missing_fields: + fields = ", ".join(repr(field) for field in missing_fields) + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + f"missing required field(s) {fields}." + ) + + memory_id = frontmatter["id"] + if not isinstance(memory_id, str) or not memory_id.strip(): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: 'id' must be a string." + ) + + memory_type = frontmatter.get("type", frontmatter.get("kind")) + if not isinstance(memory_type, str) or not memory_type.strip(): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + "'type' or 'kind' must be a string." + ) + if ( + "type" in frontmatter + and "kind" in frontmatter + and frontmatter["type"] != frontmatter["kind"] + ): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + "'type' and 'kind' must match when both are provided." + ) + + created_at = self._parse_markdown_datetime( + frontmatter["created_at"], "created_at", source + ) + updated_at = self._parse_markdown_datetime( + frontmatter["updated_at"], "updated_at", source + ) + + nested_metadata = frontmatter.get("metadata", {}) + if nested_metadata is None: + nested_metadata = {} + if not isinstance(nested_metadata, dict): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + "'metadata' must be a mapping." + ) + if any(not isinstance(key, str) for key in nested_metadata): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + "metadata field names must be strings." + ) + conflicting_metadata = {"type", "updated_at"}.intersection(nested_metadata) + if conflicting_metadata: + fields = ", ".join(sorted(conflicting_metadata)) + raise ValueError( + f"Invalid Markdown frontmatter in {source}: nested metadata " + f"duplicates reserved field(s): {fields}." + ) + + entities = frontmatter.get("entities", []) + relationships = frontmatter.get("relationships", []) + self._validate_markdown_mapping_list(entities, "entities", source) + self._validate_markdown_mapping_list(relationships, "relationships", source) + + metadata = dict(nested_metadata) + for key, value in frontmatter.items(): + if key not in self._MARKDOWN_RESERVED_FIELDS: + if key in metadata: + raise ValueError( + f"Invalid Markdown frontmatter in {source}: metadata field " + f"{key!r} is defined both at the top level and in 'metadata'." + ) + metadata[key] = value + metadata["type"] = memory_type + metadata["updated_at"] = updated_at.isoformat() + + body = "".join(lines[closing_index + 1 :]) + if body.startswith("\r\n"): + body = body[2:] + elif body.startswith("\n"): + body = body[1:] + + return { + "memory_id": memory_id, + "content": body, + "timestamp": created_at, + "metadata": metadata, + "entities": entities, + "relationships": relationships, + } + + def _import_markdown_records( + self, memories: List[Tuple[str, Dict[str, Any]]] + ) -> int: + if not memories: + return 0 + + state = self._snapshot_memory_state() + imported = 0 + current_source = "markdown document" + current_memory_id = "unknown" + try: + for current_source, memory in memories: + current_memory_id = memory["memory_id"] + if self._markdown_record_matches(current_memory_id, memory): + imported += 1 + continue + + if self.exists(current_memory_id): + success = self._replace_memory_item( + current_memory_id, + memory["content"], + metadata=memory["metadata"], + entities=memory["entities"], + relationships=memory["relationships"], + timestamp=memory["timestamp"], + skip_graph=True, + ) + else: + stored_id = self.store( + memory["content"], + metadata=memory["metadata"], + entities=memory["entities"], + relationships=memory["relationships"], + memory_id=current_memory_id, + timestamp=memory["timestamp"], + skip_graph=True, + ) + success = stored_id == current_memory_id and self.exists( + current_memory_id + ) + + if not success: + raise RuntimeError("memory store did not confirm the requested ID") + imported += 1 + except Exception as exc: + self._restore_memory_state(state) + raise RuntimeError( + f"Failed to import Markdown memory {current_memory_id!r} " + f"from {current_source}: {exc}" + ) from exc + + return imported + + def _markdown_record_matches(self, memory_id: str, memory: Dict[str, Any]) -> bool: + existing = self.memory_items.get(memory_id) + if existing is None: + return False + + return ( + existing.content == memory["content"] + and existing.timestamp == memory["timestamp"] + and existing.metadata == memory["metadata"] + and existing.entities == memory["entities"] + and existing.relationships == memory["relationships"] + ) + + def _parse_markdown_datetime( + self, value: Any, field_name: str, source: str + ) -> datetime: + if isinstance(value, datetime): + parsed = value + elif isinstance(value, date): + parsed = datetime.combine(value, datetime.min.time()) + elif isinstance(value, str) and value.strip(): + candidate = value.strip() + if candidate.endswith("Z"): + candidate = candidate[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(candidate) + except ValueError as exc: + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + f"'{field_name}' must be an ISO-8601 datetime." + ) from exc + else: + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + f"'{field_name}' must be an ISO-8601 datetime." + ) + + return parsed + + def _validate_markdown_mapping_list( + self, value: Any, field_name: str, source: str + ) -> None: + if not isinstance(value, list): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + f"'{field_name}' must be a list." + ) + for index, item in enumerate(value): + if not isinstance(item, dict): + raise ValueError( + f"Invalid Markdown frontmatter in {source}: " + f"'{field_name}[{index}]' must be a mapping." + ) + # Statistics def stats(self, **filters) -> Dict[str, Any]: """ diff --git a/tests/context/test_agent_memory_markdown.py b/tests/context/test_agent_memory_markdown.py new file mode 100644 index 00000000..cf2c3dbf --- /dev/null +++ b/tests/context/test_agent_memory_markdown.py @@ -0,0 +1,406 @@ +from copy import deepcopy +from datetime import datetime +from unittest.mock import MagicMock, patch + +import pytest +import yaml + +from semantica.context.agent_memory import AgentMemory + + +def markdown_document(frontmatter, body=""): + yaml_text = yaml.safe_dump(frontmatter, sort_keys=False, allow_unicode=True) + return f"---\n{yaml_text}---\n\n{body}" + + +def required_frontmatter(memory_id="mem_test", **overrides): + frontmatter = { + "id": memory_id, + "created_at": "2026-07-22T09:00:00+00:00", + "updated_at": "2026-07-22T10:00:00+00:00", + "type": "note", + } + frontmatter.update(overrides) + return frontmatter + + +def test_markdown_round_trip_is_lossless_and_deterministic(): + memory = AgentMemory() + content = "\n# Editable memory\n\n\n\n" + memory.store( + content, + metadata={ + "type": "note", + "updated_at": "2026-07-22T12:30:00+00:00", + "source": "caf\u00e9 review", + "tags": ["alpha", "beta"], + "id": "metadata-id", + "kind": "metadata-kind", + }, + entities=[{"id": "ent_1", "type": "topic", "text": "Memory"}], + relationships=[ + { + "source_id": "ent_1", + "target_id": "ent_2", + "type": "related_to", + "confidence": 0.9, + } + ], + memory_id="mem_round_trip", + timestamp=datetime.fromisoformat("2026-07-22T12:00:00+00:00"), + ) + + exported = memory.export(format="markdown") + restored = AgentMemory() + + assert restored.import_data(exported, format="markdown") == 1 + assert restored.get("mem_round_trip") == memory.get("mem_round_trip") + assert restored.export(format="markdown") == exported + assert "caf\u00e9 review" in exported + + +def test_markdown_directory_round_trip_uses_one_stable_file_per_memory(tmp_path): + memory = AgentMemory() + for index in range(2): + timestamp = f"2026-07-22T0{index + 8}:00:00" + memory.store( + f"Memory {index}", + metadata={ + "type": "reference", + "updated_at": timestamp, + "source": f"source-{index}", + }, + memory_id=f"mem_{index}", + timestamp=datetime.fromisoformat(timestamp), + ) + + first_export = tmp_path / "first" + assert memory.export(format="markdown", destination=first_export) == str( + first_export + ) + first_files = sorted(first_export.glob("*.md")) + assert len(first_files) == 2 + + restored = AgentMemory() + assert restored.import_data(first_export, format="markdown") == 2 + assert set(restored.memory_items) == {"mem_0", "mem_1"} + + second_export = tmp_path / "second" + restored.export(format="markdown", destination=second_export) + assert {path.name: path.read_text(encoding="utf-8") for path in first_files} == { + path.name: path.read_text(encoding="utf-8") + for path in sorted(second_export.glob("*.md")) + } + + +def test_multiple_memories_require_a_destination_but_filters_can_select_one(): + memory = AgentMemory() + for memory_type in ("note", "reference"): + memory.store( + memory_type, + metadata={ + "type": memory_type, + "updated_at": "2026-07-22T10:00:00", + }, + memory_id=f"mem_{memory_type}", + timestamp=datetime(2026, 7, 22, 9, 0, 0), + ) + + with pytest.raises(ValueError, match="destination"): + memory.export(format="markdown") + + exported = memory.export(format="markdown", type="note") + assert "id: mem_note" in exported + assert "id: mem_reference" not in exported + + +def test_markdown_update_replaces_supported_fields_and_preserves_id(): + memory = AgentMemory() + memory.store( + "Original content", + metadata={ + "type": "note", + "updated_at": "2026-07-22T09:30:00", + "remove_me": True, + }, + memory_id="mem_existing", + timestamp=datetime(2026, 7, 22, 9, 0, 0), + ) + edited = markdown_document( + required_frontmatter( + "mem_existing", + created_at="2026-07-22T09:15:00", + updated_at="2026-07-22T11:00:00", + type="decision", + source="manual edit", + ), + "Updated content", + ) + + assert memory.import_data(edited, format="markdown") == 1 + + updated = memory.get("mem_existing") + assert memory.count() == 1 + assert updated["memory_id"] == "mem_existing" + assert updated["content"] == "Updated content" + assert updated["timestamp"] == "2026-07-22T09:15:00" + assert updated["metadata"] == { + "type": "decision", + "source": "manual edit", + "updated_at": "2026-07-22T11:00:00", + } + assert memory.stats["items_by_type"] == {"decision": 1} + assert [item.memory_id for item in memory.short_term_memory] == ["mem_existing"] + + +def test_public_update_preserves_id_and_statistics(): + memory = AgentMemory() + memory.store( + "Original", + metadata={"type": "note"}, + memory_id="mem_update", + timestamp=datetime(2026, 7, 22, 9, 0, 0), + ) + + assert memory.update("mem_update", content="Updated", metadata={"type": "decision"}) + assert memory.get("mem_update")["content"] == "Updated" + assert set(memory.memory_items) == {"mem_update"} + assert list(memory.memory_index) == ["mem_update"] + assert memory.stats["total_items"] == 1 + assert memory.stats["items_by_type"] == {"decision": 1} + + +def test_reimporting_unchanged_markdown_does_not_rewrite_memory(): + memory = AgentMemory() + document = markdown_document(required_frontmatter(), "Unchanged") + assert memory.import_data(document, format="markdown") == 1 + before = memory.get("mem_test") + + with patch.object( + memory, "_replace_memory_item", wraps=memory._replace_memory_item + ) as replace: + assert memory.import_data(document, format="markdown") == 1 + + replace.assert_not_called() + assert memory.get("mem_test") == before + assert memory.count() == 1 + + +def test_kind_is_accepted_as_the_type_alias(): + frontmatter = required_frontmatter() + frontmatter["kind"] = frontmatter.pop("type") + + memory = AgentMemory() + assert ( + memory.import_data( + markdown_document(frontmatter, "Kind alias"), format="markdown" + ) + == 1 + ) + assert memory.get("mem_test")["metadata"]["type"] == "note" + + +@pytest.mark.parametrize("missing_field", ["id", "created_at", "updated_at", "type"]) +def test_required_frontmatter_fields_are_enforced(missing_field): + frontmatter = required_frontmatter() + frontmatter.pop(missing_field) + + with pytest.raises(ValueError, match="missing required"): + AgentMemory().import_data( + markdown_document(frontmatter, "Invalid"), format="markdown" + ) + + +@pytest.mark.parametrize( + ("overrides", "message"), + [ + ({"id": 123}, "'id' must be a string"), + ({"created_at": "yesterday"}, "ISO-8601"), + ({"updated_at": "later"}, "ISO-8601"), + ({"type": ["note"]}, "must be a string"), + ({"kind": "decision"}, "must match"), + ({"metadata": []}, "must be a mapping"), + ({"entities": {}}, "must be a list"), + ({"entities": ["entity"]}, "must be a mapping"), + ({"relationships": [1]}, "must be a mapping"), + ({"metadata": {"type": "nested"}}, "duplicates reserved"), + ({"metadata": {"source": "nested"}, "source": "top"}, "defined both"), + ], +) +def test_invalid_frontmatter_values_return_actionable_errors(overrides, message): + document = markdown_document(required_frontmatter(**overrides), "Invalid") + + with pytest.raises(ValueError, match=message): + AgentMemory().import_data(document, format="markdown") + + +@pytest.mark.parametrize( + "document", + [ + "not frontmatter", + "---\nid: broken", + "---\nid: [unterminated\n---\n", + ( + "---\nid: first\nid: second\ncreated_at: 2026-07-22T09:00:00\n" + "updated_at: 2026-07-22T10:00:00\ntype: note\n---\n" + ), + ], +) +def test_malformed_frontmatter_is_rejected(document): + with pytest.raises(ValueError, match="frontmatter"): + AgentMemory().import_data(document, format="markdown") + + +def test_directory_is_fully_validated_before_any_memory_is_changed(tmp_path): + valid = markdown_document(required_frontmatter("mem_valid"), "Valid") + invalid_fields = required_frontmatter("mem_invalid") + invalid_fields.pop("updated_at") + invalid = markdown_document(invalid_fields, "Invalid") + (tmp_path / "a-valid.md").write_text(valid, encoding="utf-8") + (tmp_path / "z-invalid.md").write_text(invalid, encoding="utf-8") + + memory = AgentMemory() + with pytest.raises(ValueError, match="updated_at"): + memory.import_data(tmp_path, format="markdown") + + assert memory.count() == 0 + + +def test_duplicate_ids_in_a_directory_are_rejected_before_import(tmp_path): + document = markdown_document(required_frontmatter("mem_duplicate"), "Body") + (tmp_path / "first.md").write_text(document, encoding="utf-8") + (tmp_path / "second.markdown").write_text(document, encoding="utf-8") + + memory = AgentMemory() + with pytest.raises(ValueError, match="Duplicate Markdown memory ID"): + memory.import_data(tmp_path, format="markdown") + + assert memory.count() == 0 + + +def test_markdown_import_keeps_provenance_out_of_the_context_graph(): + knowledge_graph = MagicMock() + memory = AgentMemory(knowledge_graph=knowledge_graph) + fields = required_frontmatter( + entities=[{"id": "entity_1", "type": "topic"}], + relationships=[ + { + "source_id": "entity_1", + "target_id": "entity_2", + "type": "related_to", + } + ], + ) + + assert ( + memory.import_data(markdown_document(fields, "Provenance"), format="markdown") + == 1 + ) + assert memory.get("mem_test")["entities"][0]["id"] == "entity_1" + knowledge_graph.add_nodes.assert_not_called() + knowledge_graph.add_edges.assert_not_called() + + +def test_operational_failure_restores_existing_in_memory_state(): + memory = AgentMemory() + memory.store( + "Original", + metadata={ + "type": "note", + "updated_at": "2026-07-22T09:30:00", + }, + memory_id="mem_existing", + timestamp=datetime(2026, 7, 22, 9, 0, 0), + ) + before_memory = deepcopy(memory.get("mem_existing")) + before_stats = deepcopy(memory.stats) + edited_fields = required_frontmatter("mem_existing") + original_store = memory.store + + def fail_after_store(*args, **kwargs): + original_store(*args, **kwargs) + raise RuntimeError("store unavailable") + + with patch.object(memory, "store", side_effect=fail_after_store): + with pytest.raises(RuntimeError, match="store unavailable"): + memory.import_data( + markdown_document(edited_fields, "Edited"), format="markdown" + ) + + assert memory.get("mem_existing") == before_memory + assert memory.stats == before_stats + assert list(memory.memory_index) == ["mem_existing"] + assert [item.memory_id for item in memory.short_term_memory] == ["mem_existing"] + + +def test_import_does_not_report_retention_pruned_memory_as_successful(): + memory = AgentMemory(retention_policy="1_days") + fields = required_frontmatter( + "mem_expired", + created_at="2000-01-01T00:00:00", + updated_at="2000-01-01T00:00:00", + ) + + with pytest.raises(RuntimeError, match="did not confirm"): + memory.import_data(markdown_document(fields, "Expired"), format="markdown") + + assert memory.count() == 0 + assert memory.stats["total_items"] == 0 + + +def test_exported_filenames_cannot_escape_or_collide_with_destination(tmp_path): + memory = AgentMemory() + for memory_id in ("../outside", "a/b", "a\\b", "A/B"): + memory.store( + memory_id, + metadata={ + "type": "note", + "updated_at": "2026-07-22T10:00:00", + }, + memory_id=memory_id, + timestamp=datetime(2026, 7, 22, 9, 0, 0), + ) + + destination = tmp_path / "export" + memory.export(format="markdown", destination=destination) + exported_files = list(destination.glob("*.md")) + + assert len(exported_files) == 4 + assert len({path.name.casefold() for path in exported_files}) == 4 + assert all(path.parent == destination for path in exported_files) + assert not (tmp_path / "outside.md").exists() + + +def test_empty_markdown_export_and_import_are_no_ops(): + memory = AgentMemory() + + assert memory.export(format="markdown") == "" + assert memory.import_data("", format="markdown") == 0 + + +def test_legacy_dict_import_behavior_is_unchanged(): + memory = AgentMemory() + data = { + "memories": [ + { + "memory_id": "source_id", + "content": "Legacy import", + "metadata": {"type": "note"}, + "timestamp": "2000-01-01T00:00:00", + } + ] + } + + assert memory.import_data(data, format="dict") == 1 + assert not memory.exists("source_id") + imported = next(iter(memory.memory_items.values())) + assert imported.content == "Legacy import" + assert imported.metadata == {"type": "note"} + + +def test_markdown_export_destination_must_be_a_directory(tmp_path): + destination = tmp_path / "memory.md" + destination.write_text("occupied", encoding="utf-8") + + with pytest.raises(ValueError, match="not a directory"): + AgentMemory().export(format="markdown", destination=destination)