takana/scripts/yupana.py

#!/usr/bin/env python3
# yupana.py — la YUPANA del build: el ábaco que RECKONA sobre el estado de la distro.
#
# DOS OBJETOS ANDINOS, DOS ROLES. En los Andes el khipu GUARDABA (cuerdas anudadas que registran
# cantidades y relaciones) y la yupana CALCULABA sobre ese registro (el ábaco inca). Acá es igual y
# la distinción es la arquitectura, no adorno:
#     · `docs/state/build-state.json` es el KHIPU — el registro firme y versionado: los nudos
#       (recetas) y las cuerdas (deps), la única fuente de verdad de QUÉ hay y QUÉ falta.
#     · este módulo es la YUPANA — el motor que reckona sobre el khipu: a QUIÉN rompo si toco algo
#       (radio), qué imágenes dependen de un nudo (perfiles), en qué orden masticar (ondas).
# Es la BASE metodológica y la PUERTA ÚNICA; el resto de los scripts de estado son sus órganos.
# Antes este saber vivía disperso (en la cabeza, en `grep`, en docs que envejecen) y cada medición
# a mano mentía distinto.
#
# POR QUÉ EXISTE ESTE FICHERO Y NO SÓLO build-state.py. El 2026-07-22 un cambio a `libdrm`
# (default_library=both, para gtk4) invalidó 118 recetas KDE. El radio se midió contra
# `recipes/*.toml` y NO contra `recipes/incoming-kde/` ⇒ dio 8 en vez de 118. `build-state.py` sin
# `--kde` cometía EXACTAMENTE el mismo error: cargaba sólo el corpus, y su campo `perfiles` llegaba
# a AFIRMAR que tocar `libdrm` sólo afectaba a `escritorio-mirada` — una mentira con confianza. La
# causa es estructural: las dependencias INVERSAS son un hecho del disco entero, no de la vista que
# uno cargó. Este módulo las calcula SIEMPRE sobre TODAS las colas, y es de donde build-state saca
# `perfiles`/`dependientes`. La regla, en una línea:
#
#     QUÉ NODOS PUNTÚO es una decisión de vista. QUIÉN ME CONSUME es un hecho — nunca depende de la vista.
#
# RESOLUCIÓN SIBLING-FIRST. Una dep `libdrm` desde una receta de `incoming-kde` resuelve a
# `incoming-kde/libdrm` si existe, y si no cae al `corpus/libdrm` (igual que el sandbox al montar las
# capas). Por eso los nodos se identifican por PAR `(cola, nombre)`, no por nombre: `corpus/fontconfig`
# y `incoming-kde/fontconfig` son nudos distintos y romper uno no rompe a los consumidores del otro.
#
# yupana es la PUERTA ÚNICA de la metodología. Sus verbos son los órganos del ciclo, en el orden en
# que se usan (el mismo que la cadena del plan docs/plan-catalogo-objetivo.md):
#
#   radio <receta>    ¿a QUIÉN rompo si toco esto? (nativo; cruza TODAS las colas) ← ANTES de cambiar nada
#   perfiles <receta> ¿qué imágenes dependen de esto? (nativo)
#   estado            regenera el grafo firme          → build-state.py
#   objetivo          qué debe tener cada imagen        → targets.py --lista
#   frontera <perfil> qué pide upstream y no tenemos    → seed-graph.py --frontera
#   triaje            clasificar la frontera            → triaje.py
#   drenar [--perfil] el orden de masticado, en ondas   → drenar.py
#
# Uso:  scripts/yupana.py radio <receta> [--json]
#       scripts/yupana.py estado [--kde] | drenar [--todos] | triaje | frontera <perfil> | objetivo
# Como módulo:  from yupana import cargar_colas, membresia, dependientes, resolver
import glob, json, os, subprocess, sys, tomllib
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "scripts"))
import targets as targets_mod


def cargar_colas():
    """{cola: {nombre: [deps de build]}} sobre TODAS las colas del disco. `corpus` = recipes/;
    `incoming-<x>` = recipes/incoming-<x>/. Es el grafo completo, sin recortar por vista."""
    colas = {}
    for f in sorted(glob.glob(str(ROOT / "recipes/*.toml"))):
        n = os.path.basename(f)[:-5]
        colas.setdefault("corpus", {})[n] = _deps(f)
    for f in sorted(glob.glob(str(ROOT / "recipes/incoming-*/*.toml"))):
        q = os.path.basename(os.path.dirname(f))
        colas.setdefault(q, {})[os.path.basename(f)[:-5]] = _deps(f)
    return colas


