api/agent_loop_routes.py

"""agent_loop_routes.py — Route SSE legacy, persona helpers, reason/unified/loop, agent-kernel.

Estratto da agent.py (split 2026-06-30).
Route coperte:
  POST /run_loop                  (deprecated 410)
  POST /api/agent/run-stream      (SSE legacy loop)
  POST /api/reason/loop
  POST /api/unified/loop
  GET  /api/agent-kernel/status
  POST /api/agent-kernel/dispatch
"""
from __future__ import annotations
import os, asyncio, json, uuid, time, re
import re as _re_persona
from fastapi import APIRouter, HTTPException, Request, Body
router = APIRouter()
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, field_validator
from typing import Literal
from .state import (
    _agent_tasks, _task_checkpoints, _loop_registry, _run_stream_tasks,
    _prune_agent_tasks, _prune_checkpoints, _prune_loop_registry,
    _get_mem_manager, _get_mem_manager_async, _get_executor, _get_planner, _get_ai_client,
    ReasonLoopIn, AgentTaskIn,
    write_ahead_task_created,
)
from .speculative import fire_speculative_tools
try:
    from .quality_guardian import run_quality_check as _run_quality_check
except Exception as _qg_err:
    import logging as _qg_log; _qg_log.getLogger(__name__).warning("[routes] quality_guardian import failed: %s", _qg_err)
    _run_quality_check = None
import logging
_logger = logging.getLogger("api.agent")
from .persistence import (
    sb_upsert_task, sb_update_status, sb_append_event,
    sb_restore_task, sb_get_events, sb_delete_task_events,
    sb_list_tasks, sb_save_checkpoint, sb_get_checkpoint,
    sb_restore_handoff_context, sb_upsert_handoff, sb_delete_handoff,
)
from ._agent_helpers import (
    _RE_SURROGATES, _ss, _log_task_exc,
    _PERSONA_KEYWORD_MAP, _PERSONA_CLIENT_CACHE,
    _build_persona_kw_map, _classify_persona_server, _get_persona_llm_client,
)
@router.post('/run_loop')
async def run_loop_removed():
    """S352: endpoint rimosso. Usare POST /api/agent/tasks + GET /api/agent/tasks/{id}/stream."""
    raise HTTPException(
        status_code=410,
        detail={
            "error": "Gone",
            "message": "Endpoint rimosso. Usare POST /api/agent/tasks + GET /api/agent/tasks/{id}/stream",
            "migration": "/api/agent/tasks",
        },
    )


# ── SSE run-stream ────────────────────────────────────────────────────────────

