takana/scripts/targets.py

#!/usr/bin/env python3
# targets.py — lee `docs/state/targets.toml` (el manifiesto de objetivo) y expande un perfil a su
# lista de RAÍCES. Un solo parser para dos consumidores: los scripts de imagen (shell) y el grafo
# de estado (`build-state.py`), para que el set de paquetes de la distro no vuelva a vivir en tres
# strings de shell distintos.
#
# Uso:  scripts/targets.py cli               # raíces expandidas, una línea separada por espacios
#       scripts/targets.py --lista           # perfiles disponibles, con su descripción
#       scripts/targets.py --json            # todo el manifiesto expandido
#       scripts/targets.py --services cli    # los servicios que ESA imagen arranca, resueltos
#       scripts/targets.py --selftest        # roturas a propósito + un control que TIENE que pasar
#       scripts/targets.py --service-paths gnome  # rutas de receta de los servicios de SISTEMA
#
# ⚠ La FLAG va en inglés (regla 4 del CLAUDE.md: la superficie de CLI es contrato) aunque `--lista`
# sea deuda vieja del mismo fichero. La CLAVE del TOML sí es `servicios`, en castellano, porque ahí
# la convención local es esa (`paquetes`, `hereda`, `cola`, `descripcion`) y mezclar idiomas DENTRO
# del manifiesto sería peor que la coherencia con el resto de la superficie.
# Como módulo:  from targets import load  →  {perfil: {descripcion, cola, hereda, paquetes, raices}}
#
# `hereda` es transitivo y el orden se PRESERVA (primero las raíces de lo heredado, en el orden
# declarado, después las propias), porque hay imágenes donde el orden de hidratación importa
# (mirada: libs antes que binarios). Duplicados: se queda la primera aparición.
import json, sys, tomllib
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
DEFAULT = ROOT / "docs/state/targets.toml"


def load(path=DEFAULT):
    """Devuelve {perfil: {descripcion, cola, hereda, paquetes, raices}} con `raices` ya expandido."""
    doc = tomllib.load(open(path, "rb"))
    crudos = doc.get("perfil", {})
    perfiles = {}
    for n, p in crudos.items():
        perfiles[n] = dict(
            descripcion=p.get("descripcion", ""),
            cola=p.get("cola", "corpus"),
            hereda=list(p.get("hereda", [])),
            paquetes=list(p.get("paquetes", [])),
            # `servicios` = los LABELS que esta imagen arranca. Es el «enable» de la distro: el
            # paquete declara QUÉ es el servicio (`[[service]]` en su receta) y el perfil decide SI
            # arranca, igual que systemd parte [Service] de [Install]. Ver SDD 30 §3.
            servicios=list(p.get("servicios", [])),
        )

    # Expansión de `hereda` con detección de ciclo (un manifiesto con un ciclo es un error del
    # humano, no un caso a tolerar: sale ruidoso).
    def expandir(n, camino):
        if n in camino:
            raise ValueError(f"ciclo en `hereda`: {' → '.join(camino + [n])}")
        if n not in perfiles:
            raise KeyError(f"perfil desconocido en `hereda`: {n}")
        p = perfiles[n]
        if "raices" in p:
            return p["raices"]
        vistos, raices = set(), []
        vistos_s, svcs = set(), []
        for h in p["hereda"] + [None]:
            hp = perfiles[h] if h else None
            for pkg in (expandir(h, camino + [n]) if h else p["paquetes"]):
                if pkg not in vistos:
                    vistos.add(pkg); raices.append(pkg)
            # los servicios heredados viajan con el perfil padre (ya expandido por la llamada de
            # arriba): una imagen que hereda `cli` arranca también lo que `cli` arranca.
            for s in (hp["servicios_exp"] if hp else p["servicios"]):
                if s not in vistos_s:
                    vistos_s.add(s); svcs.append(s)
        p["raices"] = raices
        p["servicios_exp"] = svcs
        return raices

    for n in perfiles:
        expandir(n, [])
    return perfiles


