Spaces:

d1d4c
/

retro-cpu

Runtime error

feat: v0.8.0 — run_bash tool (subprocess, 15s timeout)
5286aae verified 4 months ago
31 kB
"""
retro-cpu — Online Retro instance (Telegram webhook bridge).
ADR-562 pilot. A CPU-only HF Docker Space that lets Dídac operate a lightweight
Retro from his phone (Telegram) when the home `satelite` node is offline.
Flow:
  Telegram update -> /webhook -> (text | voice)
    text  : retro-space  /v1/chat/completions   (Qwen2.5-7B)
    voice : retro-tts     /v1/audio/transcriptions (Whisper) -> same chat pipeline
  reply -> Telegram sendMessage
No GPU here. Inference is delegated to the two ZeroGPU Spaces. Conversation
memory is an ontology-governed graph (bruno/v1) on the /data bucket volume —
see bruno_graph.py (sovereign adapter, ADR-562/563 Fase 1).
Version: 0.8.0
"""

import os
import re
import json
import ipaddress
import logging
import asyncio
import tempfile
import subprocess
from pathlib import Path
from typing import Any, Dict, List, Optional

from bruno_graph import BrunoGraph

import httpx
from fastapi import FastAPI, Request, Header, HTTPException
from fastapi.responses import JSONResponse, PlainTextResponse

# ──────────────────────────────────────────────────────────────────────────
# Config
# ──────────────────────────────────────────────────────────────────────────
VERSION = "0.8.0"

TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
HF_TOKEN = os.environ.get("HF_TOKEN", "")
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "")
ALLOWED_CHAT_IDS = {
    s.strip() for s in os.environ.get("ALLOWED_CHAT_IDS", "").split(",") if s.strip()
}

# Inference backends (private Spaces — reachable via https + Bearer HF_TOKEN)
RETRO_SPACE_URL = os.environ.get(
    "RETRO_SPACE_URL", "https://d1d4c-retro-space.hf.space"
)
RETRO_TTS_URL = os.environ.get(
    "RETRO_TTS_URL", "https://d1d4c-retro-tts.hf.space"
)

MODEL = os.environ.get("MODEL", "7b")
MAX_HISTORY = int(os.environ.get("MAX_HISTORY", "20"))      # messages kept per chat
MAX_TOKENS = int(os.environ.get("MAX_TOKENS", "1024"))
TEMPERATURE = float(os.environ.get("TEMPERATURE", "0.3"))
INFER_TIMEOUT = float(os.environ.get("INFER_TIMEOUT", "120"))

DATA_DIR = Path(os.environ.get("DATA_DIR", "/data"))

TELEGRAM_API = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}"
TELEGRAM_FILE_API = f"https://api.telegram.org/file/bot{TELEGRAM_BOT_TOKEN}"