@router.post('/api/agent/run-stream')
async def agent_run_stream(body: ReasonLoopIn, request: Request):
    # S-BENCH: auth guard — consistente con /api/exec e /api/execute-shell
    _itok = os.getenv('INTERNAL_TOKEN', '')
    if _itok and request.headers.get('X-Internal-Token') != _itok:
        raise HTTPException(401, 'Unauthorized')
    async def generate():
        queue: asyncio.Queue = asyncio.Queue()

        async def step_cb(step: dict) -> None:
            await queue.put(step)

        async def run_loop() -> None:
            try:
                from agents.unified_loop import UnifiedAgentLoop
                # S388: usa singleton _get_ai_client() — nessuna re-istanziazione OpenAI() per request
                client = _get_ai_client()
                try:
                    from agents.critic import Critic
                    from agents.response_verifier import ResponseVerifier
                    _critic   = Critic(llm_client=client)
                    _verifier = ResponseVerifier()
                except Exception as _cv_err:
                    _logger.warning("[routes] Critic/Verifier init failed: %s", _cv_err)
                    _critic   = None
                    _verifier = None
                # Resume automatico: inietta contesto checkpoint se disponibile (Case 2.5 fall-through)
                _resume_ctx = getattr(body, '_resume_context', None)
                _resume_max = getattr(body, '_resume_max_steps', None) or body.max_steps
                context_str = '\n'.join(m.get('content', '') for m in body.context) if body.context else ''
                # Bug-5-FIX: resume context iniettato DOPO che context_str è definito (era NameError)
                if _resume_ctx:
                    context_str = f"[RIPRESA AUTOMATICA]\n{_resume_ctx}\n\n{context_str}".strip()

                loop = UnifiedAgentLoop(
                    llm_client=client, critic=_critic, verifier=_verifier,
                    memory=await _get_mem_manager_async(), executor=_get_executor(), planner=_get_planner(),
                )
                # S456-X5: prepend project context (projectMemory.getContext() dal frontend)
                if body.project_context:
                    context_str = f"[PROGETTO CORRENTE]\n{body.project_context}\n\n{context_str}".strip()
                # S456-X4: inject top failure patterns appresi dal selfLearning frontend
                if body.learning_hints:
                    # S591: learning_hints[:3]→[:5] — più pattern appresi nel context
                    hints_str = "\n".join(f"- {h}" for h in body.learning_hints[:5])
                    context_str = f"{context_str}\n\n[PATTERN DI ERRORE APPRESI]\n{hints_str}".strip()
                # P35: vincoli negativi dal frontend (agentConstraints.ts → VFS /.agent/constraints.json)
                _neg_c = getattr(body, 'negative_constraints', '') or ''
                if _neg_c:
                    context_str = f"[VINCOLI OPERATIVI APPRESI — NON VIOLARE]\n{_neg_c}\n\n{context_str}".strip()
                result = await loop.run(
                    goal=body.goal, context=context_str,
                    max_steps=body.max_steps, on_step=step_cb,
                    session_id=getattr(body, "session_id", "") or "",
                )
                await queue.put({
                    '__done__': True,
                    'result':   result.get('output',  ''),
                    'engine':   result.get('engine',  'fallback'),
                    'success':  result.get('success', False),
                })
            except Exception as exc:
                # GAP-A1: log incident in registry (fire-and-forget, non-blocking)
                try:
                    from api.incident_registry import log_incident as _log_inc
                    asyncio.create_task(_log_inc(
                        task_id=body.goal[:32].replace(' ', '_'),
                        goal=body.goal, error=str(exc), source="agent",
                    )).add_done_callback(_log_task_exc)
                except Exception as _exc:
                    _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
                await queue.put({'__error__': str(exc)})

        task    = asyncio.create_task(run_loop())
        task_id = body.goal[:32].replace(' ', '_')
        # ABORT-1: registra task + queue per permettere cancellazione via POST /api/agent/abort
        _run_stream_tasks[task_id] = {"task": task, "queue": queue}
        yield "retry: 3000\n\n"
        yield f"data: {json.dumps({'type': 'task_start', 'taskId': task_id})}\n\n"

        # S386: fast-fail — se tutti i provider sono down (heartbeat lo sa già),
        # non aspettare 120s di tentativi: rispondi subito con errore chiaro.
        try:
            from api.state import _heartbeat_state
            _providers = _heartbeat_state.get("providers", [])
            if _providers and not any(p.get("ok") for p in _providers):
                task.cancel()
                _names = ", ".join(p["name"] for p in _providers)
                yield f"data: {json.dumps({'type': 'task_aborted', 'taskId': task_id, 'abort_reason': 'system', 'abort_source': 'no_providers', 'error': f'Nessun provider AI disponibile ({_names})'})}\n\n"  # MX18-ABORT: no providers → system abort
                yield "data: [DONE]\n\n"
                return
        except Exception as _hb_err:
            _logger.debug("[routes] heartbeat skip silenced: %s", type(_hb_err).__name__)  # non inizializzato, prosegui normalmente

        # S386: timeout ridotto 120→60s — risposta entro 1 minuto o errore esplicito
        timeout_secs    = float(os.getenv('AGENT_STREAM_TIMEOUT', '60'))
        heartbeat_secs  = 15.0
        elapsed         = 0.0
        try:
            while True:
                try:
                    item    = await asyncio.wait_for(queue.get(), timeout=heartbeat_secs)
                    elapsed = 0.0
                except asyncio.TimeoutError:
                    elapsed += heartbeat_secs
                    if elapsed >= timeout_secs:
                        yield f"data: {json.dumps({'type': 'task_aborted', 'taskId': task_id, 'abort_reason': 'timeout', 'abort_source': 'stream_timeout'})}\n\n"  # MX18-ABORT: timeout → task_aborted
                        break
                    yield 'data: {"type":"ping"}\n\n'
                    continue
                # ABORT-2: segnale abort dall'endpoint POST /api/agent/abort
                if "__abort__" in item:
                    _ar  = item.get('abort_reason', 'user_stop')   # MX18-ABORT: dynamic reason
                    _src = item.get('abort_source', 'backend_abort_queue')
                    yield f"data: {json.dumps({'type': 'task_aborted', 'taskId': task_id, 'abort_reason': _ar, 'abort_source': _src})}\n\n"  # MX16+MX18-ABORT
                    break
                if '__error__' in item:
                    yield f"data: {json.dumps({'type': 'task_error', 'taskId': task_id, 'error': _ss(item['__error__'])})}\n\n"
                    break
                # S420: streaming token — emetti subito al frontend senza accumulare
                if item.get('action') == 'text_chunk':
                    yield f"data: {json.dumps({'type': 'text_chunk', 'token': _ss(item.get('token', '')), 'taskId': task_id})}\n\n"
                    continue
                # S758-P4.1: tool_use — chip pre-esecuzione (agent_run_stream path)
                _rs_act = item.get('action', '')
                _rs_st  = item.get('status', '')
                if ((_rs_act == 'tool_start' and _rs_st == 'running') or
                        (_rs_act.startswith('executor:') and _rs_st == 'started')):
                    _rs_tool = _rs_act.replace('executor:', '') if _rs_act.startswith('executor:') else _rs_act
                    yield f"data: {json.dumps({'type': 'tool_use', 'taskId': task_id, 'tool': _rs_tool, 'name': _rs_tool, 'label': item.get('title', _rs_tool.replace('_', ' ').capitalize())})}\n\n"
                if '__done__' in item:
                    yield f"data: {json.dumps({'type': 'task_done', 'taskId': task_id, 'result': _ss(item['result']), 'engine': item['engine'], 'success': item['success']})}\n\n"
                    break
                # S393 Priority 1: Narrative Streaming — arricchisce step_done con explanation
                _NARR_QUICK = {
                    'llm': 'Elaborazione risposta AI',
                    'direct_tools': 'Strumenti diretti',
                    'web_search': 'Ricerca web', 'get_weather': 'Dati meteo',
                    'read_page': 'Lettura pagina', 'calculate': 'Calcolo matematico',
                    'generate_image': 'Generazione immagine AI',
                    'execution_validator_fix': 'Auto-correzione codice (S393)',
                    'tool_governor_skip': 'Tool già eseguito — risultato riutilizzato',
                    # S661: label narrative per tool aggiunti in S648-S659 — prima usavano
                    # _act_q.replace('_',' ').capitalize() → "Apply patch", "Call api" (generico)
                    'apply_patch':   'Applico patch al file…',
                    'call_api':      'Chiamo API REST…',
                    'send_email':    'Invio email…',
                    'create_pdf':    'Genero documento PDF…',
                    'web_research':  'Ricerca multi-fonte…',
                    'write_file':    'Scrivo file…',
                    'read_file':     'Leggo file…',
                    'execute_shell': 'Eseguo comando shell…',
                    'analyze_image': 'Analizzo immagine…',
                    'run_python':    'Eseguo Python (Pyodide)…',
                    # S-GAP1: narrative fasi strategiche
                    'plan':             'Analizzo la richiesta e preparo un piano di esecuzione…',
                    'reflective_debug': 'Ho incontrato un ostacolo — ricalcolo una strategia più efficiente…',
                    'fallback':         'Adotto un approccio alternativo per completare il task…',
                    'smolagents':       'Orchestro gli strumenti necessari…',
                }
                _act_q = item.get('action', '')
                if 'explanation' not in item:
                    item['explanation'] = _NARR_QUICK.get(_act_q, _act_q.replace('_', ' ').capitalize())
                if 'title' not in item:
                    item['title'] = item['explanation']

                # S403: SSE Visibility Guard — classifica ogni step event:
                # "internal"  → mai visibile (pipeline internals: planner, llm, reflection)
                # "progress"  → visibile come progress card (tool reali, auto-fix)
                # "debug"     → visibile solo in dev mode (direct_tools, fast_path)
                # Il frontend filtra per visibility — solo "progress" mostrato all'utente.
                _STEP_VISIBILITY: dict[str, str] = {
                    # Internal pipeline — never shown to user
                    'plan':              'progress',   # S-GAP1
                    'llm':               'internal',
                    'smolagents':        'internal',
                    'fallback':          'progress',   # S-GAP1
                    'reflective_debug':  'progress',   # S-GAP1
                    'fast_path':         'internal',
                    'executor':          'internal',
                    # Progress — shown as step cards (user-visible)
                    'tool_start':              'progress',
                    'execution_validator_fix': 'progress',
                    'goal_verifier':           'progress',
                    'web_search':              'progress',
                    'get_weather':             'progress',
                    'read_page':               'progress',
                    'calculate':               'progress',
                    'generate_image':          'progress',
                    'run_python':              'progress',
                    'tool_governor_skip':      'progress',
                    # S660: tool aggiunti in S648-S659 mancanti da _STEP_VISIBILITY →
                    # fallback rule: _act_q.startswith('tool_') era False per questi →
                    # classificati 'debug' → nascosti all'utente durante esecuzione.
                    'apply_patch':   'progress',
                    'call_api':      'progress',
                    'send_email':    'progress',
                    'create_pdf':    'progress',
                    'web_research':  'progress',
                    'write_file':    'progress',
                    'read_file':     'progress',
                    'execute_shell': 'progress',
                    'analyze_image': 'progress',
                    # Debug — shown only when devMode active
                    'direct_tools':     'debug',
                    # S-LOOP2: fase esecuzione avanzata — visibili come progress card
                    'reasoning_core':    'progress',  # S-LOOP2: ReasoningCore multi-step
                    'browser_verifier':  'progress',  # S-LOOP2: Browser Goal Verification live
                }
                # Fallback: azioni sconosciute con "tool_" prefix → progress; resto → debug
                _vis = _STEP_VISIBILITY.get(_act_q)
                if _vis is None:
                    _vis = 'progress' if _act_q.startswith('tool_') or _act_q.startswith('executor:') else 'debug'
                item['visibility'] = _vis

                yield f"data: {json.dumps({'type': 'step_done', 'step': item, 'taskId': task_id})}\n\n"
        finally:
            task.cancel()
            # ABORT-3: cleanup registro — libera memoria e impedisce abort su task già terminati
            _run_stream_tasks.pop(task_id, None)
            yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