# ── El lado «habilitar»: perfil → servicios resueltos ────────────────────────────────────────────
#
# Un servicio se declara en la receta (`[[service]]`, SDD 30 §3.1) y se HABILITA en el perfil
# (`servicios = [...]`). Este resolutor cruza las dos mitades y, sobre todo, hace la comprobación
# INVERSA — la que ninguna métrica hacía: un paquete del perfil que TRAE un demonio y que el perfil
# no arranca. Esa es la versión servicios de la lección de `foot` que este manifiesto ya aprendió
# caro: la métrica mide la clausura de lo DECLARADO y no puede ver lo que falta en la declaración.
# Una imagen puede salir 121/121 sellada, con el demonio adentro, y sin que nada lo levante nunca.

def declaraciones(recipes_dir=None):
    """Escanea el corpus y devuelve {label: [(receta, servicio_dict), ...]}.

    Devuelve la LISTA por label (no el primero) a propósito: dos recetas que declaren el mismo
    label es un error que hay que ver, y quedarse con una lo escondería."""
    recipes_dir = Path(recipes_dir) if recipes_dir else ROOT / "recipes"
    por_label = {}
    for f in sorted(recipes_dir.rglob("*.toml")):
        try:
            doc = tomllib.load(open(f, "rb"))
        except tomllib.TOMLDecodeError:
            continue  # recetas rotas las reporta el grafo, no este resolutor
        for s in doc.get("service", []):
            lbl = s.get("label")
            if lbl:
                por_label.setdefault(lbl, []).append((doc.get("name", f.stem), s))
    return por_label


# Los grafos por COLA. `build-state.json` es sólo el corpus: las recetas de un escritorio viven en
# `recipes/incoming-<x>/` y `build-state.py` sólo las carga con su flag, así que su membresía de
# perfil está en el grafo de SU cola. Mirar sólo el del corpus daría "no puedo comprobar" para todo
# GNOME/KDE/COSMIC — que es peor que no comprobar: parece una respuesta.
GRAFOS = ["build-state.json", "build-state-gnome.json", "build-state-kde.json",
          "build-state-cosmic.json", "build-state-wlr.json"]


def _ruta_de(nombre, recipes_dir=None):
    """La ruta de la receta que se llama `nombre`. Se busca por el campo `name`, no por el nombre de
    fichero: son casi siempre iguales, pero el que manda es el campo."""
    recipes_dir = Path(recipes_dir) if recipes_dir else ROOT / "recipes"
    for f in sorted(recipes_dir.rglob("*.toml")):
        try:
            if tomllib.load(open(f, "rb")).get("name") == nombre:
                return str(f.relative_to(ROOT))
        except tomllib.TOMLDecodeError:
            continue
    return None


def _membresia():
    """{receta: [perfiles]} fusionando TODOS los grafos de estado. `{}` si no hay ninguno (no es
    fatal: degrada a no poder comprobar la pertenencia, y se dice)."""
    memb = {}
    for nombre in GRAFOS:
        try:
            d = json.load(open(ROOT / "docs/state" / nombre))
        except (FileNotFoundError, json.JSONDecodeError):
            continue
        for k, v in d.get("nodes", {}).items():
            # unión: una receta del corpus puede aparecer en varios grafos con perfiles distintos
            ya = memb.setdefault(k, [])
            for perf in v.get("perfiles", []):
                if perf not in ya:
                    ya.append(perf)
    return memb


