Coverage for src \ truenex_memory \ core \ memory_service.py: 94%

1"""High-level local memory service used by CLI and adapters.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6 

7from truenex_memory.core.config import ensure_project_dirs, resolve_project_config 

8from truenex_memory.core.indexer import index_path 

9from truenex_memory.retrieval.semantic import HashingEmbedder 

10from truenex_memory.store.models import MemoryNode, RetrievalLog, SearchHit 

11from truenex_memory.store.qdrant_store import QdrantVectorStore, VectorStoreUnavailable 

12from truenex_memory.store.repository import MemoryRepository 

13 

14 

15class MemoryService: 

16 """Facade around local configuration, indexing and repository operations.""" 

17 

18 def __init__(self, project_root: Path | str = ".") -> None: 

19 self.config = resolve_project_config(project_root) 

20 self.embedder = HashingEmbedder() 

21 self.vector_store_status: dict[str, object] = { 

22 "backend": self.config.vector_backend, 

23 "active_backend": "sqlite", 

24 "qdrant_url": self.config.qdrant_url, 

25 "qdrant_collection": self.config.qdrant_collection, 

26 "available": False, 

27 "error": None, 

28 } 

29 vector_store = None 

30 if self.config.vector_backend == "qdrant": 

31 try: 

32 vector_store = QdrantVectorStore( 

33 collection_name=self.config.qdrant_collection, 

34 dimensions=self.embedder.dimensions, 

35 url=self.config.qdrant_url, 

36 ) 

37 vector_store.initialize() 

38 self.vector_store_status.update({"active_backend": "qdrant", "available": True}) 

39 except VectorStoreUnavailable as exc: 

40 self.vector_store_status.update({"error": str(exc)}) 

41 self.repository = MemoryRepository( 

42 self.config.db_path, 

43 embedder=self.embedder, 

44 vector_store=vector_store, 

45 ) 

46 

47 def init_project(self) -> None: 

48 ensure_project_dirs(self.config) 

49 self.repository.initialize() 

50 

51 def add(self, content: str, *, memory_type: str = "note") -> str: 

52 self.init_project() 

53 return self.repository.add_memory(content, memory_type=memory_type) 

54 

55 def index(self, path: Path | str) -> int: 

56 self.init_project() 

57 return index_path(Path(path), project_root=self.config.project_root, repository=self.repository) 

58 

59 def search(self, query: str, *, top_k: int = 5, include_inactive: bool = False) -> list[SearchHit]: 

60 self.init_project() 

61 return self.repository.search(query, top_k=top_k, include_inactive=include_inactive) 

62 

63 def list_memory_nodes(self, *, status: str | None = None) -> list[MemoryNode]: 

64 self.init_project() 

65 return self.repository.list_memory_nodes(status=status) 

66 

67 def set_memory_status(self, memory_id: str, status: str) -> None: 

68 self.init_project() 

69 self.repository.set_memory_status(memory_id, status) 

70 

71 def stats(self) -> dict[str, int]: 

72 self.init_project() 

73 return self.repository.stats() 

74 

75 def list_retrieval_logs(self, *, limit: int = 20) -> list[RetrievalLog]: 

76 self.init_project() 

77 return self.repository.list_retrieval_logs(limit=limit) 

78 

79 def get_retrieval_log(self, trace_id: str) -> RetrievalLog | None: 

80 self.init_project() 

81 return self.repository.get_retrieval_log(trace_id) 

82 

83 @property 

84 def last_trace_id(self) -> str | None: 

85 return self.repository.last_trace_id 

86 

87 def vector_status(self) -> dict[str, object]: 

88 """Return vector backend status without exposing project content.""" 

89 

90 return dict(self.vector_store_status)