Coverage for src / lilbee / server / handlers / memory.py: 100%

23 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-06-28 01:01 +0000

1"""Memory CRUD handlers shared by the HTTP memory routes. 

2 

3Each handler delegates to :mod:`lilbee.app.memory` so the embedding, owner 

4scoping, and store access live in one place. All operate on the human's 

5``owner=local`` memories; agent-scoped memory is reached through MCP only. 

6""" 

7 

8from __future__ import annotations 

9 

10from litestar.exceptions import NotFoundException 

11 

12from lilbee.app.memory import ( 

13 MEMORY_DISABLED_HINT, 

14 forget, 

15 list_memories, 

16 memory_enabled, 

17 remember, 

18 set_memory_shared, 

19) 

20from lilbee.data.store import MemoryKind 

21from lilbee.server.models import ( 

22 MemoryFlagsResponse, 

23 MemoryItem, 

24 MemoryListResponse, 

25 MemoryRemoveResponse, 

26 RememberResponse, 

27) 

28 

29 

30def _require_memory() -> None: 

31 """Raise 404 if the memory subsystem is disabled (off by default).""" 

32 if not memory_enabled(): 

33 raise NotFoundException(detail=MEMORY_DISABLED_HINT) 

34 

35 

36async def remember_memory(text: str, kind: MemoryKind, shared: bool) -> RememberResponse: 

37 """Store a memory and return its id and kind.""" 

38 _require_memory() 

39 memory_id = remember(text, kind=kind, shared=shared) 

40 return RememberResponse(id=memory_id, kind=kind) 

41 

42 

43async def list_local_memories() -> MemoryListResponse: 

44 """Return the human's stored memories, newest first.""" 

45 _require_memory() 

46 return MemoryListResponse( 

47 memories=[ 

48 MemoryItem( 

49 id=m.id, 

50 kind=m.kind, 

51 shared=m.shared, 

52 text=m.text, 

53 ) 

54 for m in list_memories() 

55 ] 

56 ) 

57 

58 

59async def update_memory_shared(memory_id: str, shared: bool) -> MemoryFlagsResponse: 

60 """Set a memory's shared-with-agents flag.""" 

61 _require_memory() 

62 updated = set_memory_shared(memory_id, shared=shared) 

63 return MemoryFlagsResponse(id=memory_id, updated=updated) 

64 

65 

66async def remove_memory(memory_id: str) -> MemoryRemoveResponse: 

67 """Delete a memory by id.""" 

68 _require_memory() 

69 forget(memory_id) 

70 return MemoryRemoveResponse(removed=memory_id)