sergio/takana

`aplicar.py` ejecuta un plan. La regla que lo define: **un paso que no se ejecutó no se marca como
hecho**. Un plan tiene pasos ejecutables y pasos MANUALES cuyo `cmd` son comentarios («instalá el
paquete», «cambiá el registro A»); un aplicador que ejecuta un bloque de comentarios obtiene exit 0 y
lo marca «ok» — la peor mentira posible, porque deja el servicio caído con el informe en verde.

Acá quedan `pendiente-humano`, la corrida sale con ≠0, y `--hecho N` los confirma — negándose si el
paso sí tenía comandos («corrélo, no lo marques»).

Se le cree a la VERIFICACIÓN, no al exit code: el rsync que llenó el disco devolvió 0 y dejó 1367
artefactos vacíos. Si el comando sale bien y la verificación falla, queda `sospechoso`. Y las
verificaciones van TIPADAS (`cmd`/`humano`): «la columna Available debe ser > X» no es un comando.

Estado reanudable en `<plan>.estado.json`, escrito con temporal + fsync + rename — la lección que
costó un upgrade entero en el SDD 28.

**Y un fallo propio, encontrado en la primera corrida real**: la verificación del paso de datos
imprimía el número de directorios vacíos y devolvía 0 igual. Dio «✓ verificado: 2» donde ese 2 eran
dos vacíos en destino. Un guardián que siempre pasa no es un guardián. Ahora COMPARA los ficheros de
los dos lados y falla si difieren. Probado con rotura a propósito y con el control que debe pasar:
copia no hecha ⇒ origen=5 destino=0 ⇒ FALLA; copia hecha ⇒ 5 y 5 ⇒ PASA.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RomoxEGZUhaT4pob1QSX5x

diff --git a/docs/29-mudanza.md b/docs/29-mudanza.md
index 44e1510e..88a5b261 100644
--- a/docs/29-mudanza.md
+++ b/docs/29-mudanza.md
@@ -151,7 +151,50 @@ comando entero salía ≠0 y se descartaba **también el `cmdline`**, que sí se
 servicios quedaban sin invocación —el dato que da sentido al censo— y el fallo se leía como «no se
 pudo». **Una sonda que falla es un dato; no puede arrastrar a las que funcionaron.**
 
-### 4.1 El aplicador *(pendiente)*
+### 4.1 El aplicador *(implementado: `scripts/mudanza/aplicar.py`)*
+
+`aplicar.py --plan plan.toml [--dry-run] [--only <clase>] [--paso N] [--hecho N]`
+
+**La regla que lo define: un paso que no se ejecutó NO se marca como hecho.** Un plan tiene pasos de
+dos naturalezas y confundirlas es la forma más fácil de dar una mudanza por buena sin estarlo:
+
+- **ejecutable** — tiene comandos de verdad (copiar, comprobar espacio);
+- **manual** — su `cmd` son COMENTARIOS («instalá el paquete», «cambiá el registro A»). No hay nada
+  que correr, y **un aplicador que ejecuta un bloque de comentarios obtiene exit 0 y lo marca «ok»**.
+  Ésa es la peor mentira posible: deja el servicio caído con el informe en verde.
+
+Acá un paso manual queda `pendiente-humano`, la corrida **no** se considera completa (sale con ≠0), y
+se confirma con `--hecho N` — que además **se niega** si el paso sí tenía comandos: «corrélo, no lo
+marques».
+
+**Se le cree a la verificación, no al exit code.** El `rsync` que llenó el disco devolvió **0** y
+dejó 1367 artefactos vacíos. Un paso pasa a `ok` sólo si su verificación pasa; si el comando salió
+bien y la verificación falló, queda `sospechoso`, que se informa como peor que un fallo.
+
+**Y las verificaciones están TIPADAS** (`cmd` / `humano`): «la columna Available debe ser > X» no es
+un comando, y darla por buena porque «no dio error» es exactamente cómo un aplicador miente.
+
+**Reanudable**: el estado vive en `<plan>.estado.json`, escrito con temporal + `fsync` + `rename` —
+la misma lección que costó un upgrade entero en el SDD 28 §6.13. Re-correr saltea lo ya hecho.
+
+#### ⚠ Y una verificación que era decorativa
+
+La primera versión del paso de datos verificaba así:
+
+```
+ssh … 'du -sh <p>; find <p> -maxdepth 1 -type d -empty | wc -l'
+```
+
+Imprimía el número de vacíos **y devolvía 0 igual**. En la primera corrida real dio «✓ verificado: 2»
+— donde ese 2 eran dos directorios vacíos en destino. Un guardián que siempre pasa no es un guardián.
+
+Ahora **compara**: cuenta los ficheros de los dos lados y falla si no coinciden. Probado con rotura a
+propósito y con el control que tiene que pasar:
+
+```
+copia NO hecha  → ficheros: origen=5 destino=0  ⇒ FALLA
+copia hecha     → ficheros: origen=5 destino=5  ⇒ PASA
+```
 
 Ejecuta el plan contra la máquina nueva. Requisitos, todos pagados en el SDD 28:
 
