ai-agent/MemoryAgents/backend/app/adapters/base.py

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any

from ..config import settings
from ..db import repository, utcnow
from ..llm import llm
from ..schemas import AuditEvent, ChatResponse, MemoryItem, SystemDescriptor


class AdapterUnavailable(RuntimeError):
    pass


class MemoryAgent(ABC):
    id: str
    descriptor: SystemDescriptor

    def __init__(self) -> None:
        self.repo = repository
        self.llm = llm

    @property
    def status(self) -> SystemDescriptor:
        return self.descriptor

    async def memories(self, query: str | None = None) -> list[MemoryItem]:
        return await self.repo.list_memories(self.id, query)

    async def audit(self) -> list[AuditEvent]:
        return await self.repo.list_audit(self.id)

    async def reset(self) -> None:
        await self.repo.reset(self.id)

    async def delete_memory(self, memory_id: str) -> bool:
        deleted = await self.repo.delete_memory(self.id, memory_id)
        if deleted:
            await self.repo.delete_embedding(self.id, memory_id)
            await self._audit("DELETE/Memory", target=memory_id)
        return deleted

    async def _audit(
        self,
        operation: str,
        target: str | None = None,
        status: str = "ok",
        details: dict[str, Any] | None = None,
    ) -> AuditEvent:
        event = AuditEvent(
            id=self.repo.memory_id("audit"),
            system=self.id,
            workspace_id=settings.workspace_id,
            operation=operation,
            target=target,
            status=status,
            details=details or {},
            created_at=utcnow(),
        )
        return await self.repo.add_audit(event)

    @abstractmethod
    async def chat(self, message: str) -> ChatResponse:
        raise NotImplementedError