# ── Reason loop / Unified loop ─────────────────────────────────────────────────

@router.post('/api/reason/loop')
async def reason_loop(body: ReasonLoopIn):
    try:
        from agents.unified_loop import UnifiedAgentLoop
        # S388: singleton — riusa il client già inizializzato
        client = _get_ai_client()
        try:
            from agents.critic import Critic
            from agents.response_verifier import ResponseVerifier
            _critic   = Critic(llm_client=client)
            _verifier = ResponseVerifier()
        except Exception as _cv_err:
            _logger.warning("[routes] Critic/Verifier init failed: %s", _cv_err)
            _critic   = None
            _verifier = None
        loop = UnifiedAgentLoop(
            llm_client=client, critic=_critic, verifier=_verifier,
            memory=await _get_mem_manager_async(), executor=_get_executor(), planner=_get_planner(),
        )
        context_str = '\n'.join(m.get('content', '') for m in body.context) if body.context else ''
        # N-2-FIX: accumula step intermedi tramite on_step — inclusi nel response JSON per debug frontend
        _steps_log: list[dict] = []
        async def _on_step(step_data: dict) -> None:
            _steps_log.append({
                'action': step_data.get('action', ''),
                'output': str(step_data.get('output', ''))[:400],  # S577: 200→400
            })
        result = await loop.run(goal=body.goal, context=context_str, max_steps=body.max_steps, on_step=_on_step, session_id=getattr(body, "session_id", "") or "")
        if isinstance(result, dict):
            output_text = result.get('output', '') or result.get('answer', '') or ''
            engine_used = result.get('engine', 'ambiguity-gate' if result.get('answer') else 'unknown')
            errors_list = result.get('errors', [])
        else:
            output_text = str(result)
            engine_used = 'unknown'
            errors_list = []
        return {
            'ok':     bool(output_text and output_text.strip()),
            'success': bool(output_text and output_text.strip()),  # alias compat frontend
            'output':  output_text,                                  # alias compat frontend
            'result': output_text,
            'source': 'backend_loop',
            'engine': engine_used,
            'errors': errors_list,
            'steps':  _steps_log,  # N-2-FIX: step intermedi per debug/telemetria frontend
        }
    except Exception as e:
        _logger.error("[reason/loop] Error: %s", e)
        return {
            'ok':     False,
            'result': f'Backend reasoning non disponibile: {e}. Il loop browser continua normalmente.',
            'source': 'fallback',
            'steps':  [],
        }


