api/hf_storage.py

"""
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