Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-memory/src/lexigram/ai/memory/episodic/store.py: 68%

1"""Episodic memory store — records and recalls past interaction turns.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.contracts.core import HealthCheckResult, HealthStatus 

8from lexigram.logging import ( 

9 get_logger, 

10) 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.ai.memory import ( 

14 MemoryEntry, 

15 MemoryQuery, 

16 MemorySearchResult, 

17 MemoryStoreProtocol, 

18 ) 

19 

20logger = get_logger(__name__) 

21 

22 

23class EpisodicMemoryStore: 

24 """Episodic memory layer backed by a pluggable MemoryStoreProtocol.""" 

25 

26 def __init__(self, backend: MemoryStoreProtocol) -> None: 

27 self._backend = backend 

28 

29 async def record(self, entry: MemoryEntry) -> None: 

30 await self._backend.store(entry) 

31 logger.debug("episodic_recorded", entry_id=entry.id, role=entry.role) 

32 

33 async def recall(self, query: MemoryQuery) -> list[MemorySearchResult]: 

34 return await self._backend.retrieve(query) 

35 

36 async def forget(self, entry_id: str, owner_id: str) -> None: 

37 """Forget a specific episode within an owner scope. 

38 

39 Args: 

40 entry_id: ID of the entry to forget. 

41 owner_id: Owner scope preventing cross-owner deletion. 

42 """ 

43 await self._backend.delete(entry_id, owner_id) 

44 logger.debug("episodic_forgotten", entry_id=entry_id, owner_id=owner_id) 

45 

46 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

47 return HealthCheckResult( 

48 component="memory.episodic", 

49 status=HealthStatus.HEALTHY, 

50 details={"timeout": timeout}, 

51 ) 

52 

53 

54__all__ = ["EpisodicMemoryStore"]