api/kernel.py

"""
backend/api/kernel.py — AI Kernel (ARCH-K2.1)

Interfaccia unica tra Brain/Executor e tutti i servizi sottostanti.
Il Brain NON importa mai direttamente job_queue, event_bus, session_manager, providers.
Usa SOLO le 4 primitive Kernel:

  submitTask(payload, priority, session_id)    → TaskResult
  chat(messages, model_hint, session_id)       → ChatResult
  memory(op, session_id, **kwargs)             → MemoryResult
  publishEvent(topic, payload, ...)            → EventResult

Invarianti ADR rispettati:
  S1:  stateless — nessuno stato locale (stato in Queue + DB)
  S4:  Brain non conosce l'infrastruttura
  S5:  ogni Tool sostituibile senza toccare agentLoop
  S9:  nessun servizio conosce l'impl. interna di un altro
  S10: ogni comunicazione è asincrona
  S20: ogni Provider è sostituibile
  S21: Brain dipende solo dal Kernel
  S27: ogni decisione tracciabile via correlation_id
  S30: nessuna API pubblica dipende da provider specifico

HTTP Endpoints (auth: MACHINE):
  POST /api/kernel/submit          — submitTask()
  POST /api/kernel/chat            — chat()
  POST /api/kernel/memory          — memory()
  POST /api/kernel/event/publish   — publishEvent()
  GET  /api/kernel/status          — diagnostica servizi

Python-importable (uso interno brain/executor):
  from api.kernel import kernel
  result = await kernel.submit_task(payload={...})
"""
from __future__ import annotations

import asyncio
import logging
import time
import uuid
from typing import Any, Literal

from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field

from .auth_guard import AuthRole, require_role

_logger = logging.getLogger("api.kernel")

# ── Router ─────────────────────────────────────────────────────────────────────
router = APIRouter(
    prefix="/api/kernel",
    tags=["kernel"],
    dependencies=[Depends(require_role(AuthRole.MACHINE))],
)

# ── Priority ───────────────────────────────────────────────────────────────────
TaskPriority = Literal["HIGH", "NORMAL", "LOW", "BACKGROUND"]

# ── Result models ──────────────────────────────────────────────────────────────

class TaskResult(BaseModel):
    task_id:        str
    correlation_id: str
    status:         str           # "queued" | "error"
    queue_backend:  str           # "redis" | "memory" | "none"
    ts:             float = Field(default_factory=time.time)
    error:          str | None = None


class ChatResult(BaseModel):
    correlation_id: str
    content:        str
    provider:       str
    model:          str
    cached:         bool = False
    ts:             float = Field(default_factory=time.time)
    error:          str | None = None


class MemoryResult(BaseModel):
    correlation_id: str
    op:             str
    data:           Any = None
    backend:        str
    ts:             float = Field(default_factory=time.time)
    error:          str | None = None


class EventResult(BaseModel):
    event_id:       str
    correlation_id: str
    topic:          str
    delivered_to:   int  = 0
    redis_fanout:   bool = False
    ts:             float = Field(default_factory=time.time)
    error:          str | None = None


# ── Request bodies ─────────────────────────────────────────────────────────────

class SubmitTaskRequest(BaseModel):
    payload:        dict         = Field(..., description="Job payload per il Worker")
    priority:       TaskPriority = "NORMAL"
    session_id:     str | None   = None
    correlation_id: str | None   = None
    timeout_s:      int          = 300


class ChatRequest(BaseModel):
    messages:       list[dict]   = Field(..., description="Array OpenAI-style [{role, content}]")
    model_hint:     str | None   = None   # "vision" | "chat" | "embedding" | provider name
    session_id:     str | None   = None
    correlation_id: str | None   = None
    max_tokens:     int          = 4096
    temperature:    float        = 0.7


class MemoryRequest(BaseModel):
    op:             Literal["read", "write", "search", "compress", "clear"]
    session_id:     str | None   = None
    correlation_id: str | None   = None
    query:          str | None   = None
    limit:          int          = 10
    layer:          Literal["working", "episodic", "semantic", "reflection", "all"] = "all"
    content:        str | None   = None
    role:           str          = "assistant"
    metadata:       dict         = Field(default_factory=dict)