diff --git a/scripts/mudanza/aplicar.py b/scripts/mudanza/aplicar.py
new file mode 100755
index 00000000..8d3eca57
--- /dev/null
+++ b/scripts/mudanza/aplicar.py
@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+"""aplicar.py — ejecuta un PLAN de mudanza. Idempotente, reanudable, y sin mentir (SDD 29 etapa 3).
+
+── LA REGLA QUE LO DEFINE: UN PASO QUE NO SE EJECUTÓ NO SE MARCA COMO HECHO ────────────────────
+Un plan tiene pasos de dos naturalezas, y confundirlas es la forma más fácil de que una mudanza se dé
+por buena sin estarlo:
+
+  · **ejecutable**  — tiene comandos de verdad (copiar datos, comprobar espacio).
+  · **manual**      — su `cmd` son COMENTARIOS: «instalá el paquete», «cambiá el registro A». No hay
+                      nada que correr. Un aplicador que ejecuta un bloque de comentarios obtiene
+                      exit 0 y lo marca «ok». Eso es una mentira, y es la peor: deja el servicio
+                      caído con el informe en verde.
+
+Acá un paso manual se marca `pendiente-humano`, **la corrida NO se considera completa**, y hay que
+confirmarlo explícitamente (`--hecho N`) tras haberlo hecho.
+
+── LAS OTRAS DOS ───────────────────────────────────────────────────────────────────────────────
+· **Reanudable**: el estado vive al lado del plan (`<plan>.estado.json`). Re-correr saltea lo que ya
+  salió bien. Cada copia grande de esta mudanza se cortó al menos una vez — reanudar no es un lujo.
+· **La verificación es aparte del comando, y se cree a ella, no al exit code.** El `rsync` que llenó
+  el disco devolvió **0** y dejó 1367 artefactos VACÍOS. Un paso sólo pasa a `ok` si su verificación
+  pasa; si el comando salió bien y la verificación falla, queda `sospechoso`, que es peor que fallar.
+
+Uso:
+    scripts/mudanza/aplicar.py --plan plan.toml [--dry-run] [--only datos] [--hecho 7] [--paso 3]
+"""
+import argparse, json, os, subprocess, sys, time, tomllib
+
+
+def limpio(cmd):
+    """Un `cmd` que después de sacar comentarios y blancos no tiene nada, es MANUAL."""
+    return "\n".join(l for l in (cmd or "").splitlines()
+                     if l.strip() and not l.strip().startswith("#")).strip()
+
+
+class Estado:
+    def __init__(self, path):
+        self.path = path
+        self.d = json.load(open(path)) if os.path.exists(path) else {}
+
+    def get(self, n):
+        return self.d.get(str(n), {})
+
+    def set(self, n, **kw):
+        e = self.d.setdefault(str(n), {})
+        e.update(kw)
+        e["ts"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+        # Escritura durable: temporal + fsync + rename. Un corte a mitad no puede dejar el estado en
+        # cero bytes — pasó con el estado de `takana upgrade` y costó un upgrade entero (SDD 28).
+        tmp = self.path + ".tmp"
+        with open(tmp, "w") as f:
+            json.dump(self.d, f, indent=1, sort_keys=True)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp, self.path)
+        d = os.open(os.path.dirname(os.path.abspath(self.path)), os.O_RDONLY)
+        try:
+            os.fsync(d)
+        finally:
+            os.close(d)
+
+
+def correr(cmd, timeout=None):
+    p = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, timeout=timeout)
+    return p.returncode, (p.stdout + p.stderr).strip()
+
+
+def main():
+    ap = argparse.ArgumentParser(description="Ejecuta un plan de mudanza. Idempotente y reanudable.")
+    ap.add_argument("--plan", required=True)
+    ap.add_argument("--dry-run", action="store_true", help="mostrar qué se correría, sin correr")
+    ap.add_argument("--only", help="ejecutar sólo una clase de paso (preflight/datos/servicio/…)")
+    ap.add_argument("--paso", type=int, action="append", help="ejecutar sólo estos pasos (repetible)")
+    ap.add_argument("--hecho", type=int, action="append",
+                    help="marcar un paso MANUAL como hecho, tras haberlo hecho de verdad (repetible)")
+    ap.add_argument("--timeout", type=int, default=7200, help="segundos máx por paso")
+    a = ap.parse_args()
+
+    with open(a.plan, "rb") as f:
+        plan = tomllib.load(f)
+    pasos = plan.get("paso", [])
+    est = Estado(a.plan + ".estado.json")
+
+    if a.hecho:
+        for n in a.hecho:
+            p = next((x for x in pasos if x["n"] == n), None)
+            if not p:
+                print(f"!! no hay paso {n}", file=sys.stderr); sys.exit(2)
+            if limpio(p.get("cmd", "")):
+                print(f"!! el paso {n} NO es manual: tiene comandos. Corrélo, no lo marques.",
+                      file=sys.stderr); sys.exit(2)
+            est.set(n, estado="ok-manual", nota="confirmado a mano por el operador")
+            print(f"  paso {n} marcado hecho a mano: {p['titulo'].splitlines()[0]}")
+        return
+
+    print(f"\n══ APLICANDO {plan.get('origen','?')} → {plan.get('destino','?')} ══")
+    print(f"   {len(pasos)} paso(s) · estado en {est.path}\n")
+
+    hechos = fallidos = sospechosos = manuales = saltados = 0
+    for p in pasos:
+        n, clase, titulo = p["n"], p["clase"], p["titulo"].strip()
+        if a.only and clase != a.only:
+            continue
+        if a.paso and n not in a.paso:
+            continue
+        ya = est.get(n).get("estado")
+        if ya in ("ok", "ok-manual"):
+            saltados += 1
+            continue
+
+        cmd = limpio(p.get("cmd", ""))
+        if not cmd:
+            manuales += 1
+            est.set(n, estado="pendiente-humano")
+            print(f"  ⏸ {n:3} [{clase}] {titulo}")
+            print("       MANUAL — no hay nada que ejecutar. Lo que hay que hacer:")
+            for l in (p.get("cmd") or "").strip().splitlines():
+                print(f"       {l}")
+            print(f"       cuando esté hecho: --hecho {n}")
+            continue
+
+        print(f"  ▶ {n:3} [{clase}] {titulo}")
+        if a.dry_run:
+            for l in cmd.splitlines():
+                print(f"       $ {l}")
+            continue
+        try:
+            rc, out = correr(cmd, a.timeout)
+        except subprocess.TimeoutExpired:
+            rc, out = 124, f"(timeout de {a.timeout}s)"
+        if rc != 0:
+            fallidos += 1
+            est.set(n, estado="falla", rc=rc, salida=out[-2000:])
+            print(f"       ✗ falló (rc={rc}): {out.splitlines()[-1][:120] if out else ''}")
+            print("       se detiene acá: los pasos siguientes pueden depender de éste.")
+            break
+
+        # ── la verificación manda, no el exit code ──────────────────────────────────────────────
+        ver, vtipo = (p.get("verifica") or "").strip(), p.get("verifica_tipo", "cmd")
+        if not ver:
+            est.set(n, estado="ok", rc=0, nota="sin verificación declarada")
+            hechos += 1
+            print("       ✓ (sin verificación declarada)")
+            continue
+        if vtipo == "humano":
+            sospechosos += 1
+            est.set(n, estado="pendiente-humano", rc=0, salida=out[-2000:])
+            print(f"       ✓ comando ok — pero la verificación es HUMANA, no se da por buena sola:")
+            print(f"         {ver}")
+            print(f"         salida: {out.splitlines()[-1][:120] if out else '(vacía)'}")
+            print(f"         cuando la mires: --hecho {n}")
+            continue
+        vrc, vout = correr(ver, a.timeout)
+        if vrc == 0:
+            hechos += 1
+            est.set(n, estado="ok", rc=0, verifica_salida=vout[-2000:])
+            print(f"       ✓ verificado: {vout.splitlines()[-1][:120] if vout else ''}")
+        else:
+            sospechosos += 1
+            est.set(n, estado="sospechoso", rc=0, verifica_rc=vrc, verifica_salida=vout[-2000:])
+            print(f"       ⚠ el comando salió BIEN y la verificación FALLÓ (rc={vrc}).")
+            print("         Eso es peor que un fallo: el rsync que llenó el disco también devolvió 0")
+            print("         y dejó 1367 artefactos vacíos. NO se marca como hecho.")
+
+    print(f"\n── resumen ──")
+    print(f"   ok {hechos} · ya estaban {saltados} · MANUALES pendientes {manuales} · "
+          f"sospechosos {sospechosos} · fallidos {fallidos}")
+    if manuales or sospechosos or fallidos:
+        print("   ⚠ la mudanza NO está completa: hay pasos sin confirmar. `aplicar.py --plan … ` de")
+        print("     nuevo saltea lo ya hecho; los manuales se confirman con `--hecho N`.")
+        sys.exit(1)
+    print("   ✅ todos los pasos ejecutados y verificados.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/mudanza/planear.py b/scripts/mudanza/planear.py
index b60c2774..ec6d5a96 100755
--- a/scripts/mudanza/planear.py
+++ b/scripts/mudanza/planear.py
@@ -126,9 +126,12 @@ def pasos(censo, target, key):
     ssh = f"ssh {ssh_opts} {target}"
     P = []
 
-    def paso(clase, titulo, cmd, verifica=None, nota=None):
+    def paso(clase, titulo, cmd, verifica=None, nota=None, verifica_tipo="cmd"):
+        # `verifica_tipo` no es decoración: el aplicador NO puede tratar igual «corré esto» que «un
+        # humano tiene que mirar esto». Dar por buena una verificación humana porque «no dio error»
+        # es exactamente cómo un aplicador miente.
         P.append({"clase": clase, "titulo": titulo, "cmd": cmd,
-                  "verifica": verifica or "", "nota": nota or ""})
+                  "verifica": verifica or "", "verifica_tipo": verifica_tipo, "nota": nota or ""})
 
     datos_muda = [d for d in censo.get("datos", []) if d.get("decision") == "muda"]
     svc_muda = [s for s in censo.get("servicio", []) if s.get("decision") == "muda"]
