"""RLM Neo-Cortex -- Module 1: Memory Gateway FastAPI Router. Exposes the MemoryGateway as a REST API mounted under /api/v1/memory. Routes: POST /api/v1/memory/write Write a new memory POST /api/v1/memory/search Vector-search memories GET /api/v1/memory/health Health-check all backends DELETE /api/v1/memory/{tid}/{mid} Delete a specific memory Story 1.09. """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional from uuid import UUID from .gateway import MemoryGateway, QuotaExceededError, PartialDeleteError logger = logging.getLogger("core.rlm.gateway_router") # --------------------------------------------------------------------------- # Pydantic models — defined at module level so FastAPI can resolve them # --------------------------------------------------------------------------- try: from pydantic import BaseModel, Field, field_validator class WriteMemoryRequest(BaseModel): """Request body for POST /write.""" tenant_id: UUID content: str = Field(..., min_length=10, description="Memory content (min 10 chars)") source: str = Field(..., min_length=1) domain: str = Field(..., min_length=1) metadata: Optional[Dict[str, Any]] = Field(default=None) @field_validator("content") @classmethod def content_not_blank(cls, v: str) -> str: if not v or not v.strip(): raise ValueError("content must not be blank") return v class WriteMemoryResponse(BaseModel): """Response body for POST /write.""" memory_id: Optional[str] = None tenant_id: UUID tier: str surprise_score: float created_at: str pg_id: Optional[int] = None class SearchRequest(BaseModel): """Request body for POST /search.""" tenant_id: UUID query: str = Field(..., min_length=1) limit: int = Field(default=10, ge=1, le=100) min_score: float = Field(default=0.0, ge=0.0, le=1.0) class MemoryResponse(BaseModel): """Single memory item in search results.""" memory_id: Optional[str] = None tenant_id: UUID content: str source: str domain: str tier: str surprise_score: float created_at: str class HealthResponse(BaseModel): """Health check response.""" status: str pg: bool qdrant: bool redis: bool _PYDANTIC_OK = True except ImportError: _PYDANTIC_OK = False WriteMemoryRequest = None # type: ignore[assignment,misc] WriteMemoryResponse = None # type: ignore[assignment,misc] SearchRequest = None # type: ignore[assignment,misc] MemoryResponse = None # type: ignore[assignment,misc] HealthResponse = None # type: ignore[assignment,misc] # --------------------------------------------------------------------------- # FastAPI router factory # --------------------------------------------------------------------------- def create_router(gateway: MemoryGateway) -> Any: """Build and return a FastAPI APIRouter wired to *gateway*. Args: gateway: A MemoryGateway instance (initialised or not — the endpoints themselves require it to be initialised). Returns: fastapi.APIRouter with all memory endpoints registered. Raises: ImportError: If fastapi or pydantic are not installed. """ if not _PYDANTIC_OK: raise ImportError( "pydantic is required for gateway_router. " "Install with: pip install pydantic" ) try: from fastapi import APIRouter, HTTPException except ImportError as exc: raise ImportError( "fastapi is required for gateway_router. " "Install with: pip install fastapi" ) from exc router = APIRouter(prefix="/api/v1/memory", tags=["memory"]) # ------------------------------------------------------------------ # Endpoint: POST /api/v1/memory/write # ------------------------------------------------------------------ @router.post("/write", response_model=WriteMemoryResponse, status_code=200) async def write_memory(req: WriteMemoryRequest) -> WriteMemoryResponse: """Write a new memory through the routing pipeline.""" try: record = await gateway.write_memory( tenant_id=req.tenant_id, content=req.content, source=req.source, domain=req.domain, metadata=req.metadata, ) return WriteMemoryResponse( memory_id=record.vector_id, tenant_id=record.tenant_id, tier=record.memory_tier.value, surprise_score=record.surprise_score, created_at=record.created_at.isoformat(), pg_id=record.pg_id, ) except QuotaExceededError as exc: raise HTTPException( status_code=429, detail={ "error": "quota_exceeded", "tenant_id": str(exc.tenant_id), "limit": exc.limit, "current": exc.current, }, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) except Exception as exc: logger.error("write_memory endpoint error: %s", exc) raise HTTPException(status_code=500, detail="Internal gateway error") # ------------------------------------------------------------------ # Endpoint: POST /api/v1/memory/search # ------------------------------------------------------------------ @router.post("/search", response_model=List[MemoryResponse], status_code=200) async def search_memories(req: SearchRequest) -> List[MemoryResponse]: """Vector-search memories by semantic query.""" try: records = await gateway.search_memories( tenant_id=req.tenant_id, query=req.query, limit=req.limit, min_score=req.min_score, ) return [ MemoryResponse( memory_id=r.vector_id, tenant_id=r.tenant_id, content=r.content, source=r.source, domain=r.domain, tier=r.memory_tier.value, surprise_score=r.surprise_score, created_at=r.created_at.isoformat(), ) for r in records ] except Exception as exc: logger.error("search_memories endpoint error: %s", exc) raise HTTPException(status_code=500, detail="Internal gateway error") # ------------------------------------------------------------------ # Endpoint: GET /api/v1/memory/health # ------------------------------------------------------------------ @router.get("/health", response_model=HealthResponse, status_code=200) async def health() -> HealthResponse: """Return health status for all three backends.""" result = await gateway.health_check() return HealthResponse( status=result["status"], pg=result.get("pg", False), qdrant=result.get("qdrant", False), redis=result.get("redis", False), ) # ------------------------------------------------------------------ # Endpoint: DELETE /api/v1/memory/{tenant_id}/{memory_id} # ------------------------------------------------------------------ @router.delete( "/{tenant_id}/{memory_id}", status_code=200, ) async def delete_memory(tenant_id: UUID, memory_id: str) -> Dict[str, Any]: """Delete a memory. Returns 404 if not found.""" try: deleted = await gateway.delete_memory( tenant_id=tenant_id, memory_id=memory_id, ) if not deleted: raise HTTPException( status_code=404, detail=f"Memory {memory_id} not found for tenant {tenant_id}", ) return {"deleted": True, "memory_id": memory_id} except HTTPException: raise except PartialDeleteError as exc: raise HTTPException( status_code=500, detail={ "error": "partial_delete", "memory_id": exc.memory_id, "backends_failed": exc.backends_failed, }, ) except Exception as exc: logger.error("delete_memory endpoint error: %s", exc) raise HTTPException(status_code=500, detail="Internal gateway error") return router # VERIFICATION_STAMP # Story: 1.09 # Verified By: parallel-builder # Verified At: 2026-02-26T10:00:00Z # Tests: see tests/rlm/test_gateway.py # Coverage: >=85%