api/blackboard.py

"""
backend/api/blackboard.py — S-BB: Blackboard condiviso cross-sessione via Upstash Redis REST.

Riusa UPSTASH_REDIS_REST_URL e UPSTASH_REDIS_REST_TOKEN già configurati
in backend/api/llm_cache.py — zero setup aggiuntivo.

TTL: 600s (10 minuti) — session-scoped.
Endpoint:
  POST /api/blackboard/{session_id}/write  — scrive una entry
  GET  /api/blackboard/{session_id}/read   — legge tutte le entry
  DEL  /api/blackboard/{session_id}        — pulisce il blackboard

Pattern: i sub-agenti scrivono via frontend in-memory (agentBlackboard.ts);
il backend persiste su Upstash per cross-tab/cross-reload continuity.
"""
import os
import json
import httpx
import asyncio
from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter(prefix="/api/blackboard", tags=["blackboard"])

_URL   = os.getenv("UPSTASH_REDIS_REST_URL", "")
_TOKEN = os.getenv("UPSTASH_REDIS_REST_TOKEN", "")
_TTL   = 600  # 10 minuti


class BBEntry(BaseModel):
    agentId:  str
    key:      str
    value:    str
    severity: str = "info"
    ts:       float = 0.0


async def _redis_post(command: list) -> dict | None:
    """Esegue un comando Redis via Upstash REST API."""
    if not _URL or not _TOKEN:
        return None
    try:
        async with httpx.AsyncClient(timeout=2.0) as c:
            r = await c.post(
                _URL,
                json=command,
                headers={
                    "Authorization": f"Bearer {_TOKEN}",
                    "Content-Type": "application/json",
                },
            )
            return r.json() if r.is_success else None
    except Exception:
        return None


async def _redis_scan(pattern: str) -> list[str]:
    """Scansiona le chiavi Redis con SCAN paginato via POST (Upstash REST API).

    P40-H: la versione precedente usava cursor fisso "0" → solo il primo batch
    di 100 chiavi veniva letto. Con >100 entry Redis per sessione le chiavi
    extra venivano omesse silenziosamente. Fix: loop cursor finché cursor == "0".

    FIX G3: la versione GET usava URL-encoding del pattern (: → %3A, * → %2A).
    Upstash router path-decodifica %3A → ':' ma il pattern risultante veniva
    spezzato alla prima '/' → SCAN trovava 0 chiavi su pattern con ':'.
    La versione POST invia il pattern come stringa JSON → nessun encoding → corretto.
    """
    all_keys: list[str] = []
    cursor = "0"
    while True:
        result = await _redis_post(["SCAN", cursor, "MATCH", pattern, "COUNT", "100"])
        if not result or "result" not in result:
            break
        scan_result = result["result"]
        # Upstash SCAN ritorna [next_cursor, [keys]] o [[next_cursor, [keys]]] (pipeline)
        if not isinstance(scan_result, list) or len(scan_result) < 2:
            break
        cursor = str(scan_result[0])
        keys = scan_result[1]
        if isinstance(keys, list):
            all_keys.extend(keys)
        # cursor "0" segnala fine iterazione
        if cursor == "0":
            break
    return all_keys


@router.post("/{session_id}/write")
async def bb_write(session_id: str, entry: BBEntry):
    """Scrive una entry nel blackboard Upstash. Fail-safe: ritorna ok=True anche se Upstash non disponibile."""
    rkey    = f"bb:{session_id}:{entry.agentId}:{entry.key}"
    payload = json.dumps({
        "agentId":  entry.agentId,
        "key":      entry.key,
        "value":    entry.value,
        "severity": entry.severity,
        "ts":       entry.ts,
    })
    # SET key value EX ttl
    await _redis_post(["SET", rkey, payload, "EX", _TTL])
    return {"ok": True}


@router.get("/{session_id}/read")
async def bb_read(session_id: str):
    """Legge tutte le entries del blackboard per la sessione."""
    keys = await _redis_scan(f"bb:{session_id}:*")
    if not keys:
        return {"entries": [], "session_id": session_id}

    mget = await _redis_post(["MGET"] + keys)
    entries = []
    if mget and "result" in mget:
        for v in mget["result"]:
            if v:
                try:
                    entries.append(json.loads(v))
                except Exception:
                    pass
    return {"entries": entries, "session_id": session_id}


@router.delete("/{session_id}")
async def bb_clear(session_id: str):
    """Pulisce il blackboard al termine della sessione."""
    keys = await _redis_scan(f"bb:{session_id}:*")
    if keys:
        await _redis_post(["DEL"] + keys)
    return {"ok": True, "deleted": len(keys)}