@@ -143,6 +146,7 @@ def pasos(censo, target, key):
     paso("preflight", f"el destino responde y tiene sitio (hacen falta {humano(need)})",
          f"{ssh} 'df -Pm / | tail -1'",
          verifica=f"la columna 'Available' debe ser > {int(need)} MiB",
+         verifica_tipo="humano",
          nota=("⚠ tamaños ilegibles, NO contados en el total: " + ", ".join(ilegibles)) if ilegibles else
               "Dimensionado con el tamaño de COPIA (hardlinks expandidos), no con `du`.")
 
@@ -152,10 +156,15 @@ def pasos(censo, target, key):
         paso("datos", f"copiar {p}  ({d.get('size','?')})",
              f"rsync -aH --partial --info=stats2 --exclude='.dmerge' "
              f"-e {shlex.quote('ssh ' + ssh_opts)} {shlex.quote(p + '/')} {target}:{shlex.quote(p + '/')}",
-             verifica=f"{ssh} 'du -sh {shlex.quote(p)}; find {shlex.quote(p)} -maxdepth 1 -type d -empty | wc -l'",
+             verifica=(f"src=$(find {shlex.quote(p)} -type f | wc -l); "
+                       f"dst=$({ssh} 'find {shlex.quote(p)} -type f | wc -l'); "
+                       f'echo "ficheros: origen=$src destino=$dst"; [ "$src" = "$dst" ]'),
              nota="`-H` obligatorio: sin él los hardlinks se expanden (60 G medidos → 85 G). "
-                  "La verificación cuenta EN DESTINO: un rsync que se corta deja directorios VACÍOS "
-                  "y devuelve 0.")
+                  "Y la verificación COMPARA, no informa: cuenta los ficheros de los dos lados y "
+                  "FALLA si no coinciden. La versión anterior imprimía el número de directorios "
+                  "vacíos y devolvía 0 igual — un guardián que siempre pasa no es un guardián. "
+                  "Hace falta porque un rsync que se corta deja el destino a medias y devuelve 0: "
+                  "así se llenó un disco y quedaron 1367 artefactos vacíos.")
 
     # ── 2. servicios ────────────────────────────────────────────────────────────────────────────
     for s in svc_muda:
@@ -197,6 +206,7 @@ def pasos(censo, target, key):
              "# NO SE BORRA NADA ACÁ. Esta lista existe para leerse antes de apagar el origen:\n"
              + "\n".join(det),
              verifica="revisión humana: leer la lista completa antes de destruir el origen",
+         verifica_tipo="humano",
              nota="El día que se apague la máquina vieja esto no se puede recuperar. Que esté "
                   "escrito ES el paso.")
     return P
@@ -220,6 +230,7 @@ def emitir(censo, P, target):
         L.append(f'cmd = """\n{p["cmd"]}\n"""')
         if p["verifica"]:
             L.append(f'verifica = """\n{p["verifica"]}\n"""')
+            L.append(f'verifica_tipo = "{p["verifica_tipo"]}"')
         if p["nota"]:
             L.append(f'nota = """{p["nota"]}"""')
         L.append("")