""" backend/api/brain_planner.py — Brain Planner (ARCH-I4.1) Componente Planner del Brain: reasoning, planning, decomposition, reflection. Produce WorkflowPlan (lista ordinata di PlanStep) senza sapere come eseguirli. Il Brain è ora separato in: BrainPlanner → "cosa fare e in che ordine" (questo file) BrainExecutor → "come eseguire lo stato" (brain_executor.py) Flusso: Kernel.submit_task(payload, capability="plan") → BrainPlanner.plan(goal) → WorkflowPlan(steps=[PlanStep,...]) → BrainExecutor.execute_plan(plan) → [Kernel.submit_task per ogni step] Invarianti ADR: S4: Brain non conosce l'infrastruttura S9: Planner ignora come Executor esegue S21: Brain dipende solo dal Kernel S27: ogni piano tracciato via plan_id """ from __future__ import annotations import asyncio import logging import time import uuid from typing import Any, Literal from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from .auth_guard import AuthRole, require_role _logger = logging.getLogger("api.brain_planner") # ── Resolver guard ───────────────────────────────────────────────────────────── try: from .capability_resolver import resolver as _resolver, ResolveRequest as _RReq _RESOLVER_AVAILABLE = True except Exception: _resolver = None # type: ignore[assignment] _RESOLVER_AVAILABLE = False # ── Models ────────────────────────────────────────────────────────────────────── class PlanStep(BaseModel): """Singolo step del piano — capability + payload + dipendenze.""" step_id: str = Field(default_factory=lambda: str(uuid.uuid4())) capability: str = Field(..., description="Capability richiesta per questo step") description: str = Field("", description="Descrizione human-readable") payload: dict[str, Any] = Field(default_factory=dict) depends_on: list[str] = Field(default_factory=list, description="step_id degli step da completare prima") timeout_s: int = Field(60) retry_max: int = Field(2) optional: bool = Field(False, description="Se True, fallimento non blocca il piano") requires_gpu: bool = Field(False) metadata: dict[str, Any] = Field(default_factory=dict) class WorkflowPlan(BaseModel): """Piano di esecuzione prodotto dal Planner — input per il Workflow Engine.""" plan_id: str = Field(default_factory=lambda: str(uuid.uuid4())) goal: str = Field(..., description="Obiettivo originale") strategy: Literal["sequential", "parallel", "dag"] = Field( "sequential", description="sequential=step uno alla volta, " "parallel=tutti in parallelo, " "dag=dipendenze esplicite tra step") steps: list[PlanStep] = Field(default_factory=list) context: dict[str, Any] = Field(default_factory=dict, description="Contesto condiviso tra gli step") created_at: float = Field(default_factory=time.time) metadata: dict[str, Any] = Field(default_factory=dict) def validate_dag(self) -> list[str]: """Verifica che il DAG non abbia cicli. Ritorna errori (lista vuota = ok).""" step_ids = {s.step_id for s in self.steps} errors = [] for step in self.steps: for dep in step.depends_on: if dep not in step_ids: errors.append(f"Step {step.step_id}: dipende da {dep} che non esiste nel piano") return errors class PlanRequest(BaseModel): goal: str = Field(..., description="Obiettivo da pianificare") context: dict[str, Any] = Field(default_factory=dict) strategy: Literal["sequential", "parallel", "dag"] = "sequential" max_steps: int = Field(10, description="Numero massimo di step nel piano") session_id: str | None = None correlation_id: str = Field(default_factory=lambda: str(uuid.uuid4())) hints: list[str] = Field(default_factory=list, description="Capability suggerite dal caller") class ReflectRequest(BaseModel): """Richiesta di reflection su un piano eseguito (per migliorare piani futuri).""" plan_id: str outcome: Literal["success", "partial", "failure"] failed_steps: list[str] = Field(default_factory=list) notes: str = "" correlation_id: str = Field(default_factory=lambda: str(uuid.uuid4())) # ── BrainPlanner singleton ────────────────────────────────────────────────────── class BrainPlanner: """ Planner del Brain: data un goal, produce un WorkflowPlan. Strategia di planning (senza LLM, rule-based per ora): 1. Decomposizione goal → capability necessarie (via Resolver + hints) 2. Ordinamento step (sequenziale per default, DAG se dipendenze esplicite) 3. Validazione piano (no cicli, capabilities disponibili) 4. Reflection store (memorizza piani e outcome per futuro miglioramento) Nota: la versione LLM-based (reasoning con modello) sarà ARCH-I4.1 Phase 2. Questa implementazione è rule-based e serve a definire il contratto. """ def __init__(self) -> None: self._plans: dict[str, WorkflowPlan] = {} # plan_id → plan self._reflections: list[dict[str, Any]] = [] # storia reflection self._lock = asyncio.Lock() # ── Plan ────────────────────────────────────────────────────────────────── async def plan(self, req: PlanRequest) -> WorkflowPlan: """ Produce un WorkflowPlan dal goal. Step logic (rule-based): - Se hints presenti: crea uno step per ogni hint - Altrimenti: step singolo con capability "llm" (Brain decide runtime) - Verifica che ogni capability sia risolvibile dal Resolver """ capabilities = req.hints if req.hints else ["llm"] # Verifica capabilities via Resolver verified_caps = [] for cap in capabilities[:req.max_steps]: if _RESOLVER_AVAILABLE and _resolver is not None: if _resolver.can_resolve(cap): verified_caps.append(cap) else: _logger.warning("[planner] capability '%s' non risolvibile — esclusa dal piano", cap) else: verified_caps.append(cap) # no resolver: trust hints if not verified_caps: verified_caps = ["llm"] # fallback sempre # Costruisci steps steps = [] prev_id: str | None = None for cap in verified_caps: step = PlanStep( capability = cap, description = f"Esegui capability '{cap}' per goal: {req.goal[:80]}", payload = {"goal": req.goal, **req.context}, depends_on = [prev_id] if (req.strategy == "sequential" and prev_id) else [], timeout_s = 120, ) steps.append(step) prev_id = step.step_id plan = WorkflowPlan( goal = req.goal, strategy = req.strategy, steps = steps, context = req.context, metadata = {"session_id": req.session_id, "correlation_id": req.correlation_id}, ) # Validate DAG errors = plan.validate_dag() if errors: raise ValueError("Piano DAG non valido: " + "; ".join(errors)) async with self._lock: self._plans[plan.plan_id] = plan _logger.info("[planner] plan created plan_id=%s goal=%s steps=%d strategy=%s", plan.plan_id, req.goal[:40], len(steps), req.strategy) return plan def get_plan(self, plan_id: str) -> WorkflowPlan | None: return self._plans.get(plan_id) # ── Reflect ─────────────────────────────────────────────────────────────── async def reflect(self, req: ReflectRequest) -> dict: """Registra l'outcome di un piano — base per miglioramento futuro.""" plan = self._plans.get(req.plan_id) entry = { "plan_id": req.plan_id, "goal": plan.goal if plan else "unknown", "outcome": req.outcome, "failed_steps": req.failed_steps, "notes": req.notes, "ts": time.time(), } async with self._lock: self._reflections.append(entry) if len(self._reflections) > 500: # cap self._reflections = self._reflections[-500:] _logger.info("[planner] reflect plan=%s outcome=%s", req.plan_id, req.outcome) return {"reflected": True, "plan_id": req.plan_id, "outcome": req.outcome} def status(self) -> dict: return { "total_plans": len(self._plans), "total_reflections": len(self._reflections), "resolver_available": _RESOLVER_AVAILABLE, } # ── Singleton ──────────────────────────────────────────────────────────────────── planner = BrainPlanner() # ── HTTP Router ────────────────────────────────────────────────────────────────── router = APIRouter( prefix="/api/brain", tags=["brain-planner"], dependencies=[Depends(require_role(AuthRole.MACHINE))], ) @router.post("/plan", summary="Pianifica un goal → WorkflowPlan") async def route_plan(req: PlanRequest) -> WorkflowPlan: try: return await planner.plan(req) except ValueError as exc: raise HTTPException(400, str(exc)) @router.get("/plan/{plan_id}", summary="Recupera un piano esistente") async def route_get_plan(plan_id: str) -> WorkflowPlan: p = planner.get_plan(plan_id) if not p: raise HTTPException(404, f"Piano '{plan_id}' non trovato") return p @router.post("/reflect", summary="Reflection su un piano eseguito") async def route_reflect(req: ReflectRequest) -> dict: return await planner.reflect(req) @router.get("/planner/status", summary="Stato del BrainPlanner") async def route_status() -> dict: return planner.status()