| """
|
| 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
|
|
|
|
|
|
|
| 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()
|
|
|
