takana/scripts/test-llama-cpp.py

#!/usr/bin/env python3
"""Guardián de `recipes/llama-cpp.toml` — el motor de inferencia local del §6.7 del SDD 26.

Mide CUATRO afirmaciones sobre el ARTEFACTO VIGENTE, y ninguna se deduce de que el build haya
salido 0:

  1. ISA        el binario trae AVX2/FMA/SSE4.2 de verdad.
  2. hermético  estáticamente enlazado, sin un solo NEEDED.
  3. inferencia una pasada real contra un modelo pineado por sha256 ⇒ salen tokens.
  4. embeddings `llama-server --embedding` + `POST /v1/embeddings` ⇒ sale un vector, y además:
                  · el MISMO texto dos veces da el MISMO vector (si no, no hay nada determinista),
                  · dos textos DISTINTOS dan vectores DISTINTOS (si no, es un stub de ceros).
                El punto 4 es el que le importa al §6.3: la mitad semántica del archivo personal
                se apoya EXACTAMENTE en ese endpoint.

⚠ EL CONTROL NEGATIVO DEL PUNTO 1 NO ES HIPOTÉTICO, ESTÁ MEDIDO. El PRIMER sello de esta receta
(`b3:e3c62185`) llevaba sólo `-DGGML_NATIVE=OFF`, selló, corrió, contestó `--version` — y tenía
**cero** `%ymm`, **cero** `vfmadd` y **cero** `roundps` en 1.634.770 líneas de `objdump -d`. La causa
está en la cabecera de la receta: el sandbox exporta `SOURCE_DATE_EPOCH`, ggml lee esa variable para
decidir `GGML_NATIVE_DEFAULT` y termina apagando `INS_ENB`. O sea: la variable que existe para que
el build REPRODUZCA apagaba todas las instrucciones vectoriales del motor, sin un aviso.

El control negativo VIVO es `--sin-modelo`: las mismas sondas contra un modelo que no existe tienen
que FALLAR. Sin eso, un arnés roto («no contestó») se lee igual que un motor que anda.

El modelo de prueba es `stories260K` (1,1 MB, 260 mil parámetros — el mismo que usa el CI de
llama.cpp). Se pinea por **sha256**, que es la identidad; la URL no lo es (ADR 0013) y por eso se
puede servir desde donde sea. No es un modelo útil: es un modelo REAL, que es lo que hace falta para
ejercitar la cadena entera. El modelo de producción es su propia unidad de trabajo.

    scripts/test-llama-cpp.py
    scripts/test-llama-cpp.py --sin-modelo     # control negativo: TIENE que fallar
"""
import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

RAIZ = Path(__file__).resolve().parent.parent
RECETA = RAIZ / "recipes/llama-cpp.toml"
TAKANA = RAIZ / "target/release/takana"

MODELO_SHA = "270cba1bd5109f42d03350f60406024560464db173c0e387d91f0426d3bd256d"
MODELO_URL = "https://huggingface.co/ggml-org/models/resolve/main/tinyllamas/stories260K.gguf"
MODELO_CACHE = RAIZ / "work/llama-modelos" / f"{MODELO_SHA[:16]}-stories260K.gguf"

fallos = []


def ok(msg):
    print(f"  \033[32m✓\033[0m {msg}")


def mal(msg):
    print(f"  \033[31m✗\033[0m {msg}")
    fallos.append(msg)


def artefacto():
    """El artefacto VIGENTE, resuelto por `takana hash` — nunca por `ls store/*`.

    Un `ls` agarra el primero que empiece igual, y el corpus es compartido: el de al lado puede ser
    de un hash viejo que ya nadie construye. Pasó con `gmp` en el guardián del foco, y el síntoma
    fue una librería que no relocaba.
    """
    if os.environ.get("LLAMA_DIR"):
        d = Path(os.environ["LLAMA_DIR"])
        print(f"\033[33m⚠ LLAMA_DIR: corriendo contra {d} A SABIENDAS — el resultado NO se puede\033[0m")
        print("\033[33m  citar como «el llama-cpp de hoy pasa».\033[0m")
        return d
    h = subprocess.run([str(TAKANA), "--store", str(RAIZ / "store"), "hash", str(RECETA)],
                       capture_output=True, text=True, cwd=RAIZ)
    if h.returncode != 0:
        sys.exit(f"no se pudo hashear la receta: {h.stderr.strip()}")
    d = RAIZ / "store" / f"{h.stdout.strip().removeprefix('b3:')}-llama-cpp"
    if not d.is_dir():
        sys.exit(f"el artefacto vigente NO está en el store: {d.name}\n"
                 f"  construilo:  flock -o work/.farm-build.lock ./target/release/takana "
                 f"--store ./store build recipes/llama-cpp.toml")
    return d


