"""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), }