class PublishEventRequest(BaseModel):
    topic:          str          = Field(..., description="Es. task.created, tool.finished")
    payload:        dict         = Field(default_factory=dict)
    correlation_id: str | None   = None
    session_id:     str | None   = None
    source:         str          = "kernel"


# ── KernelAPI — Python-importable ──────────────────────────────────────────────

class KernelAPI:
    """
    Interfaccia Python del Kernel. Importabile dai moduli Brain/Executor
    senza dipendenze dirette ai servizi sottostanti.

    Usage:
        from api.kernel import kernel
        result = await kernel.submit_task(payload={"type": "llm_call", ...})
    """

    # ── submitTask ─────────────────────────────────────────────────────────────

    async def submit_task(
        self,
        payload:        dict,
        priority:       TaskPriority = "NORMAL",
        session_id:     str | None   = None,
        correlation_id: str | None   = None,
        timeout_s:      int          = 300,
    ) -> TaskResult:
        """
        Accoda un job alla Queue con priorità.
        Routing: Redis (JQ_ENABLED=1) → fallback none (stateless, S1).
        (S10, S16: asincrono, task_id globale univoco)
        """
        corr  = correlation_id or str(uuid.uuid4())
        t_id  = str(uuid.uuid4())

        job = {
            "task_id":        t_id,
            "correlation_id": corr,
            "session_id":     session_id,
            "priority":       priority,
            "timeout_s":      timeout_s,
            "payload":        payload,
            "submitted_at":   time.time(),
        }

        queue_backend = "none"
        try:
            from .job_queue import _rpush, _K_PENDING, _redis_ok
            if _redis_ok():
                import json as _json
                ok = await _rpush(_K_PENDING, _json.dumps(job), ttl=timeout_s + 60)
                queue_backend = "redis" if ok else "none"
        except Exception as exc:
            _logger.warning("[kernel.submit_task] queue err: %s", exc)

        # Fire-and-forget: pubblica task.created sull'Event Bus (S26, S27)
        asyncio.create_task(self._emit("task.created", {
            "task_id":    t_id,
            "priority":   priority,
            "session_id": session_id,
        }, corr))

        _logger.info("[kernel] submitTask id=%s priority=%s backend=%s", t_id, priority, queue_backend)
        return TaskResult(
            task_id=t_id,
            correlation_id=corr,
            status="queued",
            queue_backend=queue_backend,
        )

    # ── chat ───────────────────────────────────────────────────────────────────

    async def chat(
        self,
        messages:       list[dict],
        model_hint:     str | None = None,
        session_id:     str | None = None,
        correlation_id: str | None = None,
        max_tokens:     int        = 4096,
        temperature:    float      = 0.7,
    ) -> ChatResult:
        """
        Invia chat al provider selezionato. model_hint richiede una capability
        senza nominare il provider specifico. (S20, S30: provider-agnostic)
        LLM cache applicata automaticamente (S9).
        """
        corr      = correlation_id or str(uuid.uuid4())
        cache_key = f"k:{model_hint}:{hash(str(messages))}"

        # Cache read  (GAP-3-fix: get_cached returns str|None → JSON-parse)
        try:
            import json as _json
            from .llm_cache import get_cached, set_cached
            _cached_raw = await get_cached(cache_key)
            cached = _json.loads(_cached_raw) if _cached_raw else None
            if cached:
                _logger.debug("[kernel.chat] cache hit corr=%s", corr)
                return ChatResult(
                    correlation_id=corr,
                    content=cached.get("content", ""),
                    provider=cached.get("provider", "cache"),
                    model=cached.get("model", ""),
                    cached=True,
                )
        except Exception:
            cached = None

        provider_name = "unknown"
        model_name    = "unknown"
        content       = ""
        error         = None

        try:
            # ARCH-I4.4: Provider Layer LLM — usa CapabilityRouter per selezione dinamica
            from models.provider_router import capability_router as _cap_router
            
            _ai_obj = await _cap_router.get_client_for_capability(model_hint or "default")

            # Determina provider/model per logging e ChatResult
            if hasattr(_ai_obj, "providers") and _ai_obj.providers:
                _p = _ai_obj.providers[0]
                provider_name = _p.name
                model_name    = _p.default_model
            else:
                provider_name = model_hint or "ai_client"
                model_name    = "unknown"

            # AIClient.chat() restituisce str direttamente (non un completions object)
            content = await asyncio.wait_for(
                _ai_obj.chat(messages, max_tokens=max_tokens, temperature=temperature),
                timeout=60.0,
            )

            # Cache write (fail-open: non blocca il caller)  (GAP-3-fix)
            if cached is None:
                try:
                    import json as _json
                    from .llm_cache import set_cached
                    await set_cached(cache_key, _json.dumps({
                        "content": content,
                        "provider": provider_name,
                        "model": model_name,
                    }))
                except Exception:
                    pass

        except Exception as exc:
            error = str(exc)
            _logger.warning("[kernel.chat] provider err corr=%s: %s", corr, exc)

        asyncio.create_task(self._emit("response.generated", {
            "provider": provider_name,
            "model":    model_name,
            "session_id": session_id,
            "cached":   False,
        }, corr))

        return ChatResult(
            correlation_id=corr,
            content=content,
            provider=provider_name,
            model=model_name,
            cached=False,
            error=error,
        )

    # ── memory ─────────────────────────────────────────────────────────────────

    async def memory(
        self,
        op:             str,
        session_id:     str | None  = None,
        correlation_id: str | None  = None,
        query:          str | None  = None,
        content:        str | None  = None,
        role:           str         = "assistant",
        layer:          str         = "all",
        limit:          int         = 10,
        metadata:       dict | None = None,
    ) -> MemoryResult:
        """
        Operazioni unificate su tutti i layer di memoria.
        Router: working → episodic → semantic → reflection.
        (S3: solo Queue e DB contengono stato; S9: interfaccia opaca)
        """
        corr = correlation_id or str(uuid.uuid4())
        meta = metadata or {}

        try:
            from .state import _get_mem_manager_async as _gmm
            mem = await _gmm()
            if mem is None:
                return MemoryResult(
                    correlation_id=corr, op=op, backend="none",
                    error="MemoryManager not initialized",
                )

            data    = None
            backend = layer

            if op == "read":
                ctx = await asyncio.to_thread(mem.working.get_context) \
                    if layer in ("working", "all") else ""
                data    = {"context": ctx}
                backend = "working"

            elif op == "write" and content:
                await asyncio.to_thread(mem.working.add_entry, role, content, meta)
                await asyncio.to_thread(mem.episodic.add, content, meta)
                backend = "working+episodic"
                asyncio.create_task(self._emit("memory.updated", {
                    "op": "write", "session_id": session_id, "layer": backend,
                }, corr))

            elif op == "search" and query:
                results = await asyncio.to_thread(mem.semantic.search, query, limit)
                data    = {"results": results}
                backend = "semantic"

            elif op == "compress":
                summary = await asyncio.to_thread(mem.working.compress)
                data    = {"summary": summary}
                backend = "working"

            elif op == "clear":
                await asyncio.to_thread(mem.working.clear)
                data    = {"cleared": True}
                backend = "working"

            else:
                return MemoryResult(
                    correlation_id=corr, op=op, backend="none",
                    error=f"op '{op}' non valida o parametri mancanti",
                )

            return MemoryResult(correlation_id=corr, op=op, data=data, backend=backend)

        except Exception as exc:
            _logger.warning("[kernel.memory] op=%s err: %s", op, exc)
            return MemoryResult(correlation_id=corr, op=op, backend="error", error=str(exc))

    # ── publishEvent ───────────────────────────────────────────────────────────

    async def publish_event(
        self,
        topic:          str,
        payload:        dict,
        correlation_id: str | None = None,
        session_id:     str | None = None,
        source:         str        = "kernel",
    ) -> EventResult:
        """
        Pubblica un evento sull'Event Bus (in-memory + Redis fanout).
        (S10, S13, S19, S27: asincrono, idempotente, persistente, tracciabile)
        """
        corr     = correlation_id or str(uuid.uuid4())
        event_id = str(uuid.uuid4())

        try:
            from .event_bus import publish as _publish_internal  # GAP-4-fix
            result = await _publish_internal(
                topic,
                payload,
                correlation_id=corr,
                session_id=session_id,
                source=source,
            )
            return EventResult(
                event_id=event_id,
                correlation_id=corr,
                topic=topic,
                delivered_to=result.get("delivered_to", 0) if isinstance(result, dict) else 0,
                redis_fanout=result.get("redis_fanout", False) if isinstance(result, dict) else False,
            )
        except ImportError:
            # Fallback: usa il router HTTP interno via publish endpoint
            try:
                from .event_bus import publish as _bus_pub, BusEvent  # type: ignore[attr-defined]
                evt = BusEvent(
                    topic=topic,
                    payload=payload,
                    correlation_id=corr,
                    session_id=session_id,
                    source=source,
                )
                r = await _bus_pub(evt)
                return EventResult(
                    event_id=event_id,
                    correlation_id=corr,
                    topic=topic,
                    delivered_to=getattr(r, "delivered_to", 0),
                    redis_fanout=getattr(r, "redis_fanout", False),
                )
            except Exception as exc2:
                _logger.warning("[kernel.publish_event] fallback err topic=%s: %s", topic, exc2)
                return EventResult(event_id=event_id, correlation_id=corr, topic=topic, error=str(exc2))
        except Exception as exc:
            _logger.warning("[kernel.publish_event] topic=%s err: %s", topic, exc)
            return EventResult(event_id=event_id, correlation_id=corr, topic=topic, error=str(exc))

    # ── internal helper ────────────────────────────────────────────────────────

    async def _emit(self, topic: str, payload: dict, correlation_id: str) -> None:
        """Fire-and-forget — non blocca mai il caller (S10)."""
        try:
            await self.publish_event(topic=topic, payload=payload, correlation_id=correlation_id)
        except Exception as exc:
            _logger.debug("[kernel._emit] topic=%s err=%s", topic, exc)

    # ── resolveCapability ──────────────────────────────────────────────────────

    async def resolve_capability(
        self,
        capability: str,
        constraints: dict | None = None,
        correlation_id: str | None = None,
    ) -> dict:
        """
        ARCH-E3.2: Mappa una capability al miglior Worker disponibile.
        (S4: Brain non conosce l'infrastruttura, chiede solo capacità)
        """
        corr = correlation_id or str(uuid.uuid4())
        try:
            from .marketplace import resolve_capability as _resolve
            res = await _resolve(capability, constraints)
            _logger.info("[kernel] resolveCapability cap=%s corr=%s -> %s", capability, corr, res.get("status"))
            return res
        except Exception as exc:
            _logger.warning("[kernel.resolve_capability] err: %s", exc)
            return {"status": "error", "message": str(exc)}

    # ── executePlugin ──────────────────────────────────────────────────────────

    async def execute_plugin(
        self,
        plugin_id: str,
        input_data: Any,
        session_id: str | None = None,
        correlation_id: str | None = None,
    ) -> dict:
        """
        ARCH-E3.3: Esegue un plugin sandboxato via Kernel.
        (S5: Plugin sostituibili senza toccare agentLoop)
        """
        corr = correlation_id or str(uuid.uuid4())
        try:
            from .plugins import plugin_manager
            res = await plugin_manager.execute(plugin_id, input_data, session_id or "default")
            _logger.info("[kernel] executePlugin id=%s corr=%s -> %s", plugin_id, corr, res.get("status"))
            return res
        except Exception as exc:
            _logger.warning("[kernel.execute_plugin] err: %s", exc)
            return {"status": "error", "message": str(exc)}