def bajar_modelo():
    if MODELO_CACHE.is_file():
        got = hashlib.sha256(MODELO_CACHE.read_bytes()).hexdigest()
        if got == MODELO_SHA:
            return MODELO_CACHE
        MODELO_CACHE.unlink()
    MODELO_CACHE.parent.mkdir(parents=True, exist_ok=True)
    print(f"  · bajando el modelo de prueba ({MODELO_URL.rsplit('/', 1)[-1]}, 1,1 MB)…")
    tmp = MODELO_CACHE.with_suffix(".incoming")
    with urllib.request.urlopen(MODELO_URL, timeout=120) as r, open(tmp, "wb") as f:
        f.write(r.read())
    got = hashlib.sha256(tmp.read_bytes()).hexdigest()
    if got != MODELO_SHA:
        tmp.unlink()
        sys.exit(f"el modelo bajado NO es el pineado:\n  esperado {MODELO_SHA}\n  obtenido {got}")
    tmp.rename(MODELO_CACHE)
    return MODELO_CACHE


# ── 1. ISA ────────────────────────────────────────────────────────────────────────────────────
def probar_isa(art):
    print("\n\033[1m1. ISA — ¿el motor trae instrucciones vectoriales?\033[0m")
    lib = art / "usr/lib/libggml-cpu.a"
    if not lib.is_file():
        return mal("no hay libggml-cpu.a en el artefacto")
    d = subprocess.run(["objdump", "-d", str(lib)], capture_output=True, text=True)
    if d.returncode != 0:
        return mal(f"objdump falló sobre libggml-cpu.a: {d.stderr.strip()[:120]}")
    for patron, quien in (("%ymm", "AVX2 (registros de 256 bits)"),
                          ("vfmadd", "FMA"),
                          ("roundps", "SSE4.2")):
        n = d.stdout.count(patron)
        (ok if n else mal)(f"{quien}: {n} apariciones de `{patron}`"
                           + ("" if n else "  ← INS_ENB quedó apagado (ver la cabecera)"))


# ── 2. hermético ──────────────────────────────────────────────────────────────────────────────
def probar_hermetico(art):
    print("\n\033[1m2. Hermético — ¿el binario pide algo de afuera?\033[0m")
    for nombre in ("llama-server", "llama-cli"):
        b = art / "usr/bin" / nombre
        if not b.is_file():
            mal(f"{nombre} no está en el artefacto")
            continue
        r = subprocess.run(["readelf", "-d", str(b)], capture_output=True, text=True)
        needed = re.findall(r"\(NEEDED\).*\[(.+?)\]", r.stdout)
        (ok if not needed else mal)(
            f"{nombre}: {'sin NEEDED (estático)' if not needed else 'pide ' + ', '.join(needed)}")


# ── 3. inferencia ─────────────────────────────────────────────────────────────────────────────
def probar_inferencia(art, modelo):
    print("\n\033[1m3. Inferencia — ¿sale un token de un modelo de verdad?\033[0m")
    r = subprocess.run([str(art / "usr/bin/llama-cli"), "-m", str(modelo),
                        "-p", "Once upon a time", "-n", "24", "--no-warmup",
                        "--single-turn", "--seed", "1"],
                       capture_output=True, text=True, timeout=180)
    if r.returncode != 0:
        return mal(f"llama-cli salió {r.returncode}: {r.stderr.strip().splitlines()[-1:] or ''}")
    # La salida del turno viene después del prompt echoado; nos alcanza con que haya texto nuevo.
    cuerpo = r.stdout.split("Once upon a time", 1)[-1]
    generado = "".join(l for l in cuerpo.splitlines()
                       if l.strip() and not l.startswith(("[", ">", "Exiting")))
    (ok if len(generado.strip()) >= 10 else mal)(
        f"generó {len(generado.strip())} caracteres: {generado.strip()[:70]!r}")


# ── 4. embeddings ─────────────────────────────────────────────────────────────────────────────
def vector(puerto, texto):
    req = urllib.request.Request(
        f"http://127.0.0.1:{puerto}/v1/embeddings",
        data=json.dumps({"input": texto, "model": "prueba"}).encode(),
        headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=60) as r:
        return json.load(r)["data"][0]["embedding"]


