Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-governance/src/lexigram/ai/governance/audit/memory.py: 19%

1from __future__ import annotations 

2 

3from lexigram.ai.governance.audit.models import AIAuditEvent, AuditQuery, AuditSummary 

4 

5 

6class InMemoryAuditStore: 

7 """In-memory audit store for testing and development. 

8 

9 Not suitable for production — events are lost on process restart. 

10 The store is intentionally simple: append-only list with linear scan 

11 queries. For production use, implement :class:`AIAuditStore` with a 

12 durable backend (e.g. via ``DatabaseProviderProtocol``). 

13 """ 

14 

15 def __init__(self) -> None: 

16 self._events: list[AIAuditEvent] = [] 

17 

18 async def record(self, event: AIAuditEvent) -> None: 

19 """Append event to the in-memory list.""" 

20 self._events.append(event) 

21 

22 async def query(self, query: AuditQuery) -> list[AIAuditEvent]: 

23 """Linear scan with filter, ordered by timestamp descending.""" 

24 results = self._filter(query) 

25 results.sort(key=lambda e: e.timestamp, reverse=True) 

26 return results[query.offset : query.offset + query.limit] 

27 

28 async def aggregate(self, query: AuditQuery) -> AuditSummary: 

29 """Compute summary statistics over matching events.""" 

30 events = self._filter(query) 

31 summary = AuditSummary(total_events=len(events)) 

32 

33 for event in events: 

34 if event.cost is not None: 

35 summary.total_spend += event.cost 

36 if event.tokens is not None: 

37 summary.total_tokens += event.tokens 

38 if event.status == "denied": 

39 summary.denied_count += 1 

40 

41 model_key = event.model or "unknown" 

42 summary.by_model[model_key] = summary.by_model.get(model_key, 0) + 1 

43 

44 user_key = event.user_id or "anonymous" 

45 summary.by_user[user_key] = summary.by_user.get(user_key, 0) + 1 

46 

47 type_key = event.event_type.value 

48 summary.by_event_type[type_key] = summary.by_event_type.get(type_key, 0) + 1 

49 

50 return summary 

51 

52 def _filter(self, query: AuditQuery) -> list[AIAuditEvent]: 

53 """Apply filter criteria and return matching events.""" 

54 results: list[AIAuditEvent] = [] 

55 for event in self._events: 

56 if query.start and event.timestamp < query.start: 

57 continue 

58 if query.end and event.timestamp > query.end: 

59 continue 

60 if query.event_types and event.event_type not in query.event_types: 

61 continue 

62 if query.user_id and event.user_id != query.user_id: 

63 continue 

64 if query.model and event.model != query.model: 

65 continue 

66 if query.provider and event.provider != query.provider: 

67 continue 

68 if query.status and event.status != query.status: 

69 continue 

70 results.append(event) 

71 return results 

72 

73 

74# --------------------------------------------------------------------------- 

75# Database-backed implementation 

76# ---------------------------------------------------------------------------