def _deps(f):
    try:
        return tomllib.load(open(f, "rb")).get("deps", {}).get("build", []) or []
    except Exception:
        return []


# --- INVARIANTE (c): CANÓNICO — sin duplicados accidentales entre orígenes ------------------------
# Sufijos que en hammer marcan una VARIANTE deliberada del mismo paquete (misma fuente, otra decisión
# de build: estático/shared, kernel por perfil, mesa por backend). Compartir sha con estos NO es un
# duplicado a fundir — es la diversificación coherente [[etapa-g-gui-chain-boundary]]. `-hello`/`-hola`
# son demos que vendorean la fuente de otro a propósito.
SUFIJOS_VARIANTE = ("-shared", "-static", "-libs", "-tiny", "-dev", "-headers", "-minimal",
                    "-generic", "-metal-dual", "-metal", "-swrast", "-llvmpipe", "-hello", "-hola")


def _base(n):
    """Reduce un nombre a su base de familia quitando sufijos de variante, repetido: `zlib-shared`→
    `zlib`, `mesa-llvmpipe`→`mesa`, `linux-generic`→`linux`. NO toca el `lib` inicial a propósito
    (`libpng`→`png` colapsaría paquetes sin relación). Para decidir si dos nudos que comparten fuente
    son la MISMA familia (variante ok) o paquetes SIN relación (colisión real)."""
    prev = None
    while prev != n:
        prev = n
        for s in SUFIJOS_VARIANTE:
            if n.endswith(s) and len(n) > len(s):
                n = n[: -len(s)]
    return n


def _fuente(f):
    """(sha, url, version) de una receta — la identidad de lo que BAJA, no de cómo se llama."""
    try:
        d = tomllib.load(open(f, "rb"))
    except Exception:
        return (None, "", None)
    s = d.get("source", {})
    return (s.get("sha256") or s.get("output_hash"),
            s.get("tarball") or s.get("url") or s.get("git") or "",
            d.get("version"))


def _recetas_con_fuente():
    """{(cola, nombre): {sha, url, version, deps}} sobre todo el disco."""
    out = {}
    for pat, es_incoming in (("recipes/*.toml", False), ("recipes/incoming-*/*.toml", True)):
        for f in sorted(glob.glob(str(ROOT / pat))):
            n = os.path.basename(f)[:-5]
            q = os.path.basename(os.path.dirname(f)) if es_incoming else "corpus"
            sha, url, ver = _fuente(f)
            out[(q, n)] = dict(sha=sha, url=url, version=ver, deps=_deps(f))
    return out


def duplicados(como_json=False):
    """INVARIANTE (c): detecta nudos que son el MISMO paquete bajo nombres/colas distintos, por
    EVIDENCIA de fuente (sha/url), no por nombre. Clasifica cada grupo:
      colisión-fuente — mismo sha, nombres de familias DISTINTAS ⇒ casi seguro un sha copiado por
                        error (itstool tenía el tarball de gettext-tiny). ACCIONABLE.
      variante        — mismo sha, misma familia (`zlib`/`zlib-shared`, `mesa-*`) ⇒ deliberado, OK.
      sombra          — mismo NOMBRE en >1 cola. Redundante si versión+sha+deps coinciden; si no,
                        es una variante justificada de la cola.
    NO funde nada solo: reporta y deja el juicio al humano (como el triaje). El radio de cada nudo
    ayuda a elegir el canónico."""
    recs = _recetas_con_fuente()
    from collections import defaultdict
    rev = dependientes()

    # 1) grupos por sha (identidad de fuente)
    por_sha = defaultdict(list)
    for k, r in recs.items():
        if r["sha"]:
            por_sha[r["sha"]].append(k)
    colisiones, variantes = [], []
    for sha, ks in por_sha.items():
        nombres = {n for _, n in ks}
        if len(nombres) < 2:
            continue
        familias = {_base(n) for _, n in ks}
        grupo = dict(sha=sha[:16], nudos=sorted(f"{q}/{n}" for q, n in ks),
                     radio={f"{q}/{n}": len(rev.get((q, n), ())) for q, n in ks})
        (variantes if len(familias) == 1 else colisiones).append(grupo)

    # 2) sombras: mismo nombre en >1 cola
    por_nombre = defaultdict(list)
    for (q, n) in recs:
        por_nombre[n].append(q)
    sombras = []
    for n, qs in por_nombre.items():
        if len(qs) < 2:
            continue
        firmas = {(recs[(q, n)]["version"], recs[(q, n)]["sha"],
                   tuple(sorted(recs[(q, n)]["deps"]))) for q in qs}
        sombras.append(dict(nombre=n, colas=sorted(qs),
                            redundante=len(firmas) == 1,  # idénticas ⇒ la sombra no aporta nada
                            radio={f"{q}/{n}": len(rev.get((q, n), ())) for q in qs}))

    doc = dict(schema="hammer-duplicados/1",
               colisiones_fuente=sorted(colisiones, key=lambda g: -max(g["radio"].values() or [0])),
               variantes=sorted(variantes, key=lambda g: g["nudos"]),
               sombras=sorted(sombras, key=lambda s: (not s["redundante"], s["nombre"])))
    (ROOT / "docs/state/duplicados.json").write_text(
        json.dumps(doc, indent=1, ensure_ascii=False) + "\n")

    if como_json:
        print(json.dumps(doc, indent=1, ensure_ascii=False))
        return 0
    print(f"== INVARIANTE (c) CANÓNICO — duplicados por evidencia de fuente\n")
    print(f"⚠ COLISIONES DE FUENTE: {len(colisiones)}  (mismo sha, familias distintas — revisá el sha)")
    for g in doc["colisiones_fuente"]:
        print(f"   sha {g['sha']}…  {g['nudos']}")
        print(f"        radio: " + "  ".join(f"{k.split('/')[-1]}={v}" for k, v in g["radio"].items()))
    red = [s for s in sombras if s["redundante"]]
    print(f"\n⚠ SOMBRAS REDUNDANTES: {len(red)}  (mismo nombre+versión+sha+deps en 2 colas — una sobra)")
    for s in red:
        print(f"   {s['nombre']:24} {s['colas']}  radio " +
              " ".join(f"{q}={v}" for (q, _), v in
                       ((k.split('/'), v) for k, v in s["radio"].items())))
    print(f"\n✓ variantes deliberadas (misma fuente, familia OK): {len(variantes)}  — no son problema")
    print(f"✓ sombras justificadas (difieren en versión/deps): {len(sombras)-len(red)}")
    print(f"\n   → docs/state/duplicados.json  ·  el radio ayuda a elegir el canónico (el de mayor radio)")
    return 0


