"""Endpoint HTTP del Memory Gateway.""" from __future__ import annotations import hashlib import time import uuid from collections import Counter from datetime import datetime from typing import Any, Optional from fastapi import APIRouter, Depends, Header, HTTPException, Request from qdrant_client.http import models as qm import config import embed import guardrail import metrics import rerank import state import store from audit import audit, now_iso, require_auth from config import ( API_KEYS, COLLECTION, EMBED_DIM, EMBED_MODEL, GATEWAY_VERSION, GUARDRAIL_BLOCK_THRESHOLD, GUARDRAIL_ENABLED, GUARDRAIL_RERANK_BLOCK, GUARDRAIL_SUPERSEDE_CHECK, GUARDRAIL_SUPERSEDE_MIN, GUARDRAIL_VERSION, GUARDRAIL_WARN_THRESHOLD, MAX_TEXT_LEN, RERANK_CANDIDATES, SCORE_DECAY_HALF_LIFE_DAYS, SCORE_W_AUTHORITY, SCORE_W_IMPORTANCE, SCORE_W_RECENCY, SCORE_W_RELEVANCE, ) from models import MemoryIn, ScoreIn, SearchIn router = APIRouter() @router.post("/v1/memories") async def add_memory( body: MemoryIn, key: str = Depends(require_auth), idempotency_key: Optional[str] = Header(default=None, alias="Idempotency-Key"), ) -> dict: idem_key = f"{key}:{idempotency_key}" if idempotency_key else None if idem_key: state.idempotency_cleanup() existing = state.idempotency.get(idem_key) if existing: if existing["hash"] != state.payload_hash(body): raise HTTPException(status_code=409, detail="Idempotency-Key già usata con payload diverso") audit(key, "create_replay", idempotency_key=idempotency_key[:16]) return existing["response"] memory_id = str(uuid.uuid4()) superseded_id: Optional[str] = None supersede_warning: Optional[dict] = None if body.supersedes_id: old = state.qdrant.retrieve(collection_name=COLLECTION, ids=[body.supersedes_id], with_payload=True) if not old: raise HTTPException(status_code=404, detail="Memoria da supersedere non trovata") if old[0].payload.get("superseded_by"): raise HTTPException(status_code=409, detail="La memoria è già stata superseduta: correggi la versione attiva") superseded_id = body.supersedes_id # Verifica lineage (B): la correzione deve parlare dello stesso fatto del record vecchio if config.GUARDRAIL_SUPERSEDE_CHECK and rerank.enabled() and (old[0].payload or {}).get("text"): rr = await rerank.rerank(body.text, [(old[0].payload or {}).get("text", "")]) if rr: cross = rerank.normalize_score(rr[0][0]) if cross < config.GUARDRAIL_SUPERSEDE_MIN: supersede_warning = { "cross_score": round(cross, 4), "message": "La correzione non sembra riguardare lo stesso fatto del record originale: verifica il lineage.", } audit(key, "supersede_weak_cross", old_id=superseded_id, cross_score=round(cross, 4)) vector = await state.embed(body.text) sparse = state.sparse_encode(body.text) similarity_guardrail: Optional[dict] = None if config.GUARDRAIL_ENABLED and not body.supersedes_id: similarity_guardrail = await guardrail.decide(state.qdrant, COLLECTION, body.text, vector, topic=body.topic, parent_id=body.parent_id) if similarity_guardrail["decision"] == "BLOCK": audit(key, "create_blocked", kind=body.kind, agent_id=body.agent_id or "shared", reason=similarity_guardrail["reason"], matches=[m["memory_id"] for m in similarity_guardrail["matches"]]) raise HTTPException( status_code=409, detail={ "error": "duplicate_memory", "reason": similarity_guardrail["reason"], "matches": similarity_guardrail["matches"], "message": "Memoria già presente o quasi identica: usa supersedes_id per correggere la versione attiva, oppure riformula il contenuto.", }, ) payload: dict[str, Any] = { "text": body.text, "kind": body.kind, "agent_id": body.agent_id or "shared", "project_id": body.project_id, "scope": body.scope, "source": body.source, "confidence": body.confidence, "private": body.private, "created_at": now_iso(), "expires_at": guardrail.parse_ts(body.expires_at), "supersedes_id": superseded_id, "supersede_reason": body.supersede_reason, "parent_id": body.parent_id, "level": body.level, "topic": body.topic, "importance": body.importance, "links": [link.model_dump() for link in body.links] if body.links else None, "embedding_model": EMBED_MODEL, "text_hash": guardrail.text_hash(body.text), } if similarity_guardrail: payload["guardrail"] = { "version": GUARDRAIL_VERSION, "decision": similarity_guardrail["decision"], "reason": similarity_guardrail["reason"], "matches": similarity_guardrail["matches"], } if similarity_guardrail.get("suggestion"): payload["guardrail"]["suggestion"] = similarity_guardrail["suggestion"] point_vector: dict[str, Any] = {"": vector} if sparse is not None: point_vector["bm25"] = sparse state.qdrant.upsert(collection_name=COLLECTION, points=[qm.PointStruct(id=memory_id, vector=point_vector, payload=payload)]) state.invalidate_meta() reparented_count = 0 if superseded_id: state.qdrant.set_payload( collection_name=COLLECTION, payload={"superseded_by": memory_id, "superseded_at": now_iso(), "supersede_reason": body.supersede_reason}, points=[superseded_id], ) audit(key, "supersede", old_id=superseded_id, new_id=memory_id, kind=body.kind, agent_id=payload["agent_id"]) reparented_count = store.reparent_active_children(state.qdrant, COLLECTION, superseded_id, memory_id) if reparented_count: audit(key, "reparent", old_id=superseded_id, new_id=memory_id, count=reparented_count) else: audit(key, "create", memory_id=memory_id, kind=body.kind, agent_id=payload["agent_id"], guardrail=payload.get("guardrail", {}).get("decision", "ALLOW")) response = {"memory_id": memory_id, "created_at": payload["created_at"], "supersedes_id": superseded_id, "reparented": reparented_count} if supersede_warning: response["supersede_warning"] = supersede_warning if idem_key: state.idempotency[idem_key] = {"hash": state.payload_hash(body), "response": response, "ts": time.time()} return response def _composite_score(r: dict, now: float) -> float: """Score composito (C): rerank + importance + recency-decay + authority, pesi normalizzati.""" try: age_days = max(0.0, (now - datetime.fromisoformat(str(r.get("created_at")).replace("Z", "+00:00")).timestamp()) / 86400.0) except (ValueError, TypeError, AttributeError): age_days = 0.0 recency = pow(0.5, age_days / SCORE_DECAY_HALF_LIFE_DAYS) authority = {"high": 1.0, "medium": 0.7, "low": 0.4}.get(r.get("confidence"), 0.7) importance = float(r.get("importance", 0.5) or 0.5) total_w = SCORE_W_RELEVANCE + SCORE_W_IMPORTANCE + SCORE_W_RECENCY + SCORE_W_AUTHORITY raw = ( SCORE_W_RELEVANCE * float(r["rerank_score"]) + SCORE_W_IMPORTANCE * importance + SCORE_W_RECENCY * recency + SCORE_W_AUTHORITY * authority ) return raw / total_w if total_w else raw @router.post("/v1/memories:search") async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> dict: use_rerank = rerank.enabled() and body.rerank is not False # Con reranking attivo recuperiamo più candidati di top_k per dare margine allo stadio di rerank limit = max(body.top_k, RERANK_CANDIDATES) if use_rerank else body.top_k # Multi-query (E): varianti della stessa query, pool unito con dedup (la prima ha priorità) queries = list(dict.fromkeys([body.query] + [q for q in (body.queries or []) if q]))[:3] merged: dict[str, Any] = {} for q in queries: vector = await state.embed(q) sparse = state.sparse_encode(q) if body.hybrid else None hits = store.search(state.qdrant, COLLECTION, body, vector, sparse, limit=limit) for h in hits: merged.setdefault(h.id, h) results = store.format_results(list(merged.values())[:limit]) rerank_info: dict = {"enabled": use_rerank, "used": False, "queries_used": len(queries)} if use_rerank and len(results) >= 2: rr = await rerank.rerank(body.query, [r["text"] or "" for r in results]) if rr: scores, backend, took_ms = rr now = time.time() for r, s in zip(results, scores): r["rerank_score"] = round(rerank.normalize_score(s), 4) r["composite_score"] = round(_composite_score(r, now), 4) results.sort(key=lambda r: (r["composite_score"], r["rerank_score"]), reverse=True) rerank_info.update(used=True, backend=backend, took_ms=took_ms, candidates=len(results)) else: rerank_info["reason"] = "tutti i nodi rerank non raggiungibili (ordine di fusione preservato)" elif use_rerank: rerank_info["reason"] = "candidati insufficienti" results = results[: body.top_k] audit( key, "search", query_hash=hashlib.sha256(body.query.encode()).hexdigest()[:16], top_k=body.top_k, min_score=body.min_score, hits=len(results), rerank_backend=rerank_info.get("backend"), queries_used=len(queries), ) metrics.record_search(len(results)) return {"results": results, "min_score": body.min_score, "total_hits": len(results), "rerank": rerank_info} @router.post("/v1/score") async def score(body: ScoreIn, key: str = Depends(require_auth)) -> dict: """Primitiva cross-encoder: rilevanza (query, documento) in [0,1] via catena rerank. Building block per estensione (validazione estrattore, lineage check) e job di consolidamento; 503 se tutti i nodi della catena non raggiungibili.""" rr = await rerank.rerank(body.query, body.documents) if rr is None: raise HTTPException(status_code=503, detail="nessun nodo rerank raggiungibile") scores, backend, took_ms = rr return { "scores": [round(rerank.normalize_score(s), 4) for s in scores], "raw": [round(s, 4) for s in scores], "backend": backend, "took_ms": took_ms, } @router.get("/v1/memories/{memory_id}") async def get_memory(memory_id: str, key: str = Depends(require_auth)) -> dict: try: point = state.qdrant.retrieve(collection_name=COLLECTION, ids=[memory_id], with_payload=True) except Exception: # noqa: BLE001 — id non-UUID o payload malformato → non trovato, non 500 point = [] if not point: raise HTTPException(status_code=404, detail="Memoria non trovata") audit(key, "get", memory_id=memory_id) return {"memory_id": memory_id, **point[0].payload} @router.delete("/v1/memories/{memory_id}") async def delete_memory(memory_id: str, key: str = Depends(require_auth)) -> dict: point = state.qdrant.retrieve(collection_name=COLLECTION, ids=[memory_id], with_payload=True) if not point: raise HTTPException(status_code=404, detail="Memoria non trovata") state.qdrant.delete(collection_name=COLLECTION, points_selector=[memory_id]) state.invalidate_meta() audit(key, "delete", memory_id=memory_id) return {"deleted": memory_id} @router.get("/v1/meta/overview") async def meta_overview(key: str = Depends(require_auth)) -> dict: now = time.time() cached = state.meta_cache.get("overview") if cached and now - cached["ts"] < 60: audit(key, "meta", cached=True) return {**cached["data"], "cached": True} scope_kinds: dict[str, Counter] = {} projects: Counter = Counter() agents: Counter = Counter() total = 0 superseded = 0 offset: Any = None while True: points, next_offset = state.qdrant.scroll(collection_name=COLLECTION, limit=1000, with_payload=["scope", "kind", "project_id", "agent_id", "superseded_by"], with_vectors=False, offset=offset) for point in points: payload = point.payload total += 1 scope = payload.get("scope", "agent") kind = payload.get("kind", "fact") scope_kinds.setdefault(scope, Counter())[kind] += 1 if payload.get("project_id"): projects[payload["project_id"]] += 1 agents[payload.get("agent_id", "shared")] += 1 if payload.get("superseded_by"): superseded += 1 if not next_offset: break offset = next_offset data = { "scopes": [{"scope": scope, "count": sum(counts.values()), "kinds": [{"kind": kind, "count": count} for kind, count in sorted(counts.items())]} for scope, counts in sorted(scope_kinds.items())], "projects": [{"project_id": project, "count": count} for project, count in projects.most_common()], "agents": [{"agent_id": agent, "count": count} for agent, count in agents.most_common()], "superseded": superseded, "total": total, } state.meta_cache["overview"] = {"ts": now, "data": data} audit(key, "meta", cached=False, total=total) return {**data, "cached": False} @router.get("/v1/status") async def status(request: Request) -> dict: ip = request.client.host if request.client else "unknown" now = time.monotonic() window = state.status_ratelimit.setdefault(ip, []) window[:] = [t for t in window if now - t < 60] if len(window) >= state.STATUS_RATE_LIMIT_PER_MIN: raise HTTPException(status_code=429, detail="Rate limit superato") window.append(now) info = state.qdrant.get_collection(COLLECTION) return {"status": "ok", "collection": COLLECTION, "points": info.points_count, "embedding_model": EMBED_MODEL, "embedding_dim": EMBED_DIM, "access": "shared", "api_keys": len(API_KEYS), "version": GATEWAY_VERSION, "git_commit": __import__("config").GIT_COMMIT, "guardrail_version": GUARDRAIL_VERSION} @router.get("/v1/version") async def version() -> dict: return {"version": GATEWAY_VERSION, "git_commit": __import__("config").GIT_COMMIT, "guardrail_version": GUARDRAIL_VERSION, "guardrail_enabled": GUARDRAIL_ENABLED, "guardrail_block_threshold": GUARDRAIL_BLOCK_THRESHOLD, "guardrail_warn_threshold": GUARDRAIL_WARN_THRESHOLD, "embedding_model": EMBED_MODEL, "collection": COLLECTION, "embed_nodes": [n.name for n in embed.chain_nodes()], "rerank_enabled": rerank.enabled(), "rerank_model": config.RERANK_MODEL, "rerank_nodes": [n.name for n in rerank._get_chain()]} @router.get("/v1/metrics") async def metrics_endpoint(key: str = Depends(require_auth)) -> dict: return metrics.snapshot(state.qdrant, COLLECTION)