graph store memory

src/kontextual_engine/adapters/memory/graph_store.py

@@ -6,6 +6,7 @@ from dataclasses import dataclass, field
 from typing import Iterable
 
 from kontextual_engine.core import (
+    AuditEvent,
     LifecycleState,
     MemoryEdgeRecord,
     MemoryEventRecord,
@@ -21,6 +22,7 @@ class InMemoryMemoryGraphRepository:
     nodes: dict[str, MemoryNodeRecord] = field(default_factory=dict)
     edges: dict[str, MemoryEdgeRecord] = field(default_factory=dict)
     events: dict[str, MemoryEventRecord] = field(default_factory=dict)
+    audit_events: dict[str, AuditEvent] = field(default_factory=dict)
 
     def save_memory_profile(self, profile: MemoryProfileRecord) -> MemoryProfileRecord:
         self.profiles[str(profile.profile_id)] = profile
@@ -124,3 +126,38 @@ class InMemoryMemoryGraphRepository:
         if kind is not None:
             events = [event for event in events if event.kind == kind]
         return sorted(events, key=lambda event: (event.graph_id, event.timestamp, event.contract_event_id))
+
+    def save_memory_audit_event(self, event: AuditEvent) -> AuditEvent:
+        self.audit_events[str(event.event_id)] = event
+        return event
+
+    def get_memory_audit_event(self, event_id: str) -> AuditEvent:
+        try:
+            return self.audit_events[event_id]
+        except KeyError as exc:
+            raise NotFoundError("Memory audit event not found", details={"event_id": event_id}) from exc
+
+    def list_memory_audit_events(
+        self,
+        *,
+        graph_id: str | None = None,
+        correlation_id: str | None = None,
+        operation: str | None = None,
+    ) -> list[AuditEvent]:
+        events: Iterable[AuditEvent] = self.audit_events.values()
+        if graph_id is not None:
+            events = [event for event in events if _audit_event_matches_graph(event, graph_id)]
+        if correlation_id is not None:
+            events = [event for event in events if event.correlation_id == correlation_id]
+        if operation is not None:
+            events = [event for event in events if event.operation == operation]
+        return sorted(events, key=lambda event: (event.occurred_at, event.event_id))
+
+
+def _audit_event_matches_graph(event: AuditEvent, graph_id: str) -> bool:
+    if event.target == f"memory-graph:{graph_id}":
+        return True
+    if event.details.get("graph_id") == graph_id:
+        return True
+    request = event.details.get("request")
+    return isinstance(request, dict) and request.get("graph_id") == graph_id