""" backend/api/hf_storage.py — Zero-Cost Persistence via Hugging Face Datasets. Implementa il mirroring asincrono della memoria dell'agente su HF Datasets. Design: - Fire-and-forget: non blocca mai il flusso principale. - Sharding: supporta caricamento su diversi dataset (Account A/B/C/D). - Formato: JSONL (append-only) per massima compatibilità. FIX-HF-API (MX18-VERIFY): usa /upload/{branch} con multipart/form-data (non raw body POST). L'endpoint /upload/ esiste (401 senza auth) ma accetta multipart, non raw bytes. """ import os import json import time import httpx import asyncio import logging from typing import Any, Optional _logger = logging.getLogger("api.hf_storage") # Configurazione default (Account A / BRAIN) HF_TOKEN = os.getenv("HF_TOKEN", "") HF_DATASET_REPO = os.getenv("HF_DATASET_REPO", "Arjanit98/agent-memory") async def hf_append_record( dataset_path: str, record: dict, repo_id: Optional[str] = None, token: Optional[str] = None, ) -> bool: """ Appende un record a un file JSONL in un dataset Hugging Face via API. dataset_path: percorso del file nel repo (es. 'decisions.jsonl') Usa /api/datasets/{repo}/upload/{branch} con multipart/form-data. File giornaliero (decisions_2026-06-26.jsonl) per evitare conflitti. """ _token = token or HF_TOKEN _repo = repo_id or HF_DATASET_REPO if not _token or not _repo: return False # File giornaliero per shard naturale + evitare read-modify-write date_str = time.strftime("%Y-%m-%d") base = dataset_path.replace(".jsonl", "") filename = f"{base}_{date_str}.jsonl" try: record["_timestamp_ms"] = int(time.time() * 1000) line = json.dumps(record, ensure_ascii=False) + "\n" # HF Hub upload API: POST /api/datasets/{repo}/upload/{branch} # Multipart form-data: ogni file come campo "file" con path nel repo. # Ref: https://huggingface.co/docs/hub/api#post-apireposuploadref url = f"https://huggingface.co/api/datasets/{_repo}/upload/main" async with httpx.AsyncClient(timeout=12.0) as client: resp = await client.post( url, headers={"Authorization": f"Bearer {_token}"}, files={ # Chiave = path nel repo, valore = (nome, contenuto, mimetype) filename: (filename, line.encode("utf-8"), "text/plain"), }, ) if resp.status_code in (200, 201): return True _logger.warning( "HF Upload %s → %d: %s", filename, resp.status_code, resp.text[:200], ) except Exception as exc: _logger.error("HF Storage error for %s: %s", dataset_path, exc) return False def hf_fire_and_forget( dataset_path: str, record: dict, repo_id: Optional[str] = None, ) -> None: """Lancia il caricamento in background senza attendere.""" try: loop = asyncio.get_event_loop() if loop.is_running(): loop.create_task(hf_append_record(dataset_path, record, repo_id)) except Exception: pass # Fire-and-forget: mai propagare eccezioni al caller