SYSTEM_PROMPT = os.environ.get(
    "SYSTEM_PROMPT",
    "Eres Retro Lite, una instancia online y ligera de Retro que corre en "
    "HuggingFace para que Dídac pueda hablar contigo desde el móvil (Telegram) "
    "cuando el nodo de casa (satélite) está offline.\n"
    "\n"
    "QUIÉN ERES:\n"
    "- Eres el modelo Qwen2.5-7B sirviendo como interlocutor conversacional.\n"
    "- NO eres el Retro completo de casa: no tienes acceso a sus agentes, a su "
    "memoria/HyperGraph, ni a sus herramientas. Eres una versión de respaldo "
    "para conversar y pensar en voz alta.\n"
    "\n"
    "QUÉ PUEDES HACER:\n"
    "- Conversar, responder preguntas, razonar, redactar y resumir texto.\n"
    "- Mantener el hilo de esta conversación (recuerdas los mensajes recientes).\n"
    "- Ejecutar comandos bash en el contenedor de HuggingFace donde corres "
    "(usa la tool run_bash). Puedes hacer ls, cat, python3, curl, etc. "
    "El filesystem es el del Space, no el de casa.\n"
    "- Consultar tu memoria de conversaciones pasadas (tools get_memory y search_messages).\n"
    "\n"
    "QUÉ NO PUEDES HACER (sé honesto, NUNCA lo finjas):\n"
    "- No controlas dispositivos, domótica, ni sistemas externos.\n"
    "- No tienes acceso a internet real (salvo endpoints internos de HF), "
    "calendario, email ni a la red de casa.\n"
    "- No puedes ejecutar acciones en el mundo real ni en otros agentes de Retro.\n"
    "- El bash que ejecutas corre en el Space de HF, no en satélite ni en casa.\n"
    "Si te piden algo de lo anterior, dilo con claridad y, si procede, sugiere que "
    "lo haga el Retro de casa cuando vuelva a estar online.\n"
    "\n"
    "ESTILO:\n"
    "- Directo, honesto y conciso. Es un chat de móvil: respuestas breves por "
    "defecto (2-5 frases), sin markdown pesado ni listas largas salvo que se pidan.\n"
    "- Responde en el idioma de Dídac (normalmente español).\n"
    "- No te inventes capacidades ni datos. Ante la duda, reconoce lo que no sabes.",
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
log = logging.getLogger("retro-cpu")

app = FastAPI(title="retro-cpu", version=VERSION)

# ──────────────────────────────────────────────────────────────────────────
# "Public-but-closed" hardening
#
# HF Space visibility is all-or-nothing, so when the Space is public we close it
# at the application layer: only the authenticated Telegram webhook does anything.
#   - /webhook : Telegram IP allowlist  AND  secret token  AND  chat_id allowlist
#   - /health  : detailed only with the secret; bare {"ok":true} otherwise
#   - /        : bare 200 (required by HF init-process readiness probe)
#   - /admin/* : removed (webhook is registered out-of-band via the Telegram API)
#
# Telegram's official webhook source ranges (Bot API docs):
#   149.154.160.0/20  and  91.108.4.0/22
# NOTE: behind HF's reverse proxy the real client IP arrives in X-Forwarded-For;
# the secret token remains the strong control (XFF can be spoofed), the IP filter
# is defense-in-depth. Toggle with ENABLE_IP_FILTER=false if it ever blocks real
# Telegram delivery (we calibrate against the first real message).
# ──────────────────────────────────────────────────────────────────────────
ENABLE_IP_FILTER = os.environ.get("ENABLE_IP_FILTER", "true").lower() == "true"
TELEGRAM_IP_NETS = [
    ipaddress.ip_network("149.154.160.0/20"),
    ipaddress.ip_network("91.108.4.0/22"),
]


def _client_ip(request: Request) -> Optional[str]:
    """Original client IP. Behind HF's proxy it's the first X-Forwarded-For entry."""
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()
    return request.client.host if request.client else None


def _ip_is_telegram(ip: Optional[str]) -> bool:
    if not ip:
        return False
    try:
        addr = ipaddress.ip_address(ip)
    except ValueError:
        return False
    return any(addr in net for net in TELEGRAM_IP_NETS)


# ──────────────────────────────────────────────────────────────────────────
# Conversation memory — ontology-governed graph (bruno/v1, sovereign)
#
# Replaces the old flat-JSON transcript. Messages persist as bruno/v1 `Message`
# nodes (append-only, A1-unique) in a sovereign sqlite graph on the /data bucket;
# the context carousel is a query over the active session. ContextMemory (singleton
# per chat) is scaffolded across sessions. Type-E gate: accept-bruno-v1-gate-20260621.
# ──────────────────────────────────────────────────────────────────────────
_bruno: Optional[BrunoGraph] = None


def get_bruno() -> BrunoGraph:
    """Lazily open the sovereign bruno/v1 graph. Falls back to /tmp if /data is RO."""
    global _bruno
    if _bruno is None:
        if DATA_DIR.exists() and os.access(DATA_DIR, os.W_OK):
            db_dir = DATA_DIR
            persistent = True
        else:
            db_dir = Path(tempfile.gettempdir())
            persistent = False
            log.warning("/data not writable — bruno_graph on ephemeral %s", db_dir)
        _bruno = BrunoGraph(
            db_path=str(db_dir / "bruno_graph.db"),
            ontology_path=str(Path(__file__).with_name("bruno_v1.json")),
        )
        log.info("BrunoGraph ready (bruno/v1, schema_enforced, persistent=%s) at %s",
                 persistent, db_dir / "bruno_graph.db")
    return _bruno


def _touch_context_memory(bruno: BrunoGraph, chat_id: int, new_session: bool) -> None:
    """Ensure a ContextMemory node exists for this chat; bump session_count on /reset.
    Creates a placeholder on first contact (synthesis_strategy="pending_fase1").
    Real content is written asynchronously by synthesize_context_memory() after each turn.
    """
    try:
        cm = bruno.get_context_memory(chat_id)
        if cm is None:
            note = f"Chat {chat_id}. Memoria inicializada. Síntesis en curso."
            bruno.upsert_context_memory(chat_id, note, synthesis_strategy="pending_fase1")
        elif new_session:
            bruno.upsert_context_memory(
                chat_id, cm.get("content", ""),
                bump_session=True,
                synthesis_strategy=cm.get("synthesis_strategy", "pending_fase1"),
            )
    except Exception as exc:  # noqa: BLE001 — memory upkeep must never break a reply
        log.warning("ContextMemory upkeep failed for %s: %s", chat_id, exc)


def build_system_prompt(bruno: BrunoGraph, chat_id: int) -> str:
    """Return the base SYSTEM_PROMPT with cross-session context injected when available.
    Injects ContextMemory content only when it carries a real LLM synthesis
    (synthesis_strategy == "llm_qwen2.5-7b"). Placeholders are silently skipped.
    """
    cm = bruno.get_context_memory(chat_id)
    if cm and cm.get("synthesis_strategy") == "llm_qwen2.5-7b":
        content = cm.get("content", "").strip()
        if content:
            return (
                SYSTEM_PROMPT
                + "\n\nCONTEXTO DE SESIONES ANTERIORES (resumen de memoria):\n"
                + content
                + "\n\nUsa este contexto si es relevante para la conversación actual."
            )
    return SYSTEM_PROMPT


async def synthesize_context_memory(
    bruno: BrunoGraph, chat_id: int, history: List[Dict[str, str]]
) -> None:
    """Background task: synthesize recent conversation → upsert ContextMemory.
    Calls retro-space with a low-token meta-prompt (≤200 output tokens). Runs
    AFTER the reply is already sent (asyncio.create_task) so latency = 0 for user.
    A failed synthesis is logged but never raises (never breaks the next turn).
    """
    try:
        # Use last 10 turns to keep the synthesis prompt short
        recent = history[-10:]
        conv_text = "\n".join(
            f"{m['role'].upper()}: {m['content'][:300]}" for m in recent
        )
        messages = [
            {
                "role": "system",
                "content": (
                    "Eres un asistente de memoria. Tu única tarea es extraer de la "
                    "conversación un resumen CONCISO (máx 150 palabras) con los datos "
                    "relevantes sobre el usuario: intereses, proyectos mencionados, "
                    "preferencias, contexto personal. Solo hechos, sin florituras."
                ),
            },
            {
                "role": "user",
                "content": f"Conversación reciente:\n\n{conv_text}\n\n¿Qué debo recordar?",
            },
        ]
        payload = {
            "model": MODEL,
            "messages": messages,
            "temperature": 0.1,
            "max_tokens": 200,
        }
        headers = {"Authorization": f"Bearer {HF_TOKEN}", "Content-Type": "application/json"}
        async with httpx.AsyncClient(timeout=INFER_TIMEOUT) as client:
            r = await client.post(
                f"{RETRO_SPACE_URL}/v1/chat/completions", json=payload, headers=headers
            )
            r.raise_for_status()
            synthesis = r.json()["choices"][0]["message"]["content"].strip()

        bruno.upsert_context_memory(chat_id, synthesis, synthesis_strategy="llm_qwen2.5-7b")
        log.info("ContextMemory synthesized for chat %s (%d chars)", chat_id, len(synthesis))
    except Exception as exc:  # noqa: BLE001
        log.warning("ContextMemory synthesis failed for chat %s: %s", chat_id, exc)


# ──────────────────────────────────────────────────────────────────────────
# Tool definitions (OpenAI format) — Bruno's graph query capabilities
# ──────────────────────────────────────────────────────────────────────────
BRUNO_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_memory",
            "description": (
                "Recupera el resumen de memoria de conversaciones anteriores con este usuario. "
                "Úsala cuando el usuario pregunte sobre conversaciones pasadas, qué recuerdas "
                "de él, o cuando el contexto histórico sea relevante para responder bien."
            ),
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_messages",
            "description": (
                "Busca en el historial completo de conversaciones (todas las sesiones) los "
                "mensajes que contengan una palabra clave. Úsala cuando el usuario pregunte "
                "sobre algo específico que puede haber mencionado en sesiones anteriores."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {
                        "type": "string",
                        "description": "Palabra o frase a buscar en el historial",
                    }
                },
                "required": ["keyword"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_bash",
            "description": (
                "Ejecuta un comando bash en el contenedor de HuggingFace donde corre Bruno. "
                "Úsalo cuando el usuario pida hacer ls, cat, python3, ver archivos, comprobar "
                "el entorno, etc. El filesystem es el del Space (no el de casa/satélite). "
                "Devuelve stdout+stderr combinados. Timeout 15s."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "Comando bash a ejecutar (e.g. 'ls /home/user/app', 'python3 --version')",
                    }
                },
                "required": ["command"],
            },
        },
    },
]