def resolver_servicios(perfil, perfiles=None):
    """Devuelve (habilitados, problemas) para un perfil.

    habilitados: [(label, receta, exec)] en el orden declarado.
    problemas:   lista de strings. Distingue ERROR (rompe la imagen) de AVISO (huele mal)."""
    perfiles = perfiles or load()
    p = perfiles[perfil]
    decl = declaraciones()
    memb = _membresia()
    raices = set(p["raices"])

    def en_el_perfil(receta):
        """`True`/`False`/`None` (= no se puede saber).

        Las RAÍCES mandan sobre el grafo, y el orden importa: `targets.toml` es la fuente de verdad
        de lo que la imagen quiere y `build-state.json` es DERIVADO de él —lo regenera el cron cada
        30 min—. Preguntarle primero al derivado haría que añadir una raíz se leyera como error
        hasta la siguiente cosecha, que es castigar al que arregla el manifiesto."""
        if receta in raices:
            return True
        perfs = memb.get(receta)
        if perfs is None:
            return None
        return perfil in perfs

    habilitados, problemas = [], []
    for lbl in p["servicios_exp"]:
        cands = decl.get(lbl, [])
        if not cands:
            problemas.append(f"ERROR  {lbl}: habilitado por el perfil y NINGUNA receta lo declara")
            continue
        # La unicidad del label se juzga DENTRO del perfil, no en el corpus entero: el mismo
        # servicio existe legítimamente en dos colas (p.ej. `upower` en incoming-gnome y en
        # incoming-kde) y son la misma unidad en dos imágenes distintas. Exigir unicidad global
        # convertiría esa convivencia normal en un error.
        dentro = [(r, s) for r, s in cands if en_el_perfil(r) is True]
        incierto = [(r, s) for r, s in cands if en_el_perfil(r) is None]
        if not dentro and incierto:
            quienes = ", ".join(r for r, _ in incierto)
            problemas.append(f"AVISO  {lbl}: no puedo comprobar que `{quienes}` esté en el perfil "
                             f"(sin build-state.json y no es raíz)")
            receta, svc = incierto[0]
        elif not dentro:
            quienes = ", ".join(r for r, _ in cands)
            problemas.append(
                f"ERROR  {lbl}: lo declara(n) {quienes}, que NO pertenece(n) al perfil ⇒ el card "
                f"apuntaría a un binario que la imagen no trae")
            continue
        elif len(dentro) > 1:
            quienes = ", ".join(r for r, _ in dentro)
            problemas.append(f"ERROR  {lbl}: lo declaran {len(dentro)} recetas DEL PERFIL ({quienes})")
            continue
        else:
            receta, svc = dentro[0]
        if svc.get("scope", "system") != "system":
            problemas.append(
                f"AVISO  {lbl}: es de SESIÓN (`{receta}`) — hoy lo lanza el script de sesión, no "
                f"arje: fuera de mirada nadie entrega cards de sesión (SDD 30 §4c)")
        habilitados.append((lbl, receta, svc.get("exec", "?")))

    # La inversa: paquetes del perfil que TRAEN demonio y nadie arranca.
    hab = {l for l, _, _ in habilitados}
    for lbl, cands in sorted(decl.items()):
        if lbl in hab:
            continue
        for receta, _ in cands:
            if en_el_perfil(receta) is True:
                problemas.append(
                    f"AVISO  {lbl}: `{receta}` está en la imagen y TRAE este servicio, "
                    f"pero el perfil no lo arranca")
    return habilitados, problemas


def autotest():
    """Roturas A PROPÓSITO + un control que TIENE que pasar.

    Un guardián que nunca falló no se sabe si sirve, y uno que sólo se prueba con casos rotos puede
    estar gritando siempre. Por eso el primer caso es el BUENO: si ese no pasa en verde, el resto de
    los ✓ no prueban nada."""
    global declaraciones, _membresia
    orig_d, orig_m = declaraciones, _membresia
    SVC = {"label": "sshd", "exec": "/bin/busybox"}
    SES = {"label": "pipewire", "exec": "/usr/bin/pipewire", "scope": "session"}

    def perfil(svcs, raices=("openssh",)):
        return dict(descripcion="", cola="corpus", hereda=[], paquetes=list(raices),
                    servicios=list(svcs), raices=list(raices), servicios_exp=list(svcs))

    casos = [
        # (nombre, perfil, declaraciones, membresía, ¿espera ERROR?)
        ("CONTROL: declarado, habilitado y la receta en el perfil",
         perfil(["sshd"]), {"sshd": [("openssh", SVC)]}, {"openssh": ["prueba"]}, False),
        ("habilitado y NADIE lo declara",
         perfil(["sshd"]), {}, {}, True),
        ("dos recetas DEL PERFIL declaran el mismo label",
         perfil(["sshd"], ("openssh", "dropbear")),
         {"sshd": [("openssh", SVC), ("dropbear", SVC)]}, {}, True),
        ("lo declara una receta que NO está en el perfil",
         perfil(["sshd"], ()), {"sshd": [("openssh", SVC)]}, {"openssh": ["otro"]}, True),
        ("mismo label en DOS COLAS, sólo una en el perfil ⇒ NO es colisión",
         perfil(["sshd"]), {"sshd": [("openssh", SVC), ("openssh-kde", SVC)]},
         {"openssh": ["prueba"], "openssh-kde": ["otro"]}, False),
        ("servicio de SESIÓN ⇒ avisa pero no rompe",
         perfil(["pipewire"], ("pipewire",)), {"pipewire": [("pipewire", SES)]}, {}, False),
        ("la INVERSA: en la imagen, trae demonio, nadie lo arranca ⇒ AVISO",
         perfil([]), {"sshd": [("openssh", SVC)]}, {"openssh": ["prueba"]}, False),
    ]
    todo_ok = True
    try:
        for nombre, p_, decl, memb, espera in casos:
            declaraciones = lambda *a, **k: decl
            _membresia = lambda: memb
            _, prob = resolver_servicios("prueba", {"prueba": p_})
            hay = any(x.startswith("ERROR") for x in prob)
            ok = hay == espera
            todo_ok &= ok
            print(f"{'✓' if ok else '✗'} {nombre}")
            for x in prob:
                print(f"      {x}")
    finally:
        declaraciones, _membresia = orig_d, orig_m
    print("TODAS OK" if todo_ok else "HAY FALLOS")
    return todo_ok


