api/agent_checkpoint_routes.py

"""agent_checkpoint_routes.py — Checkpoint, debug/timing, skill-stats, circuit-status.

Estratto da agent.py (split 2026-06-30).
Route coperte:
  POST   /api/agent/tasks/{task_id}/checkpoint
  GET    /api/agent/tasks/{task_id}/checkpoint
  DELETE /api/agent/tasks/{task_id}/checkpoint
  GET    /api/agent/checkpoints
  GET    /debug/timing
  GET    /api/agent/skill-stats/{session_id}
  GET    /api/agent/skill-stats
  GET    /api/agent/circuit-status/{session_id}
"""
from __future__ import annotations
import os, asyncio, json, uuid, time, re
import re as _re_persona
from fastapi import APIRouter, HTTPException, Request, Body
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:
    _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
try:
    from .telegram_notify import notify_task_done as _tg_done, notify_task_error as _tg_error, notify_task_start as _tg_start, notify_task_step as _tg_step
except Exception:
    async def _tg_done(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_error(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_start(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_step(*_a, **_kw):  pass  # type: ignore[misc]

router = APIRouter()



# ── Task checkpoints ───────────────────────────────────────────────────────────

class CheckpointIn(BaseModel):
    taskId:     str
    step:       int
    goal:       str
    plan:       list[str] = []
    logs:       list[str] = []
    artifacts:  list[str] = []
    retryCount: int = 0
    extra:      dict = {}


@router.post('/api/agent/tasks/{task_id}/checkpoint')
async def save_checkpoint(task_id: str, body: CheckpointIn):
    _prune_checkpoints()
    _task_checkpoints[task_id] = {
        'taskId':     task_id,
        'step':       body.step,
        'goal':       body.goal,
        'plan':       body.plan,
        'logs':       body.logs[-50:],
        'artifacts':  body.artifacts,
        'retryCount': body.retryCount,
        'extra':      body.extra,
        'savedAt':    int(time.time() * 1000),
    }
    asyncio.create_task(sb_save_checkpoint(task_id, _task_checkpoints[task_id])).add_done_callback(_log_task_exc)
    return {'saved': True, 'taskId': task_id, 'step': body.step}


@router.get('/api/agent/tasks/{task_id}/checkpoint')
async def get_checkpoint(task_id: str):
    _prune_checkpoints()
    cp = _task_checkpoints.get(task_id)
    if not cp:
        cp = await sb_get_checkpoint(task_id)
    if not cp:
        raise HTTPException(404, detail={'error': 'checkpoint_not_found', 'taskId': task_id})
    return cp


@router.delete('/api/agent/tasks/{task_id}/checkpoint')
async def delete_checkpoint(task_id: str):
    _task_checkpoints.pop(task_id, None)
    return {'deleted': task_id}


@router.get('/api/agent/checkpoints')
async def list_checkpoints():
    _prune_checkpoints()
    now = int(time.time() * 1000)
    return {
        'count': len(_task_checkpoints),
        'checkpoints': [
            {'taskId': k, 'step': v['step'], 'goal': v['goal'][:300], 'age_ms': now - v['savedAt']}  # S606: 200→300
            for k, v in _task_checkpoints.items()
        ],
    }


# ─── Sprint 5 ITEM 15: /debug/timing — telemetria timing + qualità agente ────
# Usato da TelemetryDashboard.tsx (frontend) per la sezione "Qualità agente".
# Espone: timing_stats (avg/count per fase) + repair_stats (contatori qualità).
# Non richiede auth — dati aggregati, nessun dato sensibile.
@router.get('/debug/timing')
async def get_debug_timing():
    """
    Espone timing breakdown per fase (classify/plan/coder/verifier/browser)
    e contatori qualità (goal_success, repair_success, tool_failure, req_engine).
    Formato: { timing_stats: {label: {avg, count}}, repair_stats: {key: count} }
    """
    try:
        from api.state import _TIMING_STORE, _REPAIR_STATS
        timing_stats: dict = {}
        for label, samples in _TIMING_STORE.items():
            if samples:
                avg_val = round(sum(samples) / len(samples), 1)
            else:
                avg_val = None
            timing_stats[label] = {"avg": avg_val, "count": len(samples)}
        return {
            "timing_stats": timing_stats,
            "repair_stats": dict(_REPAIR_STATS),
        }
    except Exception as exc:
        return {"timing_stats": {}, "repair_stats": {}, "error": str(exc)}


# ─── GAP-SKILL-SYNC: /api/agent/skill-stats — statistiche tool adattive ──────
# Espone i dati del SkillTracker (session-scoped success/fail per tool)
# al frontend per merge con skillRegistry Dexie — vista cross-runtime unificata.
@router.get('/api/agent/skill-stats/{session_id}')
async def get_skill_stats(session_id: str):
    """Success/fail rate + Wilson score per ogni tool nella sessione.

    Il frontend usa questa API per arricchire i dati Dexie di skillRegistry.ts
    con le stats backend: confidence reale (server-side) vs contatori browser-only.
    """
    try:
        from agents.skill_tracker import get_skill_tracker
        return {
            "session_id": session_id,
            "stats":      get_skill_tracker().get_stats(session_id),
        }
    except Exception as exc:
        return {"session_id": session_id, "stats": {}, "error": str(exc)}


@router.get('/api/agent/skill-stats')
async def list_all_skill_sessions():
    """Debug: panoramica di tutte le sessioni SkillTracker attive (tool count, call count)."""
    try:
        from agents.skill_tracker import get_skill_tracker
        return get_skill_tracker().get_all_sessions()
    except Exception as exc:
        return {"error": str(exc)}

# ── /api/agent/circuit-status/{session_id} — circuit breaker live status ──────
# Espone per ogni tool tracciato in sessione: stato circuito, Wilson score,
# recovery calls effettuate — utile per debug e monitoring real-time.
@router.get('/api/agent/circuit-status/{session_id}')
async def get_circuit_status(session_id: str):
    """
    Stato real-time del circuit breaker per ogni tool di una sessione.

    Per ogni tool tracciato, classifica il circuito come:
      - open   → Wilson score < 0.15 AND total_count >= 3 AND tool ha fallback
                 (il tool viene bypassato — routing automatico ai fallback)
      - closed → performance sufficiente o dati insufficienti per aprire il circuit

    Campi per tool:
      wilson_score:     lower bound dell'intervallo di confidenza al 95% (0–1)
      success_count:    successi registrati nella sessione
      fail_count:       fallimenti registrati nella sessione
      total_count:      chiamate totali
      success_rate:     raw rate (NON usato dal circuit — solo informativo)
      avg_latency_ms:   latenza media (ms)
      has_fallbacks:    True se TOOL_REGISTRY definisce fallback per il tool
      recovery_calls:   quante volte il recovery credit ha concesso un tentativo
      circuit_state:    "open" | "closed" | "no_data" | "insufficient_data"

    Thresholds (from executor.py):
      circuit_open_threshold: 0.15   (Wilson score sotto cui il circuit si apre)
      min_calls_for_circuit:  3      (chiamate minime prima che il circuit possa aprirsi)
      recovery_interval:      5      (ogni N call con circuit open → recovery attempt)
    """
    try:
        from agents.skill_tracker import get_skill_tracker
        from tools.registry import TOOL_REGISTRY
        from api.state import _get_executor
        from agents.executor import (
            _CIRCUIT_OPEN_THRESHOLD,
            _MIN_CALLS_FOR_CIRCUIT,
            _RECOVERY_INTERVAL,
        )

        stats = get_skill_tracker().get_stats(session_id)

        # Recovery counts vivono nell'istanza Executor singleton
        executor    = _get_executor()
        rec_counts: dict = {}
        if executor is not None:
            rec_counts = getattr(executor, '_circuit_recovery_counts', {})

        circuits_open: list[dict]   = []
        circuits_closed: list[dict] = []

        for tool_name, s in stats.items():
            has_fallbacks = bool(TOOL_REGISTRY.get(tool_name, {}).get('fallbacks'))
            recovery_calls = rec_counts.get(tool_name, 0)

            # Replica logica _is_circuit_open() di executor.py
            if s['total_count'] == 0:
                state = 'no_data'
            elif s['total_count'] < _MIN_CALLS_FOR_CIRCUIT:
                state = 'insufficient_data'
            elif s['wilson_score'] < _CIRCUIT_OPEN_THRESHOLD and has_fallbacks:
                state = 'open'
            else:
                state = 'closed'

            entry = {
                'tool':            tool_name,
                'circuit_state':   state,
                'wilson_score':    s['wilson_score'],
                'success_count':   s['success_count'],
                'fail_count':      s['fail_count'],
                'total_count':     s['total_count'],
                'success_rate':    s['success_rate'],
                'avg_latency_ms':  s['avg_latency_ms'],
                'has_fallbacks':   has_fallbacks,
                'recovery_calls':  recovery_calls,
            }
            if state == 'open':
                circuits_open.append(entry)
            else:
                circuits_closed.append(entry)

        # Ordina open per Wilson score asc (peggiori prima), closed per desc (migliori prima)
        circuits_open.sort(key=lambda x: x['wilson_score'])
        circuits_closed.sort(key=lambda x: x['wilson_score'], reverse=True)

        return {
            'session_id':            session_id,
            'total_tools_tracked':   len(stats),
            'circuits_open_count':   len(circuits_open),
            'circuits_closed_count': len(circuits_closed),
            'circuits_open':         circuits_open,
            'circuits_closed':       circuits_closed,
            'thresholds': {
                'circuit_open_threshold': _CIRCUIT_OPEN_THRESHOLD,
                'min_calls_for_circuit':  _MIN_CALLS_FOR_CIRCUIT,
                'recovery_interval':      _RECOVERY_INTERVAL,
            },
        }
    except Exception as exc:
        return {
            'session_id':          session_id,
            'total_tools_tracked': 0,
            'circuits_open_count': 0,
            'circuits_open':       [],
            'circuits_closed':     [],
            'error':               str(exc),
        }