takana/scripts/test-atuq-archivo-semantico.py

#!/usr/bin/env python3
"""¿Se le puede preguntar al archivo personal por lo que DECÍA? (SDD 26 §6.3, mitad semántica)

    python3 scripts/test-atuq-archivo-semantico.py                    # con el modelo de la imagen
    python3 scripts/test-atuq-archivo-semantico.py --negative-control  # sin modelo: lo DICE

── QUÉ SE EJERCITA ───────────────────────────────────────────────────────────────────────────
El navegador archiva tres páginas de temas distintos —eso es el §6.3 literal, que ya existía— y
después se le pregunta al host **desde dentro de la jaula y por el cable de verdad** (marco de 4
bytes), con los paths de PRODUCCIÓN. La respuesta tiene que ordenar por parecido: la página del
tema preguntado primero.

Y se pregunta DOS cosas distintas, que ganan páginas distintas. Con una sola pregunta, un ranking
constante —o uno que siempre devuelve la más reciente— pasaría el test sin haber entendido nada.

── POR QUÉ EL HOST SE INVOCA A MANO Y NO POR LA EXTENSIÓN ────────────────────────────────────
Porque así se comprueba lo que de verdad importa del archivo: **que quedó en disco y se consulta
sin el navegador**. Es el mismo camino que usa `scripts/test-atuq-archivo.py` para la búsqueda
literal.

REQUISITOS: rootfs de `atuq-nested.sh`, y `atuq` + `puriy-costura` + `llama-cpp` +
`ia-modelo-embeddings` sellados.
"""
import http.server
import json
import os
import shutil
import struct
import subprocess
import sys
import tempfile
import threading

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RFS = os.environ.get("RFS", "/mnt/cosecha/escritorios/atuq-rootfs")
ALPINE = os.path.join(ROOT, ".dev-fs/alpine")
HAMMER = os.path.join(ROOT, "target/release/takana")
STORE = os.environ.get("HAMMER_STORE", os.path.join(ROOT, "store"))
MODELO_EN_IMAGEN = "/usr/share/takana/ia/embeddings.gguf"

# Tres páginas de temas bien separados, y dos preguntas que NO comparten palabras con el texto de
# su página: si compartieran, la literal también acertaría y esto no probaría nada semántico.
PAGINAS = {
    "/gato.html": ("El felino en el tejado",
                   "Un minino atigrado dormía sobre las tejas calientes toda la siesta."),
    "/red.html": ("Cerrar la salida de una aplicación",
                  "Reglas de egress por cgroup, denegar por defecto y permitir sólo lo declarado."),
    "/pan.html": ("Masa madre",
                  "Harina, agua y sal; fermentación lenta durante la noche y horno muy caliente."),
}
PREGUNTAS = [
    ("¿dónde se echó a dormir el gato?", "/gato.html"),
    ("cómo bloqueo internet a un programa", "/red.html"),
]


def fatal(msg):
    print(f"✗ {msg}", file=sys.stderr)
    raise SystemExit(1)


def artefacto(receta, nombre, escotilla=None):
    forzado = os.environ.get(escotilla) if escotilla else None
    if forzado:
        if not os.path.isdir(forzado):
            fatal(f"{escotilla}={forzado} no existe")
        print(f"⚠ {escotilla}: se mide {os.path.basename(forzado)}, que NO es necesariamente el vigente")
        return forzado
    r = subprocess.run([HAMMER, "--store", STORE, "hash", f"recipes/{receta}.toml"],
                       cwd=ROOT, capture_output=True, text=True)
    if r.returncode != 0:
        fatal(f"no pude hashear {receta}: {r.stderr.strip()}")
    h = r.stdout.strip().splitlines()[-1].removeprefix("b3:")
    d = os.path.join(STORE, f"{h}-{nombre}")
    if not os.path.isdir(d):
        fatal(f"{nombre} b3:{h[:8]} no está sellado — construilo antes")
    if not os.listdir(d):
        fatal(f"{nombre} b3:{h[:8]} está VACÍO: eso no es un artefacto, es un nombre")
    return d