def probar_embeddings(art, modelo):
    print("\n\033[1m4. Embeddings — el endpoint del que cuelga la mitad semántica del §6.3\033[0m")
    puerto = 18131
    srv = subprocess.Popen(
        # ⚠ `--pooling mean` NO es un gusto, y la razón conviene saberla ANTES de pinear el
        # modelo de producción: un GGUF declara su tipo de pooling en los metadatos, y los modelos
        # de EMBEDDING de verdad (e5, bge, nomic) lo traen. `stories260K` es un LM causal de
        # juguete y no lo trae, así que llama.cpp cae en `none` y el endpoint compatible con OpenAI
        # contesta **400: "Pooling type \'none\' is not OAI compatible"** — no un vector vacío, no
        # un 500: un 400 que parece un error del cliente. Se fuerza `mean`, que es lo que hace
        # cualquier motor de embeddings sobre un LM causal.
        [str(art / "usr/bin/llama-server"), "-m", str(modelo), "--embedding",
         "--pooling", "mean",
         "--host", "127.0.0.1", "--port", str(puerto), "-c", "512", "--no-warmup"],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        listo = False
        for _ in range(120):
            if srv.poll() is not None:
                break
            # ⚠ `/health` contesta MIENTRAS carga, con 503 y `{"status":"loading model"}`. Tomar
            # «contestó» por «listo» manda la primera consulta contra un servidor a medio levantar
            # y devuelve un 503 que se lee como «el endpoint no existe». Se espera el `ok`.
            try:
                with urllib.request.urlopen(f"http://127.0.0.1:{puerto}/health", timeout=2) as r:
                    if json.load(r).get("status") == "ok":
                        listo = True
                        break
            except Exception:
                pass
            time.sleep(0.5)
        if not listo:
            return mal(f"llama-server no levantó (salió {srv.poll()})")
        ok(f"llama-server --embedding escucha en :{puerto}")

        a1 = vector(puerto, "el zorro salta sobre el perro")
        a2 = vector(puerto, "el zorro salta sobre el perro")
        b1 = vector(puerto, "la reunión de contabilidad es el martes")

        (ok if len(a1) > 8 else mal)(f"devuelve un vector de {len(a1)} dimensiones")
        (ok if a1 == a2 else mal)(
            "el MISMO texto da el MISMO vector" if a1 == a2
            else "el mismo texto dio DOS vectores distintos — no es determinista")
        (ok if a1 != b1 else mal)(
            "dos textos DISTINTOS dan vectores distintos" if a1 != b1
            else "dos textos distintos dieron el MISMO vector — es un stub, no un modelo")
        if any(x != 0.0 for x in a1):
            ok("el vector no es todo ceros")
        else:
            mal("el vector es TODO CEROS — pasaría cualquier prueba de «¿hay un vector?»")
    finally:
        srv.terminate()
        try:
            srv.wait(timeout=10)
        except subprocess.TimeoutExpired:
            srv.kill()


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--sin-modelo", action="store_true",
                   help="control negativo: las sondas 3 y 4 corren contra un modelo inexistente "
                        "y TIENEN que fallar")
    a = p.parse_args()

    art = artefacto()
    print(f"\033[1martefacto:\033[0m {art.name}")

    if a.sin_modelo:
        print("\n\033[33m── CONTROL NEGATIVO: modelo inexistente; lo correcto es que FALLE ──\033[0m")
        modelo = RAIZ / "work/llama-modelos/NO-EXISTE.gguf"
        probar_inferencia(art, modelo)
        probar_embeddings(art, modelo)
        if fallos:
            print(f"\n\033[32m✓ CONTROL NEGATIVO OK\033[0m — {len(fallos)} sondas fallaron, "
                  "como tenían que fallar. El arnés mide algo.")
            return 0
        print("\n\033[31m✗ CONTROL NEGATIVO ROTO\033[0m — todo pasó SIN modelo: el arnés no mide nada.")
        return 1

    probar_isa(art)
    probar_hermetico(art)
    modelo = bajar_modelo()
    probar_inferencia(art, modelo)
    probar_embeddings(art, modelo)

    print()
    if fallos:
        print(f"\033[31m✗ {len(fallos)} fallo(s)\033[0m")
        for f in fallos:
            print(f"    · {f}")
        return 1
    print("\033[32m✓ el motor de inferencia local del corpus infiere y vectoriza\033[0m")
    return 0


if __name__ == "__main__":
    sys.exit(main())