def resolver(dep, desde_cola, indice):
    """Sibling-first: (cola, nombre) del nudo que provee `dep` visto desde `desde_cola`, o None si
    ninguna cola lo tiene (dep externa / nodo `wanted`). `indice` = {cola: set(nombres)}."""
    if dep in indice.get(desde_cola, ()):
        return (desde_cola, dep)
    if dep in indice.get("corpus", ()):
        return ("corpus", dep)
    return None


def dependientes(colas=None):
    """{(cola,nombre): set((cola,nombre) que lo consumen)} — el mapa de dependencias INVERSAS sobre
    el grafo entero. Este es el cálculo que, hecho sobre una sola cola, produjo el bug de libdrm."""
    colas = colas or cargar_colas()
    indice = {q: set(m) for q, m in colas.items()}
    rev = {}
    for q, miembros in colas.items():
        for nombre, deps in miembros.items():
            for d in deps:
                objetivo = resolver(d, q, indice)
                if objetivo:
                    rev.setdefault(objetivo, set()).add((q, nombre))
    return rev


def membresia(colas=None, perfiles=None):
    """{(cola,nombre): set(perfiles que lo alcanzan)} sobre el grafo entero, respetando sibling-first.
    Un nudo pertenece a una imagen si la clausura de sus raíces —resolviendo cada cuerda desde la
    cola del consumidor— llega hasta él. Esto es lo que arregla la mentira de `perfiles`."""
    colas = colas or cargar_colas()
    perfiles = perfiles if perfiles is not None else targets_mod.load()
    indice = {q: set(m) for q, m in colas.items()}
    memb = {}
    for pn, p in perfiles.items():
        cola_p = p.get("cola", "corpus")
        vistos, pila = set(), []
        for raiz in p["raices"]:
            nodo = resolver(raiz, cola_p, indice)
            pila.append(nodo if nodo else ("wanted", raiz))  # raíz sin receta = nodo wanted
        while pila:
            nodo = pila.pop()
            if nodo in vistos:
                continue
            vistos.add(nodo)
            memb.setdefault(nodo, set()).add(pn)
            q, nombre = nodo
            if q == "wanted":
                continue
            for d in colas.get(q, {}).get(nombre, []):
                sig = resolver(d, q, indice)
                pila.append(sig if sig else ("wanted", d))
    return memb


def _estado(nodo):
    """Estado del nudo desde el grafo firme, si está regenerado. Best-effort: el radio funciona sin él."""
    q, nombre = nodo
    fichero = "docs/state/build-state-kde.json" if q != "corpus" else "docs/state/build-state.json"
    try:
        d = json.loads((ROOT / fichero).read_text())
        return d["nodes"].get(nombre, {}).get("state", "?")
    except Exception:
        return "?"