TOOL_MAX_ROUNDS = 3   # safety cap: at most N tool-call rounds per user turn


def _execute_tool(name: str, args: dict, bruno: BrunoGraph, chat_id: int) -> str:
    """Execute a Bruno tool call locally (SQLite — microseconds). Never raises."""
    try:
        if name == "get_memory":
            cm = bruno.get_context_memory(chat_id)
            if cm and cm.get("synthesis_strategy") == "llm_qwen2.5-7b":
                content = cm.get("content", "").strip()
                if content:
                    return content
            return "No hay memoria sintetizada todavía de conversaciones anteriores."

        if name == "search_messages":
            keyword = str(args.get("keyword", "")).strip().lower()
            if not keyword:
                return "Error: se requiere una palabra clave para buscar."
            rows = bruno.conn.execute(
                "SELECT properties FROM nodes "
                "WHERE graph_id = ? AND domain_type = 'Message' "
                "AND lower(json_extract(properties, '$.content')) LIKE ? "
                "ORDER BY created_at DESC LIMIT 5",
                (bruno.graph_id, f"%{keyword}%"),
            ).fetchall()
            if not rows:
                return f"No encontré mensajes que contengan '{keyword}'."
            parts = []
            for r in rows:
                p = json.loads(r[0])
                parts.append(
                    f"[{p['role']}] sesión {p.get('session_ref', '?')}: "
                    f"{p['content'][:300]}"
                )
            return "\n---\n".join(parts)

        if name == "run_bash":
            command = args.get("command", "").strip()
            if not command:
                return "Error: se requiere un comando bash."
            try:
                proc = subprocess.run(
                    command,
                    shell=True,
                    capture_output=True,
                    text=True,
                    timeout=15,
                    cwd="/home/user/app",
                )
                output = (proc.stdout + proc.stderr).strip()
                if not output:
                    output = f"(sin salida, exit code {proc.returncode})"
                elif len(output) > 2000:
                    output = output[:2000] + "\n…(truncado)"
                return output
            except subprocess.TimeoutExpired:
                return "Error: el comando superó el límite de 15 segundos."

        return f"Herramienta '{name}' no reconocida."
    except Exception as exc:  # noqa: BLE001
        log.warning("_execute_tool %s error: %s", name, exc)
        return f"Error ejecutando '{name}': {exc}"