@router.post('/api/unified/loop')
async def unified_loop(body: ReasonLoopIn):
    """Alias di /api/reason/loop — compatibilità con tutte le versioni frontend."""
    return await reason_loop(body)


# ── Agent kernel ───────────────────────────────────────────────────────────────

@router.get('/api/agent-kernel/status')
async def agent_kernel_status():
    gh_token = os.getenv('GITHUB_TOKEN') or os.getenv('GH_TOKEN', '')
    return {
        'dispatch_available': bool(gh_token),
        'workflow_url': 'https://github.com/Baida98/AI/actions/workflows/agent-kernel.yml',
        'mobile_url':   'https://github.com/Baida98/AI/actions',
        'secrets_needed': ['OPENROUTER_API_KEY', 'GROQ_API_KEY', 'GEMINI_API_KEY', 'HF_TOKEN', 'NVIDIA_API_KEY'],
        'usage': 'Vai su GitHub Actions → Agent Kernel — no PC → Run workflow → inserisci il goal',
    }


# S442-FIX3: modello Pydantic per agent_kernel_dispatch.
# Prima: body: dict grezzo → mode non validato, goal controllato solo dopo estrazione.
# Ora: validazione in ingresso → 422 chiaro invece di 500 a runtime.
class AgentKernelDispatchIn(BaseModel):
    goal: str
    mode: Literal["plan", "execute", "analyze"] = "plan"

    @field_validator('goal', mode='before')
    @classmethod
    def validate_goal(cls, v: object) -> str:
        if not isinstance(v, str) or not str(v).strip():
            raise ValueError('goal must be a non-empty string')
        return str(v).strip()


