Sacar del bucle de eventos la recuperación RAG, SQLite y scrypt

ARCHITECTURE_REVIEW §3.1. `interpretar` es `async`, pero sus dos operaciones más
caras eran SÍNCRONAS y corrían en el bucle de eventos, así que mientras una
interpretación recuperaba, el proceso entero no contestaba a NADA: ni health
check, ni el login de otro veterinario, ni una ingesta del puente. La
concurrencia efectiva era 1 en una app que se anuncia multiusuario.

- `ai/service.py`: `recuperar` / `recuperar_multi` a un hilo. Es lo más caro del
request —embedding bge-m3, búsqueda LanceDB y cross-encoder sobre hasta
`rag_candidatos` filas—, del orden de segundos en cpu-basic.
- `routers/auth.py`: SQLite y scrypt de login y alta. scrypt (n=2**14) es caro a
propósito: es justo el trabajo que no puede vivir en el bucle.
- `routers/lab.py`: la escritura de `lab_persistir`, que corre por cada muestra
con el analizador enviando en ráfaga (120/minute).

`security/authz.py` no toca la BD (la sesión es una cookie firmada), así que las
peticiones autenticadas ya no pasaban por SQLite. `main.py` se deja como está: su
carga corre en el lifespan, antes de que haya tráfico que bloquear.

Las pruebas miden la PROPIEDAD, no la implementación: cuentan cuántas veces
consigue despertarse el bucle mientras hay trabajo en curso, y comprueban que dos
interpretaciones concurrentes se solapan en vez de serializarse. Verificado que
fallan sin el arreglo: 1 latido en vez de ~80, y 0.81s para dos recuperaciones de
0.4s. Márgenes holgados a propósito —miden concurrencia, no latencia— para que no
parpadeen en CI.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/ai/service.py

@@ -7,6 +7,7 @@ Un reintento ante fallo de validación; si persiste, error tipado (nunca texto c
 
 from __future__ import annotations
 
+import asyncio
 import logging
 
 from ..config import obtener_config
@@ -208,13 +209,20 @@ async def interpretar(pet: PeticionInterpretacion) -> RespuestaInterpretacion:
     # consulta concatenada).
     nombres_patrones = [p.nombre for p in pet.patrones]
     nombres_hallazgos = [h.nombre for h in pet.hallazgos]
+    # `asyncio.to_thread` y no llamada directa: la recuperación es SÍNCRONA y cara —embedding
+    # bge-m3, búsqueda en LanceDB y cross-encoder bge-reranker-v2-m3 sobre hasta
+    # `rag_candidatos` filas—, del orden de segundos en cpu-basic. Ejecutada en el bucle de
+    # eventos dejaba el proceso entero sordo mientras durase: ni health check, ni el login de
+    # otro veterinario, ni una ingesta del puente. La concurrencia efectiva era 1.
     if cfg.rag_multiconsulta:
-        fragmentos = recuperar_multi(
+        fragmentos = await asyncio.to_thread(
+            recuperar_multi,
             construir_consultas(nombres_patrones, nombres_hallazgos),
             especie=pet.paciente.especie,
         )
     else:
-        fragmentos = recuperar(
+        fragmentos = await asyncio.to_thread(
+            recuperar,
             construir_consulta(nombres_patrones, nombres_hallazgos),
             especie=pet.paciente.especie,
         )

backend/app/routers/auth.py

@@ -9,6 +9,8 @@ Mejoras de seguridad frente a auth.php:
 
 from __future__ import annotations
 
+import asyncio
+
 from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 from pydantic import BaseModel, EmailStr, Field
 
@@ -75,17 +77,21 @@ async def estado(request: Request) -> dict:
 @router.post("/auth/login")
 @limiter.limit(obtener_config().limite_login)
 async def login(request: Request, body: LoginBody, response: Response) -> dict:
+    # Todo lo que toca SQLite o scrypt va a un hilo: son llamadas SÍNCRONAS dentro de un
+    # endpoint `async`, así que en el bucle de eventos bloqueaban el proceso entero. scrypt es
+    # además caro A PROPÓSITO (n=2**14, decenas de ms): es justo el trabajo que no puede vivir
+    # en el bucle, y el login es el endpoint que más veces lo ejecuta.
     ip = request.client.host if request.client else "?"
-    if intentos_recientes(body.email, ip, _VENTANA_THROTTLE_S) >= _MAX_INTENTOS:
+    if await asyncio.to_thread(intentos_recientes, body.email, ip, _VENTANA_THROTTLE_S) >= _MAX_INTENTOS:
         raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "Demasiados intentos. Espera unos minutos.")
 
-    usuario = buscar_usuario(body.email)
-    if not usuario or not verificar_password(body.password, usuario["password"]):
-        registrar_intento(body.email, ip)
+    usuario = await asyncio.to_thread(buscar_usuario, body.email)
+    if not usuario or not await asyncio.to_thread(verificar_password, body.password, usuario["password"]):
+        await asyncio.to_thread(registrar_intento, body.email, ip)
         # Mensaje genérico: no revela si el email existe.
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Email o contraseña incorrectos.")
 
-    limpiar_intentos(body.email)
+    await asyncio.to_thread(limpiar_intentos, body.email)
     csrf = _emitir_sesion(response, usuario["email"], usuario["nombre"])
     return {"ok": True, "nombre": usuario["nombre"], "csrf": csrf}
 
@@ -101,9 +107,10 @@ async def registro(request: Request, body: RegistroBody, response: Response) ->
             status.HTTP_403_FORBIDDEN,
             "El alta de cuentas está restringida. Solicita acceso al administrador.",
         )
-    if buscar_usuario(body.email):
+    if await asyncio.to_thread(buscar_usuario, body.email):
         raise HTTPException(status.HTTP_409_CONFLICT, "Ya existe una cuenta con ese email.")
-    crear_usuario(body.nombre, body.apellido, body.email, body.password)
+    # `crear_usuario` hashea con scrypt además de escribir: doble motivo para salir del bucle.
+    await asyncio.to_thread(crear_usuario, body.nombre, body.apellido, body.email, body.password)
     csrf = _emitir_sesion(response, body.email, body.nombre)
     return {"ok": True, "nombre": body.nombre, "csrf": csrf}
 

backend/app/routers/lab.py

@@ -10,6 +10,8 @@ la auth); la consulta usa la sesión existente. El mapeo código→analito ocurr
 
 from __future__ import annotations
 
