"""backend/api/telegram_webhook.py — Ricezione comandi dal bot Telegram. Riceve aggiornamenti dal bot Telegram via webhook e risponde a comandi: /start /help — menu + lista comandi /status — stato backend + task attivi (in-memory + Supabase fallback) /tasks — ultimi task con elapsed time /do — lancia un nuovo task tramite loop /scan_now — health-check completo (AI + Supabase + Telegram) /health — alias /scan_now /score — score immediato dall'ultimo benchmark (no run) callback_query — gestisce inline buttons notify_task_done ARCHITETTURA: il bot usa getUpdates polling via scripts/session-daemon.mjs. Il webhook NON è registrato su Telegram — l'endpoint /webhook/setup è DISABILITATO per impedire conflitti con il daemon. Per passare a webhook mode: ferma il daemon, poi riabilita manualmente. """ from __future__ import annotations import asyncio, html, logging, os, time from fastapi import APIRouter, BackgroundTasks, Request, HTTPException from pydantic import BaseModel import httpx # top-level — era lazy in 20+ funzioni _logger = logging.getLogger("api.telegram_webhook") # unico logger (duplicato rimosso) def _log_tg_exc(task: "asyncio.Task[None]") -> None: """Gap-2.6: log exceptions from fire-and-forget tasks.""" try: exc = task.exception() if exc: _logger.warning("tg_webhook bg task error: %s: %s", type(exc).__name__, exc) except (asyncio.CancelledError, asyncio.InvalidStateError): pass router = APIRouter(prefix="/api/telegram", tags=["telegram"]) # ── Helpers ─────────────────────────────────────────────────────────────────── def _get_bot_token() -> str: return os.getenv("TELEGRAM_BOT_TOKEN", "").strip() async def _tg_reply(chat_id: str | int, text: str, token: str | None = None, keyboard: dict | None = None) -> None: """Invia risposta al chat_id con HTML + opzionale inline keyboard.""" bot_token = token or _get_bot_token() if not bot_token: return payload: dict = { "chat_id": chat_id, "text": text, "parse_mode": "HTML", "link_preview_options": {"is_disabled": True}, } if keyboard: payload["reply_markup"] = keyboard try: import httpx async with httpx.AsyncClient(timeout=8.0) as c: await c.post( f"https://api.telegram.org/bot{bot_token}/sendMessage", json=payload, ) except Exception as exc: _logger.warning("tg_reply error: %s", exc) async def _tg_answer_callback(callback_query_id: str, text: str = "", token: str | None = None) -> None: """Risponde a un callback_query (obbligatorio per chiudere il loading sui buttons).""" bot_token = token or _get_bot_token() if not bot_token: return try: import httpx async with httpx.AsyncClient(timeout=5.0) as c: await c.post( f"https://api.telegram.org/bot{bot_token}/answerCallbackQuery", json={"callback_query_id": callback_query_id, "text": text, "show_alert": False}, ) except Exception as exc: _logger.debug("answer_callback error: %s", exc) async def _tg_send(chat_id: str | int, text: str, token: str | None = None, keyboard: dict | None = None) -> str | None: """Invia messaggio e ritorna il message_id (per editMessageText streaming).""" bot_token = token or _get_bot_token() if not bot_token: return None payload: dict = { "chat_id": chat_id, "text": text, "parse_mode": "HTML", "link_preview_options": {"is_disabled": True}, } if keyboard: payload["reply_markup"] = keyboard try: import httpx async with httpx.AsyncClient(timeout=8.0) as c: r = await c.post( f"https://api.telegram.org/bot{bot_token}/sendMessage", json=payload, ) j = r.json() return str(j.get("result", {}).get("message_id", "")) if j.get("ok") else None except Exception as exc: _logger.warning("tg_send error: %s", exc) return None async def _tg_edit(chat_id: str | int, message_id: str, text: str, token: str | None = None, keyboard: dict | None = None) -> bool: """Aggiorna messaggio esistente — streaming live via editMessageText. Ritorna True se successo. Rate-limit: max 20 edit/min per chat Telegram.""" bot_token = token or _get_bot_token() if not bot_token or not message_id: return False payload: dict = { "chat_id": chat_id, "message_id": int(message_id), "text": text[:4000], "parse_mode": "HTML", "link_preview_options": {"is_disabled": True}, } if keyboard: payload["reply_markup"] = keyboard try: import httpx async with httpx.AsyncClient(timeout=8.0) as c: r = await c.post( f"https://api.telegram.org/bot{bot_token}/editMessageText", json=payload, ) return r.json().get("ok", False) except Exception as exc: _logger.debug("tg_edit error: %s", exc) return False async def _tg_photo( chat_id: str | int, photo_url: str, caption: str = "", token: str | None = None, keyboard: dict | None = None, ) -> None: """Invia foto/chart via sendPhoto Telegram. Strategia anti URL-lungo: 1. POST a quickchart.io → scarica PNG bytes → multipart sendPhoto (no limite URL). 2. Fallback: invia URL direttamente (funziona se URL < ~2000 chars). """ bot_token = token or _get_bot_token() if not bot_token: return caption_safe = (caption or "")[:1024] import httpx as _hx_p, json as _j_p, urllib.parse as _ul_p, re as _re_p png_bytes: bytes | None = None if "quickchart.io/chart" in photo_url: try: m = _re_p.search(r"[?&]c=([^&]+)", photo_url) if m: cfg_dict = _j_p.loads(_ul_p.unquote(m.group(1))) async with _hx_p.AsyncClient(timeout=20.0) as c: qr = await c.post( "https://quickchart.io/chart", json={"chart": cfg_dict, "width": 720, "height": 420, "backgroundColor": "white", "format": "png"}, ) if qr.status_code == 200 and qr.headers.get("content-type", "").startswith("image/"): png_bytes = qr.content _logger.debug("tg_photo: quickchart POST ok, %d bytes", len(png_bytes)) except Exception as exc: _logger.debug("tg_photo: quickchart POST fallback: %s", exc) try: import httpx as _hx_s async with _hx_s.AsyncClient(timeout=15.0) as c: if png_bytes: import json as _j_s data: dict = {"chat_id": str(chat_id), "parse_mode": "HTML"} if caption_safe: data["caption"] = caption_safe if keyboard: data["reply_markup"] = _j_s.dumps(keyboard) files = {"photo": ("chart.png", png_bytes, "image/png")} await c.post(f"https://api.telegram.org/bot{bot_token}/sendPhoto", data=data, files=files) else: payload: dict = {"chat_id": chat_id, "photo": photo_url, "parse_mode": "HTML"} if caption_safe: payload["caption"] = caption_safe if keyboard: payload["reply_markup"] = keyboard await c.post(f"https://api.telegram.org/bot{bot_token}/sendPhoto", json=payload) except Exception as exc: _logger.warning("tg_photo error: %s", exc) async def _tg_typing(chat_id: str | int, action: str = "typing", token: str | None = None) -> None: """Invia sendChatAction — mostra '⌨️ digitando…' prima di operazioni pesanti. Dura 5 secondi o fino al prossimo messaggio del bot. Azioni: typing, upload_photo, upload_document, find_location, record_video_note. """ bot_token = token or _get_bot_token() if not bot_token: return try: async with httpx.AsyncClient(timeout=3.0) as c: await c.post( f"https://api.telegram.org/bot{bot_token}/sendChatAction", json={"chat_id": chat_id, "action": action}, ) except Exception: pass async def _tg_react( chat_id: str | int, message_id: int | str, emoji: str = "👍", token: str | None = None, ) -> None: """Aggiunge reazione emoji a un messaggio (Bot API 7.1+, Feb 2024). Emoji supportate: 👍 👎 ❤ 🔥 🥰 👏 😁 🤔 🤯 😱 🎉 🤩 🏆 ✅ 💯 ⚡ 🚀 🎯 """ bot_token = token or _get_bot_token() if not bot_token or not message_id: return try: async with httpx.AsyncClient(timeout=3.0) as c: await c.post( f"https://api.telegram.org/bot{bot_token}/setMessageReaction", json={ "chat_id": chat_id, "message_id": int(message_id), "reaction": [{"type": "emoji", "emoji": emoji}], "is_big": False, }, ) except Exception: pass def _fmt_elapsed(created_at_ms: int) -> str: """Formatta elapsed time da un timestamp ms → stringa leggibile.""" diff = int(time.time() * 1000) - created_at_ms s = diff // 1000 if s < 60: return f"{s}s fa" if s < 3600: return f"{s // 60}m{s % 60:02d}s fa" return f"{s // 3600}h{(s % 3600) // 60:02d}m fa" _MAIN_KB = { "inline_keyboard": [ [{"text": "🚀 Nuovo Task", "callback_data": "agent"}, {"text": "📊 Status", "callback_data": "tgw_status"}], [{"text": "📋 Tasks", "callback_data": "tgw_tasks"}, {"text": "🩺 Health", "callback_data": "tgw_health"}], [{"text": "🔧 AutoFix", "callback_data": "tgw_autofix"}, {"text": "🧠 Chiedi AI", "callback_data": "tgw_ask"}], [{"text": "📊 Bench", "callback_data": "tgw_bench"}, {"text": "🏆 Score", "callback_data": "tgw_score"}], [{"text": "📡 Telemetria", "callback_data": "tgw_telemetry"}, {"text": "⚙️ Migliora", "callback_data": "tgw_improve"}], [{"text": "📋 Log errori", "callback_data": "tgw_logs"}, {"text": "❓ Aiuto", "callback_data": "tgw_help"}], [{"text": "🔗 Coord", "callback_data": "tgw_coord"}, {"text": "🔀 Git log", "callback_data": "tgw_git"}], [{"text": "🌐 Dashboard", "url": "https://agente-ai.pages.dev"}], ] } _WEBAPP_KB = { "inline_keyboard": [ [{"text": "🚀 Apri Dashboard", "web_app": {"url": "https://agente-ai.pages.dev"}}], [{"text": "🏠 Menu", "callback_data": "tgw_help"}], ] } _BACK_KB = { "inline_keyboard": [ [{"text": "🏠 Menu", "callback_data": "tgw_help"}, {"text": "📊 Status", "callback_data": "tgw_status"}], ] } # ── Bench cache + keyboard (GAP-TGB) ───────────────────────────────────────── # Salva l'ultimo run bench per chat_id → usato dai callback tgw_bench_fix/run _BENCH_CACHE: dict[int, dict] = {} _BENCH_ACTION_KB = { "inline_keyboard": [ [{"text": "🔧 Applica Fix", "callback_data": "tgw_bench_fix"}, {"text": "🔄 Riesegui", "callback_data": "tgw_bench_run"}], [{"text": "⚙️ Migliora", "callback_data": "tgw_improve"}], [{"text": "🏠 Menu", "callback_data": "tgw_help"}], ] } # ── Reply Keyboard persistente — navigazione principale ────────────────────── _REPLY_MAIN_KB = { "keyboard": [ [{"text": "🚀 Lancia Task"}, {"text": "📊 Stato"}, {"text": "🩺 Health"}], [{"text": "📈 Performance"}, {"text": "🛠 Dev Tools"}, {"text": "💬 Chiedi AI"}], ], "resize_keyboard": True, "persistent": True, "input_field_placeholder": "Scrivi un goal o scegli dal menu ↑", } # ── Sub-menu inline keyboards ───────────────────────────────────────────────── _TASK_MENU_KB = { "inline_keyboard": [ [{"text": "🤖 Nuovo Task", "callback_data": "tgw_do"}, {"text": "🔧 AutoFix", "callback_data": "tgw_autofix"}], [{"text": "⚙️ Migliora AI", "callback_data": "tgw_improve"}, {"text": "📋 Task recenti", "callback_data": "tgw_tasks"}], [{"text": "🧠 Briefing", "callback_data": "tgw_briefing"}, {"text": "📝 Salva Nota", "callback_data": "tgw_nota"}], [{"text": "🔍 Cerca web", "callback_data": "tgw_cerca"}, {"text": "🌤 Meteo", "callback_data": "tgw_meteo"}], ] } _STATUS_MENU_KB = { "inline_keyboard": [ [{"text": "📊 Daemon+Task", "callback_data": "tgw_status"}, {"text": "🔌 Provider AI", "callback_data": "tgw_providers"}], [{"text": "🔗 Coord sessioni", "callback_data": "tgw_coord"}, {"text": "🔀 Git log", "callback_data": "tgw_git"}], [{"text": "🌐 Dashboard", "url": "https://agente-ai.pages.dev"}], ] } _PERF_MENU_KB = { "inline_keyboard": [ [{"text": "📊 Benchmark", "callback_data": "tgw_bench"}, {"text": "🏆 Score", "callback_data": "tgw_score"}], [{"text": "📡 Telemetria", "callback_data": "tgw_telemetry"}, {"text": "⚙️ Migliora", "callback_data": "tgw_improve"}], [{"text": "🔧 Fix gap bench", "callback_data": "tgw_bench_fix"}], ] } _HEALTH_MENU_KB = { "inline_keyboard": [ [{"text": "🔍 Scan completo", "callback_data": "tgw_health"}, {"text": "📝 Log errori", "callback_data": "tgw_logs"}], [{"text": "📊 Status", "callback_data": "tgw_status"}, {"text": "🔌 Provider AI", "callback_data": "tgw_providers"}], ] } _DEV_MENU_KB = { "inline_keyboard": [ [{"text": "📸 Snapshot", "callback_data": "tgw_snap"}, {"text": "✅ Verify", "callback_data": "tgw_verify"}], [{"text": "💊 Heal", "callback_data": "tgw_heal"}, {"text": "📝 Log", "callback_data": "tgw_logs"}], [{"text": "🔀 Git commits", "callback_data": "tgw_git"}, {"text": "🏓 Ping", "callback_data": "tgw_ping"}], ] } # ── Command handlers ────────────────────────────────────────────────────────── async def _cmd_help(chat_id: int) -> None: """Menu principale: Reply Keyboard persistente + inline quick actions.""" await _tg_typing(chat_id) welcome = ( "👋 Sono Agente AI, il tuo assistente autonomo per lo sviluppo.\n\n" "Scrivi quello che vuoi fare — in italiano, liberamente.\n" "Analizzo, codifico, faccio push e ti aggiorno in tempo reale.\n\n" "💡 Qualche idea:\n" " analizza i bug in providers.py\n" " ottimizza le query Supabase più lente\n" " fai autofix degli errori nel log\n\n" "🌐 Dashboard completa →" ) await _tg_reply(chat_id, welcome, keyboard=_REPLY_MAIN_KB) await _tg_reply(chat_id, "⚡ Cosa vuoi fare?", keyboard=_MAIN_KB) async def _cmd_logs(chat_id: int, level: str = "WARNING") -> None: """Mostra ultimi log dal backend Railway filtrando per livello.""" import httpx as _hx railway_url = os.getenv("RAILWAY_URL", "https://ai-production-4c06.up.railway.app").rstrip("/") await _tg_reply(chat_id, f"📋 Log Railway{level.upper()}\n⏳ Fetching…") try: async with _hx.AsyncClient(timeout=10.0) as c: r = await c.get(f"{railway_url}/api/telegram/logs", params={"level": level.upper(), "n": 20}) data = r.json() if r.status_code == 200 else {} except Exception as exc: await _tg_reply(chat_id, "❌ Log non disponibili\n" + html.escape(str(exc)[:200]) + "\n" "Controlla Railway dashboard.", keyboard=_BACK_KB) return records = data.get("records", []) if not records: msg = ("✅ Nessun " + level.upper() + " nei log!\nSistema stabile." if level.upper() in ("WARNING","ERROR") else "📋 Log vuoti — nessun record disponibile") await _tg_reply(chat_id, msg, keyboard=_BACK_KB) return import datetime as _dt lines = [f"📋 Log ({data.get('count',0)} rec — {level.upper()})\n"] for rec in records[:15]: ts = _dt.datetime.fromtimestamp(rec.get("ts",0), tz=_dt.timezone.utc).strftime("%H:%M:%S") lvl = rec.get("level","?") lgr = rec.get("logger","").split(".")[-1][:18] msg = html.escape(str(rec.get("msg",""))[:100]) icon = "🔴" if lvl=="ERROR" else "🟡" if lvl=="WARNING" else "⚪" lines.append(f"{icon} {ts} [{lgr}] {msg}") await _tg_reply(chat_id, "\n".join(lines), keyboard=_BACK_KB) async def _cmd_status(chat_id: int) -> None: await _tg_typing(chat_id) try: from api.state import _agent_tasks, _loop_registry # noqa: F401 total = len(_agent_tasks) running = sum(1 for t in _agent_tasks.values() if t.get("status") == "RUNNING") success = sum(1 for t in _agent_tasks.values() if t.get("status") == "SUCCESS") error = sum(1 for t in _agent_tasks.values() if t.get("status") == "ERROR") queued = sum(1 for t in _agent_tasks.values() if t.get("status") == "QUEUED") # Supabase fallback: se in-memory è vuoto (restart backend) legge dal DB sb_line = "" if total == 0: try: from api.state import _sb if _sb: res = await asyncio.to_thread( lambda: _sb.table("agent_tasks") .select("status") .order("created_at", desc=True) .limit(50) .execute() ) rows = res.data or [] if rows: db_run = sum(1 for r in rows if r.get("status") == "RUNNING") db_done = sum(1 for r in rows if r.get("status") == "SUCCESS") db_err = sum(1 for r in rows if r.get("status") == "ERROR") sb_line = ( "\n📦 Supabase (ultimi 50): " + str(db_run) + " in corso / " + str(db_done) + " ok / " + str(db_err) + " err" + " (backend riavviato)" ) except Exception as _exc: _logger.debug("[telegram_webhook] silenced %s", type(_exc).__name__) # noqa: BLE001 from api.scheduler import _tasks as sched_tasks, _loop_task sched_ok = _loop_task is not None and not _loop_task.done() sched_pending = sum(1 for t in sched_tasks.values() if t.get("status") == "pending") sched_label = "✅ attivo" if sched_ok else "❌ fermo" ts_now = time.strftime("%Y-%m-%d %H:%M:%S") railway_url = os.getenv("RAILWAY_URL","https://ai-production-4c06.up.railway.app") ry_line = "" try: import httpx as _hx async with _hx.AsyncClient(timeout=4.0) as c: rv = await c.get(f"{railway_url}/api/info") if rv.status_code == 200: rj = rv.json() ry_line = ("\n🚂 Railway: v" + rj.get("version","?") + " — " + rj.get("sprint","")) except Exception: pass # ── NEW-1: HEAD git + ultimo commit ────────────────────────────────── # Chiama GitHub API con GITHUB_TOKEN (Railway env) — timeout 4s, silent fail. # Mostra: sha corto + prima riga commit message + età ("3h fa"). git_line = "" try: import httpx as _hx_g, datetime as _dt _gh_token = os.getenv("GITHUB_TOKEN", "").strip() _gh_repo = os.getenv("GITHUB_REPO", "Baida98/AI").strip() if _gh_token and _gh_repo: async with _hx_g.AsyncClient(timeout=4.0) as _gc: _gr = await _gc.get( f"https://api.github.com/repos/{_gh_repo}/commits/main", headers={"Authorization": f"Bearer {_gh_token}", "User-Agent": "agente-ai"}, params={"per_page": 1}, ) if _gr.status_code == 200: _cj = _gr.json() _sha = (_cj.get("sha") or "")[:7] _cmsg = ((_cj.get("commit") or {}).get("message") or "").split("\n")[0][:45] _date = ((_cj.get("commit") or {}).get("committer") or {}).get("date", "") _age = "" if _date: _ts = _dt.datetime.fromisoformat(_date.replace("Z", "+00:00")) _secs = int((_dt.datetime.now(_dt.timezone.utc) - _ts).total_seconds()) if _secs < 3600: _age = f"{_secs // 60}m fa" elif _secs < 86400: _age = f"{_secs // 3600}h fa" else: _age = f"{_secs // 86400}g fa" git_line = ( f"\n🔀 HEAD: {html.escape(_sha)}" f" {html.escape(_cmsg)} ({_age})" ) except Exception: pass # ── NEW-2: task live — goal + step corrente dal loop_registry ──────── # Per il primo task RUNNING: mostra goal + azione corrente (dal buffer SSE) # + tempo trascorso. Zero overhead se non c'è task in corso. live_line = "" try: _running_list = [t for t in _agent_tasks.values() if t.get("status") == "RUNNING"] if _running_list: import json as _lj _rt = _running_list[0] _tid = _rt.get("id") or _rt.get("task_id") or "" _goal_s = html.escape((_rt.get("goal") or "")[:38]) _elapsed = "" _ca = _rt.get("created_at", 0) if isinstance(_ca, int) and _ca > 0: _es = int(time.time() * 1000 - _ca) // 1000 _elapsed = f" · {_es // 60}m{_es % 60:02d}s" if _es >= 60 else f" · {_es}s" # Legge ultimo evento SSE dal buffer (action/type corrente) _act = "" _ebuf = (_loop_registry.get(_tid) or {}).get("event_buffer", []) for _ev in reversed(_ebuf[-30:]): try: _raw = _ev[6:] if _ev.startswith("data: ") else _ev _ed = _lj.loads(_raw) _a = _ed.get("action") or _ed.get("type") or "" if _a and _a not in ("ping", "connected", "keepalive"): _act = f" → {html.escape(str(_a)[:20])}" break except Exception: pass live_line = f"\n⚙️ Live: {_goal_s}{_act}{_elapsed}" except Exception: pass # ── NEW-3: coord mini — sessioni agent-coord attive ────────────────── # 1 riga: chi sta lavorando, su quale sprint, su quali file. # Timeout aggressivo 3s — /status deve essere veloce. coord_line = "" try: import httpx as _hx_c2, json as _jc2, time as _tc2 _SB_URL2 = os.getenv("SUPABASE_URL", "https://zwdoplodbdsxfrddoxmo.supabase.co").rstrip("/") _SB_KEY2 = os.getenv("SUPABASE_KEY", "") if _SB_KEY2: async with _hx_c2.AsyncClient(timeout=3.0) as _cc: _cr = await _cc.get( f"{_SB_URL2}/rest/v1/agent_tasks", params={"goal": "ilike.*__session__*", "select": "context"}, headers={"apikey": _SB_KEY2, "Authorization": f"Bearer {_SB_KEY2}"}, ) if _cr.is_success: _now_ms = int(_tc2.time() * 1000) _active = [] for _row in (_cr.json() or []): try: _ctx = _jc2.loads(_row.get("context") or "{}") if _now_ms - int(_ctx.get("lastHeartbeat", 0)) < 300_000: _active.append(_ctx) except Exception: pass if _active: _s = _active[0] _files = [f.split("/")[-1] for f in _s.get("claimedFiles", [])] _fstr = ", ".join(_files[:3]) or "—" _extra = f" (+{len(_active)-1})" if len(_active) > 1 else "" coord_line = ( f"\n🔗 Coord: {html.escape(_s.get('sessionName','?'))}" f" [{html.escape(_s.get('sprint','—'))}]" f" · {html.escape(_fstr)}{_extra}" ) else: coord_line = "\n🔗 Coord: nessuna sessione attiva" except Exception: pass # ── Icona salute sistema ────────────────────────────────────────────── if running > 0: _sys_icon, _sys_label = "⚙️", f"{running} task in esecuzione" elif error > 0 and success == 0 and total > 0: _sys_icon, _sys_label = "🔴", "ultimi task terminati con errore" elif total == 0: _sys_icon, _sys_label = "💤", "nessun task recente" else: _sys_icon, _sys_label = "✅", "tutto operativo" parts = [f"📊 Sistema {_sys_icon} — {_sys_label}", ""] if running or queued: parts.append(f"⚙️ In esecuzione: {running} · In coda: {queued}") if success or error or total: parts.append(f"✅ Completati: {success} · ❌ Errori: {error} · Totale: {total}") for _extra_line in [sb_line, ry_line, git_line, live_line, coord_line]: if _extra_line: parts.append(_extra_line) parts += [ "", f"🗓 Scheduler: {sched_label}" + (f" · {sched_pending} in coda" if sched_pending else ""), "", f"🕐 {ts_now} UTC", ] await _tg_reply(chat_id, "\n".join(p for p in parts if p is not None), keyboard=_MAIN_KB) except Exception as exc: await _tg_reply(chat_id, "⚠️ Errore lettura stato: " + html.escape(str(exc)[:200])) async def _cmd_tasks(chat_id: int) -> None: await _tg_typing(chat_id) STATUS_EMOJI = { "SUCCESS": "✅", "ERROR": "❌", "RUNNING": "⚙️", "QUEUED": "⏳", "CANCELLED": "🚫", } try: from api.state import _agent_tasks mem_tasks = sorted( _agent_tasks.values(), key=lambda t: t.get("created_at", 0), reverse=True, )[:8] # Supabase fallback se memoria vuota if not mem_tasks: try: from api.state import _sb if _sb: res = await asyncio.to_thread( lambda: _sb.table("agent_tasks") .select("task_id,goal,status,created_at") .order("created_at", desc=True) .limit(8) .execute() ) rows = res.data or [] if rows: lines = ["📋 Ultimi task (Supabase) ⚠️ backend riavviato\n"] for r in rows: em = STATUS_EMOJI.get(r.get("status", ""), "•") gol = html.escape((r.get("goal") or "")[:55]) tid = html.escape(str(r.get("task_id") or "")[:8]) ts = r.get("created_at", 0) age = _fmt_elapsed(ts) if isinstance(ts, int) and ts > 1_000_000 else "?" lines.append(em + " " + tid + " " + gol + " " + age + "") return await _tg_reply(chat_id, "\n".join(lines)) except Exception as _exc: _logger.debug("[telegram_webhook] silenced %s", type(_exc).__name__) # noqa: BLE001 return await _tg_reply(chat_id, "📋 Nessun task recente in memoria.") _ST_LABEL = { "SUCCESS": "completato", "ERROR": "fallito", "RUNNING": "in corso", "QUEUED": "in attesa", "CANCELLED": "annullato", } lines = ["📋 Ultimi task\n"] for t in mem_tasks: st = t.get("status", "?") em = STATUS_EMOJI.get(st, "•") gol = html.escape((t.get("goal") or "")[:65]) tid = html.escape(str(t.get("id") or t.get("task_id") or "")[:8]) ca = t.get("created_at", 0) age = _fmt_elapsed(ca) if isinstance(ca, int) and ca > 1_000_000 else "?" lbl = _ST_LABEL.get(st, st.lower()) lines.append(f"{em} {gol}\n {lbl} · {age} · {tid}") await _tg_reply(chat_id, "\n".join(lines)) except Exception as exc: await _tg_reply(chat_id, "⚠️ " + html.escape(str(exc)[:200])) async def _cmd_do(chat_id: int, goal: str) -> None: """Lancia task AI con streaming progressivo via editMessageText. Flusso: invia msg iniziale → salva message_id → on_step accumula token → edit throttled ogni 1.5s → flush finale con output + keyboard. Rate-limit sicuro: max ~40 edit/min totali, Telegram consente 20 edit/min/chat. """ if not goal.strip(): await _tg_reply(chat_id, "⚠️ Uso: /do <il tuo goal>") return await _tg_typing(chat_id) msg_id = await _tg_send( chat_id, "🚀 Avvio task…\n\n" "🎯 " + html.escape(goal[:200]) + "\n\n" "⏳ Sto elaborando, un momento…", ) _buf: list[str] = [] _last_edit: list[float] = [0.0] # list per mutabilità in closure _EDIT_INTERVAL = 1.5 _MAX_LEN = 3600 async def _flush(final: bool = False) -> None: if not msg_id: return content = "".join(_buf).strip() if not content: return now = time.monotonic() if not final and (now - _last_edit[0]) < _EDIT_INTERVAL: return prefix = "🧠 Risposta AI\nGoal: " + html.escape(goal[:80]) + "\n\n" suffix = "" if final else "\n⏳" body = html.escape(content[: _MAX_LEN - len(prefix) - len(suffix)]) await _tg_edit(chat_id, msg_id, prefix + body + suffix) _last_edit[0] = time.monotonic() async def _on_step(event: dict) -> None: if event.get("action") == "text_chunk": tok = event.get("token", "") if tok: _buf.append(tok) await _flush(final=False) try: from agents.unified_loop import UnifiedAgentLoop from api.state import _get_ai_client, _get_mem_manager_async, _get_executor, _get_planner client = _get_ai_client() memory = await _get_mem_manager_async() executor = _get_executor() planner = _get_planner() try: from agents.critic import Critic from agents.response_verifier import ResponseVerifier critic = Critic(llm_client=client) verifier = ResponseVerifier() except Exception: critic = verifier = None loop = UnifiedAgentLoop( llm_client=client, critic=critic, verifier=verifier, memory=memory, executor=executor, planner=planner, ) result = await asyncio.wait_for( loop.run(goal=goal, context="", max_steps=8, on_step=_on_step), timeout=120.0, ) output = result.get("output", "") if isinstance(result, dict) else str(result) await _flush(final=True) if msg_id and output: _out_str = str(output) _out_esc = html.escape(_out_str[:3000]) # Usa blockquote espandibile per output lunghi (Bot API 7.4+, Jul 2024) _out_body = ( f"
{_out_esc}
" if len(_out_str) > 400 else _out_esc ) await _tg_edit( chat_id, msg_id, "✅ Fatto!\n\n" "🎯 " + html.escape(goal[:100]) + "\n\n" + _out_body, keyboard=_MAIN_KB, ) await _tg_react(chat_id, msg_id, "🎉") try: from .telegram_notify import notify_task_done task_id = result.get("task_id", "tgw-do") if isinstance(result, dict) else "tgw-do" await notify_task_done(str(task_id), goal, str(output)) except Exception: if not msg_id: _fb_esc = html.escape(str(output)[:800]) _fb_body = ( f"
{_fb_esc}
" if len(str(output)) > 400 else _fb_esc ) await _tg_reply( chat_id, "✅ Fatto!\n\n" "🎯 " + html.escape(goal[:80]) + "\n\n" + _fb_body, ) except asyncio.TimeoutError: await _flush(final=True) err = ("⏱ Timeout — task >120s.\nGoal: " + html.escape(goal[:120]) + "\n\nUsa il pannello web.") if msg_id: await _tg_edit(chat_id, msg_id, err, keyboard=_MAIN_KB) else: await _tg_reply(chat_id, err) except Exception as exc: await _flush(final=True) err = "❌ Errore\n" + html.escape(str(exc)[:300]) + "" if msg_id: await _tg_edit(chat_id, msg_id, err, keyboard=_MAIN_KB) else: await _tg_reply(chat_id, err) async def _cmd_autofix(chat_id: int, hint: str = "") -> None: """Comando /autofix — legge errori dai log backend, genera patch via AI, pusha su GitHub. Flusso: 1. GET /api/telegram/logs?level=ERROR — raccoglie ultimi errori 2. AI loop con streaming (on_step) — analizza e genera patch 3. Parsa blocco ```autofix\nFILE: path\n---\ncontent``` dall'output AI 4. Git Data API blob->tree->commit->PATCH ref — push automatico 5. Riporta commit SHA al chat + link GitHub Env vars Railway: GITHUB_TOKEN (gia' presente), GITHUB_REPO, GITHUB_BRANCH. """ import httpx, base64 as _b64, json as _json await _tg_typing(chat_id, "upload_document") msg_id = await _tg_send( chat_id, "🔧 AutoFix avviato\n\n" + (f"Hint: {html.escape(hint[:100])}\n\n" if hint else "") + "⏳ Step 1/4 — Lettura log errori…", ) async def _edit(text: str, final: bool = False) -> None: if msg_id: await _tg_edit(chat_id, msg_id, text, keyboard=_MAIN_KB if final else None) # ── Step 1: leggi errori dal log endpoint ───────────────────────────────────────── railway_url = os.getenv("RAILWAY_URL","https://ai-production-4c06.up.railway.app").rstrip("/") try: async with httpx.AsyncClient(timeout=10.0) as c: resp = await c.get(f"{railway_url}/api/telegram/logs", params={"level": "ERROR", "n": 30}) log_data = resp.json() if resp.status_code == 200 else {} except Exception as e: log_data = {} _logger.warning("autofix: log fetch: %s", e) records = log_data.get("records", []) if not records and not hint: await _edit( "✅ AutoFix\n\n" "🟢 Nessun errore nei log recenti!\n\n" "Sistema stabile. Usa /autofix <descrizione bug> " "per fix su un errore specifico.", final=True, ) return log_lines = "\n".join( f"[{r.get('level','?')}] {r.get('logger','?')}: {r.get('msg','')}" for r in records[:15] ) if records else f"(nessun errore nei log — hint: {hint})" log_preview = "\n".join( f"• [{r.get('level','?')}] {r.get('logger','?')}: {r.get('msg','')[:80]}" for r in records[:5] ) or f"hint: {html.escape(hint[:100])}" await _edit( "🔧 AutoFix\n\n" "📋 Errori trovati:\n" + log_preview + "\n\n⏳ Step 2/4 — Analisi AI in corso…", ) # ── Step 2: AI loop genera patch ───────────────────────────────────────────────────── BT3 = "```" # triple backtick — non usare literal per evitare syntax issues goal = ( "Sei un senior engineer. Analizza questi errori dal log backend Python " "e genera un patch preciso per risolvere il problema principale.\n\n" "LOG ERRORI:\n" + log_lines[:2000] + (f"\n\nHINT UTENTE: {hint}" if hint else "") + "\n\n" "FORMATO RISPOSTA RICHIESTO (tassativo):\n" + BT3 + "autofix\n" + "FILE: backend/api/.py\n" + "---\n" + "\n" + BT3 + "\n\n" "Regole:\n" "1. Scrivi SOLO il blocco con FILE: e contenuto completo (non un diff)\n" "2. Per piu' file includi un blocco per file\n" "3. Se non riesci a determinare il file, scrivi FILE: UNKNOWN e spiega" ) context = f"Backend: {railway_url} Repo: {os.getenv('GITHUB_REPO','Baida98/AI')}" _buf: list[str] = [] _last: list[float] = [0.0] async def _on_step(event: dict) -> None: if event.get("action") == "text_chunk": tok = event.get("token", "") if tok: _buf.append(tok) now = time.monotonic() if now - _last[0] > 2.0: preview = "".join(_buf).strip()[-300:] await _edit( "🔧 AutoFix — Step 2/4 AI analizza…\n\n" "" + html.escape(preview) + "\n\n⏳" ) _last[0] = now try: from agents.unified_loop import UnifiedAgentLoop from api.state import _get_ai_client, _get_mem_manager_async, _get_executor, _get_planner client = _get_ai_client() memory = await _get_mem_manager_async() executor = _get_executor() planner = _get_planner() try: from agents.critic import Critic from agents.response_verifier import ResponseVerifier critic = Critic(llm_client=client) verifier = ResponseVerifier() except Exception: critic = verifier = None loop = UnifiedAgentLoop( llm_client=client, critic=critic, verifier=verifier, memory=memory, executor=executor, planner=planner, ) result = await asyncio.wait_for( loop.run(goal=goal, context=context, max_steps=5, on_step=_on_step), timeout=150.0, ) ai_output = result.get("output", "") if isinstance(result, dict) else str(result) except asyncio.TimeoutError: await _edit("⏱ AutoFix timeout — AI non ha risposto in 150s.\n\n" "Riprova con un hint più specifico.", final=True) return except Exception as exc: await _edit("❌ AutoFix — errore AI\n" + html.escape(str(exc)[:300]) + "", final=True) return # ── Step 3: parsa blocco autofix dall'output AI ─────────────────────────────────── await _edit( "🔧 AutoFix\n\n⏳ Step 3/4 — Parsing patch…\n\n" "" + html.escape(ai_output.strip()[-300:]) + "" ) import re as _re BLOCK_RE = _re.compile( r"```autofix\s*\nFILE:\s*(\S+)\s*\n---\s*\n([\s\S]*?)```", _re.MULTILINE, ) patches: list[tuple[str, str]] = [ (m.group(1).strip(), m.group(2)) for m in BLOCK_RE.finditer(ai_output) if m.group(1).strip() not in ("", "UNKNOWN") and m.group(2).strip() ] if not patches: await _edit( "⚠️ AutoFix — nessuna patch parsata\n\n" "Output AI:\n" + html.escape(ai_output.strip()[:600]) + "\n\n" "L'AI non ha prodotto un blocco autofix valido.\n" "Prova: /autofix descrizione precisa del bug", final=True, ) return # ── Step 4: push su GitHub via Git Data API ───────────────────────────────────────── gh_token = os.getenv("GITHUB_TOKEN", "") gh_repo = os.getenv("GITHUB_REPO", "Baida98/AI") gh_branch = os.getenv("GITHUB_BRANCH", "main") if not gh_token: await _edit( "❌ AutoFix — GITHUB_TOKEN non configurato\n\n" "Aggiungi GITHUB_TOKEN nelle Railway env vars.", final=True, ) return await _edit( "🔧 AutoFix\n\n" f"⏳ Step 4/4 — Push {len(patches)} file su GitHub…\n\n" + "\n".join(f"• {p[0]}" for p in patches[:6]) ) gh_hdr = { "Authorization": f"token {gh_token}", "Accept": "application/vnd.github.v3+json", "Content-Type": "application/json", "User-Agent": "agente-ai-autofix", } gh_base = f"https://api.github.com/repos/{gh_repo}" try: async with httpx.AsyncClient(timeout=20.0) as c: r = await c.get(f"{gh_base}/git/refs/heads/{gh_branch}", headers=gh_hdr) r.raise_for_status() h_sha = r.json()["object"]["sha"] r = await c.get(f"{gh_base}/git/commits/{h_sha}", headers=gh_hdr) r.raise_for_status() t_sha = r.json()["tree"]["sha"] tree_items = [] for fpath, content in patches: b64 = _b64.b64encode(content.encode()).decode() r = await c.post(f"{gh_base}/git/blobs", headers=gh_hdr, json={"content": b64, "encoding": "base64"}) r.raise_for_status() tree_items.append({"path": fpath, "mode": "100644", "type": "blob", "sha": r.json()["sha"]}) r = await c.post(f"{gh_base}/git/trees", headers=gh_hdr, json={"base_tree": t_sha, "tree": tree_items}) r.raise_for_status() new_t = r.json()["sha"] err_summary = ("; ".join(rc.get("msg","")[:60] for rc in records[:2]) or hint[:60] or "autofix via /autofix command") commit_msg = ( f"fix(autofix): {err_summary}\n\n" f"Generato da /autofix — {len(patches)} file patchati\n" + "Patch: " + ", ".join(p[0] for p in patches) ) r = await c.post(f"{gh_base}/git/commits", headers=gh_hdr, json={"message": commit_msg, "tree": new_t, "parents": [h_sha]}) r.raise_for_status() c_sha = r.json()["sha"] r = await c.patch(f"{gh_base}/git/refs/heads/{gh_branch}", headers=gh_hdr, json={"sha": c_sha, "force": False}) r.raise_for_status() files_list = "\n".join(f"• {p[0]}" for p in patches[:6]) await _edit( "✅ AutoFix completato!\n\n" f"Commit: {c_sha[:10]}\n" f"Branch: {gh_branch}\n" f"File patchati:\n{files_list}\n\n" "U0001f680 Railway deploy: avviato automaticamente\n" f"U0001f517 Vedi commit", final=True, ) _logger.info("autofix: pushed %s — %d files", c_sha[:10], len(patches)) except Exception as exc: await _edit( "❌ AutoFix — push GitHub fallito\n" + html.escape(str(exc)[:400]) + "\n\n" "Verifica GITHUB_TOKEN nelle Railway env vars.", final=True, ) _logger.error("autofix: push failed: %s", exc) async def _cmd_nota(chat_id: int, text: str) -> None: """Salva nota rapida nella memoria persistente dell'agente.""" goal = ( f"Salva questa nota nella tua memoria persistente usando il tool remember: " f"«{text}». Poi confermami con '✅ Nota salvata' e ripeti il testo della nota." ) await _cmd_do(chat_id, goal) async def _cmd_cerca(chat_id: int, query: str) -> None: """Ricerca web + sintesi AI in italiano.""" goal = ( f"Cerca su web: {query}. " "Usa web_search e sintetizza i risultati più rilevanti in 4-5 bullet points concisi " "in italiano. Per ogni punto includi la fonte (dominio) tra parentesi." ) await _cmd_do(chat_id, goal) async def _cmd_meteo(chat_id: int, city: str) -> None: """Meteo per città specifica via tool get_weather.""" goal = ( f"Usa il tool get_weather per {city} e dimmi: temperatura attuale, " "condizioni meteo, umidità, previsioni per le prossime ore. Risposta concisa in italiano." ) await _cmd_do(chat_id, goal) async def _cmd_riepilogo(chat_id: int) -> None: """Briefing completo: task recenti + deploy + score + prossimi passi. GAP-TGB: usa benchmark_handler.get_smart_summary per briefing strutturato con analisi gap da ultimo report. Fallback: agent-loop se import fallisce. """ try: from .benchmark_handler import get_smart_summary as _gsummary text = await _gsummary(chat_id) await _tg_reply(chat_id, text, keyboard=_BENCH_ACTION_KB) except Exception as _exc: _logger.debug("benchmark_handler fallback: %s", _exc) goal = ( "Dammi un briefing completo dello stato attuale come assistente personale. " "Struttura ESATTA con questi header ## obbligatori:\n" "## 📋 Task recenti (usa recall per recuperare gli ultimi 5)\n" "## 🚀 Deploy status (CF Pages + Railway + HF Space — dati reali via tool)\n" "## 📊 Score AI (ultimo benchmark disponibile)\n" "## ⚠️ Problemi aperti (errori, timeout, regressioni)\n" "## ✅ Prossimi passi (3 priorità concrete)\n" "Usa i tool per dati reali — mai inventare status." ) await _cmd_do(chat_id, goal) async def _cmd_scan_now(chat_id: int) -> None: """Health-check via health_full() — AI + Supabase + Telegram.""" await _tg_reply(chat_id, "🔍 Scan in corso…") try: from .providers import health_full h = await health_full() ai = h.get("ai", {}) sb = h.get("supabase", {}) tg = h.get("telegram", {}) bk = h.get("backend", {}) ai_ok = "✅" if ai.get("ok") else "❌" sb_ok = "✅" if sb.get("ok") else "❌" tg_ok = "✅" if tg.get("ok") else "❌" over = "✅" if h.get("ok") else "⚠️" sb_msg = "ok" if sb.get("ok") else str(sb.get("error", "?"))[:40] tg_msg = ("@" + str(tg.get("username", ""))) if tg.get("username") else str(tg.get("error", "?"))[:40] avail = str(ai.get("available", 0)) + "/" + str(ai.get("total", 0)) best = str(ai.get("best", "?")) ntasks = str(bk.get("active_tasks", 0)) ms_str = str(h.get("elapsed_ms", "?")) st_str = str(h.get("status", "?")) parts = [ over + " Health Check — " + st_str, "", ai_ok + " AI — " + avail + " provider | best: " + best, sb_ok + " Supabase — " + sb_msg, tg_ok + " Telegram — " + tg_msg, "📦 Task attivi: " + ntasks + " 🕒 " + ms_str + "ms", ] await _tg_reply(chat_id, "\n".join(parts), keyboard=_MAIN_KB) except Exception as exc: await _tg_reply( chat_id, "❌ Scan errore\n" + html.escape(str(exc)[:300]) + "", ) async def _cmd_coord(chat_id: int) -> None: """🔗 /coord — sessioni agent-coord attive e file claimati (sola lettura). Legge la tabella Supabase 'agent_tasks' dove le sessioni agent-coord.mjs persistono con goal='__session__'. TTL = 5 minuti senza heartbeat. Non modifica nulla — pura lettura. Non interferisce con il lavoro agente. """ import httpx as _hx_c, time as _tc, json as _jc SB_URL = os.getenv("SUPABASE_URL", "https://zwdoplodbdsxfrddoxmo.supabase.co").rstrip("/") SB_KEY = os.getenv("SUPABASE_KEY", "") if not SB_KEY: await _tg_reply(chat_id, "⚠️ SUPABASE_KEY non configurata su Railway.") return await _tg_reply(chat_id, "🔗 Agent Coord — lettura sessioni…") try: async with _hx_c.AsyncClient(timeout=8.0) as _c: r = await _c.get( f"{SB_URL}/rest/v1/agent_tasks", params={"goal": "ilike.*__session__*", "select": "goal,context,updated_at"}, headers={ "apikey": SB_KEY, "Authorization": f"Bearer {SB_KEY}", "Content-Type": "application/json", }, ) if not r.is_success: await _tg_reply(chat_id, f"❌ Supabase {r.status_code}: {r.text[:200]}") return rows = r.json() or [] except Exception as exc: await _tg_reply(chat_id, "❌ " + html.escape(str(exc)[:200]) + "") return SESSION_TTL_MS = 5 * 60 * 1000 # 5 minuti senza heartbeat = morta now_ms = int(_tc.time() * 1000) active, stale = [], [] for row in rows: try: ctx = _jc.loads(row.get("context") or "{}") except Exception: continue hb = int(ctx.get("lastHeartbeat", 0)) age_ms = now_ms - hb entry = { "name": ctx.get("sessionName", "?"), "sprint": ctx.get("sprint", "—"), "files": ctx.get("claimedFiles", []), "age_ms": age_ms, } (active if age_ms < SESSION_TTL_MS else stale).append(entry) if not rows: await _tg_reply(chat_id, "🔗 Agent Coord\n\n" "ℹ️ Nessuna sessione registrata.\nTutti i file sono liberi.", keyboard=_BACK_KB) return def _age_str(ms: int) -> str: s = ms // 1000 return f"{s // 60}m{s % 60:02d}s" if s >= 60 else f"{s}s" out = ["🔗 Agent Coord", ""] if active: out.append(f"✅ Attive ({len(active)})") for s in active: fnames = ", ".join( "" + html.escape(f.split("/")[-1]) + "" for f in s["files"] ) or "nessun file" out.append( f" {html.escape(s['name'])} " f"[{html.escape(s['sprint'])}] · {_age_str(s['age_ms'])} fa" ) out.append(f" 📎 {fnames}") else: out.append("✅ Nessuna sessione attiva — file liberi") if stale: out.append("") out.append(f"⚫ Inattive >5min ({len(stale)})") for s in stale: out.append(f" {html.escape(s['name'])} · {_age_str(s['age_ms'])} fa") await _tg_reply(chat_id, "\n".join(out), keyboard=_BACK_KB) async def _cmd_git(chat_id: int, n: int = 5) -> None: """🔀 /git [n] — ultimi N commit su main con SHA, messaggio, autore, eta e link diff. Chiama GitHub API (GITHUB_TOKEN Railway). Default: ultimi 5 commit. Max: 10. Timeout 6s. Silent fail se GITHUB_TOKEN assente. """ import httpx as _hx_git2, datetime as _dt2 _gh_token = os.getenv("GITHUB_TOKEN", "").strip() _gh_repo = os.getenv("GITHUB_REPO", "Baida98/AI").strip() if not _gh_token: await _tg_reply(chat_id, "⚠️ GITHUB_TOKEN non configurato su Railway — /git non disponibile.") return n = max(1, min(n, 10)) await _tg_reply(chat_id, f"🔀 Git log — ultimi {n} commit…") try: async with _hx_git2.AsyncClient(timeout=6.0) as _gc: _gr = await _gc.get( f"https://api.github.com/repos/{_gh_repo}/commits", headers={"Authorization": f"Bearer {_gh_token}", "User-Agent": "agente-ai"}, params={"sha": "main", "per_page": str(n)}, ) if _gr.status_code != 200: await _tg_reply(chat_id, f"❌ GitHub API {_gr.status_code}: {html.escape(_gr.text[:200])}") return commits = _gr.json() or [] except Exception as exc: await _tg_reply(chat_id, "❌ " + html.escape(str(exc)[:200]) + "") return if not commits: await _tg_reply(chat_id, "ℹ️ Nessun commit trovato.", keyboard=_BACK_KB) return _now_utc = _dt2.datetime.now(_dt2.timezone.utc) def _age(iso: str) -> str: try: _ts = _dt2.datetime.fromisoformat(iso.replace("Z", "+00:00")) _secs = int((_now_utc - _ts).total_seconds()) if _secs < 3600: return f"{_secs // 60}m fa" if _secs < 86400: return f"{_secs // 3600}h fa" if _secs < 604800: return f"{_secs // 86400}g fa" return _ts.strftime("%d/%m") except Exception: return "?" lines = [f"🔀 Git log — main ({_gh_repo})\n"] for c in commits: _sha = (c.get("sha") or "")[:7] _commit = c.get("commit") or {} _msg_raw = (_commit.get("message") or "").split("\n")[0][:52] _msg = html.escape(_msg_raw) _author = html.escape((_commit.get("author") or {}).get("name", "?")[:18]) _date = (_commit.get("committer") or {}).get("date", "") _when = _age(_date) _url = f"https://github.com/{_gh_repo}/commit/{c.get('sha','')}" lines.append( f"{html.escape(_sha)} {_msg}\n" f" {_author} · {_when} " f'diff →' ) await _tg_reply(chat_id, "\n".join(lines), keyboard=_BACK_KB) async def _cmd_telemetry(chat_id: int) -> None: """📡 Metriche runtime live: /api/telemetry + /debug/timing da Railway.""" import httpx as _hx_t rw_url = os.getenv("RAILWAY_URL", "https://ai-production-4c06.up.railway.app").rstrip("/") await _tg_reply(chat_id, "⏳ Telemetria — interrogo Railway…") try: async with _hx_t.AsyncClient(timeout=8.0) as _c: tel_r, tim_r = await asyncio.gather( _c.get(f"{rw_url}/api/telemetry"), _c.get(f"{rw_url}/debug/timing"), return_exceptions=True, ) except Exception as e: await _tg_reply(chat_id, f"❌ Railway non raggiungibile\n{html.escape(str(e)[:120])}", keyboard=_BACK_KB) return tel_d = tel_r.json() if not isinstance(tel_r, Exception) and tel_r.status_code == 200 else {} tim_d = tim_r.json() if not isinstance(tim_r, Exception) and tim_r.status_code == 200 else {} msg = "📡 Telemetria Runtime\n\n" timing = tel_d.get("timing", {}) if timing: msg += "⏱ Latenze per fase (avg / p90 / n):\n" for phase, v in list(timing.items())[:10]: avg_v = v.get("avg", 0); p90_v = v.get("p90", 0); n_v = v.get("n", 0) icon = "🟢" if avg_v < 3000 else "🟡" if avg_v < 10000 else "🔴" msg += f"{icon} {phase[:18]:<18} {avg_v:>7}ms p90:{p90_v:>7}ms ×{n_v}\n" msg += "\n" else: msg += "⚠️ Nessun dato timing — backend idle o prima run.\n" repair = tel_d.get("repair", tim_d.get("repair_stats", {})) if repair: msg += "\n🔧 Quality & Repair:\n" for k, v in list(repair.items())[:10]: msg += f" {k[:22]:<22} {v:>6}\n" msg += "\n" ts_d = tim_d.get("timing_stats", {}) if ts_d: msg += "\n📊 Breakdown /debug/timing:\n" for label, v in list(ts_d.items())[:8]: avg_v = v.get("avg") or 0; icon = "🟢" if avg_v < 3000 else "🟡" if avg_v < 10000 else "🔴" msg += f"{icon} {label[:18]:<18} {avg_v:>8}ms ×{v.get('count',0)}\n" msg += "" if len(msg) > 4000: msg = msg[:4000] + "…" await _tg_reply(chat_id, msg, keyboard=_BACK_KB) async def _cmd_score(chat_id: int) -> None: """🏆 Score card dettagliata — chart + ranking 4 competitor + nodes + gaps + runtime telemetry.""" import httpx as _hx_sc, base64 as _b64_sc, json as _j_sc, urllib.parse as _ul_sc gh_token = os.getenv("GITHUB_TOKEN", "").strip() rw_url = os.getenv("RAILWAY_URL", "https://ai-production-4c06.up.railway.app").rstrip("/") await _tg_reply(chat_id, "⏳ Score — carico report + metriche runtime…") report: dict | None = None if gh_token: try: async with _hx_sc.AsyncClient(timeout=8.0) as _c: _r = await _c.get( "https://api.github.com/repos/Baida98/AI/contents/benchmark-report.json?ref=main", headers={"Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "AgenteAI-Bot"}, ) if _r.status_code == 200: report = _j_sc.loads(_b64_sc.b64decode(_r.json()["content"]).decode()) except Exception as _e: _logger.warning("cmd_score fetch: %s", _e) if not report: await _tg_reply(chat_id, "❌ Score — benchmark-report.json non trovato.\n" "Avvia prima /bench per generare i dati.", keyboard=_BENCH_ACTION_KB) return s = report.get("summary", {}) tasks = report.get("tasks", []) gaps = report.get("gapCards", []) nodes = report.get("orchestrationNodeMap", {}) ts = (report.get("timestamp") or "")[:16].replace("T", " ") seed = report.get("seed", "?") ver = report.get("version", "?") replay = report.get("replayCli", f"node benchmark-extended.mjs --seed {seed}") mft = s.get("mft") gap_cnt = s.get("gapCount", len(gaps)) verdict = s.get("verdict", "").replace("_", " ") avg_ai = s.get("avgScore", 0) avg_rpl = s.get("avgReplit", 57.9) avg_cur = s.get("avgCursor", 64.1) avg_dev = s.get("avgDevin", 70.1) avg_mns = s.get("avgManus", 71.2) w_rpl = s.get("wins_replit", "?") w_dev = s.get("wins_devin", "?") w_mns = s.get("wins_manus", "?") # Runtime telemetria da Railway (best-effort) rt_timing: dict = {} rt_repair: dict = {} try: async with _hx_sc.AsyncClient(timeout=5.0) as _c: _tr = await _c.get(f"{rw_url}/api/telemetry") if _tr.status_code == 200: _td = _tr.json() rt_timing = _td.get("timing", {}) rt_repair = _td.get("repair", {}) except Exception: pass # ── Chart per categoria ─────────────────────────────────────── cat_map: dict = {} for t in tasks: cat = (t.get("cat") or "other").replace("_", " ")[:14] cat_map.setdefault(cat, []).append(float(t.get("score", 0))) labels_ = list(cat_map.keys()) scores_ = [round(sum(v)/len(v)) for v in cat_map.values()] colors_ = ["#4CAF50" if sc >= avg_rpl else "#FF9800" if sc >= 50 else "#F44336" for sc in scores_] chart_cfg = { "type": "horizontalBar", "data": {"labels": labels_, "datasets": [ {"label": "Agente AI", "data": scores_, "backgroundColor": colors_, "borderWidth": 1}, {"label": f"Replit {avg_rpl}", "data": [avg_rpl]*len(labels_), "type": "line", "borderColor": "#2196F3", "borderDash": [5,3], "pointRadius": 0, "fill": False, "borderWidth": 2}, {"label": f"Devin {avg_dev}", "data": [avg_dev]*len(labels_), "type": "line", "borderColor": "#FF9800", "borderDash": [5,3], "pointRadius": 0, "fill": False, "borderWidth": 2}, {"label": f"Manus {avg_mns}", "data": [avg_mns]*len(labels_), "type": "line", "borderColor": "#9C27B0", "borderDash": [5,3], "pointRadius": 0, "fill": False, "borderWidth": 2}, ]}, "options": { "title": {"display": True, "text": f"Agente AI {avg_ai}% | Replit {avg_rpl}% | Devin {avg_dev}% | Manus {avg_mns}%"}, "scales": {"xAxes": [{"ticks": {"min": 0, "max": 100, "stepSize": 20}}]}, "legend": {"display": True, "position": "bottom"}, "plugins": {"datalabels": {"display": False}}, }, } chart_url = ("https://quickchart.io/chart?c=" + _ul_sc.quote(_j_sc.dumps(chart_cfg, separators=(",",":"))) + "&width=760&height=440&backgroundColor=white") # ── Caption messaggio 1 (foto, max 1024) ───────────────────── bar_g = "█" * round(avg_ai/10) + "░" * (10 - round(avg_ai/10)) d_rpl = round(avg_ai - avg_rpl); s_rpl = ("+" if d_rpl >= 0 else "") + str(d_rpl) d_dev = round(avg_ai - avg_dev); s_dev = ("+" if d_dev >= 0 else "") + str(d_dev) d_mns = round(avg_ai - avg_mns); s_mns = ("+" if d_mns >= 0 else "") + str(d_mns) d_cur = round(avg_ai - avg_cur); s_cur = ("+" if d_cur >= 0 else "") + str(d_cur) caption = f"🏆 Score — {ts} UTC v{ver}\n" caption += f"{bar_g} {avg_ai}% {verdict}\n\n" caption += f"{'Modello':<10} {'Score':>5} {'Δ':>4} Wins\n" caption += f"{'Agente AI':<10} {str(avg_ai)+'%':>5} {'─':>4} ─\n" caption += f"{'Replit':<10} {str(avg_rpl)+'%':>5} {s_rpl:>4} {w_rpl}/10\n" caption += f"{'Cursor':<10} {str(avg_cur)+'%':>5} {s_cur:>4} ─\n" caption += f"{'Devin':<10} {str(avg_dev)+'%':>5} {s_dev:>4} {w_dev}/10\n" caption += f"{'Manus':<10} {str(avg_mns)+'%':>5} {s_mns:>4} {w_mns}/10\n\n" for cat, vals in sorted(cat_map.items(), key=lambda x: -sum(x[1])/len(x[1])): sc = round(sum(vals)/len(vals)) bar = "█" * round(sc/10) + "░" * (10 - round(sc/10)) d = round(sc - avg_rpl) vs = ("+" if d >= 0 else "") + str(d) icon = "✅" if sc >= avg_rpl else "⚠️" if sc >= 50 else "❌" line = f"{cat[:12]:<12} {bar} {sc:>3}% {vs:>4} {icon}\n" if len(caption) + len(line) < 1010: caption += line caption += f"\n🎲 Seed {seed} MFT {mft}s Gaps: {gap_cnt}" await _tg_photo(chat_id, chart_url, caption=caption[:1024], keyboard=_BENCH_ACTION_KB) # ── Messaggio 2 — dettaglio completo ───────────────────────── det = "📊 Score — Dettaglio\n\n" # Orchestration nodes NODE_ICONS = {"planner":"🧠","executor":"⚙️","reasoner":"🔬", "recovery_manager":"🛡","robustness_layer":"🔒","memory_module":"💾"} if nodes: det += "⚡ Orchestration Nodes:\n" for nk, nv in nodes.items(): sr = str(nv.get("success_rate", "?")) lat = nv.get("avg_latency_s") ntsk = nv.get("tasks", "") try: icon = "✅" if float(sr.rstrip("%")) >= 60 else "⚠️" if float(sr.rstrip("%")) >= 30 else "❌" except Exception: icon = "❓" lat_s = f" {lat}s" if lat is not None else "" tsk_s = f" ×{ntsk}" if ntsk else "" det += f"{NODE_ICONS.get(nk,'•')} {nk[:20]:<20} {icon} {sr:>5}{lat_s}{tsk_s}\n" det += "\n" # Runtime telemetria (live da Railway) if rt_timing: det += "\n⏱ Runtime Latenze (live):\n" for phase, v in list(rt_timing.items())[:8]: avg_v = v.get("avg", 0); p90_v = v.get("p90", 0); n_v = v.get("n", 0) icon = "🟢" if avg_v < 3000 else "🟡" if avg_v < 10000 else "🔴" det += f"{icon} {phase[:16]:<16} avg:{avg_v:>7}ms p90:{p90_v:>7}ms ×{n_v}\n" det += "\n" if rt_repair: det += "🔧 Quality counters:\n" for k, v in list(rt_repair.items())[:6]: det += f" {k[:20]:<20} {v}\n" det += "\n" else: det += "ℹ️ Telemetria runtime non disponibile (Railway idle)\n" # Top 3 best + Top 3 worst sorted_tasks = sorted([t for t in tasks if t.get("score") is not None], key=lambda t: -t["score"]) if sorted_tasks: det += "\n🥇 Migliori task:\n" for t in sorted_tasks[:3]: ref_r = (t.get("ref") or {}).get("replit", avg_rpl) d_ = round(t["score"] - ref_r) ds_ = ("+" if d_ >= 0 else "") + str(d_) ms_ = f" {round(t['agentMs']/1000)}s" if t.get("agentMs") else "" det += f" {(t.get('id') or '?'):>3} {t['score']}% ({ds_} vsRpl){ms_}\n" det += f" {(t.get('label') or '')[:60]}\n" det += "\n⚠️ Task critici:\n" for t in sorted_tasks[-3:]: ref_r = (t.get("ref") or {}).get("replit", avg_rpl) ref_m = (t.get("ref") or {}).get("manus", avg_mns) d_ = round(t["score"] - ref_r) dm_ = round(t["score"] - ref_m) ds_ = ("+" if d_ >= 0 else "") + str(d_) dms_ = ("+" if dm_ >= 0 else "") + str(dm_) det += f" {(t.get('id') or '?'):>3} {t['score']}% vsRpl {ds_} vsManus {dms_}\n" det += f" {(t.get('label') or '')[:60]}\n" # Gap cards top 3 if gaps: det += "\n🔥 Gap da correggere:\n" for g in gaps[:3]: gicon = "🔴" if g.get("gravita") == "critical" else "🟡" det += f"{gicon} {(g.get('modulo_coinvolto') or '').replace('_',' ')} — {(g.get('causa_radice') or '')[:55]}\n" det += f" Fix: {(g.get('fix_immediato') or '')[:65]}\n" det += f" Atteso: {(g.get('beneficio_atteso') or '')[:40]}\n" det += f"\n{replay}" det += f"\n🌐 Dashboard →" _TG_MAX = 4000 if len(det) <= _TG_MAX: await _tg_reply(chat_id, det, keyboard=_BENCH_ACTION_KB) else: await _tg_reply(chat_id, det[:_TG_MAX]) await _tg_reply(chat_id, det[_TG_MAX:_TG_MAX*2][:_TG_MAX], keyboard=_BENCH_ACTION_KB) async def _cmd_bench(chat_id: int, mode: str = "default") -> None: """📊 Benchmark via bench.yml (benchmark-extended.mjs) + quickchart.io. GAP-TGB: workflow_dispatch su bench.yml — usa benchmark-extended.mjs (20 categorie, seed canonico 1337, tutte le fix v5). Risultati inviati via Telegram da ab-bench.mjs --notify al completamento. """ gh_token = os.getenv("GITHUB_TOKEN", "").strip() # ── Tenta fetch ultimo run completato da GitHub Actions artifact ───────── last_report: dict | None = None if gh_token: try: import httpx as _hx async with _hx.AsyncClient(timeout=8.0) as _c: _r = await _c.get( "https://api.github.com/repos/Baida98/AI/actions/workflows/bench.yml/runs" "?status=completed&per_page=1", headers={"Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "AgenteAI-Bot"}, ) if _r.status_code == 200: _runs = _r.json().get("workflow_runs", []) if _runs: last_report = { "run_id": _runs[0]["id"], "run_url": _runs[0]["html_url"], "conclusion":_runs[0].get("conclusion","?"), "updated": _runs[0].get("updated_at",""), } except Exception as _exc: _logger.debug("bench fetch last run: %s", _exc) # ── Trigger nuovo run via workflow_dispatch ─────────────────────────────── run_url = "https://github.com/Baida98/AI/actions/workflows/bench.yml" if gh_token: try: import httpx as _hx async with _hx.AsyncClient(timeout=10.0) as _c: _r = await _c.post( "https://api.github.com/repos/Baida98/AI/actions/workflows/bench.yml/dispatches", json={"ref": "main", "inputs": { "mode": mode, "run_improve": "false", "force_update_baseline": "false", }}, headers={"Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "AgenteAI-Bot"}, ) if _r.status_code == 204: _logger.info("[bench] workflow_dispatch OK (mode=%s)", mode) # Attendi 2s e leggi il run ID appena creato await asyncio.sleep(2.0) async with _hx.AsyncClient(timeout=8.0) as _c2: _r2 = await _c2.get( "https://api.github.com/repos/Baida98/AI/actions/workflows/" "bench.yml/runs?per_page=1", headers={"Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "AgenteAI-Bot"}, ) if _r2.status_code == 200: _rr = _r2.json().get("workflow_runs", []) if _rr: run_url = _rr[0]["html_url"] else: _logger.warning("[bench] workflow_dispatch status=%d", _r.status_code) except Exception as _exc: _logger.warning("[bench] workflow_dispatch error: %s", _exc) # ── Costruisci messaggio con quickchart dell'ultimo run (se disponibile) ── _BENCH_CACHE[chat_id] = {"mode": mode, "run_url": run_url} # ── Fetch benchmark-report.json dal repo per quickchart reale ────────────── bench_report: dict | None = None if gh_token: try: import httpx as _hx, base64 as _b64, json as _json async with _hx.AsyncClient(timeout=8.0) as _c: _br = await _c.get( "https://api.github.com/repos/Baida98/AI/contents/benchmark-report.json?ref=main", headers={"Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "AgenteAI-Bot"}, ) if _br.status_code == 200: _content = _b64.b64decode(_br.json()["content"]).decode() bench_report = _json.loads(_content) except Exception as _exc: _logger.debug("bench fetch benchmark-report.json: %s", _exc) chart_url: str | None = None def _build_quickchart(report: dict) -> str: """Costruisce URL quickchart.io da benchmark-report.json.""" import json as _j, urllib.parse as _ul tasks = report.get("tasks", []) summary = report.get("summary", {}) avg_ai = summary.get("avgScore", 0) avg_rpl = summary.get("avgReplit", 57.9) avg_mns = summary.get("avgManus", 71.2) cat_map: dict[str, list[float]] = {} for t in tasks: cat = (t.get("cat") or "other").replace("_", " ")[:14] cat_map.setdefault(cat, []).append(t.get("score", 0)) if not cat_map: return "" labels = list(cat_map.keys()) scores = [round(sum(v)/len(v)) for v in cat_map.values()] colors = ["#4CAF50" if s >= avg_rpl else "#FF9800" if s >= 50 else "#F44336" for s in scores] cfg = { "type": "horizontalBar", "data": { "labels": labels, "datasets": [ {"label": "Agente AI", "data": scores, "backgroundColor": colors, "borderColor": colors, "borderWidth": 1}, {"label": f"Replit {avg_rpl}", "data": [avg_rpl]*len(labels), "type": "line", "borderColor": "#2196F3", "borderDash": [5,3], "pointRadius": 0, "fill": False, "borderWidth": 2}, {"label": f"Manus {avg_mns}", "data": [avg_mns]*len(labels), "type": "line", "borderColor": "#9C27B0", "borderDash": [5,3], "pointRadius": 0, "fill": False, "borderWidth": 2}, ], }, "options": { "title": {"display": True, "text": f"Agente AI {avg_ai}% | Replit {avg_rpl}% | Manus {avg_mns}%"}, "scales": {"xAxes": [{"ticks": {"min": 0, "max": 100, "stepSize": 20}}]}, "legend": {"display": True, "position": "bottom"}, "plugins": {"datalabels": {"display": False}}, }, } return ("https://quickchart.io/chart?c=" + _ul.quote(_j.dumps(cfg, separators=(",",":"))) + "&width=720&height=420&backgroundColor=white") if bench_report: chart_url = _build_quickchart(bench_report) summary = (bench_report or {}).get("summary", {}) avg_ai = summary.get("avgScore") avg_rpl = summary.get("avgReplit") avg_mns = summary.get("avgManus") # ── Tabella ASCII con barre per caption Telegram ────────────────────────── def _text_table_bench(report: dict, rpl: float) -> str: tasks = report.get("tasks", []) cat_map: dict[str, list[float]] = {} for t in tasks: cat = (t.get("cat") or "other").replace("_", " ")[:12] cat_map.setdefault(cat, []).append(float(t.get("score", 0))) if not cat_map: return "" rows = [] for cat, vals in sorted(cat_map.items(), key=lambda x: -sum(x[1]) / len(x[1])): sc = round(sum(vals) / len(vals)) bar = "█" * round(sc / 10) + "░" * (10 - round(sc / 10)) delta_rpl = sc - rpl vs = ("+" if delta_rpl >= 0 else "") + str(round(delta_rpl)) + "vsRpl" rows.append(f"{cat:<12} {bar} {sc:>3}% {vs}") return "\n".join(rows) text_table = "" if bench_report and avg_rpl is not None: text_table = _text_table_bench(bench_report, float(avg_rpl)) score_line = "" if avg_ai is not None: score_line = ( f"\n📈 Score: AI {avg_ai}%" + (f" | Replit {avg_rpl}%" if avg_rpl else "") + (f" | Manus {avg_mns}%" if avg_mns else "") + "\n" ) def _build_bench_caption(header: str) -> str: tbl = ("\n" + text_table + "") if text_table else "" link = f'\n🔗 GitHub Actions' full = header + score_line + tbl + link if len(full) > 1020 and text_table: avail = max(0, 1020 - len(header) - len(score_line) - len(link) - 14) tbl = "\n" + text_table[:avail] + "…" full = header + score_line + tbl + link return full[:1024] if last_report: _conclusion = last_report.get("conclusion", "?") _em = "✅" if _conclusion == "success" else ("❌" if _conclusion == "failure" else "⚠️") _upd = last_report.get("updated", "")[:16].replace("T", " ") header = f"📊 Benchmark avviato — {_em} {_conclusion}\n🕐 {_upd} UTC" else: header = "📊 Benchmark avviato (benchmark-extended.mjs)" caption = _build_bench_caption(header) if chart_url: await _tg_photo(chat_id, chart_url, caption=caption, keyboard=_BENCH_ACTION_KB) else: await _tg_reply(chat_id, caption, keyboard=_BENCH_ACTION_KB) # ── _cmd_improve — ciclo miglioramento: bench → gap → patch ───────────────── async def _cmd_improve(chat_id: int) -> None: """⚙️ Ciclo di miglioramento: bench.yml con --improve → gap → patch automatica. Triggera bench.yml con run_improve=true (benchmark-extended.mjs --improve). Il ciclo: few-shot retry sui task falliti → identifica pattern di miglioramento → genera regole → propone patch via notify Telegram. """ gh_token = os.getenv("GITHUB_TOKEN", "").strip() await _tg_reply(chat_id, "⚙️ Ciclo miglioramento avviato\n\n" "• Avvio benchmark-extended.mjs --improve\n" "• Identifica gap · genera regole migliorate · propone patch\n" "Riceverai notifica Telegram al completamento (~10-25 min).", keyboard=_BACK_KB) run_url = "https://github.com/Baida98/AI/actions/workflows/bench.yml" if not gh_token: await _tg_reply(chat_id, "⚠️ GITHUB_TOKEN non configurato — impossibile avviare workflow.", keyboard=_BACK_KB) return try: import httpx as _hx async with _hx.AsyncClient(timeout=10.0) as _c: _r = await _c.post( "https://api.github.com/repos/Baida98/AI/actions/workflows/bench.yml/dispatches", json={"ref": "main", "inputs": { "mode": "default", "run_improve": "true", "force_update_baseline": "false", }}, headers={"Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "AgenteAI-Bot"}, ) if _r.status_code == 204: _logger.info("[improve] workflow_dispatch OK") await asyncio.sleep(2.0) async with _hx.AsyncClient(timeout=8.0) as _c2: _r2 = await _c2.get( "https://api.github.com/repos/Baida98/AI/actions/workflows/" "bench.yml/runs?per_page=1", headers={"Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "AgenteAI-Bot"}, ) if _r2.status_code == 200: _rr = _r2.json().get("workflow_runs", []) if _rr: run_url = _rr[0]["html_url"] await _tg_reply(chat_id, f"✅ Ciclo miglioramento avviato\n" f'🔗 Segui su GitHub Actions', keyboard=_BENCH_ACTION_KB) else: await _tg_reply(chat_id, f"⚠️ GitHub API status {_r.status_code} — riprova tra poco.", keyboard=_BACK_KB) except Exception as _exc: _logger.warning("[improve] error: %s", _exc) await _tg_reply(chat_id, f"❌ Errore: {html.escape(str(_exc)[:150])}", keyboard=_BACK_KB) # ── callback_query handler — inline buttons da notify_task_done ─────────────── async def _handle_inline(iq: dict, token: str) -> None: """Gestisce inline query: @ARJagent_ap_bot in qualsiasi chat. Abilita tramite @BotFather → /setinline → ARJagent_ap_bot. """ bot_token = token or _get_bot_token() if not bot_token: return iq_id = iq.get("id","") query = (iq.get("query") or "").strip() q60 = query[:60] if not query: results = [ {"type":"article","id":"help","title":"🤖 Comandi Agente AI", "description":"Lista tutti i comandi", "input_message_content":{"message_text": "🤖 Agente AI\n/do <goal> — task streaming\n/ask <q> — risposta veloce\n" "/autofix — fix errori auto\n/status — stato\n\nagente-ai.pages.dev","parse_mode":"HTML"}}, {"type":"article","id":"do","title":"🚀 Nuovo Task AI", "description":"/do — lancia task complesso", "input_message_content":{"message_text":"/do "}}, {"type":"article","id":"ask","title":"🧠 Domanda AI", "description":"/ask ", "input_message_content":{"message_text":"/ask "}}, ] else: results = [ {"type":"article","id":"ask","title":f"🧠 Chiedi: {q60}", "description":f"/ask {q60}", "input_message_content":{"message_text":f"/ask {query}"}}, {"type":"article","id":"do","title":f"🚀 Task: {q60}", "description":f"/do {q60}", "input_message_content":{"message_text":f"/do {query}"}}, {"type":"article","id":"fix","title":f"🔧 AutoFix: {q60}", "description":f"/autofix {q60}", "input_message_content":{"message_text":f"/autofix {query}"}}, ] try: import httpx as _hx async with _hx.AsyncClient(timeout=5.0) as c: await c.post(f"https://api.telegram.org/bot{bot_token}/answerInlineQuery", json={"inline_query_id":iq_id,"results":results,"cache_time":30,"is_personal":True}) except Exception as exc: _logger.debug("inline answer error: %s", exc) async def _handle_callback(callback_query: dict, token: str) -> None: """Gestisce i callback_data dei bottoni inline (notify_task_done, menu).""" cq_id = callback_query.get("id", "") chat_id = callback_query.get("message", {}).get("chat", {}).get("id", 0) data = (callback_query.get("data") or "").strip() # Risponde subito per chiudere il loading Telegram await _tg_answer_callback(cq_id, token=token) if not chat_id or not data: return # ── task_sum: — riepilogo task da Supabase ────────────────────────── if data.startswith("task_sum:"): tid = data[9:].strip() try: from api.state import _sb if _sb: res = await asyncio.to_thread( lambda: _sb.table("agent_tasks") .select("task_id,goal,status,created_at") .eq("task_id", tid) .limit(1) .execute() ) rows = res.data or [] if rows: r = rows[0] st = r.get("status", "?") gol = html.escape((r.get("goal") or "")[:200]) em = {"SUCCESS":"✅","ERROR":"❌","RUNNING":"⚙️","QUEUED":"⏳"}.get(st,"•") ca = r.get("created_at", 0) age = _fmt_elapsed(ca) if isinstance(ca, int) and ca > 1_000_000 else "?" await _tg_reply( chat_id, "📋 Riepilogo task\n\n" "ID: " + html.escape(tid) + "\n" "Stato: " + em + " " + st + "\n" "Goal: " + gol + "\n" "Avviato: " + age, token=token, ) return await _tg_reply(chat_id, "⚠️ Task " + html.escape(tid) + " non trovato in Supabase.", token=token) except Exception as exc: await _tg_reply(chat_id, "❌ Errore: " + html.escape(str(exc)[:200]), token=token) # ── task_wins: — quick wins (non persisti — riepilogo generico) ───── elif data.startswith("task_wins:"): tid = data[10:].strip() await _tg_reply( chat_id, "💡 Quick Wins" + html.escape(tid) + "\n\n" "I quick wins vengono inclusi nella notifica del task.\n" "Avvia /tasks per vedere i task recenti o usa il pannello web per dettagli completi.\n\n" "Apri Dashboard →", token=token, ) # ── agent — nuovo task ───────────────────────────────────────────────────── elif data == "agent": await _tg_reply( chat_id, "🚀 Nuovo Task AI\n\n" "Scrivi liberamente o usa /do <goal>:\n\n" "Qualsiasi messaggio viene elaborato come task AI.\n\n" "💡 Esempi:\n" " analizza providers.py e suggerisci fix\n" " /do ottimizza le query Supabase lente", token=token, keyboard=_BACK_KB, ) # ── tgw_status / tgw_tasks / tgw_health / tgw_help — menu comandi ──────── elif data == "tgw_chart": await _tg_reply(chat_id, "📈 Grafici\n\nHeatmap commit, burndown sprint e molto altro nella Dashboard:", token=token, keyboard=_WEBAPP_KB) elif data == "tgw_ask": await _tg_reply(chat_id, "🧠 Chiedi all'AI\n\n" "Scrivi la domanda direttamente o usa:\n" "/ask <domanda>\n\n" "Esempio: /ask come ottimizzare una query PostgreSQL?", token=token, keyboard=_BACK_KB) elif data == "tgw_bench": asyncio.create_task(_cmd_bench(chat_id)).add_done_callback(_log_tg_exc) elif data == "tgw_autofix": asyncio.create_task(_cmd_autofix(chat_id,"")).add_done_callback(_log_tg_exc) elif data == "tgw_logs": asyncio.create_task(_cmd_logs(chat_id,"WARNING")).add_done_callback(_log_tg_exc) elif data == "tgw_bench_fix": # Legge gap dall'ultimo run (cache) e chiede all'agente un fix mirato cached = _BENCH_CACHE.get(chat_id, {}) goal = ( "Analizza i gap del benchmark-extended (bench.yml) più recente. " "Identifica le categorie con score più basso e genera patch mirate " "(prompt rules, retry logic, tool selection) per migliorare ogni area debole. " "Priorità: agentic → coding → reasoning. Fai push delle modifiche su GitHub." ) asyncio.create_task(_cmd_do(chat_id, goal)).add_done_callback(_log_tg_exc) elif data == "tgw_bench_run": # Rilancia bench asyncio.create_task(_cmd_bench(chat_id)).add_done_callback(_log_tg_exc) elif data == "tgw_score": asyncio.create_task(_cmd_score(chat_id)).add_done_callback(_log_tg_exc) elif data == "tgw_telemetry": asyncio.create_task(_cmd_telemetry(chat_id)).add_done_callback(_log_tg_exc) elif data == "tgw_improve": asyncio.create_task(_cmd_improve(chat_id)).add_done_callback(_log_tg_exc) elif data == "tgw_status": await _cmd_status(chat_id) elif data == "tgw_tasks": await _cmd_tasks(chat_id) elif data == "tgw_health": await _cmd_scan_now(chat_id) elif data == "tgw_help": await _cmd_help(chat_id) elif data == "tgw_coord": asyncio.create_task(_cmd_coord(chat_id)).add_done_callback(_log_tg_exc) elif data == "tgw_git": asyncio.create_task(_cmd_git(chat_id)).add_done_callback(_log_tg_exc) # ── Sub-menu dispatchers (Reply Keyboard → inline sub-menu) ───────────────── elif data in ("menu_task",): await _tg_reply(chat_id, "🚀 Task AI — scegli un'azione:", token=token, keyboard=_TASK_MENU_KB) elif data == "menu_status": await _cmd_status(chat_id) await _tg_reply(chat_id, "📊 Stato Sistema — altre opzioni:", token=token, keyboard=_STATUS_MENU_KB) elif data == "menu_perf": await _tg_reply(chat_id, "📈 Performance — scegli:", token=token, keyboard=_PERF_MENU_KB) elif data == "menu_health": asyncio.create_task(_cmd_scan_now(chat_id)).add_done_callback(_log_tg_exc) elif data == "menu_dev": await _tg_reply(chat_id, "🛠 Dev Tools — scegli:", token=token, keyboard=_DEV_MENU_KB) elif data == "tgw_home": asyncio.create_task(_cmd_help(chat_id)).add_done_callback(_log_tg_exc) # ── Nuovi callback task/utility ──────────────────────────────────────────── elif data == "tgw_do": await _tg_reply(chat_id, "🚀 Nuovo Task AI\n\n" "Scrivi il tuo obiettivo nella chat — elaboro tutto come task AI.\n\n" "Esempi:\n" " analizza bug in backend/api/state.py\n" " genera test per providers.py\n" " ottimizza le query Supabase più lente\n\n" "Oppure: /do <goal>", token=token, keyboard=_BACK_KB) elif data == "tgw_briefing": asyncio.create_task(_cmd_riepilogo(chat_id)).add_done_callback(_log_tg_exc) elif data == "tgw_nota": await _tg_reply(chat_id, "📝 Salva Nota\n\n" "Usa: /nota <testo da ricordare>\n\n" "Esempio: /nota domani rivedere il deploy HF Space", token=token, keyboard=_BACK_KB) elif data == "tgw_cerca": await _tg_reply(chat_id, "🔍 Cerca sul web\n\n" "Usa: /cerca <query>\n\n" "Esempio: /cerca best practices FastAPI async Python", token=token, keyboard=_BACK_KB) elif data == "tgw_meteo": await _tg_reply(chat_id, "🌤 Meteo\n\n" "Usa: /meteo <città>\n\n" "Esempio: /meteo Milano", token=token, keyboard=_BACK_KB) elif data == "tgw_providers": try: from api.providers import _heartbeat_state as _hs_pv providers = (_hs_pv or {}).get("providers") or [] if providers: lines = ["🔌 Provider AI\n"] best = next((p for p in providers if p.get("ok")), None) for p in providers: ok = p.get("ok", False) icon = "✅" if ok else ("🔑" if "401" in str(p.get("error","")) else ("💳" if "429" in str(p.get("error","")) else "❌")) ms = p.get("latency_ms") ms_str = f"{ms}ms" if ms else "–" name = p.get("name","?") star = " ⭐" if (best and p.get("name") == best.get("name")) else "" lines.append(f"{icon} {html.escape(name)}{star} — {ms_str}") if not ok and p.get("error"): lines.append(f" {html.escape(str(p['error'])[:80])}") import time as _tpv last = (_hs_pv or {}).get("last_run_at",0) age = f"{int(_tpv.time()-last)}s fa" if last else "?" lines.append(f"\n⏱ Heartbeat {age}") await _tg_reply(chat_id, "\n".join(lines), token=token, keyboard=_HEALTH_MENU_KB) else: await _tg_reply(chat_id, "⚠️ Dati provider non ancora disponibili (~90s al primo heartbeat).", token=token, keyboard=_BACK_KB) except Exception as exc: await _tg_reply(chat_id, "❌ Errore: " + html.escape(str(exc)[:200]) + "", token=token, keyboard=_BACK_KB) elif data == "tgw_snap": try: from .integrity_manager import handle_snap_cmd as _hi_snap asyncio.create_task(_hi_snap(chat_id, _tg_reply, None)).add_done_callback(_log_tg_exc) except Exception: await _tg_reply(chat_id, "⚠️ Integrity manager non disponibile.", token=token, keyboard=_BACK_KB) elif data == "tgw_verify": try: from .integrity_manager import handle_verify_cmd as _hi_verify asyncio.create_task(_hi_verify(chat_id, _tg_reply, None)).add_done_callback(_log_tg_exc) except Exception: await _tg_reply(chat_id, "⚠️ Integrity manager non disponibile.", token=token, keyboard=_BACK_KB) elif data == "tgw_heal": try: from .integrity_manager import handle_heal_cmd as _hi_heal asyncio.create_task(_hi_heal(chat_id, _tg_reply, None)).add_done_callback(_log_tg_exc) except Exception: await _tg_reply(chat_id, "⚠️ Integrity manager non disponibile.", token=token, keyboard=_BACK_KB) elif data == "tgw_ping": ry = os.getenv("RAILWAY_URL","https://ai-production-4c06.up.railway.app") try: async with httpx.AsyncClient(timeout=8.0) as _hxc: r = await _hxc.get(f"{ry}/health") j = r.json() ok = "✅" if j.get("status")=="ok" else "⚠️" await _tg_reply(chat_id, f"🏓 Pong! {ok}\nv{j.get('version','?')} — Railway live", token=token, keyboard=_DEV_MENU_KB) except Exception as exc: await _tg_reply(chat_id, "❌ Railway non raggiungibile\n"+html.escape(str(exc)[:150])+"", token=token, keyboard=_BACK_KB) # ── m — menu principale ─────────────────────────────────────────────────── elif data == "m": await _tg_reply( chat_id, "🤖 Menu principale\n\nScegli un'azione:", token=token, keyboard=_MAIN_KB, ) # ── fallback — command non gestito ─────────────────────────────────────── else: _logger.debug("callback_query unhandled: %s", data) # ── Webhook endpoint (passivo — getUpdates polling attivo nel daemon) ───────── @router.post("/webhook") async def telegram_webhook(request: Request) -> dict: """Riceve aggiornamenti Telegram (POST da api.telegram.org). Endpoint passivo: processa update solo se Telegram li invia qui. NOTA: il bot usa getUpdates in session-daemon.mjs — questo endpoint non riceve traffico finché il webhook non viene registrato. Gestisce sia message che callback_query (inline buttons). """ secret = os.getenv("TELEGRAM_WEBHOOK_SECRET", "").strip() if secret: recv = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "") if recv != secret: raise HTTPException(403, "Forbidden: invalid webhook secret") try: body = await request.json() except Exception: return {"ok": True} token = _get_bot_token() # ── Gestione callback_query (inline buttons) ────────────────────────────── cq = body.get("callback_query") iq = body.get("inline_query") if iq: _t=asyncio.create_task(_handle_inline(iq, token)); _t.add_done_callback(_log_tg_exc) return {"ok": True} if cq: _t=asyncio.create_task(_handle_callback(cq, token)); _t.add_done_callback(_log_tg_exc) return {"ok": True} # ── Gestione message ────────────────────────────────────────────────────── msg = body.get("message") or body.get("edited_message") if not msg: return {"ok": True} chat_id: int = msg.get("chat", {}).get("id", 0) text: str = (msg.get("text") or "").strip() if not chat_id or not text: return {"ok": True} cmd = text.split()[0].lower().split("@")[0] # normalizza /cmd@botname → /cmd if cmd in ("/help", "/start"): _t=asyncio.create_task(_cmd_help(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd == "/status": _t=asyncio.create_task(_cmd_status(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd == "/tasks": _t=asyncio.create_task(_cmd_tasks(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/do", "/run"): goal = text[len(cmd):].strip() _t=asyncio.create_task(_cmd_do(chat_id, goal)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/scan_now", "/health", "/ping"): _t=asyncio.create_task(_cmd_scan_now(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd == "/ask": question = text[len(cmd):].strip() if question: _t=asyncio.create_task(_cmd_do(chat_id, question)); _t.add_done_callback(_log_tg_exc) else: await _tg_reply(chat_id, "🧠 Uso: /ask <domanda>", keyboard=_MAIN_KB) elif cmd == "/bench": _mode = text[len(cmd):].strip() or "default" if _mode not in ("default","full","coding-only","noncode-only","agentic-only"): _mode = "default" _t=asyncio.create_task(_cmd_bench(chat_id, _mode)); _t.add_done_callback(_log_tg_exc) elif cmd == "/score": _t=asyncio.create_task(_cmd_score(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd == "/improve": _t=asyncio.create_task(_cmd_improve(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd == "/autofix": fix_hint = text[len(cmd):].strip() _t=asyncio.create_task(_cmd_autofix(chat_id, fix_hint)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/chart", "/burndown"): await _tg_reply(chat_id, "📈 Grafici — disponibili nella Dashboard\n" "🌐 agente-ai.pages.dev", keyboard=_WEBAPP_KB) elif cmd == "/logs": lvl = text[len(cmd):].strip().upper() or "WARNING" if lvl not in ("DEBUG","INFO","WARNING","ERROR","CRITICAL"): lvl = "WARNING" _t=asyncio.create_task(_cmd_logs(chat_id, lvl)); _t.add_done_callback(_log_tg_exc) elif cmd == "/ping": ry = os.getenv("RAILWAY_URL","https://ai-production-4c06.up.railway.app") import httpx as _hx try: async with _hx.AsyncClient(timeout=8.0) as c: r = await c.get(f"{ry}/health") j = r.json() ok = "✅" if j.get("status")=="ok" else "⚠️" await _tg_reply(chat_id, f"🏓 Pong! {ok} Railway live\n" f"v{j.get('version','?')} — {ry.split('//')[1][:35]}", keyboard=_BACK_KB) except Exception as e: await _tg_reply(chat_id, "❌ Railway non raggiungibile\n"+html.escape(str(e)[:150])+"", keyboard=_BACK_KB) elif cmd in ("/nota", "/ricorda", "/remember"): note_text = text[len(cmd):].strip() if note_text: _t=asyncio.create_task(_cmd_nota(chat_id, note_text)); _t.add_done_callback(_log_tg_exc) else: await _tg_reply(chat_id, "📝 Uso: /nota <testo da ricordare>", keyboard=_BACK_KB) elif cmd in ("/cerca", "/search", "/web"): query = text[len(cmd):].strip() if query: _t=asyncio.create_task(_cmd_cerca(chat_id, query)); _t.add_done_callback(_log_tg_exc) else: await _tg_reply(chat_id, "🔍 Uso: /cerca <cosa cercare>", keyboard=_BACK_KB) elif cmd in ("/snap", "/snapshot"): # P41-INTEGRITY from .integrity_manager import handle_snap_cmd as _hi_snap conv_id = text[len(cmd):].strip() or None _t=asyncio.create_task(_hi_snap(chat_id, _tg_reply, conv_id)); _t.add_done_callback(_log_tg_exc) elif cmd == "/verify": # P41-INTEGRITY from .integrity_manager import handle_verify_cmd as _hi_verify snap_id = text[len(cmd):].strip() or None _t=asyncio.create_task(_hi_verify(chat_id, _tg_reply, snap_id)); _t.add_done_callback(_log_tg_exc) elif cmd == "/heal": # P41-INTEGRITY from .integrity_manager import handle_heal_cmd as _hi_heal snap_id = text[len(cmd):].strip() or None _t=asyncio.create_task(_hi_heal(chat_id, _tg_reply, snap_id)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/meteo", "/weather", "/tempo"): city = text[len(cmd):].strip() or "Roma" _t=asyncio.create_task(_cmd_meteo(chat_id, city)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/riepilogo", "/briefing", "/update", "/aggiorna"): _t=asyncio.create_task(_cmd_riepilogo(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/telemetria", "/telemetry"): _t=asyncio.create_task(_cmd_telemetry(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/daily", "/briefing_daily"): _t=asyncio.create_task(_cmd_riepilogo(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/coord", "/sessioni", "/lock"): _t=asyncio.create_task(_cmd_coord(chat_id)); _t.add_done_callback(_log_tg_exc) elif cmd in ("/git", "/log", "/commits"): _n = int(text.split()[1]) if len(text.split()) > 1 and text.split()[1].isdigit() else 5 _t=asyncio.create_task(_cmd_git(chat_id, _n)); _t.add_done_callback(_log_tg_exc) # ── Reply Keyboard (testo pulsanti persistenti) ──────────────────────────── elif text == "🚀 Lancia Task": await _tg_reply(chat_id, "🚀 Task AI — scegli un'azione:", keyboard=_TASK_MENU_KB) elif text == "📊 Stato": _t=asyncio.create_task(_cmd_status(chat_id)); _t.add_done_callback(_log_tg_exc) elif text == "🩺 Health": _t=asyncio.create_task(_cmd_scan_now(chat_id)); _t.add_done_callback(_log_tg_exc) elif text == "📈 Performance": await _tg_reply(chat_id, "📈 Performance — scegli:", keyboard=_PERF_MENU_KB) elif text == "🛠 Dev Tools": await _tg_reply(chat_id, "🛠 Dev Tools — scegli:", keyboard=_DEV_MENU_KB) elif text == "💬 Chiedi AI": await _tg_reply(chat_id, "💬 AI Libera\n\n" "Scrivi qualsiasi cosa — la elaboro come task AI:\n\n" "analizza providers.py e suggerisci ottimizzazioni\n" "che differenza c'è tra asyncio.Task e asyncio.gather?", keyboard=_BACK_KB) else: # Testo libero → avvia task come /do _t=asyncio.create_task(_cmd_do(chat_id, text)); _t.add_done_callback(_log_tg_exc) return {"ok": True} # ── Bot API 7.0+ setup: setMyCommands + setMyDescription + setMenuButton ───────── async def _setup_bot_commands(token: str) -> dict: """Configura comandi, descrizione e menu button via Telegram Bot API 7.0+. Idempotente — sicuro da richiamare più volte senza effetti collaterali.""" import httpx results: dict = {} base = f"https://api.telegram.org/bot{token}" commands = [ {"command": "do", "description": "🚀 Lancia task AI: /do "}, {"command": "bench", "description": "📊 Benchmark (benchmark-extended.mjs) — score + gap"}, {"command": "score", "description": "🏆 Score attuale — tabella + chart senza nuovo run"}, {"command": "improve", "description": "⚙️ Ciclo miglioramento: bench → gap → patch auto"}, {"command": "status", "description": "📊 Task attivi + stato backend"}, {"command": "tasks", "description": "📋 Ultimi task con elapsed time"}, {"command": "health", "description": "🩺 Health check AI+Supabase+TG"}, {"command": "ask", "description": "🧠 Domanda veloce all'AI"}, {"command": "autofix", "description": "🔧 AutoFix: AI legge errori e pusha patch"}, {"command": "logs", "description": "📋 /logs [error|warning] — log backend Railway"}, {"command": "chart", "description": "📈 Heatmap commit 14gg"}, {"command": "help", "description": "❓ Lista completa comandi"}, {"command": "nota", "description": "📝 Salva nota rapida: /nota "}, {"command": "cerca", "description": "🔍 Cerca su web: /cerca "}, {"command": "meteo", "description": "🌤 Meteo: /meteo "}, {"command": "riepilogo", "description": "📋 Briefing completo: task + deploy + score"}, {"command": "briefing", "description": "🧠 Briefing: task+deploy+score+prossimi passi"}, {"command": "git", "description": "🔀 Ultimi commit GitHub con SHA e diff"}, {"command": "coord", "description": "🔗 Sessioni agent-coord attive e file claimati"}, {"command": "telemetry", "description": "📡 Metriche telemetria backend"}, {"command": "snap", "description": "📸 Integrity snapshot conversazione"}, {"command": "verify", "description": "✅ Verifica integrità snapshot"}, {"command": "heal", "description": "💊 Ripristina da snapshot"}, ] async with httpx.AsyncClient(timeout=12.0) as c: r = await c.post(f"{base}/setMyCommands", json={"commands": commands}) results["setMyCommands"] = {"ok": r.json().get("ok"), "http": r.status_code} r = await c.post(f"{base}/setMyDescription", json={ "description": ( "Agente AI autonomo per sviluppo software.\n\n" "Usa /do per lanciare task complessi: codice, ricerca, analisi, deploy.\n" "Monitoraggio real-time, notifiche automatiche, multi-provider AI." )}) results["setMyDescription"] = {"ok": r.json().get("ok"), "http": r.status_code} r = await c.post(f"{base}/setMyShortDescription", json={ "short_description": "Agente AI autonomo • /do per lanciare task"}) results["setMyShortDescription"] = {"ok": r.json().get("ok"), "http": r.status_code} r = await c.post(f"{base}/setChatMenuButton", json={"menu_button": {"type": "commands"}}) results["setMenuButton"] = {"ok": r.json().get("ok"), "http": r.status_code} return results @router.get("/logs") async def telegram_get_logs(n: int = 50, level: str = "WARNING") -> dict: """Snapshot log backend per monitoring Railway build/runtime. ?n=50 (max 500) — ultimi N record ?level=WARNING — livello minimo (DEBUG/INFO/WARNING/ERROR/CRITICAL) Utile da chiamare via bot o curl dopo un redeploy per vedere errori. """ import logging as _logging _lvl_map = { "DEBUG": _logging.DEBUG, "INFO": _logging.INFO, "WARNING": _logging.WARNING, "ERROR": _logging.ERROR, "CRITICAL": _logging.CRITICAL, } min_lvl = _lvl_map.get(level.upper(), _logging.WARNING) n = min(max(1, n), 500) records: list[dict] = [] # Prova MemoryHandler sul root logger (se configurato) for handler in _logging.getLogger().handlers: if hasattr(handler, "buffer"): for rec in list(handler.buffer)[-n * 2:]: if rec.levelno >= min_lvl: records.append({ "ts": rec.created, "level": rec.levelname, "logger": rec.name, "msg": rec.getMessage()[:500], }) # Fallback leggero: step degli agent_tasks recenti if not records: try: from api.state import _agent_tasks for tid, task in list(_agent_tasks.items())[-20:]: for step in (task.get("steps") or [])[-5:]: msg = step.get("text") or step.get("error") or "" if msg and _logging.ERROR >= min_lvl: records.append({ "ts": task.get("updated_at", 0), "level": "ERROR" if step.get("error") else "INFO", "logger": f"task.{tid[:8]}", "msg": str(msg)[:300], }) except Exception: pass records = sorted(records, key=lambda r: r["ts"], reverse=True)[:n] return {"ok": True, "count": len(records), "level_filter": level.upper(), "records": records} @router.get("/setup") async def telegram_bot_setup() -> dict: """Bot API 7.0+: configura comandi, descrizione e menu button del bot assistente. Idempotente — chiama setMyCommands, setMyDescription, setMyShortDescription, setChatMenuButton. Richiede TELEGRAM_BOT_TOKEN su Railway. Sicuro da richiamare più volte.""" token = _get_bot_token() if not token: raise HTTPException(400, "TELEGRAM_BOT_TOKEN non configurato su Railway") try: results = await _setup_bot_commands(token) all_ok = all(v.get("ok") for v in results.values()) return { "ok": all_ok, "results": results, "hint": ("✅ Bot configurato — apri Telegram e premi ☰ comandi" if all_ok else "⚠️ Alcune operazioni fallite — verifica TELEGRAM_BOT_TOKEN"), } except Exception as exc: raise HTTPException(500, str(exc)) # ── Setup webhook — DISABILITATO (getUpdates mode attivo) ───────────────────── @router.post("/webhook/setup") async def setup_webhook(request: Request) -> dict: """P11: Registra webhook Telegram su agente-ai.pages.dev/api/telegram/process. CF Pages proxia /api/* → Railway (Cloudflare Worker nel CF Dashboard). Il webhook punta a /api/telegram/process (endpoint Railway) non alla CF Pages Function. Telegram autentica con X-Telegram-Bot-Api-Secret-Token → TELEGRAM_WEBHOOK_SECRET. Body (opzionale): { "webhook_url": "https://...", "secret": "..." } Default URL: https://agente-ai.pages.dev """ token = _get_bot_token() if not token: raise HTTPException(400, "TELEGRAM_BOT_TOKEN non configurato") try: body = await request.json() except Exception: body = {} cf_url = str(body.get("webhook_url") or os.getenv("CF_PAGES_URL", "https://agente-ai.pages.dev")).rstrip("/") secret = str(body.get("secret") or os.getenv("TELEGRAM_WEBHOOK_SECRET", "")) # P11-FIX: CF Pages proxia /api/* → Railway via CF Worker. # /api/telegram/process è già live; /api/webhook/telegram era per CF Pages Function # che viene scavalcata dal proxy Worker configurato nel CF Dashboard. webhook_url = f"{cf_url}/api/telegram/process" payload: dict = { "url": webhook_url, "allowed_updates": ["message", "callback_query"], "drop_pending_updates": False, "max_connections": 40, } if secret: payload["secret_token"] = secret try: import httpx async with httpx.AsyncClient(timeout=10.0) as c: r = await c.post(f"https://api.telegram.org/bot{token}/setWebhook", json=payload) data = r.json() if data.get("ok"): return { "ok": True, "webhook_url": webhook_url, "hint": "✅ Webhook registrato. Imposta USE_WEBHOOK=true nel daemon (o fermalo).", } raise HTTPException(400, f"Telegram API error: {data.get('description', str(data))}") except HTTPException: raise except Exception as exc: raise HTTPException(500, str(exc)) # GAP-E: avviso esplicito se INTERNAL_TOKEN non configurato if not os.getenv("INTERNAL_TOKEN"): import logging as _log _log.getLogger("agente_ai").warning( "⚠️ INTERNAL_TOKEN non impostato — autenticazione CF→Railway DISABILITATA. " "Impostare lo stesso valore fisso su Railway ENV e CF Pages/Worker secrets." ) @router.post("/process") async def process_telegram_update(request: Request, background_tasks: BackgroundTasks) -> dict: """P12: Riceve aggiornamento Telegram webhook (Telegram → CF Pages proxy → Railway). Auth (in ordine di priorità): - Se TELEGRAM_WEBHOOK_SECRET configurato: X-Telegram-Bot-Api-Secret-Token OPPURE Authorization: Bearer INTERNAL_TOKEN - Se TELEGRAM_WEBHOOK_SECRET non configurato: accetta senza auth (Telegram direct, no secret registrato) Body: oggetto JSON aggiornamento Telegram (message o callback_query). Comandi gestiti: /start /help /status /tasks /do /health /score /bench /scan_now """ # P12-AUTH: strategia corretta per endpoint pubblico webhook Telegram. # BUG P11: "if _int_token or _wh_secret" bloccava Telegram perché INTERNAL_TOKEN # è impostato su Railway ma Telegram non lo conosce. # FIX: TELEGRAM_WEBHOOK_SECRET governa l'accesso pubblico. # INTERNAL_TOKEN è opzionale fallback per CF Worker. _wh_secret = os.getenv("TELEGRAM_WEBHOOK_SECRET", "").strip() _int_token = os.getenv("INTERNAL_TOKEN", "").strip() _tg_hdr = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "") _auth_hdr = request.headers.get("Authorization", "") if _wh_secret: # Webhook secret configurato → richiede TG secret OPPURE INTERNAL_TOKEN valido _ok = (_tg_hdr == _wh_secret) or (_int_token and _auth_hdr == f"Bearer {_int_token}") if not _ok: raise HTTPException(401, "Unauthorized — X-Telegram-Bot-Api-Secret-Token o INTERNAL_TOKEN richiesto") elif _auth_hdr and _int_token: # Nessun webhook secret ma Authorization header esplicito → valida INTERNAL_TOKEN if _auth_hdr != f"Bearer {_int_token}": raise HTTPException(401, "Unauthorized — INTERNAL_TOKEN non valido") # Se TELEGRAM_WEBHOOK_SECRET non configurato e nessun auth header → accetta # (Telegram direct webhook senza secret, sicurezza minima, funziona subito) try: update = await request.json() except Exception: raise HTTPException(400, "Invalid JSON") token = _get_bot_token() if not token: return {"ok": False, "reason": "TELEGRAM_BOT_TOKEN not configured"} # Fire-and-forget: processa in background e ritorna subito async def _handle() -> None: try: if "callback_query" in update: # P61-BOT1: usa _handle_callback completa (task_sum/task_wins/agent/m/tgw_*) await _handle_callback(update["callback_query"], token) return msg = update.get("message") or {} chat_id = (msg.get("chat") or {}).get("id") text = (msg.get("text") or "").strip() if not chat_id or not text: return parts = text.split() raw_cmd = parts[0].split("@")[0].lower() if parts else "" if raw_cmd in ("/start", "/help"): await _tg_reply(chat_id, "🤖 Agente AI — Webhook Mode\n\n" "Comandi Railway (sempre attivi 24/7):\n" "/status — 📊 Stato backend + task attivi\n" "/tasks — 📋 Ultimi task\n" "/health — 🩺 Health check completo\n" "/score — 📈 Score benchmark\n" "/bench — 📊 Avvia benchmark\n" "/do <goal> — 🚀 Lancia task AI autonomo\n\n" "ℹ️ Per comandi avanzati (git, coord, CI…) avvia il daemon locale.", token=token) elif raw_cmd == "/status": await _cmd_status(chat_id) elif raw_cmd == "/tasks": await _cmd_tasks(chat_id) elif raw_cmd in ("/health", "/scan_now"): await _cmd_scan_now(chat_id) elif raw_cmd == "/score": await _cmd_score(chat_id) elif raw_cmd == "/bench": mode = parts[1] if len(parts) > 1 else "default" await _cmd_bench(chat_id, mode) elif raw_cmd == "/do": goal = " ".join(parts[1:]).strip() if not goal: await _tg_reply(chat_id, "❓ Uso: /do <obiettivo>", token=token) else: await _cmd_do(chat_id, goal) elif raw_cmd in ("/snap", "/snapshot"): # P41 from .integrity_manager import handle_snap_cmd as _hi_snap conv_id = parts[1] if len(parts) > 1 else None await _hi_snap(chat_id, lambda c, t: _tg_reply(c, t, token), conv_id) elif raw_cmd == "/verify": # P41 from .integrity_manager import handle_verify_cmd as _hi_ver snap_id = parts[1] if len(parts) > 1 else None await _hi_ver(chat_id, lambda c, t: _tg_reply(c, t, token), snap_id) elif raw_cmd == "/heal": # P41 from .integrity_manager import handle_heal_cmd as _hi_heal snap_id = parts[1] if len(parts) > 1 else None await _hi_heal(chat_id, lambda c, t: _tg_reply(c, t, token), snap_id) else: # Se non è un comando slash, trattalo come un obiettivo per /do (Assistente Proattivo) if not raw_cmd.startswith("/"): await _cmd_do(chat_id, text) else: await _tg_reply(chat_id, f"⚠️ Comando {html.escape(raw_cmd[:40])} non riconosciuto.\n\n" "Comandi disponibili: /status /tasks /health /score /bench /do\n" "Scrivi pure un obiettivo in linguaggio naturale per attivare l'assistente.", token=token) except Exception as exc: _logger.warning("process_telegram_update handler error: %s", exc) # GAP-A fix: BackgroundTasks (FastAPI-managed) invece di bare asyncio.create_task # I task creati con asyncio.create_task senza riferimento salvato possono essere GC'd. # BackgroundTasks esegue dopo l'invio della risposta, gestito dal framework. background_tasks.add_task(_handle) return {"ok": True, "queued": True} @router.post("/daily-briefing") async def daily_briefing(request: Request, background_tasks: BackgroundTasks) -> dict: """P12: Trigger daily briefing (GHA schedule o CF Worker cron). Auth: Authorization: Bearer {INTERNAL_TOKEN} (se configurato su Railway). Body: { "chat_id": "..." } """ # P12-AUTH: INTERNAL_TOKEN richiesto per daily-briefing (chiamato da GHA/CF Worker) # ma non viene applicato come guard su Telegram webhook diretto. _int_token = os.getenv("INTERNAL_TOKEN", "").strip() if _int_token: _auth_hdr = request.headers.get("Authorization", "") if _auth_hdr != f"Bearer {_int_token}": raise HTTPException(401, "Unauthorized — Authorization: Bearer INTERNAL_TOKEN richiesto") try: body = await request.json() except Exception: body = {} chat_id_str = str(body.get("chat_id") or os.getenv("TELEGRAM_CHAT_ID", "")) if not chat_id_str: raise HTTPException(400, "chat_id mancante") try: chat_id = int(chat_id_str) except ValueError: raise HTTPException(400, "chat_id non valido") token = _get_bot_token() if not token: raise HTTPException(400, "TELEGRAM_BOT_TOKEN non configurato") async def _send_briefing() -> None: try: await _cmd_status(chat_id) except Exception as exc: _logger.warning("daily_briefing error: %s", exc) background_tasks.add_task(_send_briefing) return {"ok": True, "chat_id": chat_id} @router.post("/test-notify") async def test_notify(request: Request) -> dict: """T8: Invia un messaggio Telegram di test via notify_task_done.""" try: body = await request.json() except Exception: body = {} task_id = str(body.get("task_id", "test0000")) goal = str(body.get("goal", "T8 automated test"))[:200] result = str(body.get("result", "✅ Test automatico da /api/telegram/test-notify"))[:500] try: from .telegram_notify import notify_task_done await notify_task_done(task_id, goal, result) return {"ok": True, "sent": True, "task_id": task_id} except Exception as exc: raise HTTPException(500, str(exc)) @router.get("/webhook/info") async def webhook_info() -> dict: """Mostra info webhook corrente registrato su Telegram (diagnostica).""" token = _get_bot_token() if not token: raise HTTPException(400, "TELEGRAM_BOT_TOKEN non configurato") try: import httpx async with httpx.AsyncClient(timeout=8.0) as c: r = await c.get(f"https://api.telegram.org/bot{token}/getWebhookInfo") return r.json() except Exception as exc: raise HTTPException(500, str(exc)) @router.post("/config/invalidate") async def invalidate_telegram_config() -> dict: """Invalida la cache config Telegram (chiamato da SettingsDialog dopo salvataggio).""" try: from .telegram_notify import invalidate_config_cache invalidate_config_cache() return {"ok": True, "message": "Cache Telegram invalidata — prossima notifica rilegge config"} except Exception as exc: raise HTTPException(500, str(exc))