# ──────────────────────────────────────────────────────────────────────────
# Backend calls
# ──────────────────────────────────────────────────────────────────────────
async def _call_retro_space(
    messages: List[Dict[str, Any]],
    tools: Optional[List] = None,
) -> dict:
    """Raw HTTP call to retro-space /v1/chat/completions. Returns full response dict."""
    payload: Dict[str, Any] = {
        "model": MODEL,
        "messages": messages,
        "temperature": 0 if tools else TEMPERATURE,
        "max_tokens": MAX_TOKENS,
    }
    if tools:
        payload["tools"] = tools
        payload["tool_choice"] = "auto"
    headers = {"Authorization": f"Bearer {HF_TOKEN}", "Content-Type": "application/json"}
    async with httpx.AsyncClient(timeout=INFER_TIMEOUT) as client:
        r = await client.post(
            f"{RETRO_SPACE_URL}/v1/chat/completions", json=payload, headers=headers
        )
        r.raise_for_status()
        return r.json()


async def infer_chat(history: List[Dict[str, str]], system_prompt: str = SYSTEM_PROMPT) -> str:
    """Single-shot inference (no tools). Returns text content."""
    messages = [{"role": "system", "content": system_prompt}] + history
    data = await _call_retro_space(messages)
    return (data["choices"][0]["message"]["content"] or "").strip()


