""" 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 from .state import get_env_secret _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