memory.py

"""
Agent-Q3 — Redis-backed shared memory layer.

Used by all services (multimodal, coder, research, monitor) to:
- Cache MCP tool call results (TTL'd)
- Store conversation history per session
- Persist monitor metrics across restarts
- Track skill activation log
- Coordinate cross-service rate limits

Falls back to a no-op in-memory store if REDIS_URL is unset.
"""

import json
import time
from typing import Any

import structlog

from .config import settings

log = structlog.get_logger(__name__)


class _InMemoryFallback:
    """Drop-in async stub used when REDIS_URL is unset."""

    def __init__(self):
        self._store: dict[str, tuple[float, str]] = {}
        self._lists: dict[str, list] = {}

    async def get(self, key: str) -> str | None:
        item = self._store.get(key)
        if not item:
            return None
        expires_at, value = item
        if expires_at and time.time() > expires_at:
            self._store.pop(key, None)
            return None
        return value

    async def set(self, key: str, value: str, ex: int | None = None) -> None:
        exp = (time.time() + ex) if ex else 0
        self._store[key] = (exp, value)

    async def delete(self, *keys: str) -> None:
        for k in keys:
            self._store.pop(k, None)

    async def lpush(self, key: str, *values: str) -> int:
        self._lists.setdefault(key, [])
        for v in values:
            self._lists[key].insert(0, v)
        return len(self._lists[key])

    async def lrange(self, key: str, start: int, stop: int) -> list:
        lst = self._lists.get(key, [])
        if stop == -1:
            return lst[start:]
        return lst[start : stop + 1]

    async def ltrim(self, key: str, start: int, stop: int) -> None:
        lst = self._lists.get(key, [])
        if stop == -1:
            self._lists[key] = lst[start:]
        else:
            self._lists[key] = lst[start : stop + 1]

    async def ping(self) -> bool:
        return True

    async def close(self) -> None:
        pass


class Memory:
    """Async memory facade. Uses redis.asyncio when REDIS_URL set, else in-memory."""

    def __init__(self):
        self._client = None
        self._backend: str = "uninitialized"

    async def connect(self) -> None:
        if settings.redis_url:
            try:
                from redis.asyncio import from_url

                self._client = from_url(settings.redis_url, decode_responses=True)
                await self._client.ping()
                self._backend = "redis"
                log.info("memory connected", backend="redis", url=settings.redis_url.split("@")[-1])
                return
            except Exception as e:
                log.warning("redis unavailable, falling back to in-memory", error=str(e))
        self._client = _InMemoryFallback()
        self._backend = "in-memory"
        log.info("memory connected", backend="in-memory")

    @property
    def backend(self) -> str:
        return self._backend

    @property
    def client(self):
        return self._client

    # ── Typed helpers ────────────────────────────────────────────────────────

    async def cache_get(self, key: str) -> Any | None:
        raw = await self._client.get(f"cache:{key}")
        return json.loads(raw) if raw else None

    async def cache_set(self, key: str, value: Any, ttl_secs: int = 300) -> None:
        await self._client.set(f"cache:{key}", json.dumps(value), ex=ttl_secs)

    async def log_event(self, channel: str, event: dict, max_history: int = 500) -> None:
        key = f"events:{channel}"
        await self._client.lpush(key, json.dumps({"ts": time.time(), **event}))
        await self._client.ltrim(key, 0, max_history - 1)

    async def get_events(self, channel: str, limit: int = 50) -> list[dict]:
        raw_items = await self._client.lrange(f"events:{channel}", 0, limit - 1)
        return [json.loads(x) for x in raw_items]

    async def session_append(self, session_id: str, message: dict, max_turns: int = 50) -> None:
        key = f"session:{session_id}"
        await self._client.lpush(key, json.dumps(message))
        await self._client.ltrim(key, 0, max_turns - 1)

    async def session_history(self, session_id: str, limit: int = 50) -> list[dict]:
        raw_items = await self._client.lrange(f"session:{session_id}", 0, limit - 1)
        return [json.loads(x) for x in reversed(raw_items)]

    async def ping(self) -> bool:
        try:
            return await self._client.ping()
        except Exception:
            return False

    async def close(self) -> None:
        try:
            await self._client.close()
        except Exception:
            pass


memory = Memory()