async def run_with_tools(
    history: List[Dict[str, str]],
    sys_prompt: str,
    bruno: BrunoGraph,
    chat_id: int,
) -> str:
    """Inference loop with tool calling. Returns final text reply.
    At most TOOL_MAX_ROUNDS rounds:
      1. Call retro-space with BRUNO_TOOLS.
      2. If finish_reason == "tool_calls" → execute locally → append results → repeat.
      3. If finish_reason == "stop" (or max rounds hit) → return text.
    """
    messages = [{"role": "system", "content": sys_prompt}] + list(history)
    msg: Dict[str, Any] = {}

    for round_num in range(TOOL_MAX_ROUNDS):
        data = await _call_retro_space(messages, tools=BRUNO_TOOLS)
        choice = data["choices"][0]
        finish_reason = choice.get("finish_reason", "stop")
        msg = choice["message"]

        if finish_reason != "tool_calls":
            return (msg.get("content") or "").strip()

        tool_calls = msg.get("tool_calls") or []
        if not tool_calls:
            return (msg.get("content") or "").strip()

        # Append assistant's tool_call message to the working context
        messages.append({
            "role": "assistant",
            "content": msg.get("content") or "",
            "tool_calls": tool_calls,
        })

        # Execute each tool and append results
        for tc in tool_calls:
            fn_name = tc["function"]["name"]
            try:
                fn_args = json.loads(tc["function"].get("arguments") or "{}")
            except (json.JSONDecodeError, KeyError):
                fn_args = {}
            result = _execute_tool(fn_name, fn_args, bruno, chat_id)
            log.info(
                "tool_call round=%d: %s(%r) → %d chars",
                round_num + 1, fn_name, fn_args, len(result),
            )
            messages.append({
                "role": "tool",
                "tool_call_id": tc["id"],
                "content": result,
            })

    # Safety: max rounds exhausted
    log.warning("run_with_tools: TOOL_MAX_ROUNDS=%d exhausted for chat %s", TOOL_MAX_ROUNDS, chat_id)
    return (msg.get("content") or "(Sin respuesta tras ejecutar herramientas)").strip()


async def transcribe_voice(file_bytes: bytes, filename: str = "voice.ogg") -> str:
    """Call retro-tts Whisper endpoint with the downloaded voice file."""
    headers = {"Authorization": f"Bearer {HF_TOKEN}"}
    files = {"file": (filename, file_bytes, "audio/ogg")}
    data = {"language": "es"}
    async with httpx.AsyncClient(timeout=INFER_TIMEOUT) as client:
        r = await client.post(
            f"{RETRO_TTS_URL}/v1/audio/transcriptions",
            files=files,
            data=data,
            headers=headers,
        )
        r.raise_for_status()
        return r.json().get("text", "").strip()