# ── Singleton per uso interno ──────────────────────────────────────────────────
kernel: KernelAPI = KernelAPI()


# ── HTTP Endpoints ─────────────────────────────────────────────────────────────

@router.post("/submit", response_model=TaskResult, summary="submitTask — accoda job con priorità")
async def http_submit_task(req: SubmitTaskRequest) -> TaskResult:
    """Brain/Executor usano questo endpoint per delegare lavoro ai Worker (S10, S21)."""
    return await kernel.submit_task(
        payload=req.payload,
        priority=req.priority,
        session_id=req.session_id,
        correlation_id=req.correlation_id,
        timeout_s=req.timeout_s,
    )


@router.post("/chat", response_model=ChatResult, summary="chat — LLM provider-agnostic")
async def http_chat(req: ChatRequest) -> ChatResult:
    """Chiama il provider selezionato senza esporre quale (S20, S30)."""
    return await kernel.chat(
        messages=req.messages,
        model_hint=req.model_hint,
        session_id=req.session_id,
        correlation_id=req.correlation_id,
        max_tokens=req.max_tokens,
        temperature=req.temperature,
    )


@router.post("/memory", response_model=MemoryResult, summary="memory — router unificato su tutti i layer")
async def http_memory(req: MemoryRequest) -> MemoryResult:
    """Read/write/search/compress su Working, Episodic, Semantic, Reflection (S9)."""
    return await kernel.memory(
        op=req.op,
        session_id=req.session_id,
        correlation_id=req.correlation_id,
        query=req.query,
        content=req.content,
        role=req.role,
        layer=req.layer,
        limit=req.limit,
        metadata=req.metadata,
    )