+import asyncio
+
 from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 
 from .. import db
@@ -39,7 +41,10 @@ async def post_ingesta(
     mapeado = mapear_resultado(cuerpo)
     almacen.guardar(mapeado)
     if obtener_config().lab_persistir:
-        db.guardar_resultado_lab(
+        # A un hilo como el resto de SQLite: el analizador manda en ráfaga
+        # (`limite_lab_ingesta` = 120/minute) y esto corre por cada muestra.
+        await asyncio.to_thread(
+            db.guardar_resultado_lab,
             mapeado.muestra_id.strip().lower(),
             mapeado.momento.isoformat(),
             mapeado.model_dump_json(),

backend/tests/test_bucle_no_bloqueante.py

@@ -0,0 +1,139 @@
+"""El trabajo síncrono y caro no puede correr en el bucle de eventos.
+
+`interpretar` es `async`, pero sus dos operaciones más caras son SÍNCRONAS: la recuperación RAG
+(embedding bge-m3 + LanceDB + cross-encoder sobre hasta `rag_candidatos` filas, segundos en
+cpu-basic) y todo SQLite/scrypt en los endpoints de auth. Llamadas directamente desde una
+corrutina, bloquean el proceso ENTERO mientras duran: ni health check, ni el login de otro
+veterinario, ni una ingesta del puente. La concurrencia efectiva era 1 en una app que se
+anuncia multiusuario.
+
+Estas pruebas miden la propiedad, no la implementación: si alguien sustituye un
+`await asyncio.to_thread(...)` por la llamada directa, vuelven a fallar. Los márgenes son
+holgados a propósito —miden concurrencia, no latencia— para que no parpadeen en CI.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+
+import pytest
+
+from app.ai import service
+from app.ai.base import ErrorModelo  # noqa: F401  (documenta la superficie que se falsea)
+from app.schemas import InterpretacionClinica, PeticionInterpretacion
+
+BLOQUEO_S = 0.4
+
+PETICION = {
+    "paciente": {"especie": "canino"},
+    "hallazgos": [
+        {
+            "clave": "hct",
+            "nombre": "Hematocrito",
+            "valor": 22.0,
+            "unidad": "%",
+            "direccion": "bajo",
+            "gravedad": "grave",
+        }
+    ],
+    "patrones": [{"nombre": "Anemia", "descripcion": "…", "gravedad": "grave"}],
+    "imagenes": [],
+}
+
+
+class ClienteInstantaneo:
+    """Cliente de modelo que responde sin coste: aquí se mide la recuperación, no la generación."""
+
+    nombre = "medgemma-hf"
+    prosa = True
+    modelo = "hf-space"
+
+    async def interpretar(self, *_a, **_k):
+        return InterpretacionClinica(interpretacion="ok " * 20, requiere_derivacion=True)
+
+
+@pytest.fixture
+def rag_lento(monkeypatch):
+    """Retriever síncrono y lento, como el real: `time.sleep` bloquea el hilo que lo ejecute."""
+
+    def _recuperar_bloqueante(*_a, **_k):
+        time.sleep(BLOQUEO_S)
+        return []
+
+    for nombre in ("recuperar", "recuperar_multi"):
+        monkeypatch.setattr(service, nombre, _recuperar_bloqueante)
+    monkeypatch.setattr(service, "_crear_cliente", lambda *_: ClienteInstantaneo())
+
+
+async def _contar_latidos(tarea: asyncio.Task, intervalo: float = 0.005) -> int:
+    """Cuántas veces consigue despertarse el bucle mientras `tarea` está en curso.
+
+    Es la medición directa de «¿puede el servidor atender a alguien más?». Con el trabajo
+    bloqueante en el bucle, el contador se queda en ~0.
+    """
+    latidos = 0
+    while not tarea.done():
+        await asyncio.sleep(intervalo)
+        latidos += 1
+    return latidos
+
+
+async def test_la_recuperacion_no_congela_el_bucle(rag_lento):
+    """Durante una interpretación, el bucle sigue despertándose para atender otras cosas."""
+    tarea = asyncio.create_task(service.interpretar(PeticionInterpretacion.model_validate(PETICION)))
+    latidos = await _contar_latidos(tarea)
+    await tarea
+
+    # Con to_thread caben ~80 latidos de 5 ms en 0.4 s; en el bucle serían 0 o 1.
+    assert latidos > 10, (
+        f"sólo {latidos} latidos durante la recuperación: el bucle estuvo bloqueado, "
+        "la recuperación volvió a ejecutarse sin asyncio.to_thread"
+    )
+
+
+async def test_dos_interpretaciones_se_solapan(rag_lento):
+    """Dos peticiones concurrentes no se serializan: comparten el tiempo de recuperación."""
+    inicio = time.perf_counter()
+    await asyncio.gather(
+        service.interpretar(PeticionInterpretacion.model_validate(PETICION)),
+        service.interpretar(PeticionInterpretacion.model_validate(PETICION)),
+    )
+    transcurrido = time.perf_counter() - inicio
+
+    # Serializadas costarían >= 2*BLOQUEO_S; solapadas, algo más de BLOQUEO_S.
+    assert transcurrido < BLOQUEO_S * 1.7, (
+        f"{transcurrido:.2f}s para dos interpretaciones de {BLOQUEO_S}s: se serializaron"
+    )
+
+
+async def test_el_alta_no_congela_el_bucle(alta_abierta):
+    """scrypt (n=2**14) es caro A PROPÓSITO; en el bucle, cada alta congela el servicio."""
+    import httpx
+
+    from app import db
+    from app.main import app
+
+    # ASGITransport no dispara el lifespan (TestClient sí), así que la tabla no existiría.
+    db.inicializar_db()
+
+    transporte = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transporte, base_url="http://test") as cliente:
+        tarea = asyncio.create_task(
+            cliente.post(
+                "/api/auth/registro",
+                json={
+                    "nombre": "Hilo",
+                    "apellido": "Vet",
+                    "email": "hilo@example.com",
+                    "password": "clave-segura-1",
+                },
+            )
+        )
+        latidos = await _contar_latidos(tarea, intervalo=0.002)
+        resp = await tarea
+
+    assert resp.status_code == 200, resp.text
+    assert latidos > 3, (
+        f"s��lo {latidos} latidos durante el alta: scrypt volvió a correr en el bucle de eventos"
+    )