# ──────────────────────────────────────────────────────────────────────────
# Telegram helpers
# ──────────────────────────────────────────────────────────────────────────
async def tg_send_message(chat_id: int, text: str) -> None:
    # Telegram hard limit 4096 chars/message — chunk if needed.
    chunks = [text[i : i + 4000] for i in range(0, len(text), 4000)] or ["(sin respuesta)"]
    async with httpx.AsyncClient(timeout=30) as client:
        for chunk in chunks:
            await client.post(
                f"{TELEGRAM_API}/sendMessage",
                json={"chat_id": chat_id, "text": chunk},
            )


async def tg_send_typing(chat_id: int) -> None:
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            await client.post(
                f"{TELEGRAM_API}/sendChatAction",
                json={"chat_id": chat_id, "action": "typing"},
            )
    except Exception:  # noqa: BLE001
        pass


async def tg_download_file(file_id: str) -> Optional[bytes]:
    async with httpx.AsyncClient(timeout=60) as client:
        r = await client.get(f"{TELEGRAM_API}/getFile", params={"file_id": file_id})
        r.raise_for_status()
        file_path = r.json()["result"]["file_path"]
        fr = await client.get(f"{TELEGRAM_FILE_API}/{file_path}")
        fr.raise_for_status()
        return fr.content


# ──────────────────────────────────────────────────────────────────────────
# Core handler
#
# IMPORTANT (HF Spaces network constraint, discovered 2026-06-20):
# HF Spaces CANNOT open outbound connections to api.telegram.org (ConnectTimeout),
# although they CAN reach HF-internal hosts (*.hf.space) — so inference works but
# sending a reply via the Bot API does not. The fix is to reply *inline* in the
# webhook HTTP response: Telegram executes a single method returned as the JSON
# body. No outbound to Telegram required.
#   -> compute_reply() returns the inline method dict (or None).
#   -> Voice notes need getFile + file download (both api.telegram.org) which are
#      unreachable, so voice is not supported in this mode.
# ──────────────────────────────────────────────────────────────────────────
def _say(chat_id: int, text: str) -> Dict[str, Any]:
    """Build an inline sendMessage method for the webhook response."""
    return {"method": "sendMessage", "chat_id": chat_id, "text": text[:4096]}