@router.post("/event/publish", response_model=EventResult, summary="publishEvent — Event Bus bridge")
async def http_publish_event(req: PublishEventRequest) -> EventResult:
    """Pubblica evento sull'Event Bus con correlazione (S10, S13, S27)."""
    return await kernel.publish_event(
        topic=req.topic,
        payload=req.payload,
        correlation_id=req.correlation_id,
        session_id=req.session_id,
        source=req.source,
    )

@router.post("/resolve", summary="resolveCapability — mappa capability a Worker")
async def http_resolve_capability(capability: str, constraints: dict | None = None) -> dict:
    """Brain/Executor usano questo per trovare il miglior worker per una capacità (ARCH-E3.2)."""
    return await kernel.resolve_capability(capability, constraints)

@router.post("/plugin/execute", summary="executePlugin — esegue plugin sandboxato")
async def http_execute_plugin(plugin_id: str, input_data: Any, session_id: str | None = None) -> dict:
    """Esegue un plugin tramite il Kernel (ARCH-E3.3)."""
    return await kernel.execute_plugin(plugin_id, input_data, session_id)


@router.get("/status", summary="diagnostica servizi Kernel")
async def http_kernel_status() -> dict:
    """Stato aggregato di tutti i servizi sottostanti (S17, S24)."""
    checks: dict[str, Any] = {}

    # Job Queue (Redis)
    try:
        from .job_queue import _redis_ok, _llen, _K_PENDING
        redis_up = _redis_ok()
        pending  = await _llen(_K_PENDING) if redis_up else -1
        checks["job_queue"] = {"redis": redis_up, "pending_jobs": pending}
    except Exception as exc:
        checks["job_queue"] = {"error": str(exc)}

    # Event Bus
    try:
        from .event_bus import _subscribers
        checks["event_bus"] = {
            "active_subscriptions": sum(len(s) for s in _subscribers.values()),
            "topics": list(_subscribers.keys()),
        }
    except Exception as exc:
        checks["event_bus"] = {"error": str(exc)}

    # Session Manager
    try:
        from .session_manager import _lru
        checks["session_manager"] = {"lru_entries": len(_lru)}
    except Exception as exc:
        checks["session_manager"] = {"error": str(exc)}

    # Memory Manager
    try:
        from .state import _get_mem_manager_async as _gmm
        mem = await _gmm()
        checks["memory"] = {
            "initialized": mem is not None,
            "working_entries": len(getattr(getattr(mem, "working", None), "_entries", [])) if mem else 0,
        }
    except Exception as exc:
        checks["memory"] = {"error": str(exc)}

    return {
        "kernel":   "ARCH-K2.1",
        "version":  "1.0.0",
        "ts":       time.time(),
        "services": checks,
    }