def main():
    args = sys.argv[1:]
    try:
        perfiles = load()
    except FileNotFoundError:
        print(f"no existe {DEFAULT.relative_to(ROOT)}", file=sys.stderr); return 1
    except (ValueError, KeyError) as e:
        print(f"manifiesto inválido: {e}", file=sys.stderr); return 1

    if not args or args[0] in ("--lista", "-l"):
        for n, p in sorted(perfiles.items()):
            print(f"{n:20} {len(p['raices']):3} raíces  [{p['cola']}]  {p['descripcion']}")
        return 0
    if args[0] == "--selftest":
        return 0 if autotest() else 1
    if args[0] == "--service-paths":
        # Las RUTAS de receta de los servicios de SISTEMA habilitados, para que el ensamblado de la
        # imagen se las pase a `takana service cards`. Sólo `system`: una Card de sesión en el
        # `genesis` arrancaría antes de que exista XDG_RUNTIME_DIR o un usuario (SDD 30 §4c).
        if len(args) < 2 or args[1] not in perfiles:
            print(f"uso: targets.py --service-paths <perfil>", file=sys.stderr)
            return 1
        decl = declaraciones()
        memb = _membresia()
        raices = set(perfiles[args[1]]["raices"])
        vistas = []
        for lbl in perfiles[args[1]]["servicios_exp"]:
            for receta, svc in decl.get(lbl, []):
                if svc.get("scope", "system") != "system":
                    continue
                if not (receta in raices or args[1] in memb.get(receta, [])):
                    continue
                ruta = _ruta_de(receta)
                if ruta and ruta not in vistas:
                    vistas.append(ruta)
        for r in vistas:
            print(r)
        return 0
    if args[0] in ("--services", "-s"):
        if len(args) < 2 or args[1] not in perfiles:
            print(f"uso: targets.py --services <perfil>  (hay: {', '.join(sorted(perfiles))})",
                  file=sys.stderr)
            return 1
        habilitados, problemas = resolver_servicios(args[1], perfiles)
        for lbl, receta, ex in habilitados:
            print(f"{lbl:24} {receta:20} {ex}")
        if not habilitados:
            print("(el perfil no arranca ningún servicio)")
        for pr in problemas:
            print(pr, file=sys.stderr)
        # Sólo un ERROR hace fallar: un AVISO es para leer, no para romper el cron.
        return 1 if any(x.startswith("ERROR") for x in problemas) else 0
    if args[0] == "--json":
        print(json.dumps(perfiles, indent=1, ensure_ascii=False)); return 0

    n = args[0]
    if n not in perfiles:
        print(f"perfil desconocido: {n} (hay: {', '.join(sorted(perfiles))})", file=sys.stderr)
        return 1
    print(" ".join(perfiles[n]["raices"]))
    return 0


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