class Servidor(http.server.SimpleHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_GET(self):
        ruta = self.path.split("?")[0]
        if ruta in PAGINAS:
            titulo, texto = PAGINAS[ruta]
            # Cada página lleva a la siguiente: tres cargas de PRIMER NIVEL con un solo arranque.
            orden = list(PAGINAS)
            i = orden.index(ruta)
            siguiente = orden[i + 1] if i + 1 < len(orden) else "/fin.html"
            datos = (f'<!doctype html><html><head><meta charset="utf-8"><title>{titulo}</title>'
                     f'<meta http-equiv="refresh" content="3;url={siguiente}">'
                     f"</head><body><h1>{titulo}</h1><p>{texto}</p></body></html>").encode()
        elif ruta == "/ir.html":
            # ⚠ La primera versión de esto metía las tres páginas en `<iframe>` y NO archivaba
            # ninguna: el archivo ignora a propósito lo que no es el marco principal (un iframe de
            # publicidad no es una página que alguien leyó). Ahora se ENCADENAN como navegación de
            # verdad, que es lo que hace una persona.
            datos = (b'<!doctype html><html><head><meta charset="utf-8">'
                     b'<meta http-equiv="refresh" content="1;url=/gato.html">'
                     # Sin texto visible: el archivo descarta lo que no tiene nada que buscar, así
                     # que esta página de tránsito no ensucia la evidencia con una entrada sin título.
                     b"</head><body></body></html>")
        else:
            # Igual que `/ir.html`: sin texto visible, para que el final de la cadena no se archive.
            datos = b"<html><body></body></html>"
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(datos)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(datos)


def puerto_libre():
    import socket
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p


def marco(obj):
    """El cable de native messaging: 4 bytes little-endian y el cuerpo."""
    cuerpo = json.dumps(obj).encode()
    return struct.pack("<I", len(cuerpo)) + cuerpo


def correr(capas, salida, guion):
    subprocess.run([
        "bwrap",
        "--overlay-src", ALPINE, "--overlay-src", RFS, *sum((["--overlay-src", c] for c in capas), []),
        "--tmp-overlay", "/",
        "--dev-bind", "/dev", "/dev", "--proc", "/proc", "--bind", "/sys", "/sys",
        "--tmpfs", "/tmp", "--tmpfs", "/run", "--unshare-pid",
        "--bind", salida, "/salida",
        "--uid", "0", "--gid", "0",
        "--setenv", "PATH", "/usr/bin:/bin:/usr/sbin:/sbin",
        "--setenv", "XDG_RUNTIME_DIR", "/run",
        "--setenv", "XDG_DATA_DIRS", "/usr/share", "--setenv", "MOZ_HEADLESS", "1",
        "--setenv", "MOZ_DISABLE_CONTENT_SANDBOX", "1",
        "--setenv", "MOZ_DISABLE_GMP_SANDBOX", "1",
        "--setenv", "MOZ_DISABLE_RDD_SANDBOX", "1",
        "--setenv", "MOZ_DISABLE_SOCKET_PROCESS_SANDBOX", "1",
        "--setenv", "MOZ_DISABLE_UTILITY_SANDBOX", "1",
        "sh", "-c", guion,
    ], check=False)


def main():
    negativo = "--negative-control" in sys.argv[1:]
    if negativo:
        print("== MODO CONTROL NEGATIVO: sin modelo de embeddings; tiene que DECIRLO")
    if not os.path.isdir(RFS):
        fatal(f"no está el rootfs en {RFS} — corré antes `scripts/atuq-nested.sh`")
    atuq = artefacto("atuq", "atuq", "ATUQ_DIR")
    costura = artefacto("puriy-costura", "puriy-costura")
    llama = artefacto("llama-cpp", "llama-cpp", "LLAMA_DIR")
    modelo = artefacto("ia-modelo-embeddings", "ia-modelo-embeddings")
    print(f"artefactos: {os.path.basename(atuq)}\n            {os.path.basename(costura)}"
          f"\n            {os.path.basename(llama)}\n            {os.path.basename(modelo)}")

    puerto = puerto_libre()
    servidor = http.server.ThreadingHTTPServer(("127.0.0.1", puerto), Servidor)
    threading.Thread(target=servidor.serve_forever, daemon=True).start()

    tmp = tempfile.mkdtemp(prefix="atuq-archivo-sem-")
    salida = os.path.join(tmp, "salida")
    os.makedirs(os.path.join(salida, "perfil"))
    try:
        with open(os.path.join(salida, "perfil", "user.js"), "w") as fh:
            fh.write('user_pref("browser.dom.window.dump.enabled", true);\n')
            fh.write('user_pref("browser.shell.checkDefaultBrowser", false);\n')

        # Las preguntas se escriben ya enmarcadas: así el guión de adentro sólo tiene que
        # volcarlas al binario, sin hacer aritmética de bytes en `sh`.
        for i, (q, _) in enumerate(PREGUNTAS):
            with open(os.path.join(salida, f"pregunta-{i}.bin"), "wb") as fh:
                fh.write(marco({"verb": "archive.ask", "query": q}))

        borrar = f"rm -f {MODELO_EN_IMAGEN}" if negativo else "true"
        guion = f"""
        mkdir -p /root /salida/hogar
        [ -s /etc/machine-id ] || tr -d - < /proc/sys/kernel/random/uuid > /etc/machine-id
        export HOME=/salida/hogar
        {borrar}
        timeout 90 /usr/bin/atuq --profile /salida/perfil --no-remote --headless \
            'http://127.0.0.1:{puerto}/ir.html' > /salida/consola.log 2>&1
        # Y ahora, SIN navegador: el archivo quedó en disco y se le pregunta por el cable.
        E=/salida/hogar/.local/state/puriy-costura
        for i in 0 1; do
            /usr/bin/puriy-costura --state $E < /salida/pregunta-$i.bin > /salida/respuesta-$i.bin 2>/salida/err-$i.txt
        done
        exit 0
        """
        capas = [atuq, costura, llama] + ([] if negativo else [modelo])
        # En el control negativo el modelo NO se monta: `rm -f` sobre una capa de sólo lectura
        # borraría el fichero de la capa temporal, pero montar y borrar es contar dos veces.
        correr(capas if not negativo else [atuq, costura, llama], salida, guion)

        log = open(os.path.join(salida, "consola.log"), errors="replace").read()
        base = next((l[len("IA BASE "):].strip() for l in log.splitlines()
                     if l.startswith("IA BASE ")), None)
        lineas = [l[len("ARCHIVO "):] for l in log.splitlines() if l.startswith("ARCHIVO ")]
        archivadas = [l for l in lineas if l.startswith("ARCHIVADA ")]
        for l in archivadas:
            print("   " + l)
        if len(archivadas) < 3:
            destino = os.path.join(ROOT, "work", "atuq-archivo-sem.log")
            os.makedirs(os.path.dirname(destino), exist_ok=True)
            shutil.copy(os.path.join(salida, "consola.log"), destino)
            fatal(f"se archivaron {len(archivadas)} páginas de 3 — log en {destino}")

        for i, (q, esperada) in enumerate(PREGUNTAS):
            crudo = open(os.path.join(salida, f"respuesta-{i}.bin"), "rb").read()
            if len(crudo) < 4:
                err = open(os.path.join(salida, f"err-{i}.txt"), errors="replace").read()
                fatal(f"el host no contestó a «{q}»: {err.strip()[:300]}")
            n = struct.unpack("<I", crudo[:4])[0]
            r = json.loads(crudo[4:4 + n])
            if negativo:
                if r.get("ok"):
                    fatal("sin modelo CONTESTÓ igual: eso es inventar un orden")
                if "no hay modelo de embeddings" not in (r.get("error") or ""):
                    fatal(f"la causa no nombra lo que falta: {r.get('error')}")
                print(f"\n   RECHAZO {r['error']}")
                print("\n✓ control negativo: sin modelo lo dice, nombrando lo que falta")
                print("✓ y no inventa un orden, que es el único fallo inaceptable en una búsqueda")
                return 0
            if not r.get("ok"):
                fatal(f"la pregunta «{q}» falló: {r.get('error')}")
            m = r.get("matches") or []
            if not m:
                fatal(f"«{q}» no devolvió ninguna coincidencia sobre {r.get('archived')} páginas")
            print(f"\n   «{q}» → indexadas {r['indexed']}/{r['archived']}")
            for c in m:
                print(f"      {c['score']:.4f}  {c['page']['title']}")
            if esperada not in m[0]["page"]["url"]:
                fatal(f"«{q}» puso primera a {m[0]['page']['url']} y esperaba {esperada}"
                      " — el orden no sigue al SIGNIFICADO")

        # ── Y LA BARRA LATERAL, que es por donde lo usa una persona ──────────────────────
        # Lo de arriba mide el host; esto mide que la extensión llegue hasta él y muestre el
        # orden. Sin esta mitad, la función existiría y no habría cómo usarla.
        if not base:
            fatal("la extensión de IA no dijo su URL base: no puedo abrir el panel")
        q, esperada = PREGUNTAS[0]
        panel = f"{base}panel.html?p={q.replace(' ', '%20').replace('¿', '%C2%BF')}"
        guion2 = f"""
        mkdir -p /root /salida/hogar
        [ -s /etc/machine-id ] || tr -d - < /proc/sys/kernel/random/uuid > /etc/machine-id
        export HOME=/salida/hogar
        timeout 90 /usr/bin/atuq --profile /salida/perfil --no-remote --headless '{panel}' \
            > /salida/panel.log 2>&1
        exit 0
        """
        correr(capas, salida, guion2)
        plog = open(os.path.join(salida, "panel.log"), errors="replace").read()
        coincidencias = [l[len("IA COINCIDENCIA "):] for l in plog.splitlines()
                         if l.startswith("IA COINCIDENCIA ")]
        print("\n   por la barra lateral:")
        for c in coincidencias[:4]:
            print("      " + c)
        if not coincidencias:
            destino = os.path.join(ROOT, "work", "atuq-panel-sem.log")
            os.makedirs(os.path.dirname(destino), exist_ok=True)
            shutil.copy(os.path.join(salida, "panel.log"), destino)
            fatal(f"el panel no mostró ninguna coincidencia — log en {destino}")
        if esperada not in coincidencias[0]:
            fatal(f"el panel puso primera a «{coincidencias[0]}» y esperaba {esperada}")

        print("\n✓ el archivo contesta por PARECIDO: cada pregunta pone primera a su página")
        print("✓ y son dos preguntas con ganadores distintos — un ranking constante no pasaría")
        print("✓ todo sin navegador: el archivo quedó en disco y se consulta por el cable")
        print("✓ y por la barra lateral también: la extensión llega al host y muestra el orden")
        return 0
    finally:
        servidor.shutdown()
        shutil.rmtree(tmp, ignore_errors=True)


sys.exit(main())