@router.post('/api/agent-kernel/dispatch')
async def agent_kernel_dispatch(body: AgentKernelDispatchIn):
    gh_token = os.getenv('GITHUB_TOKEN') or os.getenv('GH_TOKEN', '')
    if not gh_token:
        raise HTTPException(503, detail={
            'error': 'no_github_token',
            'message': 'GITHUB_TOKEN non configurato nel backend.',
        })
    goal = body.goal
    mode = body.mode
    import httpx as _httpx
    try:
        async with _httpx.AsyncClient(timeout=15) as _hc:
            _resp = await _hc.post(
                'https://api.github.com/repos/Baida98/AI/actions/workflows/agent-kernel.yml/dispatches',
                json={'ref': 'main', 'inputs': {'goal': goal, 'mode': mode, 'commit_memory': 'true'}},
                headers={
                    'Authorization':       f'Bearer {gh_token}',
                    'Accept':              'application/vnd.github+json',
                    'X-GitHub-Api-Version': '2022-11-28',
                },
            )
        if _resp.status_code >= 400:
            raise HTTPException(_resp.status_code, detail=_resp.text[:500])
        return {'ok': True, 'status': _resp.status_code, 'goal': goal, 'mode': mode}
    except _httpx.HTTPError as e:
        raise HTTPException(502, detail=str(e)[:500])


# ── Agent tasks (FASE 2.1 + S359 persistence) ──────────────────────────────────