diff --git a/scripts/pi2-backup/auto-git-backup.sh b/scripts/pi2-backup/auto-git-backup.sh index 327131c..d15ce5e 100755 --- a/scripts/pi2-backup/auto-git-backup.sh +++ b/scripts/pi2-backup/auto-git-backup.sh @@ -87,10 +87,11 @@ else fi # D. Sorgente Loogle MCP (app completa: qui vive solo su Pi1, fuori da git) -# .env resta escluso: contiene JWT secret e token Paperless/Gitea/OpenAI. +# .env e ogni sua copia (.env.bak*, .env.old...) restano esclusi: contengono JWT secret, +# token Paperless/Gitea/HA e password. Passa solo .env.example (l'include va prima). mkdir -p "$REPO_DIR/services/loogle-mcp" rsync -av --delete -e "ssh -q" \ - --exclude='.env' --exclude='data/' --exclude='__pycache__/' --exclude='*.pyc' --exclude='.cursor/' \ + --include='.env.example' --exclude='.env*' --exclude='data/' --exclude='__pycache__/' --exclude='*.pyc' --exclude='.cursor/' \ "$PI1_USER@$PI1_IP:/home/daniely/docker/loogle-mcp/" "$REPO_DIR/services/loogle-mcp/" \ || echo "⚠️ rsync sorgente MCP fallito (ignorato)" diff --git a/services/loogle-mcp/.gitignore b/services/loogle-mcp/.gitignore index 5b70d24..ab54a44 100644 --- a/services/loogle-mcp/.gitignore +++ b/services/loogle-mcp/.gitignore @@ -1,4 +1,6 @@ data/ .env +.env.* +!.env.example *.pyc __pycache__/ diff --git a/services/loogle-mcp/app/context/store.py b/services/loogle-mcp/app/context/store.py index a4c35c8..659284a 100644 --- a/services/loogle-mcp/app/context/store.py +++ b/services/loogle-mcp/app/context/store.py @@ -5,6 +5,7 @@ import json import os import re import secrets +import threading from datetime import datetime from typing import Optional @@ -13,6 +14,26 @@ from ..knowledge import gitea SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$") +# Contesti condivisi: stessa struttura dei progetti personali, sotto un namespace comune. +# Tutta la famiglia li legge e li aggiorna; archiviazione e ritorno a personale solo +# per proprietario o admin. +SHARED_NS = "_shared" +SHARED_WITH_FAMILY = "family" +_SHARED_META_KEYS = ("shared", "shared_with", "owner", "shared_at", "shared_by", "updated_by") + + +class ConflictError(ValueError): + """Il contesto è stato modificato da qualcun altro dopo l'ultima lettura.""" + +# I tool MCP girano in parallelo nel threadpool: serializza le scritture per progetto +_project_locks: dict[tuple[str, str], threading.Lock] = {} +_project_locks_guard = threading.Lock() + + +def _project_lock(username: str, project_id: str) -> threading.Lock: + with _project_locks_guard: + return _project_locks.setdefault((username, project_id), threading.Lock()) + def _context_root() -> str: return os.environ.get("MCP_CONTEXT_ROOT", "/data/context") @@ -24,19 +45,60 @@ def _user_root(username: str) -> str: return path +def _shared_root() -> str: + path = os.path.join(_context_root(), SHARED_NS, "projects") + os.makedirs(path, exist_ok=True) + return path + + def _project_dir(username: str, project_id: str) -> str: if not SLUG_RE.match(project_id): raise ValueError("ID progetto non valido") return os.path.join(_user_root(username), project_id) +def _id_taken(project_id: str, username: str, shared: bool) -> bool: + """Un ID condiviso non deve coincidere con progetti personali di nessun utente + (altrimenti per quell'utente il personale nasconderebbe il condiviso).""" + root = _context_root() + if os.path.exists(os.path.join(_shared_root(), project_id)): + return True + users = [u for u in os.listdir(root) if u != SHARED_NS] if shared else [username] + return any(os.path.exists(os.path.join(root, u, "projects", project_id)) for u in users) + + +def _read_meta(meta_path: str) -> dict: + with open(meta_path, encoding="utf-8") as f: + return json.load(f) + + +def _is_member(meta: dict) -> bool: + return meta.get("shared_with") == SHARED_WITH_FAMILY + + +def _resolve_project(username: str, project_id: str) -> tuple[str, str]: + """(namespace, cartella) del progetto: prima quello personale, poi quello condiviso.""" + personal = _project_dir(username, project_id) + if os.path.isfile(os.path.join(personal, "meta.json")): + return username, personal + shared = os.path.join(_shared_root(), project_id) + meta_path = os.path.join(shared, "meta.json") + if os.path.isfile(meta_path) and _is_member(_read_meta(meta_path)): + return SHARED_NS, shared + raise FileNotFoundError("Progetto non trovato") + + def _load_meta(username: str, project_id: str) -> tuple[str, dict]: - proj_dir = _project_dir(username, project_id) + _, proj_dir = _resolve_project(username, project_id) meta_path = os.path.join(proj_dir, "meta.json") - if not os.path.isfile(meta_path): - raise FileNotFoundError("Progetto non trovato") - meta = json.load(open(meta_path, encoding="utf-8")) - return meta_path, meta + return meta_path, _read_meta(meta_path) + + +def _require_owner_or_admin(username: str, meta: dict, is_admin: bool, action: str) -> None: + if meta.get("shared") and not is_admin and meta.get("owner") != username: + raise PermissionError( + f"Solo il proprietario ({meta.get('owner')}) o un admin può {action} un contesto condiviso" + ) def _save_meta(meta_path: str, meta: dict) -> None: @@ -67,7 +129,7 @@ def _apply_gitea_link( meta.pop("gitea_repo", None) if seed_from_gitea and repo: - proj_dir = _project_dir(username, meta["id"]) + _, proj_dir = _resolve_project(username, meta["id"]) ctx_path = os.path.join(proj_dir, "context.md") context_md = open(ctx_path, encoding="utf-8").read() if os.path.isfile(ctx_path) else "" new_md, changed = gitea_link.seed_context_from_readme(username, repo, context_md) @@ -80,18 +142,19 @@ def _apply_gitea_link( def list_projects(username: str, include_archived: bool = False) -> list: - root = _user_root(username) + """Progetti personali dell'utente seguiti dai contesti condivisi (campo shared=true).""" projects = [] - if not os.path.isdir(root): - return projects - for name in sorted(os.listdir(root)): - meta_path = os.path.join(root, name, "meta.json") - if not os.path.isfile(meta_path): - continue - meta = json.load(open(meta_path, encoding="utf-8")) - if meta.get("archived") and not include_archived: - continue - projects.append(meta) + for root, shared in ((_user_root(username), False), (_shared_root(), True)): + for name in sorted(os.listdir(root)): + meta_path = os.path.join(root, name, "meta.json") + if not os.path.isfile(meta_path): + continue + meta = _read_meta(meta_path) + if shared and not _is_member(meta): + continue + if meta.get("archived") and not include_archived: + continue + projects.append(meta) return projects @@ -101,13 +164,14 @@ def create_project( tags: Optional[list[str]] = None, gitea_repo: Optional[str] = None, seed_from_gitea: bool = False, + shared: bool = False, ) -> dict: slug_base = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "progetto" - project_id = slug_base[:40] - root = _user_root(username) - while os.path.exists(os.path.join(root, project_id)): - project_id = f"{slug_base[:32]}-{secrets.token_hex(2)}" - proj_dir = _project_dir(username, project_id) + project_id = slug_base[:40].strip("-") + # ID univoco sia tra i progetti personali sia tra i condivisi (la risoluzione li unisce) + while _id_taken(project_id, username, shared): + project_id = f"{slug_base[:32].strip('-')}-{secrets.token_hex(2)}" + proj_dir = os.path.join(_shared_root() if shared else _user_root(username), project_id) os.makedirs(os.path.join(proj_dir, "sessions"), exist_ok=True) os.makedirs(os.path.join(proj_dir, "artifacts"), exist_ok=True) now = datetime.now().astimezone().isoformat(timespec="seconds") @@ -120,6 +184,8 @@ def create_project( "created_at": now, "updated_at": now, } + if shared: + meta.update(_shared_fields(username, now)) repo = gitea_link.normalize_gitea_repo(gitea_repo) if repo: meta["gitea_repo"] = repo @@ -142,21 +208,73 @@ def create_project( return meta +def _shared_fields(username: str, now: str) -> dict: + return { + "shared": True, + "shared_with": SHARED_WITH_FAMILY, + "owner": username, + "shared_by": username, + "shared_at": now, + } + + +def share_project(username: str, project_id: str, shared: bool = True, is_admin: bool = False) -> dict: + """Sposta un progetto tra lo spazio personale del proprietario e quello condiviso. + + La storia (context.md, sessions/, artifacts/) si sposta con il progetto. + """ + ns, proj_dir = _resolve_project(username, project_id) + meta = _read_meta(os.path.join(proj_dir, "meta.json")) + if shared: + if ns == SHARED_NS: + return meta + target = os.path.join(_shared_root(), project_id) + others = [ + u for u in os.listdir(_context_root()) + if u not in (SHARED_NS, username) and os.path.exists(os.path.join(_context_root(), u, "projects", project_id)) + ] + if others: + raise ValueError( + f"L'ID '{project_id}' è usato anche da un progetto personale di {', '.join(others)}: " + "crea un contesto condiviso nuovo (create_project shared=true) e copiaci il contenuto" + ) + else: + if ns != SHARED_NS: + return meta + _require_owner_or_admin(username, meta, is_admin, "rendere personale") + target = os.path.join(_user_root(meta["owner"]), project_id) + if os.path.exists(target): + raise ValueError(f"Esiste già un progetto '{project_id}' nella destinazione: rinomina prima") + with _project_lock(ns, project_id): + os.rename(proj_dir, target) + meta_path = os.path.join(target, "meta.json") + meta = _read_meta(meta_path) + if shared: + meta.update(_shared_fields(username, datetime.now().astimezone().isoformat(timespec="seconds"))) + else: + for key in _SHARED_META_KEYS: + meta.pop(key, None) + _save_meta(meta_path, meta) + return meta + + def link_project_repo( username: str, project_id: str, gitea_repo: Optional[str] = None, seed_from_gitea: bool = False, ) -> dict: - meta_path, meta = _load_meta(username, project_id) - meta = _apply_gitea_link( - username, - meta, - gitea_repo, - verify=bool(gitea_link.normalize_gitea_repo(gitea_repo)), - seed_from_gitea=seed_from_gitea, - ) - _save_meta(meta_path, meta) + ns, _ = _resolve_project(username, project_id) + with _project_lock(ns, project_id): + meta_path, meta = _load_meta(username, project_id) + meta = _apply_gitea_link( + username, + meta, + gitea_repo, + verify=bool(gitea_link.normalize_gitea_repo(gitea_repo)), + seed_from_gitea=seed_from_gitea, + ) + _save_meta(meta_path, meta) return meta @@ -166,11 +284,8 @@ def get_project_context( session_limit: int = 5, include_gitea: bool = True, ) -> dict: - proj_dir = _project_dir(username, project_id) - meta_path = os.path.join(proj_dir, "meta.json") - if not os.path.isfile(meta_path): - raise FileNotFoundError("Progetto non trovato") - meta = json.load(open(meta_path, encoding="utf-8")) + _, proj_dir = _resolve_project(username, project_id) + meta = _read_meta(os.path.join(proj_dir, "meta.json")) ctx_path = os.path.join(proj_dir, "context.md") context_md = open(ctx_path, encoding="utf-8").read() if os.path.isfile(ctx_path) else "" sessions_dir = os.path.join(proj_dir, "sessions") @@ -189,11 +304,33 @@ def get_project_context( return result -def save_context(username: str, project_id: str, content: str, mode: str = "append") -> dict: - proj_dir = _project_dir(username, project_id) +def save_context( + username: str, + project_id: str, + content: str, + mode: str = "append", + if_revision: Optional[int] = None, +) -> dict: + """if_revision: meta.revision letto prima di modificare; se nel frattempo qualcuno + ha salvato, ConflictError invece di sovrascrivere (utile nei contesti condivisi).""" + ns, proj_dir = _resolve_project(username, project_id) + with _project_lock(ns, project_id): + return _save_context_locked(username, proj_dir, content, mode, if_revision) + + +def _save_context_locked( + username: str, proj_dir: str, content: str, mode: str, if_revision: Optional[int] +) -> dict: meta_path = os.path.join(proj_dir, "meta.json") if not os.path.isfile(meta_path): raise FileNotFoundError("Progetto non trovato") + meta = _read_meta(meta_path) + if if_revision is not None and int(meta.get("revision", 0)) != int(if_revision): + raise ConflictError( + f"Contesto modificato (revisione {meta.get('revision', 0)}) il {meta.get('updated_at')}" + + (f" da {meta['updated_by']}" if meta.get("updated_by") else "") + + ": rileggilo con get_project_context e riapplica la modifica" + ) ctx_path = os.path.join(proj_dir, "context.md") if mode == "replace": text = content @@ -202,31 +339,45 @@ def save_context(username: str, project_id: str, content: str, mode: str = "appe text = existing.rstrip() + "\n\n" + content.strip() + "\n" with open(ctx_path, "w", encoding="utf-8") as f: f.write(text) - meta = json.load(open(meta_path, encoding="utf-8")) + meta["revision"] = int(meta.get("revision", 0)) + 1 + if meta.get("shared"): + meta["updated_by"] = username _save_meta(meta_path, meta) - snapshot = datetime.now().strftime("%Y%m%d-%H%M%S") + ".md" + # Nei condivisi il nome della sessione dice chi ha scritto + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + (f"-{username}" if meta.get("shared") else "") + snapshot = stamp + ".md" + n = 1 + while os.path.exists(os.path.join(proj_dir, "sessions", snapshot)): + snapshot = f"{stamp}-{n}.md" + n += 1 with open(os.path.join(proj_dir, "sessions", snapshot), "w", encoding="utf-8") as f: f.write(content) return meta -def archive_project(username: str, project_id: str, archived: bool = True) -> dict: - meta_path, meta = _load_meta(username, project_id) - meta["archived"] = archived - meta["status"] = "archived" if archived else "active" - _save_meta(meta_path, meta) +def archive_project( + username: str, project_id: str, archived: bool = True, is_admin: bool = False +) -> dict: + ns, _ = _resolve_project(username, project_id) + with _project_lock(ns, project_id): + meta_path, meta = _load_meta(username, project_id) + _require_owner_or_admin(username, meta, is_admin, "archiviare") + meta["archived"] = archived + meta["status"] = "archived" if archived else "active" + _save_meta(meta_path, meta) return meta def list_resources(username: str) -> list: resources = [] for meta in list_projects(username, include_archived=False): - desc = f"Contesto progetto {meta['id']}" + ns = SHARED_NS if meta.get("shared") else username + desc = f"Contesto {'condiviso con la famiglia' if meta.get('shared') else 'progetto'} {meta['id']}" if meta.get("gitea_repo"): desc += f" (Gitea: {meta['gitea_repo']})" resources.append( { - "uri": f"loogle://context/{username}/{meta['id']}", + "uri": f"loogle://context/{ns}/{meta['id']}", "name": meta["title"], "description": desc, "mimeType": "text/markdown", @@ -236,10 +387,13 @@ def list_resources(username: str) -> list: def read_resource(username: str, uri: str) -> dict: - prefix = f"loogle://context/{username}/" - if not uri.startswith(prefix): + for ns in (username, SHARED_NS): + prefix = f"loogle://context/{ns}/" + if uri.startswith(prefix): + project_id = uri[len(prefix):] + break + else: raise FileNotFoundError("Risorsa non trovata") - project_id = uri[len(prefix):] data = get_project_context(username, project_id, session_limit=0, include_gitea=False) text = data["context_md"] gitea_repo = data["meta"].get("gitea_repo") diff --git a/services/loogle-mcp/app/integrations/homeassistant.py b/services/loogle-mcp/app/integrations/homeassistant.py index 6244fc2..84fb6c7 100644 --- a/services/loogle-mcp/app/integrations/homeassistant.py +++ b/services/loogle-mcp/app/integrations/homeassistant.py @@ -29,7 +29,7 @@ def _request(method: str, path: str, *, params: Optional[dict] = None) -> Any: "0", "false", "no", "off", ) url = path if path.startswith("http") else urljoin(API_URL + "/", path.lstrip("/")) - with httpx.Client(timeout=30.0, verify=verify) as client: + with httpx.Client(timeout=httpx.Timeout(30.0, connect=5.0), verify=verify) as client: resp = client.request(method, url, headers=_headers(), params=params) if resp.status_code >= 400: raise RuntimeError(f"Home Assistant {path}: HTTP {resp.status_code} {resp.text[:200]}") diff --git a/services/loogle-mcp/app/integrations/session_client.py b/services/loogle-mcp/app/integrations/session_client.py index 075aca9..1f52775 100644 --- a/services/loogle-mcp/app/integrations/session_client.py +++ b/services/loogle-mcp/app/integrations/session_client.py @@ -78,7 +78,7 @@ class SessionApiClient: f"Imposta {self.service}_PASSWORD_{username.upper()} in .env" ) url = urljoin(self.base_url + "/", self.login_path.lstrip("/")) - with httpx.Client(timeout=30.0, verify=self.verify_ssl) as client: + with httpx.Client(timeout=httpx.Timeout(30.0, connect=5.0), verify=self.verify_ssl) as client: resp = client.post( url, json={"username": service_user, "password": password}, ) @@ -104,7 +104,7 @@ class SessionApiClient: cookie = self.login(username) url = path if path.startswith("http") else urljoin(self.base_url + "/", path.lstrip("/")) headers = {"Cookie": f"session={cookie}"} - with httpx.Client(timeout=60.0, verify=self.verify_ssl) as client: + with httpx.Client(timeout=httpx.Timeout(60.0, connect=5.0), verify=self.verify_ssl) as client: resp = client.request(method, url, headers=headers, params=params, json=json_body) if resp.status_code == 401: with _sessions_lock: diff --git a/services/loogle-mcp/app/integrations/turni.py b/services/loogle-mcp/app/integrations/turni.py index 478472c..f425b72 100644 --- a/services/loogle-mcp/app/integrations/turni.py +++ b/services/loogle-mcp/app/integrations/turni.py @@ -93,7 +93,7 @@ def login(mcp_username: str) -> str: "0", "false", "no", "off", ) url = urljoin(API_URL + "/", "api/auth/login") - with httpx.Client(timeout=30.0, verify=verify) as client: + with httpx.Client(timeout=httpx.Timeout(30.0, connect=5.0), verify=verify) as client: resp = client.post(url, json={"username": service_user, "password": password}) if resp.status_code != 200: raise RuntimeError(f"Login Turni fallito: HTTP {resp.status_code}") @@ -118,7 +118,7 @@ def _request( ) url = path if path.startswith("http") else urljoin(API_URL + "/", path.lstrip("/")) headers = {"Authorization": f"Bearer {token}"} - with httpx.Client(timeout=60.0, verify=verify) as client: + with httpx.Client(timeout=httpx.Timeout(60.0, connect=5.0), verify=verify) as client: resp = client.request(method, url, headers=headers, params=params) if resp.status_code == 401: with _jwt_lock: @@ -135,7 +135,7 @@ def get_status() -> dict: "0", "false", "no", "off", ) url = urljoin(API_URL + "/", "api/status") - with httpx.Client(timeout=30.0, verify=verify) as client: + with httpx.Client(timeout=httpx.Timeout(30.0, connect=5.0), verify=verify) as client: resp = client.get(url) resp.raise_for_status() return resp.json() diff --git a/services/loogle-mcp/app/knowledge/apps_indexer.py b/services/loogle-mcp/app/knowledge/apps_indexer.py index 32b9278..a452494 100644 --- a/services/loogle-mcp/app/knowledge/apps_indexer.py +++ b/services/loogle-mcp/app/knowledge/apps_indexer.py @@ -267,7 +267,7 @@ def index_all(*, username: Optional[str] = None) -> dict: def search_apps_knowledge(query: str, limit: int = 8, source: Optional[str] = None) -> list: - vectors = embeddings.embed_texts([query]) + vectors = [embeddings.embed_query(query)] flt = {"source": source} if source else None hits = qdrant_store.search( [qdrant_store.APPS_SHARED_COLLECTION], diff --git a/services/loogle-mcp/app/knowledge/embeddings.py b/services/loogle-mcp/app/knowledge/embeddings.py index 84b6389..21d73e8 100644 --- a/services/loogle-mcp/app/knowledge/embeddings.py +++ b/services/loogle-mcp/app/knowledge/embeddings.py @@ -4,6 +4,7 @@ import logging import os import time +from functools import lru_cache from typing import Optional import httpx @@ -13,39 +14,123 @@ from . import thermal LOGGER = logging.getLogger("loogle_mcp.embeddings") -def embed_texts(texts: list[str]) -> list[list[float]]: - if not texts: - return [] +def embed_query(text: str) -> list[float]: + """Embedding interattivo per una query di ricerca. + + Ordine: OLLAMA_QUERY_URL (istanza dedicata su pi1, ~0.1s, mai in coda dietro + l'indicizzazione) → OLLAMA_URL (DS920, senza thermal gate) → OpenAI. + Timeout corti così il client MCP riceve un errore invece di un hang. + Cache LRU: gli agenti ripetono spesso le stesse query. + """ + return list(_embed_query_cached(text.strip(), os.environ.get("OLLAMA_EMBED_MODEL", "nomic-embed-text"))) + + +@lru_cache(maxsize=256) +def _embed_query_cached(text: str, _model: str) -> tuple[float, ...]: + return tuple(_embed_query_uncached(text)) + + +def _embed_query_uncached(text: str) -> list[float]: + query_url = os.environ.get("OLLAMA_QUERY_URL", "").strip() ollama_url = os.environ.get("OLLAMA_URL", "").strip() + openai_key = os.environ.get("OPENAI_API_KEY", "").strip() + last_exc: Optional[Exception] = None + if query_url: + try: + # Istanza dedicata: options proprie, il modello resta caricato (keep_alive -1) + return _embed_ollama_query( + text, query_url, read_s=_float_env("OLLAMA_QUERY_URL_TIMEOUT_S", 10.0), + keep_alive=-1, options=_ollama_options("OLLAMA_QUERY_NUM_THREAD"), + ) + except Exception as exc: + LOGGER.warning("Ollama query (%s) failed, fallback: %s", query_url, exc) + last_exc = exc if ollama_url: try: - return _embed_ollama(texts, ollama_url) + return _embed_ollama_query( + text, ollama_url, read_s=_float_env("OLLAMA_QUERY_TIMEOUT_S", 20.0), + keep_alive=int(_float_env("OLLAMA_KEEP_ALIVE_COOL", 300)), options=_ollama_options(), + ) except Exception as exc: - LOGGER.warning("Ollama embedding failed: %s", exc) - openai_key = os.environ.get("OPENAI_API_KEY", "").strip() + LOGGER.warning("Ollama query embedding failed: %s", exc) + last_exc = exc if openai_key: - return _embed_openai(texts, openai_key) + return _embed_openai([text], openai_key, timeout=_float_env("OPENAI_QUERY_TIMEOUT_S", 20.0))[0] + if last_exc is not None: + raise RuntimeError( + f"Servizio embedding (Ollama) non disponibile o lento: {last_exc}. Riprova tra poco." + ) from last_exc raise RuntimeError("Nessun provider embedding configurato (OLLAMA_URL o OPENAI_API_KEY)") -def _embed_ollama(texts: list[str], base_url: str) -> list[list[float]]: +def _float_env(name: str, default: float) -> float: + try: + return float(os.environ.get(name, str(default))) + except ValueError: + return default + + +def _ollama_options(env_name: str = "OLLAMA_NUM_THREAD") -> Optional[dict]: + # Sulla stessa istanza usare sempre le stesse options: se cambiano Ollama ricarica il modello + num_thread = os.environ.get(env_name, "").strip() + if num_thread: + try: + return {"num_thread": int(num_thread)} + except ValueError: + pass + return None + + +def _embed_ollama_query( + text: str, base_url: str, *, read_s: float, keep_alive: int, options: Optional[dict] +) -> list[float]: + model = os.environ.get("OLLAMA_EMBED_MODEL", "nomic-embed-text") + timeout = httpx.Timeout(connect=5.0, read=read_s, write=10.0, pool=5.0) + payload = {"model": model, "prompt": text, "keep_alive": keep_alive} + if options: + payload["options"] = options + with httpx.Client(timeout=timeout) as client: + resp = client.post(f"{base_url.rstrip('/')}/api/embeddings", json=payload) + resp.raise_for_status() + return resp.json()["embedding"] + + +def embed_texts(texts: list[str]) -> list[list[float]]: + if not texts: + return [] + # OLLAMA_INDEX_URL: istanza dedicata su pi1 (condivisa con le query). Nessun fallback + # sul DS920 per l'indicizzazione: lì un chunk costa minuti, meglio riprovare al ciclo dopo + index_url = os.environ.get("OLLAMA_INDEX_URL", "").strip() + ollama_url = index_url or os.environ.get("OLLAMA_URL", "").strip() + if ollama_url: + try: + return _embed_ollama(texts, ollama_url, dedicated=bool(index_url)) + except Exception as exc: + LOGGER.warning("Ollama embedding failed (%s): %s", ollama_url, exc) + openai_key = os.environ.get("OPENAI_API_KEY", "").strip() + if openai_key: + return _embed_openai(texts, openai_key) + if ollama_url: + raise RuntimeError(f"Ollama non disponibile per l'indicizzazione ({ollama_url})") + raise RuntimeError("Nessun provider embedding configurato (OLLAMA_URL o OPENAI_API_KEY)") + + +def _embed_ollama(texts: list[str], base_url: str, dedicated: bool = False) -> list[list[float]]: + """dedicated=True: istanza condivisa con le query → stesse options e modello sempre caricato.""" model = os.environ.get("OLLAMA_EMBED_MODEL", "nomic-embed-text") vectors = [] timeout = httpx.Timeout(connect=30.0, read=300.0, write=30.0, pool=30.0) with httpx.Client(timeout=timeout) as client: for i, text in enumerate(texts): status = thermal.wait_for_headroom(context=f"embed:{i+1}/{len(texts)}") - keep_alive = thermal.suggested_keep_alive(status) + keep_alive = -1 if dedicated else thermal.suggested_keep_alive(status) delay = thermal.suggested_delay_s(status) payload = {"model": model, "prompt": text, "keep_alive": keep_alive} # options.num_thread limita i thread CPU lato Ollama (se supportato) - num_thread = os.environ.get("OLLAMA_NUM_THREAD", "").strip() - if num_thread: - try: - payload["options"] = {"num_thread": int(num_thread)} - except ValueError: - pass + options = _ollama_options("OLLAMA_QUERY_NUM_THREAD" if dedicated else "OLLAMA_NUM_THREAD") + if options: + payload["options"] = options resp = client.post(f"{base_url.rstrip('/')}/api/embeddings", json=payload) if resp.status_code >= 400: @@ -74,9 +159,9 @@ def _embed_ollama(texts: list[str], base_url: str) -> list[list[float]]: return vectors -def _embed_openai(texts: list[str], api_key: str) -> list[list[float]]: +def _embed_openai(texts: list[str], api_key: str, timeout: float = 120.0) -> list[list[float]]: model = os.environ.get("OPENAI_EMBED_MODEL", "text-embedding-3-small") - with httpx.Client(timeout=120.0) as client: + with httpx.Client(timeout=timeout) as client: resp = client.post( "https://api.openai.com/v1/embeddings", headers={"Authorization": f"Bearer {api_key}"}, diff --git a/services/loogle-mcp/app/knowledge/gitea.py b/services/loogle-mcp/app/knowledge/gitea.py index 5815291..052f73a 100644 --- a/services/loogle-mcp/app/knowledge/gitea.py +++ b/services/loogle-mcp/app/knowledge/gitea.py @@ -111,7 +111,7 @@ def _request( params: Optional[dict] = None, json_body: Optional[dict] = None, ) -> Any: - with httpx.Client(timeout=60.0, verify=True) as client: + with httpx.Client(timeout=httpx.Timeout(30.0, connect=5.0), verify=True) as client: resp = client.request( method, _api_url(path), diff --git a/services/loogle-mcp/app/knowledge/gitea_indexer.py b/services/loogle-mcp/app/knowledge/gitea_indexer.py index 4cacdbd..685f60d 100644 --- a/services/loogle-mcp/app/knowledge/gitea_indexer.py +++ b/services/loogle-mcp/app/knowledge/gitea_indexer.py @@ -29,6 +29,33 @@ SKIP_PATH_PARTS = ( ) PRIORITY_PREFIXES = ("docs/", "ha/", "doc/", "README") +# File che esistono per contenere segreti: mai indicizzati (le collezioni RAG possono +# essere condivise con la famiglia). .env.example resta indicizzabile. +SECRET_FILE_RE = re.compile( + r"(^|/)(\.env(\.(?!example$)[^/]*)?|id_(rsa|dsa|ecdsa|ed25519)(\.pub)?|\.htpasswd|\.netrc|\.pgpass" + r"|\.claude/settings\.local\.json" + r"|[^/]*\.(pem|key|p12|pfx|jks|keystore))$", + re.IGNORECASE, +) +_SECRET_KEY = r"[A-Za-z0-9_.-]*(?:pass(?:word|wd)?|secret|token|api[_-]?key|private[_-]?key)[A-Za-z0-9_.-]*" +# chiave = valore / chiave: valore (valori letterali, non espressioni di codice) +_SECRET_ASSIGN_RE = re.compile( + rf"(?i)\b({_SECRET_KEY})(\s*[:=]\s*)([\"']?)([^\s\"'#(){{}}\[\]$<>,;]{{4,}})\3(?=[\s\"'#;,]|$)" +) +# stile config "chiave valore" a inizio riga (es. keepalived: auth_pass xxxx) +_SECRET_CONF_RE = re.compile(rf"(?im)^(\s*{_SECRET_KEY})([ \t]+)(\S{{4,}})[ \t]*$") +# password passate a comandi: sshpass -p X, --password X / --password=X +_SECRET_CLI_RE = re.compile(r"(sshpass\s+-p\s*|--password[=\s]\s*)(\\?[\"']?)([^\s\"'\\]{3,})") +_PRIVATE_KEY_RE = re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.S) + + +def redact_secrets(text: str) -> str: + """Maschera i valori segreti prima di embedding e payload Qdrant.""" + text = _PRIVATE_KEY_RE.sub("", text) + text = _SECRET_CLI_RE.sub(lambda m: f"{m.group(1)}{m.group(2)}", text) + text = _SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}{m.group(2)}{m.group(3)}{m.group(3)}", text) + return _SECRET_CONF_RE.sub(lambda m: f"{m.group(1)}{m.group(2)}", text) + def _enabled() -> bool: flag = os.environ.get("GITEA_INDEX_ENABLED", "yes").strip().lower() @@ -71,7 +98,7 @@ def _repos_for_user(username: str) -> list[str]: def _should_index_path(path: str) -> bool: lowered = path.lower() - if any(part in lowered for part in SKIP_PATH_PARTS): + if any(part in lowered for part in SKIP_PATH_PARTS) or SECRET_FILE_RE.search(path): return False ext = os.path.splitext(path)[1].lower() return bool(ext and ext in gitea.TEXT_EXTENSIONS) @@ -158,7 +185,7 @@ def index_file( if row and row["sha"] == sha and not force: return {"repo": repo, "path": path, "skipped": True, "sha": sha} - text = (file_data.get("content") or "").strip() + text = redact_secrets((file_data.get("content") or "").strip()) if not text: return {"repo": repo, "path": path, "chunks": 0, "sha": sha} @@ -327,7 +354,7 @@ def gitea_collections_for_user(username: str, is_admin: bool = False) -> list[st def search_gitea_knowledge(username: str, query: str, limit: int = 8, is_admin: bool = False) -> list: - vectors = embeddings.embed_texts([query]) + vectors = [embeddings.embed_query(query)] collections = gitea_collections_for_user(username, is_admin=is_admin) hits = qdrant_store.search(collections, vectors[0], limit=limit) for hit in hits: diff --git a/services/loogle-mcp/app/knowledge/indexer.py b/services/loogle-mcp/app/knowledge/indexer.py index 3ec5056..88ffef1 100644 --- a/services/loogle-mcp/app/knowledge/indexer.py +++ b/services/loogle-mcp/app/knowledge/indexer.py @@ -2,10 +2,15 @@ """Paperless → Qdrant indexing pipeline.""" import logging +import zlib +from concurrent.futures import ThreadPoolExecutor from typing import Optional LOGGER = logging.getLogger("loogle_mcp.indexer") +# Indicizzazioni avviate dai tool MCP: fuori dalla richiesta, una alla volta (thermal gate) +_background = ThreadPoolExecutor(max_workers=1, thread_name_prefix="bg-index") + from ..db import get_conn from . import embeddings, gitea_indexer, paperless, qdrant_store from .text_chunk import chunk_text as _chunk_text @@ -104,7 +109,8 @@ def index_context_snippet( doc_key = f"ctx-{username}-{project_id}" if snippet_id: doc_key = f"{doc_key}-{snippet_id}" - doc_id = hash(doc_key) % (2**31) + # crc32 e non hash(): hash() cambia a ogni processo e il delete non trovava i vecchi chunk + doc_id = zlib.crc32(doc_key.encode()) % (2**31) qdrant_store.delete_by_doc(collection, doc_id) ids = [] payloads = [] @@ -127,12 +133,82 @@ def index_context_snippet( qdrant_store.upsert_chunks(collection, ids, vectors, payloads) +def index_project_context(username: str, project_id: str) -> dict: + """Indicizza il context.md completo del progetto sostituendo la versione precedente. + + Personali → ctx_{utente}; condivisi → ctx_shared_family (visibili a tutta la famiglia). + Così un contesto che cambia (es. inventario) non lascia in ricerca informazioni superate. + """ + from ..context import store as context_store + + data = context_store.get_project_context(username, project_id, session_limit=0, include_gitea=False) + meta = data["meta"] + shared = bool(meta.get("shared")) + collection = qdrant_store.CTX_SHARED_COLLECTION if shared else qdrant_store.ctx_collection(username) + namespace = context_store.SHARED_NS if shared else username + drop_project_context(collection, project_id) + chunks = _chunk_text(data["context_md"]) + if not chunks: + return {"project_id": project_id, "collection": collection, "chunks": 0} + vectors = embeddings.embed_texts(chunks) + doc_key = f"ctx-{namespace}-{project_id}" + doc_id = zlib.crc32(doc_key.encode()) % (2**31) + ids, payloads = [], [] + for i, chunk in enumerate(chunks): + ids.append(f"{doc_key}-chunk-{i}") + payloads.append( + { + "doc_id": doc_id, + "project_id": project_id, + "chunk_index": i, + "title": meta.get("title") or project_id, + "text": chunk, + "owner": meta.get("owner") or username, + "visibility": "family" if shared else "personal", + "shared": shared, + "updated_by": meta.get("updated_by"), + "source": "context", + "snippet_id": "full", + } + ) + qdrant_store.upsert_chunks(collection, ids, vectors, payloads) + return {"project_id": project_id, "collection": collection, "chunks": len(chunks)} + + +_DIGEST_SNIPPET_PREFIXES = ("daily-", "plan-") + + +def drop_project_context(collection: str, project_id: str) -> int: + """Rimuove tutti i chunk di un progetto (anche quelli indicizzati a pezzi in passato), + tranne gli snippet del digest giornaliero (snippet_id daily-*/plan-*).""" + return qdrant_store.delete_matching( + collection, + "project_id", + project_id, + lambda payload: not str(payload.get("snippet_id") or "").startswith(_DIGEST_SNIPPET_PREFIXES), + ) + + +def submit_background(label: str, fn, *args, **kwargs) -> None: + """Esegue fn in background; gli errori finiscono nel log invece che al client.""" + + def _run(): + try: + fn(*args, **kwargs) + LOGGER.info("Indicizzazione background completata: %s", label) + except Exception: + LOGGER.exception("Indicizzazione background fallita: %s", label) + + _background.submit(_run) + + def search_knowledge(username: str, query: str, limit: int = 8, is_admin: bool = False) -> list: - vectors = embeddings.embed_texts([query]) + vectors = [embeddings.embed_query(query)] collections = [ qdrant_store.SHARED_COLLECTION, qdrant_store.kb_collection(username), qdrant_store.ctx_collection(username), + qdrant_store.CTX_SHARED_COLLECTION, ] collections.extend(gitea_indexer.gitea_collections_for_user(username, is_admin=is_admin)) collections.append(qdrant_store.APPS_SHARED_COLLECTION) @@ -156,8 +232,9 @@ def search_gitea_knowledge(username: str, query: str, limit: int = 8, is_admin: def search_context(username: str, query: str, limit: int = 8) -> list: - vectors = embeddings.embed_texts([query]) - return qdrant_store.search([qdrant_store.ctx_collection(username)], vectors[0], limit=limit) + vectors = [embeddings.embed_query(query)] + collections = [qdrant_store.ctx_collection(username), qdrant_store.CTX_SHARED_COLLECTION] + return qdrant_store.search(collections, vectors[0], limit=limit) def list_recent_documents(limit: int = 20) -> list: diff --git a/services/loogle-mcp/app/knowledge/paperless.py b/services/loogle-mcp/app/knowledge/paperless.py index c2da3c6..77208ed 100644 --- a/services/loogle-mcp/app/knowledge/paperless.py +++ b/services/loogle-mcp/app/knowledge/paperless.py @@ -75,7 +75,7 @@ def list_documents( ordering: str = "-modified", username: Optional[str] = None, ) -> dict: - with httpx.Client(timeout=60.0, verify=True) as client: + with httpx.Client(timeout=httpx.Timeout(60.0, connect=5.0), verify=True) as client: resp = client.get( f"{PAPERLESS_URL}/api/documents/", headers=_headers(username), @@ -86,7 +86,7 @@ def list_documents( def get_document(doc_id: int, username: Optional[str] = None) -> dict: - with httpx.Client(timeout=60.0, verify=True) as client: + with httpx.Client(timeout=httpx.Timeout(60.0, connect=5.0), verify=True) as client: resp = client.get( f"{PAPERLESS_URL}/api/documents/{doc_id}/", headers=_headers(username), diff --git a/services/loogle-mcp/app/knowledge/qdrant_store.py b/services/loogle-mcp/app/knowledge/qdrant_store.py index 990c4f5..fd509bf 100644 --- a/services/loogle-mcp/app/knowledge/qdrant_store.py +++ b/services/loogle-mcp/app/knowledge/qdrant_store.py @@ -6,6 +6,8 @@ import logging import math import os import sqlite3 +import threading +import time import uuid from typing import Optional @@ -13,8 +15,8 @@ LOGGER = logging.getLogger("loogle_mcp.vector_store") VECTOR_SIZE = 768 _local = sqlite3.connect(":memory:", check_same_thread=False) # placeholder _qdrant_client = None -_qdrant_checked = False -_use_fallback = False +_next_qdrant_check = 0.0 # monotonic: prossimo tentativo di connessione mentre si usa il fallback +_qdrant_lock = threading.Lock() def _fallback_path() -> str: @@ -41,28 +43,45 @@ def _fallback_conn() -> sqlite3.Connection: def _get_qdrant(): - global _qdrant_client, _qdrant_checked, _use_fallback - if _qdrant_checked: - return None if _use_fallback else _qdrant_client - _qdrant_checked = True + """Client Qdrant, oppure None (fallback SQLite). + + Se Qdrant non risponde si usa il fallback, ma si riprova ogni QDRANT_RETRY_S secondi + invece di restare sul fallback fino al riavvio. + """ + global _next_qdrant_check + if _qdrant_client is not None or time.monotonic() < _next_qdrant_check: + return _qdrant_client + with _qdrant_lock: + if _qdrant_client is not None or time.monotonic() < _next_qdrant_check: + return _qdrant_client + client = _connect_qdrant() + if client is None: + _next_qdrant_check = time.monotonic() + float(os.environ.get("QDRANT_RETRY_S", "60")) + return client + + +def _connect_qdrant(): + global _qdrant_client url = os.environ.get("QDRANT_URL", "").strip() if not url: - _use_fallback = True LOGGER.warning("QDRANT_URL non impostato — fallback SQLite") return None try: + import httpx from qdrant_client import QdrantClient from qdrant_client.http import models as qm - client = QdrantClient(url=url, timeout=60) + # Sonda breve: un tentativo con Qdrant giù non deve bloccare la ricerca per 20s + httpx.get(f"{url.rstrip('/')}/readyz", timeout=3.0).raise_for_status() + client = QdrantClient(url=url, timeout=20) client.get_collections() _qdrant_client = client globals()["qm"] = qm LOGGER.info("Qdrant connesso: %s", url) return client except Exception as exc: - _use_fallback = True - LOGGER.warning("Qdrant non disponibile (%s) — fallback SQLite", exc) + LOGGER.warning("Qdrant non disponibile (%s) — fallback SQLite, nuovo tentativo tra %ss", + exc, os.environ.get("QDRANT_RETRY_S", "60")) return None @@ -92,6 +111,7 @@ def ctx_collection(username: str) -> str: SHARED_COLLECTION = "kb_shared_family" GITEA_SHARED_COLLECTION = "gitea_shared_family" APPS_SHARED_COLLECTION = "apps_shared_family" +CTX_SHARED_COLLECTION = "ctx_shared_family" def gitea_collection(username: str) -> str: @@ -138,6 +158,11 @@ def upsert_chunks( def delete_by_doc(collection: str, doc_id: int) -> None: + delete_where(collection, "doc_id", doc_id) + + +def delete_where(collection: str, key: str, value) -> None: + """Elimina i punti il cui payload[key] == value.""" client = _get_qdrant() if client is not None: from qdrant_client.http import models as qm @@ -147,7 +172,7 @@ def delete_by_doc(collection: str, doc_id: int) -> None: collection_name=collection, points_selector=qm.FilterSelector( filter=qm.Filter( - must=[qm.FieldCondition(key="doc_id", match=qm.MatchValue(value=doc_id))] + must=[qm.FieldCondition(key=key, match=qm.MatchValue(value=value))] ) ), ) @@ -156,11 +181,42 @@ def delete_by_doc(collection: str, doc_id: int) -> None: rows = conn.execute("SELECT id,payload FROM vectors WHERE collection=?", (collection,)).fetchall() for row_id, payload_raw in rows: payload = json.loads(payload_raw) - if payload.get("doc_id") == doc_id: + if payload.get(key) == value: conn.execute("DELETE FROM vectors WHERE id=?", (row_id,)) conn.commit() +def delete_matching(collection: str, key: str, value, predicate) -> int: + """Elimina i punti con payload[key] == value per cui predicate(payload) è vero.""" + client = _get_qdrant() + if client is not None: + from qdrant_client.http import models as qm + + ensure_collection(collection) + flt = qm.Filter(must=[qm.FieldCondition(key=key, match=qm.MatchValue(value=value))]) + ids, offset = [], None + while True: + points, offset = client.scroll( + collection_name=collection, scroll_filter=flt, limit=256, offset=offset, with_payload=True + ) + ids.extend(p.id for p in points if predicate(p.payload or {})) + if offset is None: + break + if ids: + client.delete(collection_name=collection, points_selector=qm.PointIdsList(points=ids)) + return len(ids) + conn = _fallback_conn() + rows = conn.execute("SELECT id,payload FROM vectors WHERE collection=?", (collection,)).fetchall() + removed = 0 + for row_id, payload_raw in rows: + payload = json.loads(payload_raw) + if payload.get(key) == value and predicate(payload): + conn.execute("DELETE FROM vectors WHERE id=?", (row_id,)) + removed += 1 + conn.commit() + return removed + + def count_by_doc(collection: str, doc_id: int) -> int: client = _get_qdrant() if client is not None: diff --git a/services/loogle-mcp/app/knowledge/thermal.py b/services/loogle-mcp/app/knowledge/thermal.py index 3f59144..8734ff3 100644 --- a/services/loogle-mcp/app/knowledge/thermal.py +++ b/services/loogle-mcp/app/knowledge/thermal.py @@ -1,7 +1,8 @@ # -*- coding: utf-8 -*- -"""Thermal / load gate per proteggere il DS920 durante gli embedding Ollama. +"""Thermal / load gate per proteggere l'host degli embedding Ollama. -Legge temperatura CPU e load average dal NAS e regola pause/keep_alive. +Legge temperatura CPU e load average (dell'host locale, pi1, oppure del DS920 via +sonda HTTP/SSH) e regola pause/keep_alive. """ from __future__ import annotations @@ -104,8 +105,29 @@ def _read_via_ssh() -> Optional[dict]: return None +def _read_local() -> Optional[dict]: + """Temperatura e carico dell'host che esegue gli embedding (pi1): il container + vede /sys/class/thermal e /proc/loadavg dell'host.""" + try: + with open("/sys/class/thermal/thermal_zone0/temp", encoding="ascii") as f: + temp = int(f.read().strip()) / 1000 + load1, load5, _ = os.getloadavg() + return {"cpu_temp_c": temp, "load1": load1, "load5": load5, "nproc": os.cpu_count() or 4} + except Exception as exc: + LOGGER.debug("Thermal probe locale fallita: %s", exc) + return None + + def read_status() -> Optional[dict]: - """Ritorna {cpu_temp_c, load1, load5?, nproc} oppure None se non raggiungibile.""" + """Ritorna {cpu_temp_c, load1, load5?, nproc} oppure None se non raggiungibile. + + THERMAL_SOURCE=local legge l'host locale (embedding su pi1); default: DS920. + """ + if os.environ.get("THERMAL_SOURCE", "ds920").strip().lower() == "local": + data = _read_local() + if data: + data["source"] = "local" + return data data = _read_via_http() if data: data["source"] = "http" diff --git a/services/loogle-mcp/app/main.py b/services/loogle-mcp/app/main.py index 40f225d..ee6d73d 100644 --- a/services/loogle-mcp/app/main.py +++ b/services/loogle-mcp/app/main.py @@ -1,16 +1,19 @@ # -*- coding: utf-8 -*- """Loogle MCP Hub — gateway FastAPI + OAuth + MCP Streamable HTTP.""" +import asyncio import json import logging import os import secrets from typing import Optional +import anyio from fastapi import Depends, FastAPI, Form, HTTPException, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from pydantic import BaseModel +from starlette.concurrency import run_in_threadpool from . import audit, auth, jwt_utils, oauth from .db import get_conn, init_db @@ -31,6 +34,10 @@ app.add_middleware( allow_headers=["*"], ) STATIC_DIR = os.path.join(os.path.dirname(__file__), "static") +# Thread per tool MCP ed endpoint sync: i tool fanno I/O bloccante (HTTP, SQLite, Ollama) +MCP_THREAD_LIMIT = int(os.environ.get("MCP_THREAD_LIMIT", "40")) +# Sotto il timeout dei client MCP: meglio un errore chiaro che una connessione appesa +MCP_TOOL_TIMEOUT_S = float(os.environ.get("MCP_TOOL_TIMEOUT_S", "45")) @app.on_event("startup") @@ -45,6 +52,11 @@ def on_startup() -> None: LOGGER.warning("MCP_JWT_SECRET generato — salvalo in .env: %s", secret) +@app.on_event("startup") +async def configure_threadpool() -> None: + anyio.to_thread.current_default_thread_limiter().total_tokens = MCP_THREAD_LIMIT + + # ------------------------------------------------------------------ Health @app.get("/health") @@ -186,14 +198,17 @@ async def oauth_token(request: Request): grant_type = body.get("grant_type") client_id = body.get("client_id") or os.environ.get("MCP_OAUTH_CLIENT_ID", "loogle-mcp-public") if grant_type == "authorization_code": - return oauth.exchange_code( + return await run_in_threadpool( + oauth.exchange_code, body.get("code", ""), client_id, body.get("redirect_uri", ""), body.get("code_verifier"), ) if grant_type == "refresh_token": - return oauth.refresh_access_token(body.get("refresh_token", ""), client_id) + return await run_in_threadpool( + oauth.refresh_access_token, body.get("refresh_token", ""), client_id + ) raise HTTPException(400, "grant_type non supportato") @@ -204,23 +219,72 @@ async def oauth_token_root(request: Request): # ------------------------------------------------------------------ MCP endpoint +_UNAUTHENTICATED_METHODS = ("initialize", "notifications/initialized", "ping") + + @app.post("/mcp") async def mcp_post(request: Request): + header_version = request.headers.get("mcp-protocol-version") + if header_version and header_version not in mcp_server.SUPPORTED_PROTOCOL_VERSIONS: + return JSONResponse( + {"error": f"MCP-Protocol-Version non supportata: {header_version}"}, status_code=400 + ) auth_header = request.headers.get("authorization", "") - claims = oauth.bearer_claims_from_header(auth_header) + claims = await run_in_threadpool(oauth.bearer_claims_from_header, auth_header) try: payload = await request.json() except Exception: raise HTTPException(400, "JSON non valido") + messages = payload if isinstance(payload, list) else [payload] + if not all(isinstance(msg, dict) for msg in messages) or not messages: + return JSONResponse( + {"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "Invalid Request"}}, + status_code=400, + ) + if not any(mcp_server.expects_response(msg) for msg in messages): + # Solo notifiche/response dal client: Streamable HTTP prevede 202 senza corpo + if not claims and any(msg.get("method") not in _UNAUTHENTICATED_METHODS for msg in messages): + return Response(status_code=401, headers=_auth_challenge_headers()) + return Response(status_code=202) if isinstance(payload, list): - responses = mcp_server.handle_batch(payload, claims) + responses = await _run_with_timeout(mcp_server.handle_batch, payload, claims) + if responses is None: + responses = [_timeout_error(msg) for msg in payload if mcp_server.expects_response(msg)] return JSONResponse(responses) - response = mcp_server.handle_message(payload, claims) - if not claims and payload.get("method") not in ("initialize", "notifications/initialized", "ping"): + response = await _run_with_timeout(mcp_server.handle_message, payload, claims) + if response is None: + response = _timeout_error(payload) + if not claims and payload.get("method") not in _UNAUTHENTICATED_METHODS: return JSONResponse(response, status_code=401, headers=_auth_challenge_headers()) return JSONResponse(response) +async def _run_with_timeout(fn, payload, claims): + """I tool sono sincroni e bloccanti: threadpool per non fermare l'event loop. + + Allo scadere ritorna None; il thread prosegue fino alla fine ma il client è già servito. + """ + try: + return await asyncio.wait_for(run_in_threadpool(fn, payload, claims), MCP_TOOL_TIMEOUT_S) + except asyncio.TimeoutError: + return None + + +def _timeout_error(msg: dict) -> dict: + params = msg.get("params") or {} + target = params.get("name") or msg.get("method") + LOGGER.warning("MCP %s oltre %.0fs: risposta di timeout al client", target, MCP_TOOL_TIMEOUT_S) + return { + "jsonrpc": "2.0", + "id": msg.get("id"), + "error": { + "code": -32000, + "message": f"Timeout: '{target}' non ha risposto entro {MCP_TOOL_TIMEOUT_S:.0f}s " + "(backend lento o non raggiungibile). Riprova tra poco.", + }, + } + + @app.get("/mcp") def mcp_get(): return JSONResponse( diff --git a/services/loogle-mcp/app/mcp/server.py b/services/loogle-mcp/app/mcp/server.py index d286463..abc2c79 100644 --- a/services/loogle-mcp/app/mcp/server.py +++ b/services/loogle-mcp/app/mcp/server.py @@ -8,7 +8,20 @@ from typing import Any, Optional from . import tools LOGGER = logging.getLogger("loogle_mcp.server") -PROTOCOL_VERSION = "2024-11-05" +# Versioni supportate, dalla più recente: si risponde con quella chiesta dal client +# se supportata, altrimenti con la più recente (negoziazione MCP lifecycle) +SUPPORTED_PROTOCOL_VERSIONS = ("2025-06-18", "2025-03-26", "2024-11-05") + + +def negotiate_protocol_version(requested: Optional[str]) -> str: + if requested in SUPPORTED_PROTOCOL_VERSIONS: + return requested + return SUPPORTED_PROTOCOL_VERSIONS[0] + + +def expects_response(msg: Any) -> bool: + """Solo le request JSON-RPC (method + id) hanno risposta; notifiche e response no.""" + return isinstance(msg, dict) and "method" in msg and "id" in msg def _error(req_id: Any, code: int, message: str) -> dict: @@ -25,18 +38,21 @@ def handle_message(body: dict, claims: Optional[dict]) -> dict: params = body.get("params") or {} if method == "initialize": + requested = params.get("protocolVersion") + version = negotiate_protocol_version(requested) + LOGGER.info( + "initialize: client=%s protocollo richiesto=%s negoziato=%s", + (params.get("clientInfo") or {}).get("name"), requested, version, + ) return _result( req_id, { - "protocolVersion": PROTOCOL_VERSION, + "protocolVersion": version, "capabilities": {"tools": {}, "resources": {}}, "serverInfo": {"name": "loogle-mcp", "version": "1.0.0"}, }, ) - if method == "notifications/initialized": - return _result(req_id, {}) - if method == "ping": return _result(req_id, {}) @@ -75,4 +91,4 @@ def handle_message(body: dict, claims: Optional[dict]) -> dict: def handle_batch(messages: list, claims: Optional[dict]) -> list: - return [handle_message(msg, claims) for msg in messages if isinstance(msg, dict)] + return [handle_message(msg, claims) for msg in messages if expects_response(msg)] diff --git a/services/loogle-mcp/app/mcp/tools.py b/services/loogle-mcp/app/mcp/tools.py index f069c17..a584940 100644 --- a/services/loogle-mcp/app/mcp/tools.py +++ b/services/loogle-mcp/app/mcp/tools.py @@ -7,7 +7,7 @@ from typing import Any, Optional from .. import audit from ..context import store as context_store from ..jwt_utils import has_scope -from ..knowledge import apps_indexer, gitea, gitea_indexer, indexer, paperless +from ..knowledge import apps_indexer, gitea, gitea_indexer, indexer, paperless, qdrant_store from ..integrations import casa, homeassistant, irrigazione, turni @@ -25,7 +25,10 @@ def tool_definitions() -> list[dict]: }, { "name": "list_projects", - "description": "Elenca i progetti dell'utente corrente", + "description": ( + "Elenca i progetti personali dell'utente e i contesti condivisi con tutta la famiglia " + "(campo shared=true, con owner e updated_by). I condivisi sono letti e aggiornati da tutti." + ), "inputSchema": { "type": "object", "properties": {"include_archived": {"type": "boolean", "default": False}}, @@ -39,6 +42,11 @@ def tool_definitions() -> list[dict]: "required": ["title"], "properties": { "title": {"type": "string"}, + "shared": { + "type": "boolean", + "default": False, + "description": "Contesto condiviso con tutta la famiglia (tutti leggono e aggiornano)", + }, "tags": {"type": "array", "items": {"type": "string"}}, "gitea_repo": { "type": "string", @@ -91,7 +99,13 @@ def tool_definitions() -> list[dict]: }, { "name": "save_context", - "description": "Salva o appende memoria persistente in un progetto", + "description": ( + "Salva o appende memoria persistente in un progetto. Nei contesti condivisi (shared=true) " + "la modifica vale per tutta la famiglia: quando cambia un dato (acquisti, inventario, " + "prezzi, regole) aggiorna la sezione esistente con mode=replace del testo completo invece " + "di appendere informazioni che la contraddicono, e passa if_revision (meta.revision letto " + "con get_project_context) per non sovrascrivere modifiche fatte nel frattempo da altri." + ), "inputSchema": { "type": "object", "required": ["project_id", "content"], @@ -99,12 +113,32 @@ def tool_definitions() -> list[dict]: "project_id": {"type": "string"}, "content": {"type": "string"}, "mode": {"type": "string", "enum": ["append", "replace"], "default": "append"}, + "if_revision": { + "type": "integer", + "description": "meta.revision letto prima di modificare: se è cambiato, errore di conflitto", + }, + }, + }, + }, + { + "name": "share_project", + "description": ( + "Condivide con tutta la famiglia un tuo progetto personale (shared=true): tutti potranno " + "leggerlo e aggiornarlo, storia inclusa. shared=false lo riporta personale del proprietario " + "(solo proprietario o admin)." + ), + "inputSchema": { + "type": "object", + "required": ["project_id"], + "properties": { + "project_id": {"type": "string"}, + "shared": {"type": "boolean", "default": True}, }, }, }, { "name": "archive_project", - "description": "Archivia o ripristina un progetto", + "description": "Archivia o ripristina un progetto (se condiviso: solo proprietario o admin)", "inputSchema": { "type": "object", "required": ["project_id"], @@ -116,7 +150,7 @@ def tool_definitions() -> list[dict]: }, { "name": "search_context", - "description": "Ricerca semantica nei contesti salvati dell'utente", + "description": "Ricerca semantica nei contesti salvati dell'utente e in quelli condivisi con la famiglia", "inputSchema": { "type": "object", "required": ["query"], @@ -163,7 +197,7 @@ def tool_definitions() -> list[dict]: }, { "name": "reindex_gitea_repo", - "description": "Re-indicizza i file testo di un repository Gitea (admin o owner repo)", + "description": "Re-indicizza i file testo di un repository Gitea (admin o owner repo). Asincrono: risponde subito, l'indicizzazione prosegue in background", "inputSchema": { "type": "object", "required": ["repo"], @@ -195,7 +229,7 @@ def tool_definitions() -> list[dict]: }, { "name": "reindex_document", - "description": "Re-indicizza un documento Paperless (admin)", + "description": "Re-indicizza un documento Paperless (admin). Asincrono: risponde subito", "inputSchema": { "type": "object", "required": ["doc_id"], @@ -312,7 +346,7 @@ def tool_definitions() -> list[dict]: "reindex": { "type": "boolean", "default": True, - "description": "Aggiorna subito il vector store RAG per questo file", + "description": "Aggiorna il vector store RAG per questo file (in background)", }, }, }, @@ -447,7 +481,7 @@ def tool_definitions() -> list[dict]: }, { "name": "reindex_apps", - "description": "Forza re-indicizzazione RAG da Irrigazione e/o Turni (admin)", + "description": "Forza re-indicizzazione RAG da Irrigazione e/o Turni (admin). Asincrono: risponde subito", "inputSchema": { "type": "object", "properties": { @@ -498,12 +532,32 @@ def call_tool(name: str, arguments: dict, claims: dict) -> dict: arguments.get("tags"), gitea_repo=gitea_repo, seed_from_gitea=bool(arguments.get("seed_from_gitea")), + shared=bool(arguments.get("shared")), ) except RuntimeError as exc: raise PermissionError(str(exc)) from exc except ValueError as exc: raise ValueError(str(exc)) from exc - audit.log_tool(username, name, meta["id"]) + audit.log_tool(username, name, meta["id"], detail={"shared": bool(meta.get("shared"))}) + return _text_result(meta) + + if name == "share_project": + if not has_scope(claims, "context:write"): + raise PermissionError("Scope context:write richiesto") + project_id = arguments["project_id"] + shared = bool(arguments.get("shared", True)) + before = context_store.get_project_context(username, project_id, session_limit=0, include_gitea=False) + owner = before["meta"].get("owner") or username + was_shared = bool(before["meta"].get("shared")) + meta = context_store.share_project(username, project_id, shared=shared, is_admin=is_admin) + if was_shared != bool(meta.get("shared")): + # Sposta i vettori: via dalla collezione di partenza, reindicizza nella nuova + source = qdrant_store.ctx_collection(owner) if shared else qdrant_store.CTX_SHARED_COLLECTION + indexer.submit_background( + f"share {project_id} shared={shared}", + _move_project_vectors, source, owner, project_id, + ) + audit.log_tool(username, name, project_id, detail={"shared": shared}) return _text_result(meta) if name == "link_project_repo": @@ -558,13 +612,13 @@ def call_tool(name: str, arguments: dict, claims: dict) -> dict: arguments["project_id"], arguments["content"], arguments.get("mode", "append"), + if_revision=arguments.get("if_revision"), + ) + indexer.submit_background( + f"context {username}/{arguments['project_id']}", + indexer.index_project_context, + username, arguments["project_id"], ) - try: - indexer.index_context_snippet( - username, arguments["project_id"], arguments["content"], meta["title"] - ) - except Exception: - pass audit.log_tool(username, name, arguments["project_id"]) return _text_result(meta) @@ -572,7 +626,7 @@ def call_tool(name: str, arguments: dict, claims: dict) -> dict: if not has_scope(claims, "context:write"): raise PermissionError("Scope context:write richiesto") meta = context_store.archive_project( - username, arguments["project_id"], arguments.get("archived", True) + username, arguments["project_id"], arguments.get("archived", True), is_admin=is_admin ) audit.log_tool(username, name, arguments["project_id"]) return _text_result(meta) @@ -623,17 +677,16 @@ def call_tool(name: str, arguments: dict, claims: dict) -> dict: owner = repo.split("/", 1)[0].lower() if not is_admin and owner != username: raise PermissionError("Puoi re-indicizzare solo repository di cui sei owner") - try: - result = gitea_indexer.index_repo( - repo, - username=username, - force=True, - max_files=arguments.get("max_files"), - ) - except Exception as exc: - raise PermissionError(f"Re-indicizzazione Gitea fallita: {exc}") from exc + indexer.submit_background( + f"gitea repo {repo}", + gitea_indexer.index_repo, + repo, + username=username, + force=True, + max_files=arguments.get("max_files"), + ) audit.log_tool(username, name, detail={"repo": repo}) - return _text_result(result) + return _text_result({"queued": True, "repo": repo}) if name == "get_document": if not has_scope(claims, "knowledge:read"): @@ -655,9 +708,10 @@ def call_tool(name: str, arguments: dict, claims: dict) -> dict: if name == "reindex_document": if not is_admin: raise PermissionError("Scope admin richiesto") - result = indexer.index_document(int(arguments["doc_id"]), force=True) - audit.log_tool(username, name, str(arguments["doc_id"])) - return _text_result(result) + doc_id = int(arguments["doc_id"]) + indexer.submit_background(f"paperless doc {doc_id}", indexer.index_document, doc_id, force=True) + audit.log_tool(username, name, str(doc_id)) + return _text_result({"queued": True, "doc_id": doc_id}) if name in ( "list_repos", "get_file", "search_code", "list_issues", "get_issue", "create_issue", @@ -792,22 +846,14 @@ def call_tool(name: str, arguments: dict, claims: dict) -> dict: username=username, ) if arguments.get("reindex", True): - try: - owner, repo_name = gitea.parse_repo(repo) - meta = gitea._request( - "GET", f"/repos/{owner}/{repo_name}", username=username - ) - private = bool(meta.get("private")) - idx = gitea_indexer.index_file( - repo, - arguments["path"], - username=username, - private=private, - force=True, - ) - result["reindex"] = idx - except Exception: - result["reindex"] = {"skipped": True} + indexer.submit_background( + f"gitea file {repo}:{arguments['path']}", + _reindex_gitea_file, + repo, + arguments["path"], + username, + ) + result["reindex"] = {"queued": True} except PermissionError: raise except Exception as exc: @@ -941,17 +987,31 @@ def call_tool(name: str, arguments: dict, claims: dict) -> dict: raise PermissionError("Scope admin richiesto") src = arguments.get("source", "all") if src == "irrigazione": - result = apps_indexer.index_irrigazione() + job = apps_indexer.index_irrigazione elif src == "turni": - result = apps_indexer.index_turni() + job = apps_indexer.index_turni else: - result = apps_indexer.index_all() + job = apps_indexer.index_all + indexer.submit_background(f"apps {src}", job) audit.log_tool(username, name, detail={"source": src}) - return _text_result(result) + return _text_result({"queued": True, "source": src}) raise ValueError(f"Tool sconosciuto: {name}") +def _move_project_vectors(source_collection: str, owner: str, project_id: str) -> None: + indexer.drop_project_context(source_collection, project_id) + indexer.index_project_context(owner, project_id) + + +def _reindex_gitea_file(repo: str, path: str, username: str) -> None: + owner, repo_name = gitea.parse_repo(repo) + meta = gitea._request("GET", f"/repos/{owner}/{repo_name}", username=username) + gitea_indexer.index_file( + repo, path, username=username, private=bool(meta.get("private")), force=True + ) + + def list_resources(claims: dict) -> list: username = claims["sub"] return context_store.list_resources(username) diff --git a/services/loogle-mcp/docker-compose.yml b/services/loogle-mcp/docker-compose.yml index 72c18eb..e419755 100644 --- a/services/loogle-mcp/docker-compose.yml +++ b/services/loogle-mcp/docker-compose.yml @@ -1,3 +1,19 @@ +# Embedding (query + indicizzazione) su Ollama locale di pi1: il DS920 (J4125, senza AVX) +# impiega minuti per chunk. Sul DS920 restano Qdrant (base di conoscenza) e un Ollama di +# riserva per le sole query (OLLAMA_URL in .env). I valori qui sovrascrivono .env. +x-embedding-env: &embedding-env + OLLAMA_QUERY_URL: http://ollama:11434 + OLLAMA_INDEX_URL: http://ollama:11434 + OLLAMA_QUERY_NUM_THREAD: "2" + # Thermal gate sull'host locale (pi1), soglie da Pi5 (throttling a 80-85°C) + THERMAL_SOURCE: local + THERMAL_TEMP_SOFT_C: "70" + THERMAL_TEMP_HARD_C: "78" + THERMAL_CPU_TARGET_PCT: "75" + THERMAL_POLL_S: "15" + OLLAMA_EMBED_DELAY_S: "0.5" + OLLAMA_EMBED_COOL_DELAY_S: "0" + services: loogle-mcp: build: @@ -11,14 +27,17 @@ services: env_file: - .env environment: - - TZ=Europe/Rome - - MCP_DB=/data/loogle_mcp.db - - MCP_CONTEXT_ROOT=/data/context - - MCP_AUDIT_LOG=/data/audit.log - - MCP_VECTOR_FALLBACK=/data/vector_fallback.db + <<: *embedding-env + TZ: Europe/Rome + MCP_DB: /data/loogle_mcp.db + MCP_CONTEXT_ROOT: /data/context + MCP_AUDIT_LOG: /data/audit.log + MCP_VECTOR_FALLBACK: /data/vector_fallback.db volumes: - ./data:/data - /mnt/ha-apps/mcp/context:/data/context + depends_on: + - ollama healthcheck: test: ["CMD", "curl", "-f", "http://127.0.0.1:8700/health"] interval: 30s @@ -35,21 +54,39 @@ services: env_file: - .env environment: - - TZ=Europe/Rome - - MCP_DB=/data/loogle_mcp.db - - MCP_CONTEXT_ROOT=/data/context - - MCP_VECTOR_FALLBACK=/data/vector_fallback.db - - DS920_SSH_HOST=192.168.128.100 - - DS920_SSH_USER=daniely - - DS920_SSH_KEY=/run/secrets/ds920_ssh_key - - DS920_THERMAL_URL=http://192.168.128.100:9191/thermal + <<: *embedding-env + TZ: Europe/Rome + MCP_DB: /data/loogle_mcp.db + MCP_CONTEXT_ROOT: /data/context + MCP_VECTOR_FALLBACK: /data/vector_fallback.db command: ["python", "-m", "worker.indexer_main"] volumes: - ./data:/data - /mnt/ha-apps/mcp/context:/data/context - - /home/daniely/.ssh/id_ed25519:/run/secrets/ds920_ssh_key:ro depends_on: - loogle-mcp + - ollama + + # Ollama per gli embedding (stesso modello del DS920 → stesso spazio vettoriale). + # Nessuna porta pubblicata: raggiungibile solo dai servizi MCP sulla rete compose. + ollama: + image: ollama/ollama:0.32.15 + container_name: loogle-mcp-ollama + restart: always + environment: + - OLLAMA_KEEP_ALIVE=-1 + - OLLAMA_NUM_PARALLEL=2 + - OLLAMA_MAX_LOADED_MODELS=1 + volumes: + # Volume Docker (non bind nella cartella progetto: il sorgente viene copiato su Gitea) + - ollama-models:/root/.ollama + cpus: 2 + mem_limit: 1g + healthcheck: + test: ["CMD", "ollama", "list"] + interval: 60s + timeout: 10s + retries: 3 # Qdrant locale: solo su x86/DS920 — su Pi5 ARM 16K page size non supportato qdrant-local: @@ -61,3 +98,6 @@ services: - "6333:6333" volumes: - /mnt/ha-apps/mcp/qdrant:/qdrant/storage + +volumes: + ollama-models: diff --git a/services/loogle-mcp/docs/ARCHITETTURA.md b/services/loogle-mcp/docs/ARCHITETTURA.md index 353c10b..47faf6e 100644 --- a/services/loogle-mcp/docs/ARCHITETTURA.md +++ b/services/loogle-mcp/docs/ARCHITETTURA.md @@ -35,6 +35,7 @@ Tutti i dati restano in LAN/NAS; l’accesso esterno avviene solo via HTTPS + OA │ ├── /oauth/* OAuth 2.1 PKCE │ │ └── /dashboard UI famiglia │ │ Pi-1 :8700 loogle-mcp-indexer (worker Paperless→vector) │ +│ Pi-1 loogle-mcp-ollama (embedding query + indexer) │ └───────┬─────────────────────────────┬───────────────────────┘ │ │ ▼ ▼ @@ -63,6 +64,8 @@ Tutti i dati restano in LAN/NAS; l’accesso esterno avviene solo via HTTPS + OA |------------|------|-------|----------------------| | MCP Gateway | Pi-1 | 8700 | `loogle-mcp` | | Indexer | Pi-1 | — | `loogle-mcp-indexer` | +| Embedding | Pi-1 | 11434 (solo rete compose) | `loogle-mcp-ollama`, volume `loogle-mcp_ollama-models` | +| Ollama di riserva | DS920 | 11434 | solo query se l'Ollama di Pi-1 non risponde; embedding per extrema | | Qdrant (primario) | DS920 | 6333 | opzionale, compose extrema | | Vector fallback | Pi-1 | — | `/data/vector_fallback.db` | | Context NFS | NAS | — | `/mnt/ha-apps/mcp/context/` | @@ -133,16 +136,24 @@ File: `/home/daniely/docker/loogle-mcp/.env` | `PAPERLESS_API_TOKEN` | Token API admin (fallback) | vedi [PAPERLESS-TOKEN.md](PAPERLESS-TOKEN.md) | | `PAPERLESS_API_TOKEN_{USER}` | Token API per utente | `DANIELE`, `LUCIA`, `DAVIDE`, `LUCA` | | `QDRANT_URL` | Vector DB remoto | `http://192.168.128.100:6333` | -| `OLLAMA_URL` | Embedding locale | `http://192.168.128.100:11434` | +| `OLLAMA_URL` | Ollama DS920: riserva per le query, embedding in extrema | `http://192.168.128.100:11434` | | `OLLAMA_EMBED_MODEL` | Modello embedding | `nomic-embed-text` | | `OPENAI_API_KEY` | Fallback embedding cloud | *(opzionale)* | | `INDEXER_INTERVAL_MINUTES` | Intervallo sync Paperless | `30` | -| `THERMAL_TEMP_HARD_C` | Pausa embedding se CPU DS920 ≥ °C | `70` | -| `THERMAL_TEMP_SOFT_C` | Rallenta embedding sopra °C | `58` | -| `THERMAL_CPU_TARGET_PCT` | Cap load1 (nproc × %) | `40` | -| `OLLAMA_NUM_THREAD` | Thread CPU per richiesta embed | `1` | -| `OLLAMA_EMBED_DELAY_S` | Pausa tra embed (duty-cycle lento) | `10` | -| `DS920_THERMAL_URL` | Probe temp/load DS920 | `http://192.168.128.100:9191/thermal` | +| `MCP_TOOL_TIMEOUT_S` | Tempo massimo di una chiamata MCP | `45` | +| `MCP_THREAD_LIMIT` | Thread per i tool (eseguiti in parallelo) | `40` | +| `QDRANT_RETRY_S` | Riprova Qdrant ogni N s mentre usa il fallback SQLite | `60` | +| `OLLAMA_QUERY_TIMEOUT_S` | Timeout embedding query sull'Ollama DS920 | `20` | + +Impostate in `docker-compose.yml` (sovrascrivono `.env`) per usare l'Ollama di Pi-1: + +| Variabile | Valore | Note | +|-----------|--------|------| +| `OLLAMA_QUERY_URL` / `OLLAMA_INDEX_URL` | `http://ollama:11434` | query (~0,1 s) e indicizzazione (~0,9 s/chunk) | +| `OLLAMA_QUERY_NUM_THREAD` | `2` | uguale a `cpus: 2` del container; stesse options per query e indexer, altrimenti Ollama ricarica il modello | +| `THERMAL_SOURCE` | `local` | thermal gate su temperatura e carico di Pi-1 (`ds920` = sonda `DS920_THERMAL_URL`) | +| `THERMAL_TEMP_SOFT_C` / `_HARD_C` | `70` / `78` | soglie Pi 5 | +| `OLLAMA_EMBED_DELAY_S` | `0.5` | pausa tra un chunk e l'altro | --- @@ -155,7 +166,12 @@ File: `/home/daniely/docker/loogle-mcp/.env` | Davide | `davide` | user | `davide` | | Luca | `luca` | user | `luca` | -- Ogni utente vede **solo** i propri progetti in `/mnt/ha-apps/mcp/context/{username}/` +- Ogni utente vede i propri progetti in `/mnt/ha-apps/mcp/context/{username}/` più i + **contesti condivisi** con la famiglia in `/mnt/ha-apps/mcp/context/_shared/` +- Contesti condivisi: tutti li leggono e li aggiornano; archiviazione e ritorno a personale + solo per il proprietario (`owner`) o l'admin. Ogni salvataggio registra `updated_by` e una + sessione `AAAAMMGG-HHMMSS-{utente}.md`; `revision` aumenta a ogni salvataggio + (`save_context if_revision` evita di sovrascrivere modifiche concorrenti) - I token OAuth JWT contengono `sub` = username; ogni tool filtra per utente - L’admin può `reindex_document`, vedere audit completo, revocare refresh token @@ -169,8 +185,9 @@ File: `/home/daniely/docker/loogle-mcp/.env` |------|--------------|----------| | `ping` | — | Test connettività | | `whoami` | — | Utente e scope attivi | -| `list_projects` | context:read | Elenco progetti | -| `create_project` | context:write | Nuovo progetto | +| `list_projects` | context:read | Elenco progetti personali + condivisi (`shared`) | +| `create_project` | context:write | Nuovo progetto (`shared=true`: condiviso con la famiglia) | +| `share_project` | context:write | Condivide un proprio progetto / lo riporta personale | | `get_project_context` | context:read | Legge memoria progetto | | `save_context` | context:write | Salva/appende memoria | | `archive_project` | context:write | Archivia progetto | @@ -204,13 +221,19 @@ Vedi anche tool Gitea in `docs/GITEA-TOKEN.md` e app homelab in `docs/APPS-INTEG ## Storage contesto (per progetto) ``` -/mnt/ha-apps/mcp/context/{username}/projects/{project-id}/ - meta.json # titolo, tag, date, stato, gitea_repo (opz.) +/mnt/ha-apps/mcp/context/{username}/projects/{project-id}/ # personali +/mnt/ha-apps/mcp/context/_shared/projects/{project-id}/ # condivisi con la famiglia + meta.json # titolo, tag, date, stato, revision, gitea_repo (opz.) + # condivisi: shared, shared_with=family, owner, updated_by context.md # memoria persistente - sessions/ # snapshot conversazioni + sessions/ # snapshot conversazioni (condivisi: suffisso -{utente}) artifacts/ # file generati dagli agenti ``` +Gli ID dei condivisi sono univoci anche rispetto ai progetti personali di tutti gli utenti. +Nel vector store ogni progetto è indicizzato con il `context.md` **completo** (sostituito a +ogni salvataggio): `ctx_{utente}` per i personali, `ctx_shared_family` per i condivisi. + --- ## Vector store / RAG @@ -227,8 +250,10 @@ Vedi anche tool Gitea in `docs/GITEA-TOKEN.md` e app homelab in `docs/APPS-INTEG **Nota Pi 5:** Qdrant locale su ARM (page size 16K) non è supportato. Default: Qdrant su DS920 o fallback SQLite in `/data/vector_fallback.db`. **Pipeline indexer:** ogni 30 min: -1. Paperless → chunk → embedding (Ollama DS920) → upsert `kb_*` -2. Gitea (repo configurati in `GITEA_INDEX_REPOS_*`) → file `.md`/`.py`/`.sh` → chunk → upsert `gitea_*` +1. Paperless → chunk → embedding (Ollama Pi-1) → upsert `kb_*` +2. Gitea (repo configurati in `GITEA_INDEX_REPOS_*`) → file `.md`/`.py`/`.sh` → segreti mascherati + (`auth_pass`, `password=`, `token:`, chiavi private) → chunk → upsert `gitea_*`. I file + `.env*` (tranne `.env.example`), chiavi e certificati non vengono mai indicizzati. 3. Irrigazione + Turni → snapshot/storico → upsert `apps_shared_family` (vedi `docs/APPS-INTEGRATION.md`) --- @@ -282,7 +307,7 @@ sudo /home/daniely/rete/scripts/setup-mcp-dns-npm.sh ## Da completare (post-install) 1. **`PAPERLESS_API_TOKEN_*`** in `.env` — vedi [PAPERLESS-TOKEN.md](PAPERLESS-TOKEN.md) -2. **Ollama su DS920** con `nomic-embed-text` per embedding locali +2. **Ollama su DS920** con `nomic-embed-text`: riserva per le query ed embedding in extrema 3. **Qdrant su DS920** (opzionale): `rete/compose/extrema-ds920/loogle-mcp-compose.yml` 4. **Cambio password** per tutti gli utenti dal dashboard 5. **`MCP_JWT_SECRET`** — impostare valore robusto in produzione diff --git a/services/loogle-mcp/docs/GUIDA-UTENTI.md b/services/loogle-mcp/docs/GUIDA-UTENTI.md index c4a4b63..76cfcb1 100644 --- a/services/loogle-mcp/docs/GUIDA-UTENTI.md +++ b/services/loogle-mcp/docs/GUIDA-UTENTI.md @@ -277,6 +277,27 @@ L'indexer in background indicizza `.md`, script e config dai repo collegati (ved Dettagli tecnici e configurazione: `docs/APPS-INTEGRATION.md`. +### 11. Contesti condivisi con la famiglia + +Alcuni progetti sono di tutti, per esempio **Stampa 3D famiglia — Bambu Lab X2D**: stampante, +filamenti, piatti, costi e regole di lavoro sono gli stessi per chiunque progetti un oggetto. +Compaiono in `list_projects` con `shared: true` e chiunque li può leggere e aggiornare. + +> «Leggi il contesto condiviso della stampa 3D X2D e aiutami a progettare un portapenne» + +> «Ho comprato 2 kg di PETG nero a 18 €/kg: aggiorna la dotazione nel contesto X2D» + +Quando cambia qualcosa (acquisti, accessori, prezzi, regole) l'assistente aggiorna la sezione +esistente invece di aggiungere note che la contraddicono: da quel momento vale per tutti. +Ogni modifica registra chi l'ha fatta. + +Per creare un nuovo contesto condiviso: + +> «create_project "Viaggio estate 2027" con shared true» + +Per condividere un tuo progetto esistente: *«share_project rinnovo-bagno»*. Solo chi l'ha +condiviso (o Daniele) può archiviarlo o riportarlo personale. + --- ## Cosa può e non può fare diff --git a/services/loogle-mcp/scripts/test_mcp_concurrency.py b/services/loogle-mcp/scripts/test_mcp_concurrency.py new file mode 100644 index 0000000..0958d5e --- /dev/null +++ b/services/loogle-mcp/scripts/test_mcp_concurrency.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Test di concorrenza MCP: N chiamate parallele + sonde /health e ping. + +Uso (dentro il container): + docker exec loogle-mcp python /srv/scripts/test_mcp_concurrency.py [N] [tool] + +Misura se una chiamata lenta blocca le altre (event loop bloccato) e la +latenza delle ricerche semantiche. +""" + +import concurrent.futures as cf +import json +import os +import sys +import time + +import httpx + +sys.path.insert(0, "/srv") +from app.db import get_conn # noqa: E402 +from app.jwt_utils import create_access_token # noqa: E402 + +BASE = os.environ.get("TEST_BASE", "http://127.0.0.1:8700") +N = int(sys.argv[1]) if len(sys.argv) > 1 else 4 +TOOL = sys.argv[2] if len(sys.argv) > 2 else "search_knowledge" +CLIENT_TIMEOUT = 150.0 + +row = get_conn().execute("SELECT id, username FROM users WHERE username='daniele'").fetchone() +token, _ = create_access_token( + {"id": row["id"], "username": row["username"]}, + "context:read knowledge:read gitea:read admin", + "loadtest", +) +HEADERS = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + + +def rpc(method: str, params: dict, req_id: int) -> tuple[float, str]: + t0 = time.monotonic() + try: + r = httpx.post( + f"{BASE}/mcp", + headers=HEADERS, + json={"jsonrpc": "2.0", "id": req_id, "method": method, "params": params}, + timeout=CLIENT_TIMEOUT, + ) + body = r.json() + if "error" in body: + status = f"ERR {body['error'].get('code')}: {body['error'].get('message', '')[:90]}" + else: + text = body["result"]["content"][0]["text"] if method == "tools/call" else "{}" + payload = json.loads(text) + n = len(payload.get("results", [])) if isinstance(payload, dict) else "?" + status = f"OK results={n}" + except Exception as exc: # noqa: BLE001 + status = f"EXC {type(exc).__name__}: {exc}"[:100] + return time.monotonic() - t0, status + + +def health() -> float: + t0 = time.monotonic() + try: + httpx.get(f"{BASE}/health", timeout=CLIENT_TIMEOUT) + except Exception: # noqa: BLE001 + return -1.0 + return time.monotonic() - t0 + + +def main() -> None: + queries = ["failover keepalived", "irrigazione zone", "turni medici", "backup nas", + "runbook", "rete vlan", "home assistant", "paperless"] + t_start = time.monotonic() + with cf.ThreadPoolExecutor(max_workers=N + 2) as pool: + futs = { + pool.submit(rpc, "tools/call", + {"name": TOOL, "arguments": {"query": queries[i % len(queries)], "limit": 3}}, + i + 1): f"{TOOL}#{i + 1}" + for i in range(N) + } + time.sleep(0.5) + h = pool.submit(health) + p = pool.submit(rpc, "tools/call", {"name": "ping", "arguments": {}}, 999) + print(f"/health durante il carico: {h.result():.2f}s") + pt, ps = p.result() + print(f"ping durante il carico: {pt:.2f}s {ps}") + for fut in cf.as_completed(futs): + dt, status = fut.result() + print(f"{futs[fut]:<24} {dt:6.2f}s {status}") + print(f"Totale: {time.monotonic() - t_start:.2f}s (N={N})") + + +if __name__ == "__main__": + main() diff --git a/services/loogle-mcp/scripts/test_mcp_protocol.py b/services/loogle-mcp/scripts/test_mcp_protocol.py new file mode 100644 index 0000000..95faa02 --- /dev/null +++ b/services/loogle-mcp/scripts/test_mcp_protocol.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +"""Conformità Streamable HTTP / lifecycle MCP: negoziazione versione, 202 per notifiche, header. + +Uso (dentro il container): + docker exec loogle-mcp python /srv/scripts/test_mcp_protocol.py +""" + +import os +import sys + +import httpx + +sys.path.insert(0, "/srv") +from app.db import get_conn # noqa: E402 +from app.jwt_utils import create_access_token # noqa: E402 + +BASE = os.environ.get("TEST_BASE", "http://127.0.0.1:8700") + "/mcp" +row = get_conn().execute("SELECT id, username FROM users WHERE username='daniele'").fetchone() +token, _ = create_access_token(dict(row), "context:read", "protocol-test") +AUTH = {"Authorization": f"Bearer {token}"} +failures = 0 + + +def check(label: str, cond: bool, detail: str = "") -> None: + global failures + failures += 0 if cond else 1 + print(f"{'OK ' if cond else 'FAIL'} {label} {detail}") + + +def init(version): + params = {"protocolVersion": version, "capabilities": {}, "clientInfo": {"name": "protocol-test"}} + r = httpx.post(BASE, json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": params}) + return r.json()["result"]["protocolVersion"] + + +for requested, expected in [("2025-06-18", "2025-06-18"), ("2025-03-26", "2025-03-26"), + ("2024-11-05", "2024-11-05"), ("2099-01-01", "2025-06-18")]: + got = init(requested) + check(f"initialize {requested}", got == expected, f"-> {got}") + +r = httpx.post(BASE, json={"jsonrpc": "2.0", "method": "notifications/initialized"}) +check("notifica non autenticata initialized -> 202", r.status_code == 202 and not r.content, str(r.status_code)) +r = httpx.post(BASE, headers=AUTH, json={"jsonrpc": "2.0", "method": "notifications/cancelled", + "params": {"requestId": 5}}) +check("notifica cancelled -> 202", r.status_code == 202, str(r.status_code)) +r = httpx.post(BASE, json={"jsonrpc": "2.0", "method": "notifications/cancelled"}) +check("notifica non autenticata non ammessa -> 401", r.status_code == 401, str(r.status_code)) +r = httpx.post(BASE, headers=AUTH, json=[{"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "id": 7, "method": "ping"}]) +body = r.json() +check("batch notifica+request -> solo 1 risposta", r.status_code == 200 and len(body) == 1 and body[0]["id"] == 7, + str(body)) +r = httpx.post(BASE, headers={**AUTH, "MCP-Protocol-Version": "2025-06-18"}, + json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) +check("header versione valida -> 200", r.status_code == 200 and "tools" in r.json()["result"], str(r.status_code)) +r = httpx.post(BASE, headers={**AUTH, "MCP-Protocol-Version": "1999-01-01"}, + json={"jsonrpc": "2.0", "id": 3, "method": "tools/list"}) +check("header versione non supportata -> 400", r.status_code == 400, str(r.status_code)) +r = httpx.post(BASE, json={"jsonrpc": "2.0", "id": 4, "method": "tools/list"}) +check("request senza token -> 401 + WWW-Authenticate", + r.status_code == 401 and "resource_metadata" in r.headers.get("www-authenticate", ""), str(r.status_code)) +r = httpx.post(BASE, headers=AUTH, json=[]) +check("batch vuoto -> 400", r.status_code == 400, str(r.status_code)) +print(f"Errori: {failures}") +sys.exit(1 if failures else 0) diff --git a/services/loogle-mcp/scripts/test_mcp_readonly_tools.py b/services/loogle-mcp/scripts/test_mcp_readonly_tools.py new file mode 100644 index 0000000..0c88d29 --- /dev/null +++ b/services/loogle-mcp/scripts/test_mcp_readonly_tools.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +"""Regressione: chiama ogni tool MCP senza argomenti obbligatori (sola lettura) e ne misura la durata. + +Uso (dentro il container): + docker exec loogle-mcp python /srv/scripts/test_mcp_readonly_tools.py +""" + +import json +import os +import sys +import time + +import httpx + +sys.path.insert(0, "/srv") +from app.db import get_conn # noqa: E402 +from app.jwt_utils import create_access_token # noqa: E402 + +BASE = os.environ.get("TEST_BASE", "http://127.0.0.1:8700") +# Tool senza argomenti obbligatori ma con effetti collaterali: esclusi +SKIP = {"reindex_apps"} + +row = get_conn().execute("SELECT id, username FROM users WHERE username='daniele'").fetchone() +scope = ("context:read knowledge:read gitea:read home:read irrigation:read turni:read admin") +token, _ = create_access_token({"id": row["id"], "username": row["username"]}, scope, "regression") +HEADERS = {"Authorization": f"Bearer {token}"} + + +def rpc(method: str, params: dict) -> dict: + r = httpx.post(f"{BASE}/mcp", headers=HEADERS, timeout=120, + json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params}) + return r.json() + + +tools = rpc("tools/list", {})["result"]["tools"] +failures = 0 +for tool in tools: + if tool["inputSchema"].get("required") or tool["name"] in SKIP: + continue + t0 = time.monotonic() + body = rpc("tools/call", {"name": tool["name"], "arguments": {}}) + dt = time.monotonic() - t0 + if "error" in body: + status = f"ERR {body['error']['message'][:80]}" + failures += 1 + else: + status = f"OK {len(body['result']['content'][0]['text'])} bytes" + print(f"{tool['name']:<30} {dt:6.2f}s {status}") +print(f"Tool totali: {len(tools)}, errori: {failures}") diff --git a/services/loogle-mcp/scripts/test_shared_contexts.py b/services/loogle-mcp/scripts/test_shared_contexts.py new file mode 100644 index 0000000..ac41a6f --- /dev/null +++ b/services/loogle-mcp/scripts/test_shared_contexts.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""Test dei contesti condivisi su directory e DB temporanei (non tocca i dati reali). + +Uso (dentro il container): + docker exec loogle-mcp python /srv/scripts/test_shared_contexts.py +""" + +import json +import os +import sys +import tempfile + +TMP = tempfile.mkdtemp(prefix="shared-ctx-test-") +os.environ["MCP_CONTEXT_ROOT"] = os.path.join(TMP, "context") +os.environ["MCP_DB"] = os.path.join(TMP, "test.db") +sys.path.insert(0, "/srv") + +from app.db import init_db # noqa: E402 +from app.context import store # noqa: E402 +from app.knowledge import indexer # noqa: E402 +from app.mcp import tools # noqa: E402 + +init_db() +queued = [] +indexer.submit_background = lambda label, fn, *a, **k: queued.append((label, fn.__name__, a)) +failures = 0 + + +def claims(user, admin=False): + scope = "context:read context:write" + (" admin" if admin else "") + return {"sub": user, "scope": scope} + + +def call(user, name, args, admin=False): + out = tools.call_tool(name, args, claims(user, admin)) + return json.loads(out["content"][0]["text"]) + + +def check(label, cond): + global failures + failures += 0 if cond else 1 + print(f"{'OK ' if cond else 'FAIL'} {label}") + + +def raises(exc, fn, *a, **k): + try: + fn(*a, **k) + except exc: + return True + except Exception as other: # noqa: BLE001 + print(" eccezione inattesa:", type(other).__name__, other) + return False + + +# 1. Progetto personale di daniele, poi condiviso +p = call("daniele", "create_project", {"title": "Stampa 3D X2D", "tags": ["stampa-3d"]}) +pid = p["id"] +call("daniele", "save_context", {"project_id": pid, "content": "## Dotazione\n- PLA 4 kg"}) +check("lucia non vede il progetto personale di daniele", + pid not in [x["id"] for x in call("lucia", "list_projects", {})["projects"]]) +check("lucia non può leggerlo", raises(FileNotFoundError, call, "lucia", "get_project_context", {"project_id": pid})) +m = call("daniele", "share_project", {"project_id": pid}) +check("share_project -> shared/family/owner", m.get("shared") and m.get("shared_with") == "family" and m.get("owner") == "daniele") +check("la cartella è passata nello spazio condiviso", + os.path.isfile(os.path.join(TMP, "context", "_shared", "projects", pid, "context.md")) + and not os.path.exists(os.path.join(TMP, "context", "daniele", "projects", pid))) +check("reindicizzazione vettori accodata", any(q[1] == "_move_project_vectors" for q in queued)) + +# 2. Tutta la famiglia legge e aggiorna +for u in ("lucia", "davide", "luca"): + ids = {x["id"]: x for x in call(u, "list_projects", {})["projects"]} + check(f"{u} vede il contesto condiviso", pid in ids and ids[pid].get("shared")) +ctx = call("lucia", "get_project_context", {"project_id": pid}) +check("lucia legge la storia (sessione di daniele)", "PLA 4 kg" in ctx["context_md"] and ctx["recent_sessions"]) +rev = ctx["meta"].get("revision", 0) +m = call("lucia", "save_context", {"project_id": pid, "mode": "replace", + "content": "## Dotazione\n- PLA 4 kg\n- PETG 1 kg (comprato da Lucia)", + "if_revision": rev}) +check("lucia aggiorna, updated_by=lucia", m.get("updated_by") == "lucia") +check("daniele vede la modifica di lucia", "PETG 1 kg" in call("daniele", "get_project_context", {"project_id": pid})["context_md"]) +sessions = os.listdir(os.path.join(TMP, "context", "_shared", "projects", pid, "sessions")) +check("il nome della sessione indica l'autore", any(s.endswith("-lucia.md") for s in sessions)) +check("conflitto: if_revision superata -> errore e nessuna sovrascrittura", + raises(store.ConflictError, store.save_context, "davide", pid, "x", "replace", if_revision=rev) + and "PETG 1 kg" in call("davide", "get_project_context", {"project_id": pid})["context_md"]) +check("indicizzazione del contesto completo accodata", any(q[1] == "index_project_context" for q in queued)) + +# 3. Permessi: archiviare / rendere personale +check("lucia non può archiviare", raises(PermissionError, call, "lucia", "archive_project", {"project_id": pid})) +check("lucia non può riportarlo personale", raises(PermissionError, call, "lucia", "share_project", {"project_id": pid, "shared": False})) +check("admin può archiviare", call("davide", "archive_project", {"project_id": pid}, admin=True).get("archived") is True) +call("daniele", "archive_project", {"project_id": pid, "archived": False}) +m = call("daniele", "share_project", {"project_id": pid, "shared": False}) +check("il proprietario lo riporta personale", not m.get("shared") and "owner" not in m) +check("dopo: lucia non lo vede più", pid not in [x["id"] for x in call("lucia", "list_projects", {})["projects"]]) +check("dopo: daniele lo ha di nuovo personale con la storia", + "PETG 1 kg" in call("daniele", "get_project_context", {"project_id": pid})["context_md"]) + +# 4. Creazione diretta condivisa, ID unici tra i due spazi, risorse MCP +s = call("lucia", "create_project", {"title": "Stampa 3D X2D", "shared": True}) +check("create_project shared=true con ID diverso dal personale di daniele", s.get("shared") and s["id"] != pid) +check("daniele raggiunge sia il suo personale sia il nuovo condiviso", + call("daniele", "get_project_context", {"project_id": s["id"]})["meta"].get("shared") + and not call("daniele", "get_project_context", {"project_id": pid})["meta"].get("shared")) +call("luca", "create_project", {"title": "Omonimo"}) +call("daniele", "create_project", {"title": "Omonimo"}) +check("share_project rifiuta un ID usato da un personale di un altro utente", + raises(ValueError, call, "daniele", "share_project", {"project_id": "omonimo"})) +uris = [r["uri"] for r in tools.list_resources(claims("luca"))] +check("risorsa condivisa visibile a luca", f"loogle://context/_shared/{s['id']}" in uris) +res = tools.read_resource(f"loogle://context/_shared/{s['id']}", claims("luca")) +check("luca legge la risorsa condivisa", res["text"].startswith("# Stampa 3D X2D")) + +print(f"Errori: {failures}") +sys.exit(1 if failures else 0)