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