async def compute_reply(update: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    message = update.get("message") or update.get("edited_message")
    if not message:
        return None

    chat = message.get("chat", {})
    chat_id = chat.get("id")
    if chat_id is None:
        return None

    # Security: allowlist
    if str(chat_id) not in ALLOWED_CHAT_IDS:
        log.warning("Rejected message from unauthorized chat_id=%s", chat_id)
        return None

    user_text: Optional[str] = message.get("text")

    # Voice notes require outbound to api.telegram.org (blocked on HF Spaces).
    if user_text is None and "voice" in message:
        return _say(
            chat_id,
            "🎤 Las notas de voz no funcionan en esta instancia online: el Space "
            "de HuggingFace no puede descargar el audio de Telegram (outbound "
            "bloqueado). Escríbeme texto y te respondo.",
        )

    if not user_text:
        return None

    bruno = get_bruno()

    # Slash commands
    if user_text.strip() == "/start":
        return _say(chat_id, "🤖 Retro online (HF). Escríbeme y te respondo con Qwen2.5-7B.")
    if user_text.strip() == "/reset":
        bruno.reset_session(chat_id)            # rotate epoch — old turns persist (append-only)
        _touch_context_memory(bruno, chat_id, new_session=True)
        return _say(chat_id, "🧹 Nueva conversación iniciada (el historial anterior se conserva en memoria).")

    # Conversation pipeline — persist the user turn, build the carousel from the graph
    bruno.append_message(chat_id, "user", user_text, source="telegram")
    _touch_context_memory(bruno, chat_id, new_session=False)
    history = bruno.carousel(chat_id, n=MAX_HISTORY)

    # Fase 1: inject cross-session context when a real synthesis is available
    sys_prompt = build_system_prompt(bruno, chat_id)

    try:
        # Fase 2: tool-calling loop (get_memory, search_messages against bruno_graph)
        reply = await run_with_tools(history, sys_prompt, bruno, chat_id)
    except httpx.HTTPStatusError as exc:
        log.error("Inference HTTP error: %s — %s", exc, exc.response.text[:200])
        return _say(
            chat_id,
            "⚠️ El servidor de inferencia falló (puede estar arrancando). "
            "Inténtalo de nuevo en ~1 min.",
        )
    except Exception as exc:  # noqa: BLE001
        log.error("Inference failed: %s", exc)
        return _say(chat_id, "⚠️ Error procesando el mensaje.")

    bruno.append_message(chat_id, "assistant", reply, source="qwen2.5-7b")

    # Fase 1: background synthesis — runs AFTER reply is sent, zero user latency
    full_history = bruno.carousel(chat_id, n=MAX_HISTORY)
    asyncio.create_task(synthesize_context_memory(bruno, chat_id, full_history))

    return _say(chat_id, reply)


# ──────────────────────────────────────────────────────────────────────────
# Routes
# ──────────────────────────────────────────────────────────────────────────
@app.get("/")
async def root() -> JSONResponse:
    # Bare 200 for HF's init-process readiness probe — no info disclosed.
    return JSONResponse({"ok": True})


@app.get("/health")
async def health(
    request: Request,
    x_telegram_bot_api_secret_token: Optional[str] = Header(None),
) -> JSONResponse:
    # Detailed health only with the shared secret; otherwise bare ok.
    if not WEBHOOK_SECRET or x_telegram_bot_api_secret_token != WEBHOOK_SECRET:
        return JSONResponse({"ok": True})
    return JSONResponse(
        {
            "status": "ok",
            "version": VERSION,
            "ready": True,
            "model": MODEL,
            "allowlist_size": len(ALLOWED_CHAT_IDS),
            "data_writable": os.access(DATA_DIR, os.W_OK) if DATA_DIR.exists() else False,
            "ip_filter": ENABLE_IP_FILTER,
            "memory": "bruno/v1 (schema_enforced)",
        }
    )


@app.post("/webhook")
async def webhook(
    request: Request,
    x_telegram_bot_api_secret_token: Optional[str] = Header(None),
) -> JSONResponse:
    # Layer 1 — Telegram IP allowlist (defense-in-depth). Log the observed IP so
    # the filter can be calibrated against the first real Telegram delivery.
    ip = _client_ip(request)
    if ENABLE_IP_FILTER and not _ip_is_telegram(ip):
        log.warning("Rejected /webhook from non-Telegram IP: %s (xff=%s)",
                    ip, request.headers.get("x-forwarded-for"))
        raise HTTPException(status_code=403, detail="forbidden")

    # Layer 2 — Telegram secret token (strong control)
    if WEBHOOK_SECRET and x_telegram_bot_api_secret_token != WEBHOOK_SECRET:
        raise HTTPException(status_code=403, detail="bad secret token")

    update = await request.json()
    # Reply INLINE in the webhook response (HF Spaces can't reach api.telegram.org
    # for outbound sendMessage). Telegram executes the returned method. The whole
    # pipeline (incl. inference) runs within Telegram's webhook timeout (~60s);
    # warm inference is sub-second, cold start ~30-45s.
    # (Layer 3 — chat_id allowlist — is enforced inside compute_reply.)
    try:
        method = await compute_reply(update)
    except Exception as exc:  # noqa: BLE001
        log.error("compute_reply crashed: %s", exc)
        method = None
    if method is None:
        return JSONResponse({"ok": True})
    return JSONResponse(method)