diff --git a/README.md b/README.md index 78476f0..6858c02 100644 --- a/README.md +++ b/README.md @@ -116,9 +116,21 @@ Le regole vincolanti (obbligo `project_id`, punteggi, correzione/supersede, disc ## Gateway (componente server) -La cartella `gateway/` contiene il Memory Gateway FastAPI da deployare sul -server (Docker Compose con Qdrant 1.19 + Ollama BGE-M3). Vedi -`gateway/README.md` per il deploy. +Il Memory Gateway FastAPI + Qdrant **non è più duplicato in questo package**: +la fonte unica è il repository dedicato + +``` +git:git.enne2.net/enne2/qmem-gateway (privato) +``` + +che contiene il codice (`gateway/`), il deploy (`docker-compose.yml` con Qdrant +1.19 + gateway, `.env`, `.gitignore`), la suite di test e il README operativo. +Su questa macchina è clonato in `~/dev/qmem-gateway`. + +Motivo della dedup: la copia qui dentro era byte-identica al repo canonico +(GATEWAY_VERSION 2.11.0 / guardrail similarity-v2) e manteneva due fonti +potenzialmente divergenti. Storia completa del codice rimossa: +`git log -- gateway/` (ultimo commit prima della rimozione). ## Architettura diff --git a/gateway/Dockerfile b/gateway/Dockerfile deleted file mode 100644 index 73a8c1e..0000000 --- a/gateway/Dockerfile +++ /dev/null @@ -1,23 +0,0 @@ -FROM python:3.12-slim - -WORKDIR /app -COPY requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt - -# Hash del commit Git da cui è costruita l'immagine (iniettato al build: -# docker compose build --build-arg GIT_COMMIT=$(git rev-parse HEAD) gateway -# o come args nel compose). Esposto da GET /v1/version e /v1/status. -ARG GIT_COMMIT=unknown -ENV GIT_COMMIT=$GIT_COMMIT - -COPY . . - -# Utente non-root con privilegi minimi (best practice container) -RUN useradd --create-home --uid 10001 appuser -USER appuser - -# Pre-download del modello sparso BM25 (cache in /home/appuser/.cache/fastembed) -RUN python -c "from fastembed import SparseTextEmbedding; SparseTextEmbedding(model_name='Qdrant/bm25')" - -EXPOSE 8080 -CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"] diff --git a/gateway/README.md b/gateway/README.md deleted file mode 100644 index 906e537..0000000 --- a/gateway/README.md +++ /dev/null @@ -1,90 +0,0 @@ -# Memory Gateway — deploy - -Componente server di **pi-qmem**: FastAPI + Qdrant 1.19 + Ollama (BGE-M3). -Nessun LLM in scrittura: l'agente salva record deliberati e strutturati. - -``` -pi (estensione pi-qmem) ──HTTPS/VPN──▶ Memory Gateway (FastAPI:8082) ──▶ Qdrant 1.19 (6333) - │ - └──▶ Ollama BGE-M3 (11434, nativo host) -``` - -## Deploy (Docker Compose) - -```bash -# 1. Prepara l'ambiente (vedi qmem-gateway/docker-compose.yml come riferimento) -cp .env.example .env -chmod 600 .env -# genera le chiavi: -# QDRANT_ADMIN_API_KEY=$(openssl rand -hex 32) -# QDRANT_READ_ONLY_API_KEY=$(openssl rand -hex 32) -# API_KEYS=$(openssl rand -hex 32) # chiave condivisa per gli agenti - -# 2. Avvia -docker compose up -d --build - -# 3. Verifica -curl http://127.0.0.1:8082/v1/status -``` - -Requisiti: Docker + Compose v2, Ollama con modello `bge-m3` sul host -(`ollama pull bge-m3`), porta 8082 libera sull'interfaccia VPN. - -## API - -| Endpoint | Descrizione | -|---|---| -| `POST /v1/memories` | Crea record (text, kind, agent_id, scope, **project_id obbligatorio**, source, expires_at, supersedes_id, supersede_reason). Applica il guardrail di similarità pre-scrittura | -| `POST /v1/memories:search` | Ricerca semantica (query, kind, project_id, scope, top_k, include_superseded, min_score) | -| `GET /v1/memories/{id}` | Recupera per UUID | -| `DELETE /v1/memories/{id}` | Elimina per UUID | -| `GET /v1/meta/overview` | Discovery: scope×kind, progetti, agenti, superseduti (cache 60s) | -| `GET /v1/status` | Health + statistiche | - -Auth: header `X-API-Key` (chiave condivisa, accesso completo). Rate limit 120 req/min per chiave. Audit log in JSON lines (docker logs). - -## Versione del codice - -`GET /v1/version` (pubblico) espone la versione del codice in esecuzione, inclusa l'hash del commit Git da cui è stato costruito il container: - -```json -{"version": "2.7.0", "git_commit": "eccb2cb...", "guardrail_version": "similarity-v1", ...} -``` - -Anche `GET /v1/status` include `version`, `git_commit` e `guardrail_version`. L'hash è iniettato al build via `ARG GIT_COMMIT`/`ENV GIT_COMMIT` nel Dockerfile (default `unknown`). Per costruire con l'hash: - -```bash -docker compose build --build-arg GIT_COMMIT=$(git rev-parse HEAD) gateway -# o nel compose: build: { context: ./gateway, args: { GIT_COMMIT: ${GIT_COMMIT:-unknown} } } -``` - -## Guardrail di similarità (v1) - -Enforcement deterministico FUORI dall'LLM, prima di ogni scrittura su `POST /v1/memories`: - -1. **Strato 1 — hash esatto**: SHA-256 del testo normalizzato (`text_hash` nel payload). Se esiste un record attivo con lo stesso hash → `409 BLOCK (EXACT_DUPLICATE)`. -2. **Strato 2 — similarità semantica top-3**: embedding BGE-M3 cosine sui record attivi (esclusi i superseded). - - top-1 ≥ `GUARDRAIL_BLOCK_THRESHOLD` (default 0.85) → `409 BLOCK (KNOWN_SOLUTION)` - - top-1 ≥ `GUARDRAIL_WARN_THRESHOLD` (default 0.70) → `WARN`: salva con flag `guardrail` nel payload - - altrimenti → `ALLOW` - -Il **supersede esplicito** (`supersedes_id`) è una correzione intenzionale: bypassa il guardrail. - -Configurazione (env): `GUARDRAIL_ENABLED` (default true), `GUARDRAIL_BLOCK_THRESHOLD`, `GUARDRAIL_WARN_THRESHOLD`. Soglie di partenza da calibrare sul corpus reale. - -Risposta BLOCK (409): -```json -{"detail": {"error": "duplicate_memory", "reason": "KNOWN_SOLUTION", "matches": [{"memory_id": "...", "score": 0.92}], "message": "..."}} -``` - -## Sicurezza - -- Qdrant bindato su 127.0.0.1; gateway solo su interfaccia VPN -- Chiavi in `.env` (0600), mai committate -- JWT RBAC su Qdrant (admin + read-only) -- Backup: snapshot Qdrant + rotazione 7 giorni (cron: `0 3 * * * /opt/memory/backup.sh`) - -## Dettagli operativi - -Procedure complete (teardown, restore, nginx, troubleshooting): vedi -`docs/playbook.md` nel repo pi-qmem. diff --git a/gateway/audit.py b/gateway/audit.py deleted file mode 100644 index 10f11f9..0000000 --- a/gateway/audit.py +++ /dev/null @@ -1,38 +0,0 @@ -"""Audit e autenticazione del gateway.""" -from __future__ import annotations - -import time -from datetime import datetime, timezone -from typing import Any - -from fastapi import Header, HTTPException - -import config -import state - - -def require_auth(x_api_key: str = Header(...)) -> str: - if x_api_key not in config.API_KEYS: - raise HTTPException(status_code=401, detail="API key non valida") - now = time.monotonic() - window = state.ratelimit.setdefault(x_api_key, []) - window[:] = [t for t in window if now - t < 60] - if len(window) >= config.RATE_LIMIT_PER_MIN: - raise HTTPException(status_code=429, detail="Rate limit superato") - window.append(now) - return x_api_key - - -def audit(key: str, action: str, **extra: Any) -> None: - entry = { - "ts": datetime.now(timezone.utc).isoformat(), - "key": key[:8] + "...", - "action": action, - "request_id": state.request_id.get(), - **extra, - } - config.log.info(__import__("json").dumps(entry, default=str)) - - -def now_iso() -> str: - return datetime.now(timezone.utc).isoformat() diff --git a/gateway/cleanup.py b/gateway/cleanup.py deleted file mode 100644 index f367c52..0000000 --- a/gateway/cleanup.py +++ /dev/null @@ -1,28 +0,0 @@ -"""Pulizia periodica dei record scaduti.""" -from __future__ import annotations - -import time -from typing import Any, Callable - -from qdrant_client.http import models as qm - -from config import log - - -async def loop(qdrant: Any, collection: str, invalidate_meta: Callable[[], None]) -> None: - while True: - try: - scroll = qdrant.scroll( - collection_name=collection, - scroll_filter=qm.Filter(must=[qm.FieldCondition(key="expires_at", range=qm.Range(lt=time.time()))]), - limit=100, - with_payload=False, - ) - ids = [point.id for point in scroll[0]] - if ids: - qdrant.delete(collection_name=collection, points_selector=ids) - invalidate_meta() - log.info("cleanup: rimossi %d record scaduti", len(ids)) - except Exception as exc: # noqa: BLE001 - log.warning("cleanup error: %s", exc) - await __import__("asyncio").sleep(3600) diff --git a/gateway/config.py b/gateway/config.py deleted file mode 100644 index cb79cf8..0000000 --- a/gateway/config.py +++ /dev/null @@ -1,78 +0,0 @@ -"""Configurazione statica del Memory Gateway letta dall'ambiente.""" -from __future__ import annotations - -import logging -import os -from collections import Counter -from typing import Any - -QDRANT_URL = os.environ.get("QDRANT_URL", "http://127.0.0.1:6333") -QDRANT_API_KEY = os.environ.get("QDRANT_API_KEY", "") -EMBED_API = os.environ.get("EMBED_API", "ollama") -EMBED_URL = os.environ.get("EMBED_URL", os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")) -EMBED_MODEL = os.environ.get("EMBED_MODEL", "bge-m3") -EMBED_API_KEY = os.environ.get("EMBED_API_KEY", "") -EMBED_DIM = int(os.environ.get("EMBED_DIM", "1024")) -# Catena di fallback per gli embedding (JSON, formato RERANK_CHAIN + campo "api"). -# Vuota → comportamento legacy: endpoint singolo da EMBED_API/EMBED_URL/EMBED_API_KEY. -EMBED_CHAIN = os.environ.get("EMBED_CHAIN", "") -EMBED_TIMEOUT_MS = int(os.environ.get("EMBED_TIMEOUT_MS", "30000")) -EMBED_RETRY_COOLDOWN_S = int(os.environ.get("EMBED_RETRY_COOLDOWN_S", "60")) -# Retry transiente per le chiamate Qdrant (store/search inclusi) -QDRANT_RETRIES = int(os.environ.get("QDRANT_RETRIES", "3")) -COLLECTION = os.environ.get("COLLECTION", "memories") -API_KEYS: set[str] = {k.strip() for k in os.environ.get("API_KEYS", "").split(",") if k.strip()} -RATE_LIMIT_PER_MIN = int(os.environ.get("RATE_LIMIT_PER_MIN", "120")) -MAX_TEXT_LEN = int(os.environ.get("MAX_TEXT_LEN", "8000")) - -GUARDRAIL_ENABLED = os.environ.get("GUARDRAIL_ENABLED", "true").lower() == "true" -GUARDRAIL_BLOCK_THRESHOLD = float(os.environ.get("GUARDRAIL_BLOCK_THRESHOLD", "0.85")) -GUARDRAIL_WARN_THRESHOLD = float(os.environ.get("GUARDRAIL_WARN_THRESHOLD", "0.70")) -GUARDRAIL_VERSION = "similarity-v2" -# Strato 3 del guardrail: cross-encoder (richiede catena rerank attiva) -GUARDRAIL_RERANK = os.environ.get("GUARDRAIL_RERANK", "false").lower() == "true" -GUARDRAIL_RERANK_BLOCK = float(os.environ.get("GUARDRAIL_RERANK_BLOCK", "0.90")) -GUARDRAIL_RERANK_SUGGEST = float(os.environ.get("GUARDRAIL_RERANK_SUGGEST", "0.85")) -# Verifica supersede: cross-score (nuovo, vecchio) sotto soglia → warning non bloccante -GUARDRAIL_SUPERSEDE_CHECK = os.environ.get("GUARDRAIL_SUPERSEDE_CHECK", "false").lower() == "true" -GUARDRAIL_SUPERSEDE_MIN = float(os.environ.get("GUARDRAIL_SUPERSEDE_MIN", "0.50")) -# Score composito: rerank + importance + recency + authority (post-rerank) -SCORE_W_RELEVANCE = float(os.environ.get("SCORE_W_RELEVANCE", "0.55")) -SCORE_W_IMPORTANCE = float(os.environ.get("SCORE_W_IMPORTANCE", "0.20")) -SCORE_W_RECENCY = float(os.environ.get("SCORE_W_RECENCY", "0.15")) -SCORE_W_AUTHORITY = float(os.environ.get("SCORE_W_AUTHORITY", "0.10")) -SCORE_DECAY_HALF_LIFE_DAYS = float(os.environ.get("SCORE_DECAY_HALF_LIFE_DAYS", "180")) -GIT_COMMIT = os.environ.get("GIT_COMMIT", "unknown").strip() -GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.11.0").strip() - -VM_PUSH_URL = os.environ.get("VM_PUSH_URL", "http://host.docker.internal:8428/api/v1/import/prometheus") -VM_PUSH_INTERVAL = int(os.environ.get("VM_PUSH_INTERVAL", "30")) -METRICS_ENABLED = os.environ.get("METRICS_ENABLED", "true").lower() == "true" - -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") -log = logging.getLogger("memory-gateway") - -SPARSE_VECTOR_NAME = "bm25" - -# Re-ranking: catena di fallback resiliente (frigate → brain locale). -# Il default nel codice è OFF; il deploy imposta RERANK_ENABLED=true e la catena. -RERANK_ENABLED = os.environ.get("RERANK_ENABLED", "false").lower() == "true" -RERANK_MODEL = os.environ.get("RERANK_MODEL", "bge-reranker-v2-m3") -RERANK_CANDIDATES = int(os.environ.get("RERANK_CANDIDATES", "16")) -RERANK_MAX_DOC_CHARS = int(os.environ.get("RERANK_MAX_DOC_CHARS", "800")) -RERANK_TIMEOUT_MS = int(os.environ.get("RERANK_TIMEOUT_MS", "10000")) -RERANK_RETRY_COOLDOWN_S = int(os.environ.get("RERANK_RETRY_COOLDOWN_S", "60")) -RERANK_CHAIN = os.environ.get("RERANK_CHAIN", "") -_metrics: dict[str, Any] = { - "requests": Counter(), - "duration_sum": Counter(), - "duration_count": Counter(), - "errors": Counter(), - "search_queries": 0, - "search_hits": 0, - "rerank_calls": Counter(), - "rerank_duration_sum": Counter(), - "embed_calls": Counter(), - "embed_duration_sum": Counter(), - "qdrant_retries": 0, -} diff --git a/gateway/embed.py b/gateway/embed.py deleted file mode 100644 index 10b21b0..0000000 --- a/gateway/embed.py +++ /dev/null @@ -1,222 +0,0 @@ -"""Embedding denso (Ollama/llama.cpp) e sparse BM25, con catena di fallback resiliente. - -Catena da EMBED_CHAIN (JSON, stesso formato di RERANK_CHAIN + campo "api"): -il primo nodo raggiungibile vince, i nodi falliti entrano in cooldown. Se -EMBED_CHAIN è vuota si usa il comportamento legacy (endpoint singolo da -EMBED_API/EMBED_URL/EMBED_API_KEY). L'ultimo errore viene rilanciato al client -come gli endpoint precedenti: nessuna degradazione silenziosa della scrittura. -""" -from __future__ import annotations - -import json -import time -from dataclasses import dataclass -from typing import Any, Optional - -import httpx -from qdrant_client.http import models as qm - -import metrics -from config import ( - EMBED_API, - EMBED_API_KEY, - EMBED_CHAIN, - EMBED_DIM, - EMBED_MODEL, - EMBED_RETRY_COOLDOWN_S, - EMBED_TIMEOUT_MS, - EMBED_URL, - SPARSE_VECTOR_NAME, - log, -) - -try: - from fastembed import SparseTextEmbedding - _sparse_model: Optional[Any] = None - SPARSE_AVAILABLE = True -except Exception: # noqa: BLE001 - _sparse_model = None - SPARSE_AVAILABLE = False - log.warning("fastembed non disponibile: hybrid retrieval disattivato") - -_http: Optional[httpx.AsyncClient] = None - - -@dataclass(frozen=True) -class EmbedNode: - """Un endpoint embedding nella catena di fallback.""" - - name: str - url: str - api: str # "llamacpp" (/v1/embeddings) | "ollama" (/api/embed) - key: str - timeout_ms: int - - -def parse_chain(raw: str, default_api: str, default_url: str, default_key: str) -> list[EmbedNode]: - """Parsa EMBED_CHAIN (JSON); vuota o invalida → endpoint legacy singolo.""" - nodes: list[EmbedNode] = [] - if raw: - try: - entries = json.loads(raw) - for entry in entries if isinstance(entries, list) else []: - if not isinstance(entry, dict) or not entry.get("url"): - continue - url = str(entry["url"]).rstrip("/") - api = str(entry.get("api") or "llamacpp") - if api not in ("llamacpp", "ollama") or not url.startswith(("http://", "https://")): - continue - nodes.append( - EmbedNode( - name=str(entry.get("name") or url), - url=url, - api=api, - key=str(entry.get("key") or ""), - timeout_ms=int(entry.get("timeout_ms", EMBED_TIMEOUT_MS)), - ) - ) - except (json.JSONDecodeError, TypeError, ValueError): - log.error("EMBED_CHAIN non è JSON valido: uso l'endpoint legacy") - if not nodes and default_url: - # Compatibilità legacy: endpoint singolo dagli env EMBED_* - nodes = [EmbedNode(name="embed", url=default_url.rstrip("/"), api=default_api, key=default_key, timeout_ms=EMBED_TIMEOUT_MS)] - return nodes - - -_chain: Optional[list[EmbedNode]] = None -_down_until: dict[str, float] = {} # url → monotonic deadline del cooldown -_http: Optional[httpx.AsyncClient] = None - - -def _get_chain() -> list[EmbedNode]: - global _chain - if _chain is None: - _chain = parse_chain(EMBED_CHAIN, EMBED_API, EMBED_URL, EMBED_API_KEY) - return _chain - - -def reset_chain_cache() -> None: - """Forza il re-parse della catena (usato dai test).""" - global _chain - _chain = None - _down_until.clear() - - -def get_http() -> httpx.AsyncClient: - global _http - if _http is None: - _http = httpx.AsyncClient(timeout=30) - return _http - - -async def close_http() -> None: - global _http - if _http is not None: - await _http.aclose() - _http = None - - -def chain_nodes() -> list[EmbedNode]: - return _get_chain() - - -async def embed(text: str) -> list[float]: - """Embedding con catena di fallback: ritorna il vettore o rilancia dopo l'ultimo fallimento.""" - chain = _get_chain() - if not chain: - raise RuntimeError("nessun endpoint embedding configurato") - now = time.monotonic() - live = [n for n in chain if _down_until.get(n.url, 0) <= now] - if not live: - # tutti in cooldown: ritenta comunque il primo (meglio di un fallimento immediato) - live = [chain[0]] - payload = {"model": EMBED_MODEL, "input": text} - started = time.monotonic() - last_exc: Optional[Exception] = None - for node in live: - headers = {"Content-Type": "application/json"} - if node.key: - headers["Authorization"] = f"Bearer {node.key}" - path = "/v1/embeddings" if node.api == "llamacpp" else "/api/embed" - try: - t0 = time.monotonic() - response = await get_http().post( - f"{node.url}{path}", - json=payload, - headers=headers, - timeout=httpx.Timeout(node.timeout_ms / 1000.0), - ) - response.raise_for_status() - data = response.json() - vector = data["data"][0]["embedding"] if node.api == "llamacpp" else data["embeddings"][0] - if len(vector) != EMBED_DIM: - raise ValueError(f"dimensione vettore {len(vector)} != EMBED_DIM {EMBED_DIM}") - took = int((time.monotonic() - started) * 1000) - metrics.record_embed(node.name, True, took) - return vector - except (httpx.HTTPError, ValueError, KeyError, IndexError, TypeError) as exc: - took = int((time.monotonic() - t0) * 1000) - _down_until[node.url] = time.monotonic() + EMBED_RETRY_COOLDOWN_S - last_exc = exc - metrics.record_embed(node.name, False, took) - log.warning( - "embed: nodo '%s' fallito dopo %dms (%s: %s) → cooldown %ds", - node.name, - took, - exc.__class__.__name__, - exc, - EMBED_RETRY_COOLDOWN_S, - ) - raise RuntimeError(f"tutti i nodi embedding falliti ({len(live)} tentativi)") from last_exc - - -def get_sparse_model(): - global _sparse_model - if _sparse_model is None and SPARSE_AVAILABLE: - _sparse_model = SparseTextEmbedding(model_name="Qdrant/bm25") - return _sparse_model - - -def sparse_encode(text: str) -> Optional[qm.SparseVector]: - model = get_sparse_model() - if model is None: - return None - emb = next(model.embed(text)) - return qm.SparseVector(indices=emb.indices.tolist(), values=emb.values.tolist()) - - -def backfill_sparse(qdrant: Any, collection: str) -> None: - if not SPARSE_AVAILABLE: - return - offset: Any = None - updated = 0 - while True: - points, next_offset = qdrant.scroll( - collection_name=collection, - limit=100, - with_payload=["text"], - with_vectors=True, - offset=offset, - ) - batch: list[qm.PointStruct] = [] - for p in points: - vecs = p.vector or {} - if SPARSE_VECTOR_NAME in vecs: - continue - text = (p.payload or {}).get("text", "") - if not text: - continue - sparse = sparse_encode(text) - if sparse is None: - continue - batch.append(qm.PointStruct(id=p.id, vector={SPARSE_VECTOR_NAME: sparse})) - if batch: - # update_vectors: aggiorna SOLO il vettore sparso, preservando payload e vettore denso - # (upsert parziale sostituirebbe l'intero punto — incidente 2026-08-16) - qdrant.update_vectors(collection_name=collection, points=batch) - updated += len(batch) - if not next_offset: - break - offset = next_offset - if updated: - log.info("backfill sparse: %d record aggiornati", updated) \ No newline at end of file diff --git a/gateway/guardrail.py b/gateway/guardrail.py deleted file mode 100644 index 0c3a13b..0000000 --- a/gateway/guardrail.py +++ /dev/null @@ -1,135 +0,0 @@ -"""Guardrail anti-duplicati e similarità pre-scrittura. - -Strato 1: hash esatto. Strato 2: cosine (bi-encoder). Strato 3 (opzionale, -GUARDRAIL_RERANK): cross-encoder che conferma o scarta il "quasi-duplicato" — -il cosine confonde "stesso argomento" con "stesso fatto", il cross-encoder -legge le coppie e giudica se il nuovo testo sia davvero lo stesso contenuto. -Il reranker è un miglioramento: se non raggiungibile si degrada alla sola -similarità (niente fallimenti di scrittura per un reranker giù). -""" -from __future__ import annotations - -import hashlib -import unicodedata -from typing import Any, Optional - -from qdrant_client.http import models as qm - -import rerank -from config import ( - GUARDRAIL_BLOCK_THRESHOLD, - GUARDRAIL_RERANK, - GUARDRAIL_RERANK_BLOCK, - GUARDRAIL_RERANK_SUGGEST, - GUARDRAIL_WARN_THRESHOLD, - log, -) - - -def normalize_text(text: str) -> str: - s = unicodedata.normalize("NFD", text.lower()) - s = "".join(c for c in s if not unicodedata.combining(c)) - return " ".join(s.split()) - - -def text_hash(text: str) -> str: - return hashlib.sha256(normalize_text(text).encode("utf-8")).hexdigest() - - -def find_similar(qdrant: Any, collection: str, text: str, vector: list[float], top_k: int = 5) -> list[dict]: - qfilter = qm.Filter(must=[qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by"))]) - hits = qdrant.query_points(collection_name=collection, query=vector, query_filter=qfilter, limit=top_k, with_payload=True).points - return [ - { - "memory_id": h.id, - "score": round(float(h.score), 4), - "text": (h.payload or {}).get("text", ""), - "kind": (h.payload or {}).get("kind", ""), - "project_id": (h.payload or {}).get("project_id", ""), - } - for h in hits - ] - - -async def _cross_scores(text: str, matches: list[dict]) -> Optional[list[float]]: - """Cross-score (0,1) di (nuovo testo, candidato) per ogni match; None se non disponibile.""" - if not GUARDRAIL_RERANK or not rerank.enabled(): - return None - docs = [m["text"] or " " for m in matches] - try: - rr = await rerank.rerank(text, docs) - except Exception as exc: # noqa: BLE001 - log.warning("guardrail: rerank non disponibile (%s) → decisione solo cosine", exc.__class__.__name__) - return None - if rr is None: - log.warning("guardrail: tutti i nodi rerank non raggiungibili → decisione solo cosine") - return None - scores, _backend, _took = rr - return [rerank.normalize_score(s) for s in scores] - - -async def decide(qdrant: Any, collection: str, text: str, vector: list[float], topic: Optional[str] = None, parent_id: Optional[str] = None) -> dict: - exact_filter = qm.Filter(must=[ - qm.FieldCondition(key="text_hash", match=qm.MatchValue(value=text_hash(text))), - qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")), - ]) - exact = qdrant.query_points(collection_name=collection, query=vector, query_filter=exact_filter, limit=1, with_payload=True).points - if exact: - return {"decision": "BLOCK", "reason": "EXACT_DUPLICATE", "matches": [{"memory_id": exact[0].id, "score": 1.0}]} - - matches = find_similar(qdrant, collection, text, vector, top_k=3) - if not matches: - return {"decision": "ALLOW", "reason": "NO_CANDIDATE", "matches": []} - - # Strato 3: cross-encoder sulla short-list (giudice "è lo stesso fatto?") - cross = await _cross_scores(text, matches) - if cross is not None: - for m, c in zip(matches, cross): - m["cross_score"] = round(c, 4) - best_cross = max(cross) - best_match = matches[cross.index(best_cross)] - else: - best_cross = None - best_match = matches[0] - - top1 = matches[0]["score"] - hierarchical = (topic or parent_id) and any(m.get("memory_id") != parent_id for m in matches) - - if top1 >= GUARDRAIL_BLOCK_THRESHOLD: - if hierarchical: - return {"decision": "WARN", "reason": "HIERARCHICAL_SUBTOPIC", "matches": matches} - if best_cross is not None: - if best_cross >= GUARDRAIL_RERANK_BLOCK: - return {"decision": "BLOCK", "reason": "CROSS_DUP_CONFIRMED", "matches": matches} - return {"decision": "WARN", "reason": "CROSS_DUP_WEAK", "matches": matches, - "message": "Similarità alta ma il cross-encoder non conferma lo stesso fatto: probabilmente correlati, non duplicati."} - return {"decision": "BLOCK", "reason": "KNOWN_SOLUTION", "matches": matches} - - if top1 >= GUARDRAIL_WARN_THRESHOLD: - if best_cross is not None and best_cross >= GUARDRAIL_RERANK_BLOCK: - return {"decision": "BLOCK", "reason": "CROSS_DUP_CONFIRMED", "matches": matches} - d = {"decision": "WARN", "reason": "MODERATE_SIMILARITY", "matches": matches} - if best_cross is not None and best_cross >= GUARDRAIL_RERANK_SUGGEST: - d["suggestion"] = { - "supersedes_id": best_match["memory_id"], - "cross_score": round(best_cross, 4), - "message": "Sembra un aggiornamento del record indicato: valuta supersedes_id.", - } - return d - - # Cosine sotto la soglia WARN, ma cross-encoder che conferma un duplicato - # parafrasato sfuggito al bi-encoder. - if best_cross is not None and best_cross >= GUARDRAIL_RERANK_BLOCK: - return {"decision": "BLOCK", "reason": "CROSS_DUP_LOW_COSINE", "matches": matches} - - return {"decision": "ALLOW", "reason": "NEW_SOLUTION", "matches": matches} - - -def parse_ts(value: Optional[str]) -> Optional[float]: - from datetime import datetime - if not value: - return None - try: - return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() - except ValueError: - return None \ No newline at end of file diff --git a/gateway/main.py b/gateway/main.py deleted file mode 100644 index a46d879..0000000 --- a/gateway/main.py +++ /dev/null @@ -1,111 +0,0 @@ -"""Memory Gateway — bootstrap FastAPI, lifecycle e middleware. - -Gli endpoint e la logica di dominio sono separati in moduli: -config, models, state, audit, guardrail, embed, store, metrics, cleanup, -routes. Il contratto HTTP resta invariato. -""" -from __future__ import annotations - -import asyncio -import time -import uuid -from contextlib import asynccontextmanager - -import uvicorn -from fastapi import FastAPI, Request -from qdrant_client.http import models as qm - -import cleanup -import embed as embedding -import metrics -import rerank -import state -from config import ( - COLLECTION, - EMBED_DIM, - METRICS_ENABLED, - SPARSE_VECTOR_NAME, - GATEWAY_VERSION, - log, -) -from routes import router - -# Alias utili per compatibilità con import/debug locali; lo stato effettivo è in state.py. -qdrant = state.qdrant -embed = embedding.embed -state.embed = embedding.embed -state.sparse_encode = embedding.sparse_encode - - -async def _lifespan(_app: FastAPI): - """Crea collection/indici e avvia i loop periodici.""" - collections = state.qdrant.get_collections().collections - if not any(c.name == COLLECTION for c in collections): - state.qdrant.create_collection( - collection_name=COLLECTION, - vectors_config=qm.VectorParams(size=EMBED_DIM, distance=qm.Distance.COSINE), - sparse_vectors_config={SPARSE_VECTOR_NAME: qm.SparseVectorParams(modifier=qm.Modifier.IDF)}, - ) - for field in ("agent_id", "project_id", "scope", "kind", "supersedes_id", "superseded_by", "text_hash", "parent_id", "level", "topic"): - state.qdrant.create_payload_index(collection_name=COLLECTION, field_name=field, field_schema=qm.PayloadSchemaType.KEYWORD) - state.qdrant.create_payload_index(collection_name=COLLECTION, field_name="text", field_schema=qm.PayloadSchemaType.TEXT) - log.info("collection %s creata con indici (dense + sparse %s)", COLLECTION, SPARSE_VECTOR_NAME) - else: - log.info("collection %s già esistente", COLLECTION) - for field in ("parent_id", "level", "topic"): - try: - state.qdrant.create_payload_index(collection_name=COLLECTION, field_name=field, field_schema=qm.PayloadSchemaType.KEYWORD) - except Exception: # noqa: BLE001 - pass - info = state.qdrant.get_collection(COLLECTION) - sparse_vectors = (info.config.params.sparse_vectors or {}) if info.config and info.config.params else {} - if SPARSE_VECTOR_NAME not in sparse_vectors: - state.qdrant.create_vector_name(COLLECTION, SPARSE_VECTOR_NAME, qm.SparseVectorNameConfig(sparse=qm.SparseVectorConfig(modifier=qm.Modifier.IDF))) - log.info("sparse vector %s aggiunto alla collection esistente", SPARSE_VECTOR_NAME) - embedding.backfill_sparse(state.qdrant, COLLECTION) - - cleanup_task = asyncio.create_task(cleanup.loop(state.qdrant, COLLECTION, state.invalidate_meta)) - metrics_task = asyncio.create_task(metrics.push_loop(state.qdrant, COLLECTION, embedding.get_http)) if METRICS_ENABLED else None - try: - yield - finally: - cleanup_task.cancel() - try: - await cleanup_task - except asyncio.CancelledError: - pass - if metrics_task is not None: - metrics_task.cancel() - try: - await metrics_task - except asyncio.CancelledError: - pass - await embedding.close_http() - await rerank.close_http() - - -app = FastAPI(title="Memory Gateway", version=GATEWAY_VERSION, lifespan=_lifespan) -app.include_router(router) - - -@app.middleware("http") -async def request_id_middleware(request: Request, call_next): - rid = request.headers.get("X-Request-ID") or str(uuid.uuid4()) - state.request_id.set(rid) - response = await call_next(request) - response.headers["X-Request-ID"] = rid - return response - - -@app.middleware("http") -async def metrics_middleware(request: Request, call_next): - start = time.monotonic() - response = await call_next(request) - route = request.scope.get("route") - endpoint = route.path if route else request.url.path - metrics.record_request(endpoint, time.monotonic() - start, response.status_code) - return response - - -if __name__ == "__main__": - uvicorn.run(app, host="0.0.0.0", port=8080) diff --git a/gateway/metrics.py b/gateway/metrics.py deleted file mode 100644 index e505ed9..0000000 --- a/gateway/metrics.py +++ /dev/null @@ -1,94 +0,0 @@ -"""Metriche in-memory e push Prometheus/VictoriaMetrics.""" -from __future__ import annotations - -import time -from typing import Any - -from config import VM_PUSH_INTERVAL, VM_PUSH_URL, _metrics, log - - -def record_request(endpoint: str, duration: float, status_code: int) -> None: - _metrics["requests"][endpoint] += 1 - _metrics["duration_sum"][endpoint] += duration - _metrics["duration_count"][endpoint] += 1 - if status_code >= 400: - _metrics["errors"][(endpoint, status_code)] += 1 - - -def record_search(hits: int) -> None: - _metrics["search_queries"] += 1 - _metrics["search_hits"] += hits - - -def record_rerank(backend: str, ok: bool, took_ms: int) -> None: - _metrics["rerank_calls"][(backend, "ok" if ok else "fail")] += 1 - _metrics["rerank_duration_sum"][backend] += took_ms - - -def record_embed(backend: str, ok: bool, took_ms: int) -> None: - _metrics["embed_calls"][(backend, "ok" if ok else "fail")] += 1 - _metrics["embed_duration_sum"][backend] += took_ms - - -def snapshot(qdrant: Any, collection: str) -> dict: - try: - points = qdrant.get_collection(collection).points_count - except Exception: # noqa: BLE001 - points = None - return { - "requests": dict(_metrics["requests"]), - "avg_duration_ms": { - endpoint: round(_metrics["duration_sum"][endpoint] / _metrics["duration_count"][endpoint] * 1000, 2) - for endpoint in _metrics["duration_count"] - }, - "errors": {f"{endpoint}:{status}": count for (endpoint, status), count in _metrics["errors"].items()}, - "search_queries": _metrics["search_queries"], - "search_hits": _metrics["search_hits"], - "rerank_calls": {f"{backend}:{outcome}": count for (backend, outcome), count in _metrics["rerank_calls"].items()}, - "rerank_avg_ms": {backend: round(total / _metrics["rerank_calls"][(backend, "ok")], 2) for backend, total in _metrics["rerank_duration_sum"].items() if _metrics["rerank_calls"][(backend, "ok")]}, - "embed_calls": {f"{backend}:{outcome}": count for (backend, outcome), count in _metrics["embed_calls"].items()}, - "embed_avg_ms": {backend: round(total / _metrics["embed_calls"][(backend, "ok")], 2) for backend, total in _metrics["embed_duration_sum"].items() if _metrics["embed_calls"][(backend, "ok")]}, - "qdrant_retries": _metrics["qdrant_retries"], - "points": points, - } - - -def prometheus_lines(qdrant: Any, collection: str) -> list[str]: - lines: list[str] = [] - for endpoint, count in _metrics["requests"].items(): - lines.append(f'qmem_requests_total{{endpoint="{endpoint}"}} {count}') - for endpoint, total in _metrics["duration_sum"].items(): - count = _metrics["duration_count"][endpoint] - lines.append(f'qmem_request_duration_seconds_sum{{endpoint="{endpoint}"}} {total:.6f}') - lines.append(f'qmem_request_duration_seconds_count{{endpoint="{endpoint}"}} {count}') - for (endpoint, status), count in _metrics["errors"].items(): - lines.append(f'qmem_errors_total{{endpoint="{endpoint}",status="{status}"}} {count}') - lines.append(f"qmem_search_queries_total {_metrics['search_queries']}") - lines.append(f"qmem_search_hits_total {_metrics['search_hits']}") - for (backend, outcome), count in _metrics["rerank_calls"].items(): - lines.append(f'qmem_rerank_calls_total{{backend="{backend}",outcome="{outcome}"}} {count}') - for backend, s in _metrics["rerank_duration_sum"].items(): - lines.append(f'qmem_rerank_duration_seconds_sum{{backend="{backend}"}} {s / 1000:.6f}') - for (backend, outcome), count in _metrics["embed_calls"].items(): - lines.append(f'qmem_embed_calls_total{{backend="{backend}",outcome="{outcome}"}} {count}') - for backend, s in _metrics["embed_duration_sum"].items(): - lines.append(f'qmem_embed_duration_seconds_sum{{backend="{backend}"}} {s / 1000:.6f}') - lines.append(f"qmem_qdrant_retries_total {_metrics['qdrant_retries']}") - try: - lines.append(f"qmem_points {qdrant.get_collection(collection).points_count}") - except Exception: # noqa: BLE001 - pass - return lines - - -async def push_loop(qdrant: Any, collection: str, get_http) -> None: - while True: - try: - now_ms = int(time.time() * 1000) - body = "\n".join(f"{line} {now_ms}" for line in prometheus_lines(qdrant, collection)) + "\n" - response = await get_http().post(VM_PUSH_URL, content=body, headers={"Content-Type": "text/plain"}) - if response.status_code >= 300: - log.warning("metrics push: HTTP %s", response.status_code) - except Exception as exc: # noqa: BLE001 - log.warning("metrics push error: %s", exc) - await __import__("asyncio").sleep(VM_PUSH_INTERVAL) diff --git a/gateway/models.py b/gateway/models.py deleted file mode 100644 index 75b5957..0000000 --- a/gateway/models.py +++ /dev/null @@ -1,80 +0,0 @@ -"""Schemi Pydantic del Memory Gateway.""" -from __future__ import annotations - -from datetime import datetime -from typing import Literal, Optional - -from pydantic import BaseModel, Field, field_validator - -from config import MAX_TEXT_LEN - - -class MemoryLink(BaseModel): - target_id: str = Field(..., description="UUID del record target collegato") - predicate: str = Field(default="part_of", max_length=64, description="Tipo di relazione: parent_of, part_of, relates_to, supersedes...") - weight: float = Field(default=1.0, ge=0.0, le=1.0) - - -class MemoryIn(BaseModel): - text: str = Field(min_length=1, max_length=MAX_TEXT_LEN) - kind: Literal["decision", "fact", "episode", "preference"] = "fact" - agent_id: Optional[str] = Field(default=None, max_length=64, description="Solo provenienza, nessun isolamento") - project_id: str = Field(min_length=1, max_length=64, description="OBBLIGATORIO: progetto/dominio di appartenenza (kebab-case)") - scope: Literal["agent", "project", "org"] = "agent" - source: Optional[str] = Field(default=None, max_length=256) - confidence: Literal["high", "medium", "low"] = Field(default="medium", description="Affidabilità del record") - expires_at: Optional[str] = None - supersedes_id: Optional[str] = None - supersede_reason: Optional[str] = Field(default=None, max_length=512) - parent_id: Optional[str] = Field(default=None, description="UUID del record genitore per gerarchia/subtopic") - level: Optional[Literal["L1_ROOT", "L2_SUBTOPIC", "L3_DETAIL"]] = Field(default=None, description="Livello gerarchico") - topic: Optional[str] = Field(default=None, max_length=128, description="Topic gerarchico") - links: Optional[list[MemoryLink]] = Field(default=None, description="Collegamenti semantici e relazionali") - importance: float = Field(default=0.5, ge=0.0, le=1.0, description="Importanza stabile del record (usata nello score composito)") - private: bool = Field(default=False, description="Riservato: escluso dalle ricerche standard, visibile solo con include_private o topic esplicito") - - @field_validator("expires_at") - @classmethod - def _validate_expires_at(cls, v: Optional[str]) -> Optional[str]: - if v is None: - return v - try: - datetime.fromisoformat(v.replace("Z", "+00:00")) - except ValueError: - raise ValueError("expires_at deve essere una data ISO 8601 valida (es. 2026-09-01T00:00:00Z)") - return v - - -class ScoreIn(BaseModel): - """Primitiva di scoring cross-encoder (usata da estensione e job di consolidamento).""" - - query: str = Field(min_length=1, max_length=512) - documents: list[str] = Field(min_length=1, max_length=32) - - -class SearchIn(BaseModel): - query: str = Field(min_length=1, max_length=512) - kind: Optional[Literal["decision", "fact", "episode", "preference"]] = None - project_id: Optional[str] = None - scope: Optional[Literal["agent", "project", "org"]] = None - include_superseded: bool = False - min_score: Optional[float] = Field(default=None, ge=0.0, le=1.0) - top_k: int = Field(default=5, ge=1, le=20) - hybrid: bool = False - parent_id: Optional[str] = None - level: Optional[Literal["L1_ROOT", "L2_SUBTOPIC", "L3_DETAIL"]] = None - topic: Optional[str] = None - include_private: bool = Field(default=False, description="Includi i record privati (solo ricerche esplicite)") - rerank: Optional[bool] = Field(default=None, description="Override per-query dello stadio rerank (None = default server)") - queries: Optional[list[str]] = Field(default=None, max_length=3, description="Varianti di query (max 3): pool unito, dedup e rerank unico") - - @field_validator("queries") - @classmethod - def _validate_queries(cls, v: Optional[list[str]]) -> Optional[list[str]]: - if v is None: - return v - cleaned = [q.strip() for q in v if q and q.strip()] - if len(cleaned) != len(v): - raise ValueError("le query non devono essere vuote") - return cleaned - queries: Optional[list[str]] = Field(default=None, max_length=3, description="Varianti di query (max 3): pool unito, dedup e rerank unico") diff --git a/gateway/requirements-dev.txt b/gateway/requirements-dev.txt deleted file mode 100644 index 8d8ffdb..0000000 --- a/gateway/requirements-dev.txt +++ /dev/null @@ -1,2 +0,0 @@ -# Dipendenze di sviluppo (test): installare con pip install -r requirements-dev.txt -pytest==8.3.4 diff --git a/gateway/requirements.txt b/gateway/requirements.txt deleted file mode 100644 index 5904fbe..0000000 --- a/gateway/requirements.txt +++ /dev/null @@ -1,6 +0,0 @@ -fastapi==0.115.6 -uvicorn[standard]==0.34.0 -qdrant-client==1.19.0 -httpx==0.28.1 -pydantic==2.10.4 -fastembed==0.5.1 diff --git a/gateway/rerank.py b/gateway/rerank.py deleted file mode 100644 index c40e6af..0000000 --- a/gateway/rerank.py +++ /dev/null @@ -1,179 +0,0 @@ -"""Stadio di re-ranking (cross-encoder) con catena di fallback resiliente. - -La catena è definita da RERANK_CHAIN (JSON): il primo nodo raggiungibile vince. -Dopo un fallimento il nodo entra in cooldown (RERANK_RETRY_COOLDOWN_S) e la -richiesta passa al successivo; se tutti i nodi sono in cooldown si ritenta -comunque il primo (meglio di un fallimento immediato). Se nessun nodo risponde -la ricerca degrada con grazia all'ordine di fusione ibrida (nessun errore al -client): il reranking è un miglioramento, non una dipendenza. -""" -from __future__ import annotations - -import json -import math -import time -from dataclasses import dataclass -from typing import Optional - -import httpx - -import metrics -from config import ( - RERANK_CHAIN, - RERANK_ENABLED, - RERANK_MAX_DOC_CHARS, - RERANK_MODEL, - RERANK_RETRY_COOLDOWN_S, - RERANK_TIMEOUT_MS, - log, -) - - -@dataclass(frozen=True) -class RerankNode: - """Un endpoint reranker nella catena di fallback.""" - - name: str - url: str - key: str - timeout_ms: int - - -def parse_chain(raw: str) -> list[RerankNode]: - """Parsa RERANK_CHAIN: JSON [{name, url, key, timeout_ms}]. URL senza schema → scartato.""" - try: - entries = json.loads(raw) if raw else [] - except (json.JSONDecodeError, TypeError): - log.error("RERANK_CHAIN non è JSON valido: reranking disattivato") - return [] - if not isinstance(entries, list): - log.error("RERANK_CHAIN non è una lista: reranking disattivato") - return [] - nodes: list[RerankNode] = [] - for entry in entries: - if not isinstance(entry, dict) or not entry.get("url"): - continue - url = str(entry["url"]).rstrip("/") - if not url.startswith(("http://", "https://")): - continue - nodes.append( - RerankNode( - name=str(entry.get("name") or url), - url=url, - key=str(entry.get("key") or ""), - timeout_ms=int(entry.get("timeout_ms", RERANK_TIMEOUT_MS)), - ) - ) - return nodes - - -_chain: Optional[list[RerankNode]] = None -_down_until: dict[str, float] = {} # url → monotonic deadline del cooldown -_http: Optional[httpx.AsyncClient] = None - - -def _get_chain() -> list[RerankNode]: - global _chain - if _chain is None: - _chain = parse_chain(RERANK_CHAIN) - return _chain - - -def reset_chain_cache() -> None: - """Forza il re-parse della catena (usato dai test).""" - global _chain - _chain = None - _down_until.clear() - - -def get_http() -> httpx.AsyncClient: - global _http - if _http is None: - _http = httpx.AsyncClient(timeout=30) - return _http - - -async def close_http() -> None: - global _http - if _http is not None: - await _http.aclose() - _http = None - - -def enabled() -> bool: - """Reranking attivo: flag env + catena configurata non vuota.""" - return RERANK_ENABLED and bool(_get_chain()) - - -def live_nodes() -> tuple[list[RerankNode], bool]: - """Nodi fuori cooldown; all_down=True se nessun nodo è live (forza retry totale).""" - chain = _get_chain() - now = time.monotonic() - live = [n for n in chain if _down_until.get(n.url, 0) <= now] - return live, bool(chain) and len(live) < len(chain) - - -async def rerank(query: str, docs: list[str]) -> Optional[tuple[list[float], str, int]]: - """Reranka i documenti rispetto alla query tramite la catena di fallback. - - Ritorna (scores, backend_name, took_ms) dove scores è allineato a docs - (logit sigmoide in [0,1]), oppure None se tutti i nodi falliscono. - """ - chain = _get_chain() - if not chain or not docs: - return None - # Troncamento dei documenti: limita il costo di inferenza (i cross-encoder - # scala con la lunghezza della coppia query+doc) e evita input oltre il ctx. - docs = [d[:RERANK_MAX_DOC_CHARS] for d in docs] - live, all_down = live_nodes() - if not live: - # tutti in cooldown: ritenta comunque il primo (meglio di un fallimento immediato) - live = [chain[0]] - payload = {"model": RERANK_MODEL, "query": query, "documents": docs, "top_n": len(docs)} - started = time.monotonic() - for node in live: - headers = {"Content-Type": "application/json"} - if node.key: - headers["Authorization"] = f"Bearer {node.key}" - try: - t0 = time.monotonic() - response = await get_http().post( - f"{node.url}/v1/rerank", - json=payload, - headers=headers, - timeout=httpx.Timeout(node.timeout_ms / 1000.0), - ) - response.raise_for_status() - data = response.json() - # Il risultato è [{index, relevance_score}] ordinato per rilevanza: - # riportiamo ogni score alla posizione originaria del documento. - scores = [0.0] * len(docs) - for item in data.get("results", []): - idx = int(item["index"]) - if 0 <= idx < len(docs): - scores[idx] = float(item.get("relevance_score", 0.0)) - took = int((time.monotonic() - started) * 1000) - metrics.record_rerank(node.name, True, took) - return scores, node.name, took - except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc: - took = int((time.monotonic() - t0) * 1000) - _down_until[node.url] = time.monotonic() + RERANK_RETRY_COOLDOWN_S - metrics.record_rerank(node.name, False, took) - log.warning( - "rerank: nodo '%s' fallito dopo %dms (%s: %s) → cooldown %ds", - node.name, - took, - exc.__class__.__name__, - exc, - RERANK_RETRY_COOLDOWN_S, - ) - return None - - -def normalize_score(logit: float) -> float: - """Sigmoide: logit di rilevanza → punteggio [0,1] leggibile nei risultati.""" - if logit >= 0: - z = math.exp(-logit) - return 1.0 / (1.0 + z) - z = math.exp(logit) - return z / (1.0 + z) \ No newline at end of file diff --git a/gateway/routes.py b/gateway/routes.py deleted file mode 100644 index f5e54a4..0000000 --- a/gateway/routes.py +++ /dev/null @@ -1,333 +0,0 @@ -"""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) diff --git a/gateway/state.py b/gateway/state.py deleted file mode 100644 index 187d73c..0000000 --- a/gateway/state.py +++ /dev/null @@ -1,83 +0,0 @@ -"""Stato runtime condiviso tra bootstrap e route.""" -from __future__ import annotations - -import contextvars -import hashlib -import json -import time -from typing import Any - -import httpx -from qdrant_client import QdrantClient - -from config import QDRANT_API_KEY, QDRANT_URL, QDRANT_RETRIES, log, _metrics - - -class ResilientQdrant: - """Proxy del client Qdrant che ritenta i metodi su errori di transport - (connessione/timeout transienti, es. riavvio del container Qdrant). - Gli errori applicativi (404, validazione) non vengono ritentati.""" - - def __init__(self, client: Any, attempts: int = QDRANT_RETRIES, backoff_s: float = 0.4): - self._client = client - self._attempts = max(1, attempts) - self._backoff = backoff_s - - @staticmethod - def _transient(exc: Exception) -> bool: - if isinstance(exc, httpx.TransportError): - return True - return type(exc).__name__ in ("ConnectionError", "TimeoutError") - - def __getattr__(self, name: str) -> Any: - attr = getattr(self._client, name) - if not callable(attr): - return attr - - def wrapped(*args: Any, **kwargs: Any) -> Any: - for attempt in range(self._attempts): - try: - return attr(*args, **kwargs) - except Exception as exc: # noqa: BLE001 - if attempt == self._attempts - 1 or not self._transient(exc): - raise - delay = self._backoff * (2**attempt) - _metrics["qdrant_retries"] += 1 - log.warning( - "qdrant.%s: errore transiente (%s: %s) → retry %d/%d tra %.1fs", - name, - exc.__class__.__name__, - exc, - attempt + 2, - self._attempts, - delay, - ) - time.sleep(delay) - - return wrapped - - -qdrant = ResilientQdrant(QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)) -request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-") -ratelimit: dict[str, list[float]] = {} -status_ratelimit: dict[str, list[float]] = {} -idempotency: dict[str, dict[str, Any]] = {} -meta_cache: dict[str, Any] = {} -STATUS_RATE_LIMIT_PER_MIN = 30 -IDEMPOTENCY_TTL_SECONDS = 24 * 3600 - - -def payload_hash(body: Any) -> str: - canonical = json.dumps(body.model_dump(), sort_keys=True, default=str) - return hashlib.sha256(canonical.encode()).hexdigest() - - -def idempotency_cleanup() -> None: - now = time.time() - expired = [k for k, v in idempotency.items() if now - v["ts"] > IDEMPOTENCY_TTL_SECONDS] - for key in expired: - idempotency.pop(key, None) - - -def invalidate_meta() -> None: - meta_cache.clear() diff --git a/gateway/store.py b/gateway/store.py deleted file mode 100644 index 8063661..0000000 --- a/gateway/store.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Operazioni Qdrant condivise dalle route.""" -from __future__ import annotations - -import hashlib -from typing import Any, Optional - -from qdrant_client.http import models as qm - -from config import SPARSE_VECTOR_NAME -from models import SearchIn - - -def search_filter(body: SearchIn) -> qm.Filter | None: - must: list[Any] = [] - for key in ("kind", "project_id", "scope", "parent_id", "level", "topic"): - value = getattr(body, key) - if value: - must.append(qm.FieldCondition(key=key, match=qm.MatchValue(value=value))) - if not body.include_superseded: - must.append(qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by"))) - must_not: list[Any] = [] - if not body.include_private: - # default: esclude i record riservati (private=true) dalle ricerche standard - must_not.append(qm.FieldCondition(key="private", match=qm.MatchValue(value=True))) - if must or must_not: - return qm.Filter(must=must or None, must_not=must_not or None) - return None - - -def search(qdrant: Any, collection: str, body: SearchIn, vector: list[float], sparse: Any, limit: Optional[int] = None) -> list[Any]: - """Ricerca ibrida o densa. Con reranking attivo limit > top_k per dare candidati extra allo stadio di rerank.""" - eff_limit = limit if limit is not None else body.top_k - qfilter = search_filter(body) - if body.hybrid and sparse is not None: - return qdrant.query_points( - collection_name=collection, - prefetch=[ - qm.Prefetch(query=vector, using="", limit=max(body.top_k * 4, eff_limit), score_threshold=body.min_score), - qm.Prefetch(query=sparse, using=SPARSE_VECTOR_NAME, limit=max(body.top_k * 4, eff_limit)), - ], - query=qm.FusionQuery(fusion=qm.Fusion.RRF), - query_filter=qfilter, - limit=eff_limit, - with_payload=True, - ).points - return qdrant.query_points( - collection_name=collection, - query=vector, - query_filter=qfilter, - limit=eff_limit, - score_threshold=body.min_score, - with_payload=True, - ).points - - -def format_results(hits: list[Any]) -> list[dict]: - return [ - { - "memory_id": h.id, - "score": round(h.score, 4), - "text": h.payload.get("text"), - "kind": h.payload.get("kind"), - "agent_id": h.payload.get("agent_id"), - "scope": h.payload.get("scope"), - "project_id": h.payload.get("project_id"), - "confidence": h.payload.get("confidence"), - "importance": h.payload.get("importance", 0.5), - "created_at": h.payload.get("created_at"), - "source": h.payload.get("source"), - "supersedes_id": h.payload.get("supersedes_id"), - "superseded_by": h.payload.get("superseded_by"), - "supersede_reason": h.payload.get("supersede_reason"), - "parent_id": h.payload.get("parent_id"), - "level": h.payload.get("level"), - "topic": h.payload.get("topic"), - "private": h.payload.get("private", False), - "links": h.payload.get("links"), - } - for h in hits - ] - - -def reparent_active_children(qdrant: Any, collection: str, old_id: str, new_id: str) -> int: - children, _ = qdrant.scroll( - collection_name=collection, - scroll_filter=qm.Filter(must=[ - qm.FieldCondition(key="parent_id", match=qm.MatchValue(value=old_id)), - qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")), - ]), - limit=1000, - with_payload=False, - ) - if not children: - return 0 - qdrant.set_payload(collection_name=collection, payload={"parent_id": new_id}, points=[p.id for p in children]) - return len(children) diff --git a/gateway/tests/conftest.py b/gateway/tests/conftest.py deleted file mode 100644 index c9e1335..0000000 --- a/gateway/tests/conftest.py +++ /dev/null @@ -1,154 +0,0 @@ -"""Fixtures pytest per il Memory Gateway: FakeQdrant in-memory + mock di embed. - -I test non richiedono Qdrant né Ollama: il gateway viene importato con le -dipendenze reali (fastapi/pydantic/qdrant-client/httpx) ma qdrant e embed -sono sostituiti da fake deterministici. -""" - -import os -import sys -from pathlib import Path - -# Disabilita il push metriche nei test -os.environ["METRICS_ENABLED"] = "false" -os.environ["API_KEYS"] = "test-key" -# Guardrail disabilitato di default nei test esistenti (abilitato nei test del guardrail) -os.environ["GUARDRAIL_ENABLED"] = "false" - -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - -import pytest # noqa: E402 -import main as gateway # noqa: E402 - - -class FakePoint: - def __init__(self, point_id, vector=None, payload=None): - self.id = point_id - self.vector = vector or {} - self.payload = payload or {} - - -class FakeQdrant: - """Implementazione in-memory dei metodi Qdrant usati dal gateway.""" - - def __init__(self): - self.points: dict[str, FakePoint] = {} - self.collection_exists = False - self.upsert_calls = 0 - self.query_score = 0.9 # score di default per query_points (configurabile nei test) - - def get_collections(self): - class _C: - def __init__(self, names): - self.collections = [type("X", (), {"name": n})() for n in names] - - return _C(["memories"] if self.collection_exists else []) - - def create_collection(self, **kw): - self.collection_exists = True - - def create_payload_index(self, **kw): - pass - - def create_vector_name(self, *a, **kw): - pass - - def upsert(self, collection_name, points, **kw): - self.upsert_calls += 1 - for p in points: - self.points[p.id] = FakePoint(p.id, p.vector, p.payload) - - def retrieve(self, collection_name, ids, with_payload=True): - return [self.points[i] for i in ids if i in self.points] - - def set_payload(self, collection_name, payload, points, **kw): - for pid in points: - if pid in self.points: - self.points[pid].payload.update(payload) - - def delete(self, collection_name, points_selector, **kw): - for pid in points_selector: - self.points.pop(pid, None) - - def _matches(self, pl, query_filter): - """Applica i filtri metadata (FieldCondition match / IsEmptyCondition).""" - if not query_filter or not query_filter.must: - return True - for cond in query_filter.must: - if hasattr(cond, "key") and hasattr(cond, "match"): - if pl.get(cond.key) != cond.match.value: - return False - elif hasattr(cond, "is_empty"): - if pl.get(cond.is_empty.key): - return False - return True - - def scroll(self, collection_name, scroll_filter=None, limit=None, with_payload=True, **kw): - points = [p for p in self.points.values() if self._matches(p.payload, scroll_filter)] - if limit: - points = points[:limit] - return points, None - - def get_collection(self, collection_name): - class _Info: - points_count = len(self.points) - - return _Info() - - def query_points(self, collection_name, query=None, query_filter=None, limit=5, - score_threshold=None, with_payload=True, prefetch=None, **kw): - # Ritorna tutti i punti (score fisso); i filtri metadata sono applicati - # in modo semplice per testare kind/project_id/scope/superseded. - results = [] - for p in self.points.values(): - pl = p.payload - if query_filter and query_filter.must: - ok = True - for cond in query_filter.must: - if hasattr(cond, "key") and hasattr(cond, "match"): - if pl.get(cond.key) != cond.match.value: - ok = False - elif hasattr(cond, "is_empty"): - if pl.get(cond.is_empty.key): - ok = False - if not ok: - continue - results.append(type("H", (), {"id": p.id, "score": self.query_score, "payload": pl})()) - return type("R", (), {"points": results[:limit]})() - - -@pytest.fixture -def client(monkeypatch): - """TestClient con qdrant e embed finti.""" - fake = FakeQdrant() - monkeypatch.setattr(gateway.state, "qdrant", fake) - monkeypatch.setattr(gateway, "qdrant", fake) - monkeypatch.setattr(gateway, "API_KEYS", {"test-key"}, raising=False) - monkeypatch.setattr(gateway.state, "ratelimit", {}) # rate limit pulito per test - - async def fake_embed(text): - return [0.0] * 1024 - - monkeypatch.setattr(gateway.state, "embed", fake_embed) - monkeypatch.setattr(gateway, "embed", fake_embed) - - from fastapi.testclient import TestClient - - with TestClient(gateway.app) as c: - c.fake_qdrant = fake - yield c - - -def auth_headers(): - return {"X-API-Key": "test-key"} - - -def make_record(**overrides): - base = { - "text": "record di test", - "kind": "fact", - "project_id": "test-proj", - "scope": "agent", - } - base.update(overrides) - return base diff --git a/gateway/tests/test_api.py b/gateway/tests/test_api.py deleted file mode 100644 index 030bd6f..0000000 --- a/gateway/tests/test_api.py +++ /dev/null @@ -1,399 +0,0 @@ -"""Test API del Memory Gateway: validazione, auth, idempotency, supersede, ricerca.""" - -import pytest -from conftest import auth_headers, make_record - - -# --------------------------------------------------------------------------- -# Validazione input -# --------------------------------------------------------------------------- -def test_project_id_obbligatorio(client): - body = make_record() - del body["project_id"] - r = client.post("/v1/memories", json=body, headers=auth_headers()) - assert r.status_code == 422 - - -def test_kind_invalido(client): - r = client.post("/v1/memories", json=make_record(kind="boh"), headers=auth_headers()) - assert r.status_code == 422 - - -def test_expires_at_invalido(client): - r = client.post("/v1/memories", json=make_record(expires_at="non-una-data"), headers=auth_headers()) - assert r.status_code == 422 - assert "ISO 8601" in r.text - - -def test_expires_at_valido(client): - r = client.post("/v1/memories", json=make_record(expires_at="2026-09-01T00:00:00Z"), headers=auth_headers()) - assert r.status_code == 200 - - -def test_confidence_invalido(client): - r = client.post("/v1/memories", json=make_record(confidence="super"), headers=auth_headers()) - assert r.status_code == 422 - - -def test_confidence_default_medium(client): - r = client.post("/v1/memories", json=make_record(), headers=auth_headers()) - assert r.status_code == 200 - mid = r.json()["memory_id"] - g = client.get(f"/v1/memories/{mid}", headers=auth_headers()) - assert g.json()["confidence"] == "medium" - - -def test_text_troppo_lungo(client): - r = client.post("/v1/memories", json=make_record(text="x" * 9000), headers=auth_headers()) - assert r.status_code == 422 - - -# --------------------------------------------------------------------------- -# Auth e rate limit -# --------------------------------------------------------------------------- -def test_senza_chiave_422(client): - # Header X-API-Key mancante → 422 (header richiesto da FastAPI) - r = client.post("/v1/memories", json=make_record()) - assert r.status_code == 422 - - -def test_chiave_invalida_401(client): - r = client.post("/v1/memories", json=make_record(), headers={"X-API-Key": "sbagliata"}) - assert r.status_code == 401 - - -def test_rate_limit_429(client, monkeypatch): - import config - - monkeypatch.setattr(config, "RATE_LIMIT_PER_MIN", 3) - for _ in range(3): - r = client.post("/v1/memories", json=make_record(), headers=auth_headers()) - assert r.status_code == 200 - r = client.post("/v1/memories", json=make_record(), headers=auth_headers()) - assert r.status_code == 429 - - -# --------------------------------------------------------------------------- -# Idempotency -# --------------------------------------------------------------------------- -def test_idempotency_replay_stessa_risposta(client): - h = {**auth_headers(), "Idempotency-Key": "k-1"} - r1 = client.post("/v1/memories", json=make_record(), headers=h) - r2 = client.post("/v1/memories", json=make_record(), headers=h) - assert r1.status_code == 200 and r2.status_code == 200 - assert r1.json()["memory_id"] == r2.json()["memory_id"] - assert client.fake_qdrant.upsert_calls == 1 - - -def test_idempotency_payload_diverso_409(client): - h = {**auth_headers(), "Idempotency-Key": "k-2"} - client.post("/v1/memories", json=make_record(), headers=h) - r = client.post("/v1/memories", json=make_record(text="diverso"), headers=h) - assert r.status_code == 409 - - -def test_idempotency_key_diverse_record_distinti(client): - r1 = client.post("/v1/memories", json=make_record(), headers={**auth_headers(), "Idempotency-Key": "k-a"}) - r2 = client.post("/v1/memories", json=make_record(), headers={**auth_headers(), "Idempotency-Key": "k-b"}) - assert r1.json()["memory_id"] != r2.json()["memory_id"] - - -# --------------------------------------------------------------------------- -# Supersede -# --------------------------------------------------------------------------- -def test_supersede_target_inesistente_404(client): - r = client.post( - "/v1/memories", - json=make_record(supersedes_id="00000000-0000-0000-0000-000000000000"), - headers=auth_headers(), - ) - assert r.status_code == 404 - - -def test_supersede_ok_e_lineage(client): - r1 = client.post("/v1/memories", json=make_record(text="fatto falso"), headers=auth_headers()) - old_id = r1.json()["memory_id"] - r2 = client.post( - "/v1/memories", - json=make_record(text="fatto corretto", supersedes_id=old_id, supersede_reason="evidenza"), - headers=auth_headers(), - ) - assert r2.status_code == 200 - new_id = r2.json()["memory_id"] - # il vecchio è marcato superseded_by - old = client.get(f"/v1/memories/{old_id}", headers=auth_headers()).json() - assert old["superseded_by"] == new_id - # la ricerca di default esclude i superseduti - s = client.post("/v1/memories:search", json={"query": "fatto", "top_k": 10, "min_score": 0.0}, headers=auth_headers()) - ids = [x["memory_id"] for x in s.json()["results"]] - assert old_id not in ids - # include_superseded li mostra - s2 = client.post( - "/v1/memories:search", - json={"query": "fatto", "top_k": 10, "min_score": 0.0, "include_superseded": True}, - headers=auth_headers(), - ) - ids2 = [x["memory_id"] for x in s2.json()["results"]] - assert old_id in ids2 - - -def test_supersede_doppio_409(client): - r1 = client.post("/v1/memories", json=make_record(text="falso"), headers=auth_headers()) - old_id = r1.json()["memory_id"] - client.post("/v1/memories", json=make_record(text="corretto", supersedes_id=old_id), headers=auth_headers()) - r = client.post("/v1/memories", json=make_record(text="ancora", supersedes_id=old_id), headers=auth_headers()) - assert r.status_code == 409 - - -def test_supersede_root_ri_parenta_figli_attivi(client): - # L1 root + figlio L2 - r1 = client.post( - "/v1/memories", - json=make_record(text="root L1", level="L1_ROOT", topic="TEST-TOPIC/ROOT"), - headers=auth_headers(), - ) - root_id = r1.json()["memory_id"] - r2 = client.post( - "/v1/memories", - json=make_record(text="figlio L2", level="L2_SUBTOPIC", topic="TEST-TOPIC/SUB", parent_id=root_id), - headers=auth_headers(), - ) - child_id = r2.json()["memory_id"] - - # supersede il root - r3 = client.post( - "/v1/memories", - json=make_record( - text="root L1 corretto", - level="L1_ROOT", - topic="TEST-TOPIC/ROOT", - supersedes_id=root_id, - supersede_reason="aggiornamento", - ), - headers=auth_headers(), - ) - assert r3.status_code == 200 - new_root_id = r3.json()["memory_id"] - assert r3.json()["reparented"] == 1 - - # il figlio attivo ora punta al nuovo root - child = client.get(f"/v1/memories/{child_id}", headers=auth_headers()).json() - assert child["parent_id"] == new_root_id - - # search per parent_id sul nuovo root trova il figlio - s = client.post( - "/v1/memories:search", - json={"query": "*", "parent_id": new_root_id, "top_k": 10, "min_score": 0.0}, - headers=auth_headers(), - ) - ids = [x["memory_id"] for x in s.json()["results"]] - assert child_id in ids - - -def test_supersede_root_ri_parenta_solo_figli_attivi(client): - # L1 root + figlio L2 + figlio L2 già superseduto (versione attiva C1') - r1 = client.post( - "/v1/memories", - json=make_record(text="root", level="L1_ROOT", topic="T2/ROOT"), - headers=auth_headers(), - ) - root_id = r1.json()["memory_id"] - c1 = client.post( - "/v1/memories", - json=make_record(text="figlio vecchio", level="L2_SUBTOPIC", topic="T2/SUB", parent_id=root_id), - headers=auth_headers(), - ) - c1_id = c1.json()["memory_id"] - c1p = client.post( - "/v1/memories", - json=make_record( - text="figlio nuovo", - level="L2_SUBTOPIC", - topic="T2/SUB", - parent_id=root_id, - supersedes_id=c1_id, - ), - headers=auth_headers(), - ) - c1p_id = c1p.json()["memory_id"] - - r2 = client.post( - "/v1/memories", - json=make_record(text="root corretto", level="L1_ROOT", topic="T2/ROOT", supersedes_id=root_id), - headers=auth_headers(), - ) - new_root_id = r2.json()["memory_id"] - assert r2.json()["reparented"] == 1 # solo C1' (attivo) - - # C1' ri-parentato al nuovo root; C1 storico resta ancorato al vecchio - assert client.get(f"/v1/memories/{c1p_id}", headers=auth_headers()).json()["parent_id"] == new_root_id - assert client.get(f"/v1/memories/{c1_id}", headers=auth_headers()).json()["parent_id"] == root_id - - -# --------------------------------------------------------------------------- -# Ricerca e filtri -# --------------------------------------------------------------------------- -def test_search_filtro_project_id(client): - client.post("/v1/memories", json=make_record(text="uno", project_id="proj-a"), headers=auth_headers()) - client.post("/v1/memories", json=make_record(text="due", project_id="proj-b"), headers=auth_headers()) - s = client.post( - "/v1/memories:search", - json={"query": "test", "project_id": "proj-a", "top_k": 10, "min_score": 0.0}, - headers=auth_headers(), - ) - results = s.json()["results"] - assert len(results) == 1 - assert results[0]["project_id"] == "proj-a" - - -def test_search_hybrid_param_accettato(client): - client.post("/v1/memories", json=make_record(text="codice XYZ-123"), headers=auth_headers()) - s = client.post( - "/v1/memories:search", - json={"query": "XYZ-123", "top_k": 5, "min_score": 0.0, "hybrid": True}, - headers=auth_headers(), - ) - assert s.status_code == 200 - assert "results" in s.json() - - -def test_meta_overview(client): - client.post("/v1/memories", json=make_record(project_id="proj-a"), headers=auth_headers()) - r = client.get("/v1/meta/overview", headers=auth_headers()) - assert r.status_code == 200 - data = r.json() - assert data["total"] >= 1 - assert any(p["project_id"] == "proj-a" for p in data["projects"]) - - -def test_status_pubblico(client): - r = client.get("/v1/status") - assert r.status_code == 200 - assert r.json()["status"] == "ok" - - -def test_metrics_auth(client): - # Header mancante → 422; chiave invalida → 401; chiave valida → 200 - assert client.get("/v1/metrics").status_code == 422 - assert client.get("/v1/metrics", headers={"X-API-Key": "sbagliata"}).status_code == 401 - r2 = client.get("/v1/metrics", headers=auth_headers()) - assert r2.status_code == 200 - assert "requests" in r2.json() - - -# --------------------------------------------------------------------------- -# Versione / metadata del codice -# --------------------------------------------------------------------------- -def test_version_endpoint_pubblico(client): - """GET /v1/version è pubblico e espone git_commit e guardrail_version.""" - r = client.get("/v1/version") - assert r.status_code == 200 - data = r.json() - assert "git_commit" in data - assert "version" in data - assert "guardrail_version" in data - assert data["guardrail_version"] == "similarity-v2" - - -def test_status_espone_git_commit(client): - """/v1/status include version, git_commit e guardrail_version.""" - r = client.get("/v1/status") - assert r.status_code == 200 - data = r.json() - assert "git_commit" in data - assert "version" in data - assert "guardrail_version" in data - - -# --------------------------------------------------------------------------- -# Struttura Gerarchica e Relazionale -# --------------------------------------------------------------------------- -def test_create_and_retrieve_hierarchical_record(client): - """Crea un nodo Root L1 e un nodo Figlio L2 con links, parent_id, level, topic.""" - # 1. Crea Root L1 - r_root = client.post( - "/v1/memories", - json=make_record( - text="Master Topic Alfa Romeo", - level="L1_ROOT", - topic="ALFA-ROMEO/ROOT", - ), - headers=auth_headers(), - ) - assert r_root.status_code == 200 - root_id = r_root.json()["memory_id"] - - # 2. Crea Figlio L2 collegato - r_child = client.post( - "/v1/memories", - json=make_record( - text="Scheda Tecnica Bialbero 1.3", - parent_id=root_id, - level="L2_SUBTOPIC", - topic="ALFA-ROMEO/SPECS", - links=[{"target_id": root_id, "predicate": "part_of", "weight": 1.0}], - ), - headers=auth_headers(), - ) - assert r_child.status_code == 200 - child_id = r_child.json()["memory_id"] - - # 3. Recupera e verifica payload strutturato - g = client.get(f"/v1/memories/{child_id}", headers=auth_headers()) - assert g.status_code == 200 - data = g.json() - assert data["parent_id"] == root_id - assert data["level"] == "L2_SUBTOPIC" - assert data["topic"] == "ALFA-ROMEO/SPECS" - assert len(data["links"]) == 1 - assert data["links"][0]["target_id"] == root_id - - -def test_search_filters_hierarchical(client): - """Filtra per parent_id, level e topic.""" - r_root = client.post( - "/v1/memories", - json=make_record(text="Root doc", level="L1_ROOT", topic="TOPIC/ROOT"), - headers=auth_headers(), - ) - root_id = r_root.json()["memory_id"] - - client.post( - "/v1/memories", - json=make_record(text="Child A", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/A"), - headers=auth_headers(), - ) - client.post( - "/v1/memories", - json=make_record(text="Child B", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/B"), - headers=auth_headers(), - ) - - # Cerca solo L1_ROOT - s1 = client.post( - "/v1/memories:search", - json={"query": "doc", "level": "L1_ROOT", "top_k": 5, "min_score": 0.0}, - headers=auth_headers(), - ) - assert s1.status_code == 200 - assert len(s1.json()["results"]) == 1 - assert s1.json()["results"][0]["level"] == "L1_ROOT" - - # Cerca per parent_id - s2 = client.post( - "/v1/memories:search", - json={"query": "Child", "parent_id": root_id, "top_k": 5, "min_score": 0.0}, - headers=auth_headers(), - ) - assert s2.status_code == 200 - assert len(s2.json()["results"]) == 2 - - # Cerca per topic specifico - s3 = client.post( - "/v1/memories:search", - json={"query": "Child", "topic": "TOPIC/A", "top_k": 5, "min_score": 0.0}, - headers=auth_headers(), - ) - assert s3.status_code == 200 - assert len(s3.json()["results"]) == 1 - assert s3.json()["results"][0]["topic"] == "TOPIC/A" diff --git a/gateway/tests/test_embed_chain.py b/gateway/tests/test_embed_chain.py deleted file mode 100644 index 76d22d6..0000000 --- a/gateway/tests/test_embed_chain.py +++ /dev/null @@ -1,189 +0,0 @@ -"""Test della catena di fallback per gli embedding e del wrapper Qdrant resilient.""" -from __future__ import annotations - -import asyncio -import json - -import httpx -import pytest - -import embed as embed_mod -import state -from test_api import auth_headers, make_record - -CHAIN = json.dumps( - [ - {"name": "primario", "url": "http://primario:9001", "api": "llamacpp", "key": "k1", "timeout_ms": 500}, - {"name": "fallback", "url": "http://fallback:9002", "api": "ollama", "key": "k2", "timeout_ms": 5000}, - ] -) - -VEC_1024 = [0.01] * 1024 - - -@pytest.fixture(autouse=True) -def _reset_chain(monkeypatch): - embed_mod.reset_chain_cache() - embed_mod._http = None - yield - embed_mod.reset_chain_cache() - embed_mod._http = None - - -def _mock_client(handler) -> list[str]: - calls: list[str] = [] - - def tracking_handler(request): - calls.append(f"{request.url.host}{request.url.path}") - return handler(request) - - embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking_handler)) - return calls - - -# --------------------------------------------------------------------------- -# parse della catena -# --------------------------------------------------------------------------- - - -def test_parse_chain_valida(monkeypatch): - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) - embed_mod.reset_chain_cache() - nodes = embed_mod.chain_nodes() - assert [n.name for n in nodes] == ["primario", "fallback"] - assert [n.api for n in nodes] == ["llamacpp", "ollama"] - - -def test_parse_legacy_quando_catena_vuota(monkeypatch): - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", "") - monkeypatch.setattr(embed_mod, "EMBED_API", "ollama") - monkeypatch.setattr(embed_mod, "EMBED_URL", "http://legacy:11434") - monkeypatch.setattr(embed_mod, "EMBED_API_KEY", "lk") - embed_mod.reset_chain_cache() - nodes = embed_mod.chain_nodes() - assert len(nodes) == 1 - assert nodes[0].name == "embed" - assert nodes[0].api == "ollama" - assert nodes[0].url == "http://legacy:11434" - assert nodes[0].key == "lk" - - -def test_parse_chain_json_invalido_cade_su_legacy(monkeypatch): - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", "non-json") - monkeypatch.setattr(embed_mod, "EMBED_URL", "http://legacy:11434") - embed_mod.reset_chain_cache() - assert [n.url for n in embed_mod.chain_nodes()] == ["http://legacy:11434"] - - -# --------------------------------------------------------------------------- -# fallback e cooldown -# --------------------------------------------------------------------------- - - -def test_fallback_primario_llamacpp_fallito(monkeypatch): - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) - embed_mod.reset_chain_cache() - - def handler(request): - if request.url.host == "primario": - return httpx.Response(500) - # nodo ollama-style: risposta con campo "embeddings" - return httpx.Response(200, json={"embeddings": [VEC_1024]}) - - embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) - vector = asyncio.run(embed_mod.embed("test")) - assert vector == VEC_1024 - - -def test_cooldown_salta_primario(monkeypatch): - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) - embed_mod.reset_chain_cache() - calls: list[str] = [] - - def tracking(request): - calls.append(request.url.host) - if request.url.host == "primario": - return httpx.Response(500) - return httpx.Response(200, json={"embeddings": [VEC_1024]}) - - embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking)) - asyncio.run(embed_mod.embed("test")) - asyncio.run(embed_mod.embed("test")) - # il primario fallito entra in cooldown: la seconda chiamata lo salta - assert calls == ["primario", "fallback", "fallback"] - - -def test_dimensione_errata_salta_nodo(monkeypatch): - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) - embed_mod.reset_chain_cache() - - def handler(request): - if request.url.host == "primario": - return httpx.Response(200, json={"data": [{"embedding": [0.0] * 512}]}) # dim sbagliata - return httpx.Response(200, json={"embeddings": [VEC_1024]}) - - embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) - vector = asyncio.run(embed_mod.embed("test")) - assert len(vector) == 1024 - - -def test_tutti_nodi_falliti_rilancia(monkeypatch): - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) - embed_mod.reset_chain_cache() - embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(500))) - with pytest.raises(RuntimeError, match="tutti i nodi embedding falliti"): - asyncio.run(embed_mod.embed("test")) - - -def test_empty_text_comunque_chiamata(monkeypatch): - # il modello pydantic valida già la query; qui verifichiamo il passthrough - monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) - embed_mod.reset_chain_cache() - - def handler(request): - body = json.loads(request.content) - return httpx.Response(200, json={"embeddings": [VEC_1024] if body["input"] else []}) - - embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) - assert asyncio.run(embed_mod.embed("ok")) == VEC_1024 - - -# --------------------------------------------------------------------------- -# ResilientQdrant (retry transiente) -# --------------------------------------------------------------------------- - - -class FlakyQdrant: - def __init__(self, failures: int, exc: Exception): - self.calls = 0 - self.failures = failures - self.exc = exc - - def upsert(self, **kw): - self.calls += 1 - if self.calls <= self.failures: - raise self.exc - return "ok" - - -def test_retry_su_errore_transiente(monkeypatch): - fake = FlakyQdrant(2, httpx.ConnectError("conn")) - client = state.ResilientQdrant(fake, attempts=3, backoff_s=0.01) - assert client.upsert(x=1) == "ok" - assert fake.calls == 3 - - -def test_niente_retry_su_errore_applicativo(): - fake = FlakyQdrant(2, ValueError("404 logico")) - client = state.ResilientQdrant(fake, attempts=3, backoff_s=0.01) - with pytest.raises(ValueError): - client.upsert(x=1) - assert fake.calls == 1 - - -def test_retry_esaurito_rilancia(): - fake = FlakyQdrant(99, httpx.ReadTimeout("t")) - client = state.ResilientQdrant(fake, attempts=2, backoff_s=0.01) - with pytest.raises(httpx.ReadTimeout): - client.upsert(x=1) - assert fake.calls == 2 \ No newline at end of file diff --git a/gateway/tests/test_guardrail.py b/gateway/tests/test_guardrail.py deleted file mode 100644 index 04c6952..0000000 --- a/gateway/tests/test_guardrail.py +++ /dev/null @@ -1,111 +0,0 @@ -"""Test del guardrail di similarità pre-scrittura (Memory Gateway). - -Casi: duplicato esatto (hash) -> BLOCK 409; similarità alta -> BLOCK 409; -similarità moderata -> WARN (salva con flag); nessun candidato -> ALLOW; -supersede esplicito bypassa il guardrail. -""" - -import pytest -from conftest import auth_headers, make_record - - -@pytest.fixture(autouse=True) -def enable_guardrail(monkeypatch): - """Abilita il guardrail per i test di questa suite (il conftest lo disabilita di default).""" - import config - - monkeypatch.setattr(config, "GUARDRAIL_ENABLED", True) - - -def test_duplicato_esatto_bloccato_409(client): - """Stesso testo normalizzato -> hash uguale -> BLOCK (409).""" - body = make_record(text="Il cliente non accede al portale COSMO-SkyMed") - r1 = client.post("/v1/memories", json=body, headers=auth_headers()) - assert r1.status_code == 200 - - # Stesso testo con maiuscole/spazi diversi -> stesso hash normalizzato - body2 = make_record(text=" IL CLIENTE NON ACCEDE al portale COSMO-SkyMed ") - r2 = client.post("/v1/memories", json=body2, headers=auth_headers()) - assert r2.status_code == 409 - detail = r2.json()["detail"] - assert detail["error"] == "duplicate_memory" - assert detail["reason"] == "EXACT_DUPLICATE" - - -def test_similarita_alta_bloccata_409(client): - """Score top-1 >= soglia BLOCK (0.85) -> 409 KNOWN_SOLUTION.""" - client.fake_qdrant.query_score = 0.9 - r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - assert r1.status_code == 200 - - r2 = client.post("/v1/memories", json=make_record(text="secondo record simile"), headers=auth_headers()) - assert r2.status_code == 409 - assert r2.json()["detail"]["reason"] == "KNOWN_SOLUTION" - - -def test_similarita_moderata_warn_salva(client): - """Score top-1 tra 0.70 e 0.85 -> WARN: salva con flag guardrail.""" - client.fake_qdrant.query_score = 0.75 - r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - assert r1.status_code == 200 - - r2 = client.post("/v1/memories", json=make_record(text="secondo record simile"), headers=auth_headers()) - assert r2.status_code == 200 - memory_id = r2.json()["memory_id"] - # Il record salvato deve avere il flag guardrail WARN - saved = client.fake_qdrant.points[memory_id].payload - assert saved["guardrail"]["decision"] == "WARN" - assert saved["guardrail"]["reason"] == "MODERATE_SIMILARITY" - - -def test_nessun_candidato_allow(client): - """Score top-1 sotto soglia WARN -> ALLOW, nessun flag guardrail.""" - client.fake_qdrant.query_score = 0.5 - r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - assert r1.status_code == 200 - - r2 = client.post("/v1/memories", json=make_record(text="secondo record"), headers=auth_headers()) - assert r2.status_code == 200 - memory_id = r2.json()["memory_id"] - saved = client.fake_qdrant.points[memory_id].payload - assert saved["guardrail"]["decision"] == "ALLOW" - assert saved["guardrail"]["reason"] == "NEW_SOLUTION" - - -def test_supersede_bypassa_guardrail(client): - """Il supersede esplicito è una correzione intenzionale: bypassa il guardrail.""" - client.fake_qdrant.query_score = 0.9 - r1 = client.post("/v1/memories", json=make_record(text="record originale"), headers=auth_headers()) - assert r1.status_code == 200 - old_id = r1.json()["memory_id"] - - # Supersede con testo molto simile -> deve passare (correzione intenzionale) - r2 = client.post( - "/v1/memories", - json=make_record(text="record originale corretto", supersedes_id=old_id, supersede_reason="correzione"), - headers=auth_headers(), - ) - assert r2.status_code == 200 - assert r2.json()["supersedes_id"] == old_id - - -def test_guardrail_disabilitato_salva_sempre(client, monkeypatch): - """Con GUARDRAIL_ENABLED=false non si blocca nulla.""" - import config - - monkeypatch.setattr(config, "GUARDRAIL_ENABLED", False) - client.fake_qdrant.query_score = 0.9 - r1 = client.post("/v1/memories", json=make_record(text="primo"), headers=auth_headers()) - assert r1.status_code == 200 - r2 = client.post("/v1/memories", json=make_record(text="primo"), headers=auth_headers()) - assert r2.status_code == 200 - - -def test_text_hash_salvato_nel_payload(client): - """Ogni record salvato deve avere text_hash (per lo strato 1 del guardrail).""" - r = client.post("/v1/memories", json=make_record(text="record con hash"), headers=auth_headers()) - assert r.status_code == 200 - memory_id = r.json()["memory_id"] - saved = client.fake_qdrant.points[memory_id].payload - assert "text_hash" in saved - assert len(saved["text_hash"]) == 64 # SHA-256 hex diff --git a/gateway/tests/test_rerank.py b/gateway/tests/test_rerank.py deleted file mode 100644 index 49922ef..0000000 --- a/gateway/tests/test_rerank.py +++ /dev/null @@ -1,219 +0,0 @@ -"""Test dello stadio rerank: catena di fallback, cooldown, degrada con grazia.""" -from __future__ import annotations - -import asyncio -import json - -import httpx -import pytest - -import rerank -from test_api import auth_headers, make_record - -CHAIN = json.dumps( - [ - {"name": "primario", "url": "http://primario:9002", "key": "k1", "timeout_ms": 500}, - {"name": "fallback", "url": "http://fallback:9003", "key": "k2", "timeout_ms": 5000}, - ] -) - - -@pytest.fixture(autouse=True) -def _reset_chain(monkeypatch): - rerank.reset_chain_cache() - rerank._http = None - yield - rerank.reset_chain_cache() - rerank._http = None - - -def _use_chain(monkeypatch, raw=CHAIN): - # rerank.py importa i valori di config con `from config import`: si patchano - # gli attributi del modulo rerank, non config. - monkeypatch.setattr(rerank, "RERANK_CHAIN", raw) - rerank.reset_chain_cache() - - -def _mock_client(handler) -> list[str]: - """Client con transport mockato; ritorna la lista in cui registrare le chiamate.""" - calls: list[str] = [] - - def tracking_handler(request): - calls.append(f"{request.url.host}{request.url.path}") - return handler(request) - - rerank._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking_handler)) - return calls - - -# --------------------------------------------------------------------------- -# parse della catena -# --------------------------------------------------------------------------- - - -def test_parse_chain_valida(monkeypatch): - _use_chain(monkeypatch) - nodes = rerank._get_chain() - assert [n.name for n in nodes] == ["primario", "fallback"] - assert nodes[0].key == "k1" - assert nodes[0].timeout_ms == 500 - assert nodes[1].timeout_ms == 5000 - - -def test_parse_chain_json_invalido(monkeypatch): - _use_chain(monkeypatch, raw="non-json") - assert rerank._get_chain() == [] - assert not rerank.enabled() - - -def test_parse_chain_scarta_url_senza_schema(monkeypatch): - _use_chain(monkeypatch, raw=json.dumps([{"name": "x", "url": "primario:9002"}])) - assert rerank._get_chain() == [] - - -# --------------------------------------------------------------------------- -# fallback e cooldown -# --------------------------------------------------------------------------- - - -def test_fallback_primario_500(monkeypatch): - _use_chain(monkeypatch) - - def handler(request): - if request.url.host == "primario": - return httpx.Response(500) - return httpx.Response(200, json={"results": [{"index": 1, "relevance_score": 2.0}, {"index": 0, "relevance_score": -1.0}]}) - - _mock_client(handler) - scores, backend, _took = asyncio.run(rerank.rerank("q", ["docA", "docB"])) - assert backend == "fallback" - # gli score tornano allineati alla posizione originaria dei documenti - assert scores == pytest.approx([-1.0, 2.0]) - - -def test_cooldown_salta_nodo_fallito(monkeypatch): - _use_chain(monkeypatch) - - def handler(request): - if request.url.host == "primario": - return httpx.Response(500) - return httpx.Response(200, json={"results": [{"index": 0, "relevance_score": 1.0}]}) - - _mock_client(handler) - asyncio.run(rerank.rerank("q", ["a", "b"])) - scores, backend, _ = asyncio.run(rerank.rerank("q", ["a", "b"])) - assert backend == "fallback" # il primario è in cooldown e non viene richiamato - - -def test_cooldown_non_blocca_per_sempre(monkeypatch): - _use_chain(monkeypatch) - calls: list[str] = [] - - def handler(request): - calls.append(request.url.host) - if request.url.host == "primario": - return httpx.Response(500) - return httpx.Response(200, json={"results": [{"index": 0, "relevance_score": 1.0}]}) - - _mock_client(handler) - asyncio.run(rerank.rerank("q", ["a"])) - # svuota il cooldown: il primario torna eleggibile - rerank._down_until.clear() - asyncio.run(rerank.rerank("q", ["a"])) - assert calls == ["primario", "fallback", "primario", "fallback"] - - -def test_tutti_nodi_falliti_restifica_none(monkeypatch): - _use_chain(monkeypatch) - _mock_client(lambda request: httpx.Response(500)) - assert asyncio.run(rerank.rerank("q", ["a", "b"])) is None - - -def test_enabled_richiede_catena(monkeypatch): - _use_chain(monkeypatch, raw="") - assert not rerank.enabled() - _use_chain(monkeypatch) - monkeypatch.setattr(rerank, "RERANK_ENABLED", True) - assert rerank.enabled() - monkeypatch.setattr(rerank, "RERANK_ENABLED", False) - assert not rerank.enabled() - - -def test_normalize_score(): - assert rerank.normalize_score(0.0) == pytest.approx(0.5) - assert rerank.normalize_score(10.0) > 0.99 - assert rerank.normalize_score(-10.0) < 0.01 - - -def test_troncamento_documenti(monkeypatch): - _use_chain(monkeypatch) - seen: dict = {} - - def handler(request): - seen["docs"] = json.loads(request.content)["documents"] - return httpx.Response(200, json={"results": [{"index": 0, "relevance_score": 1.0}]}) - - _mock_client(handler) - asyncio.run(rerank.rerank("q", ["x" * 5000, "corto"])) - assert len(seen["docs"][0]) == 800 # default RERANK_MAX_DOC_CHARS - assert seen["docs"][1] == "corto" - - -# --------------------------------------------------------------------------- -# integrazione endpoint search -# --------------------------------------------------------------------------- - - -def test_search_rerank_riordina(client, monkeypatch): - for text in ["alpha", "beta", "gamma"]: - r = client.post("/v1/memories", json=make_record(text=text), headers=auth_headers()) - assert r.status_code == 200 - - async def fake_rerank(query, docs): - # inverte: beta (index 1) primo, poi gamma/alpha - return [0.1, 0.9, 0.5], "finto", 12 - - monkeypatch.setattr(rerank, "enabled", lambda: True) - monkeypatch.setattr(rerank, "rerank", fake_rerank) - - resp = client.post("/v1/memories:search", json={"query": "q", "top_k": 3}, headers=auth_headers()) - assert resp.status_code == 200 - data = resp.json() - assert data["rerank"]["used"] is True - assert data["rerank"]["backend"] == "finto" - texts = [r["text"] for r in data["results"]] - assert texts == ["beta", "gamma", "alpha"] - assert data["results"][0]["rerank_score"] == pytest.approx(rerank.normalize_score(0.9), abs=0.01) - - -def test_search_rerank_disattivato_per_query(client, monkeypatch): - for text in ["alpha", "beta"]: - client.post("/v1/memories", json=make_record(text=text), headers=auth_headers()) - - async def fail_rerank(query, docs): - raise AssertionError("rerank non deve essere chiamato con rerank=false") - - monkeypatch.setattr(rerank, "enabled", lambda: True) - monkeypatch.setattr(rerank, "rerank", fail_rerank) - - resp = client.post("/v1/memories:search", json={"query": "q", "top_k": 2, "rerank": False}, headers=auth_headers()) - assert resp.status_code == 200 - assert resp.json()["rerank"]["enabled"] is False - - -def test_search_rerank_fallito_degrada_con_grazia(client, monkeypatch): - for text in ["alpha", "beta"]: - client.post("/v1/memories", json=make_record(text=text), headers=auth_headers()) - - async def fail_rerank(query, docs): - return None - - monkeypatch.setattr(rerank, "enabled", lambda: True) - monkeypatch.setattr(rerank, "rerank", fail_rerank) - - resp = client.post("/v1/memories:search", json={"query": "q", "top_k": 2}, headers=auth_headers()) - assert resp.status_code == 200 - data = resp.json() - assert data["rerank"]["used"] is False - assert "non raggiungibili" in data["rerank"]["reason"] - assert len(data["results"]) == 2 # ordine di fusione preservato \ No newline at end of file diff --git a/gateway/tests/test_strategies.py b/gateway/tests/test_strategies.py deleted file mode 100644 index c96f7ab..0000000 --- a/gateway/tests/test_strategies.py +++ /dev/null @@ -1,252 +0,0 @@ -"""Test strategie rerank oltre la search: gate store (A), supersede verify (B), -score composito (C), multi-query (E), primitiva /v1/score.""" -from __future__ import annotations - -import pytest -from conftest import auth_headers, make_record - - -@pytest.fixture(autouse=True) -def enable_guardrail(monkeypatch): - import config - - monkeypatch.setattr(config, "GUARDRAIL_ENABLED", True) - - -def _patch_rerank(monkeypatch, scores, backend="finto", took=10): - """Abilita il rerank anche nel guardrail (che importa i valori da config).""" - monkeypatch.setattr("rerank.enabled", lambda: True) - monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) - recorded: dict = {} - - async def fake_rerank(query, docs): - recorded["query"] = query - recorded["docs"] = list(docs) - return scores, backend, took - - monkeypatch.setattr("rerank.rerank", fake_rerank) - return recorded - - -# --------------------------------------------------------------------------- -# A: gate store con cross-encoder -# --------------------------------------------------------------------------- - - -def test_cosine_alto_cross_basso_downgrade_a_warn(client, monkeypatch): - """Cosine 0.9 (zona BLOCK) ma cross-score basso → WARN CROSS_DUP_WEAK, salva.""" - _patch_rerank(monkeypatch, scores=[-6.0]) # sigmoid ≈ 0.0024 - client.fake_qdrant.query_score = 0.9 - client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - r2 = client.post("/v1/memories", json=make_record(text="secondo simile"), headers=auth_headers()) - assert r2.status_code == 200 - saved = client.fake_qdrant.points[r2.json()["memory_id"]].payload - assert saved["guardrail"]["decision"] == "WARN" - assert saved["guardrail"]["reason"] == "CROSS_DUP_WEAK" - - -def test_cosine_warn_cross_alto_upgrade_a_block(client, monkeypatch): - """Cosine in zona WARN (0.75) ma cross-score altissimo → BLOCK parafrasato catturato.""" - import rerank as rr_mod - monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) - monkeypatch.setattr("rerank.enabled", lambda: True) - - async def fake_rerank(query, docs): - return [3.0], "finto", 5 # sigmoid ≈ 0.953 ≥ 0.88 - - monkeypatch.setattr("rerank.rerank", fake_rerank) - client.fake_qdrant.query_score = 0.75 - client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - r2 = client.post("/v1/memories", json=make_record(text="stesso fatto riformulato"), headers=auth_headers()) - assert r2.status_code == 409 - assert r2.json()["detail"]["reason"] == "CROSS_DUP_CONFIRMED" - - -def test_cosine_basso_cross_alto_block_low_cosine(client, monkeypatch): - """Cosine sotto soglia WARN (0.5) ma cross altissimo → CROSS_DUP_LOW_COSINE.""" - monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) - monkeypatch.setattr("rerank.enabled", lambda: True) - - async def fake_rerank(query, docs): - return [4.0], "finto", 5 # sigmoid ≈ 0.982 - - monkeypatch.setattr("rerank.rerank", fake_rerank) - client.fake_qdrant.query_score = 0.5 - client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - r2 = client.post("/v1/memories", json=make_record(text="secondo riformulato"), headers=auth_headers()) - assert r2.status_code == 409 - assert r2.json()["detail"]["reason"] == "CROSS_DUP_LOW_COSINE" - - -def test_warn_con_suggerimento_supersedes(client, monkeypatch): - """Cosine in zona WARN, cross ≥ soglia suggest → WARN con suggestion.supersedes_id.""" - monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) - monkeypatch.setattr("rerank.enabled", lambda: True) - - async def fake_rerank(query, docs): - return [1.8], "finto", 5 # sigmoid ≈ 0.858: ≥ 0.85 (suggest), < 0.90 (block) - - monkeypatch.setattr("rerank.rerank", fake_rerank) - client.fake_qdrant.query_score = 0.75 - r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - old_id = r1.json()["memory_id"] - r2 = client.post("/v1/memories", json=make_record(text="secondo simile"), headers=auth_headers()) - assert r2.status_code == 200 - saved = client.fake_qdrant.points[r2.json()["memory_id"]].payload - assert saved["guardrail"]["reason"] == "MODERATE_SIMILARITY" - assert saved["guardrail"]["suggestion"]["supersedes_id"] == old_id - - -def test_rerank_giu_degrada_a_solo_cosine(client, monkeypatch): - """Reranker irraggiungibile → decisione legacy per cosine (BLOCK a 0.9).""" - monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) - monkeypatch.setattr("rerank.enabled", lambda: True) - - async def fail_rerank(query, docs): - return None - - monkeypatch.setattr("rerank.rerank", fail_rerank) - client.fake_qdrant.query_score = 0.9 - client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) - r2 = client.post("/v1/memories", json=make_record(text="secondo simile"), headers=auth_headers()) - assert r2.status_code == 409 - assert r2.json()["detail"]["reason"] == "KNOWN_SOLUTION" - - -# --------------------------------------------------------------------------- -# B: verifica supersede -# --------------------------------------------------------------------------- - - -def test_supersede_cross_basso_warning(client, monkeypatch): - monkeypatch.setattr("config.GUARDRAIL_SUPERSEDE_CHECK", True) - monkeypatch.setattr("rerank.enabled", lambda: True) - - async def fake_rerank(query, docs): - return [-8.0], "finto", 5 # sigmoid ≈ 0.0003 < 0.50 - - monkeypatch.setattr("rerank.rerank", fake_rerank) - r1 = client.post("/v1/memories", json=make_record(text="record originale"), headers=auth_headers()) - old_id = r1.json()["memory_id"] - r2 = client.post( - "/v1/memories", - json=make_record(text="contenuto del tutto diverso", supersedes_id=old_id, supersede_reason="fix"), - headers=auth_headers(), - ) - assert r2.status_code == 200 - data = r2.json() - assert data["supersede_warning"]["cross_score"] < 0.1 - assert "lineage" in data["supersede_warning"]["message"] - - -def test_supersede_cross_alto_nessun_warning(client, monkeypatch): - monkeypatch.setattr("config.GUARDRAIL_SUPERSEDE_CHECK", True) - monkeypatch.setattr("rerank.enabled", lambda: True) - - async def fake_rerank(query, docs): - return [3.0], "finto", 5 # sigmoid ≈ 0.95 ≥ 0.50 - - monkeypatch.setattr("rerank.rerank", fake_rerank) - r1 = client.post("/v1/memories", json=make_record(text="record originale"), headers=auth_headers()) - old_id = r1.json()["memory_id"] - r2 = client.post( - "/v1/memories", - json=make_record(text="record originale corretto", supersedes_id=old_id, supersede_reason="correzione"), - headers=auth_headers(), - ) - assert r2.status_code == 200 - assert "supersede_warning" not in r2.json() - - -# --------------------------------------------------------------------------- -# C: score composito -# --------------------------------------------------------------------------- - - -def test_composite_score_in_risultati(client, monkeypatch): - client.fake_qdrant.query_score = 0.5 # il guardrail cosine non blocca il seeding - client.post("/v1/memories", json=make_record(text="record alpha", importance=1.0, confidence="high"), headers=auth_headers()) - client.post("/v1/memories", json=make_record(text="record beta", importance=0.0, confidence="low"), headers=auth_headers()) - - async def fake_rerank(query, docs): - return [2.0, 2.0], "finto", 5 # rerank in parità → il composito decide - - monkeypatch.setattr("rerank.enabled", lambda: True) - monkeypatch.setattr("rerank.rerank", fake_rerank) - - resp = client.post("/v1/memories:search", json={"query": "record", "top_k": 2}, headers=auth_headers()) - assert resp.status_code == 200 - data = resp.json() - results = data["results"] - assert all("composite_score" in r for r in results) - assert results[0]["text"] == "record alpha" - assert results[0]["composite_score"] > results[1]["composite_score"] - - -# --------------------------------------------------------------------------- -# E: multi-query -# --------------------------------------------------------------------------- - - -def test_multi_query_pool_unito(client, monkeypatch): - client.fake_qdrant.query_score = 0.5 # il guardrail cosine non blocca il seeding - for text in ["alpha", "beta", "gamma"]: - client.post("/v1/memories", json=make_record(text=text), headers=auth_headers()) - - embed_calls: list[str] = [] - - async def fake_embed(text): - embed_calls.append(text) - return [0.0] * 1024 - - monkeypatch.setattr("state.embed", fake_embed) - monkeypatch.setattr("state.sparse_encode", lambda text: None) - - async def fake_rerank(query, docs): - assert query == "alpha" # il rerank usa la query principale - return [0.9, 0.5, 0.1], "finto", 5 - - monkeypatch.setattr("rerank.enabled", lambda: True) - monkeypatch.setattr("rerank.rerank", fake_rerank) - - resp = client.post( - "/v1/memories:search", - json={"query": "alpha", "queries": ["gamma", "alpha "], "top_k": 3}, - headers=auth_headers(), - ) - assert resp.status_code == 200 - data = resp.json() - assert data["rerank"]["queries_used"] == 2 # "alpha " normalizzata e deduplicata - assert len(embed_calls) == 2 - assert set(r["text"] for r in data["results"]) == {"alpha", "beta", "gamma"} - assert data["results"][0]["text"] == "alpha" - - -# --------------------------------------------------------------------------- -# Primitiva /v1/score -# --------------------------------------------------------------------------- - - -def test_score_endpoint_ok(client, monkeypatch): - async def fake_rerank(query, docs): - assert query == "q" - return [2.0, -3.0], "finto", 7 - - monkeypatch.setattr("rerank.enabled", lambda: True) - monkeypatch.setattr("rerank.rerank", fake_rerank) - resp = client.post("/v1/score", json={"query": "q", "documents": ["a", "b"]}, headers=auth_headers()) - assert resp.status_code == 200 - data = resp.json() - assert data["scores"][0] == pytest.approx(0.88, abs=0.01) - assert data["scores"][1] < 0.1 - assert data["backend"] == "finto" - - -def test_score_endpoint_503_se_catena_giu(client, monkeypatch): - async def fail_rerank(query, docs): - return None - - monkeypatch.setattr("rerank.enabled", lambda: True) - monkeypatch.setattr("rerank.rerank", fail_rerank) - resp = client.post("/v1/score", json={"query": "q", "documents": ["a"]}, headers=auth_headers()) - assert resp.status_code == 503 \ No newline at end of file