def radio(nombre, como_json=False):
    """¿A QUIÉN rompo si re-hasheo `recipes/**/<nombre>.toml`? La pregunta que, mal medida, costó 118
    recetas. Cruza TODAS las colas por construcción: es imposible que 'olvide' una cola."""
    colas = cargar_colas()
    indice = {q: set(m) for q, m in colas.items()}
    # ¿En qué cola vive la receta que se va a tocar? Puede estar en varias (variante sombreada).
    ubic = [(q, nombre) for q in colas if nombre in colas[q]]
    if not ubic:
        print(f"no hay receta `{nombre}` en ninguna cola", file=sys.stderr)
        return 1
    rev = dependientes(colas)
    memb = membresia(colas, targets_mod.load())

    resultado = {}
    for nodo in ubic:
        directos = sorted(rev.get(nodo, set()))
        # Cierre inverso: quién depende, transitivamente (el radio real de un re-hash en cascada).
        vistos, pila = set(), list(rev.get(nodo, set()))
        while pila:
            n = pila.pop()
            if n in vistos:
                continue
            vistos.add(n)
            pila.extend(rev.get(n, set()))
        imgs = set()
        for n in vistos | {nodo}:
            imgs |= memb.get(n, set())
        # Cuántos de los transitivos YA están sellados: ésos son los que HOY están al día y el
        # re-hash MANDA A DEUDA (los que ya están en deuda igual habría que reconstruirlos). El
        # radio verdadero de rebuild forzado por el cambio.
        sellados = sum(1 for n in vistos if _estado(n) == "sealed")
        resultado[f"{nodo[0]}/{nodo[1]}"] = dict(
            directos=len(directos), transitivos=len(vistos),
            imagenes=sorted(imgs), sellados_que_caen=sellados,
            dependientes_por_cola=_por_cola(vistos))

    if como_json:
        print(json.dumps(resultado, indent=1, ensure_ascii=False))
        return 0
    for nodo, r in resultado.items():
        print(f"== radio de tocar  {nodo}")
        print(f"   dependientes directos     {r['directos']}")
        print(f"   dependientes transitivos  {r['transitivos']}   (el radio real de un re-hash en cascada)")
        print(f"   por cola                  " +
              "  ".join(f"{q}={n}" for q, n in sorted(r["dependientes_por_cola"].items())))
        print(f"   IMÁGENES afectadas        {', '.join(r['imagenes']) or '(ninguna declarada)'}")
        print(f"   sellados que CAEN a deuda {r['sellados_que_caen']}   (hoy al día; el re-hash los fuerza a rebuild)")
        if len(resultado) > 1:
            print()
    print("\n   regla: esto cruza TODAS las colas. NO midas radio con `grep recipes/*.toml`.")
    return 0


def _por_cola(nodos):
    from collections import Counter
    return dict(Counter(q for q, _ in nodos))


# Delegación a los órganos: yupana es la puerta, cada script sigue siendo el músculo. Se delega por
# subproceso (no import) para no acoplar sus ciclos de vida ni sus flags — cada uno evoluciona solo.
ORGANOS = {
    "estado":   ["build-state.py"],
    "drenar":   ["drenar.py"],
    "triaje":   ["triaje.py"],
    "frontera": ["seed-graph.py", "--frontera"],
    "objetivo": ["targets.py", "--lista"],
    "sembrar":  ["seed-graph.py", "--perfil"],
}


def delegar(cmd, extra):
    return subprocess.call([sys.executable, str(ROOT / "scripts" / ORGANOS[cmd][0])]
                           + ORGANOS[cmd][1:] + extra)


def main():
    args = sys.argv[1:]
    if not args or args[0] in ("-h", "--help"):
        print(__doc__)
        return 0
    cmd, extra = args[0], args[1:]
    if cmd == "radio" and extra:
        return radio(extra[0], como_json="--json" in extra)
    if cmd == "duplicados":
        return duplicados(como_json="--json" in extra)
    if cmd == "perfiles" and extra:
        memb = membresia()
        hits = {f"{q}/{n}": sorted(ps) for (q, n), ps in memb.items() if n == extra[0]}
        print(json.dumps(hits, indent=1, ensure_ascii=False) if hits
              else f"`{extra[0]}` no lo alcanza ninguna imagen declarada")
        return 0
    if cmd in ORGANOS:
        return delegar(cmd, extra)
    print(f"verbo desconocido: {cmd}\n{__doc__}", file=sys.stderr)
    return 1


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