#!/usr/bin/env python3
"""Upgrade prover — does a real app upgrade keep the customer's data, and can it be undone?

R-449. `SPIKE-app-update-2026-09-01.md` §7 measured exactly ONE upgrade, by hand, on one app. A
one-off measurement that nothing repeats decays into a claim, and this project has paid for that
before. This is the thing that repeats it.

Method, per EDGE (one app, one FROM image set, one TO image set):

  1. render the catalog template with the FROM images and `docker compose up -d`
  2. SEED through the app's OWN INTERFACE — its HTTP API or its own CLI inside the container
  3. VERIFY the seed reads back  ← control C1. A fixture that cannot prove itself first proves
     nothing after.
  4. re-render with the TO images, `up -d`, settle
  5. VERIFY the seed reads back AGAIN  ← THE RESULT
  5b. WATCH MEMORY for --soak seconds under light load (R-635): peak against the compose limit, the
      kernel's own OOM-kill counter and the restart count. A kill or a restart turns `proven` into
      `failed`; a peak above 80 % of the limit adds the `memory_tight` mark.
  6. ABORT: put the FROM images back, `up -d`, and record what happens

WHAT SUCCESS IS, AND THE RULE THAT DOES *NOT* CARRY OVER FROM survive2.py.
`survive2.py` calls a file survived only if sha256 AND inode both match. That is right for a
redeploy and WRONG for an upgrade: a migration is SUPPOSED to rewrite files, so that rule fails
every correct upgrade. Success here is an APPLICATION-LEVEL READBACK — ask the app for the value.

THE RULE THAT DOES CARRY OVER, verbatim from survive2.py: "Nothing is ever seeded into a volume by
hand." R-156's evidence shows a root-written canary making an empty volume read as populated. Every
seed goes in through the app's own interface. An app with no non-browser route is recorded
`inconclusive`, WITH what was tried — that is a result, not a licence to plant a file.

"the container started" IS NOT A PASS. The spike measured an app that was HTTP 200
"update completed" and crash-looping at the same time.

The undo is an ABORT, never a "rollback". The word is struck — see
felhom.eu/documentation/architecture/09-update-architecture.md §4: once a migration has run, the old
image refuses to start on the migrated data, so there is no rollback to speak of.

Usage:  python3 upgrade-test.py [--soak SECONDS] <edge-id> [<edge-id> …]   (see EDGES)
        python3 upgrade-test.py [--soak SECONDS] --move <app> <svc>=<ref> [...]   (FROM = the template)
        python3 upgrade-test.py --write-ladder <verdict.json> --box <box verdict.json> \
                --catalog <checkout> --evidence <rel> [--box-evidence <rel>]   (the ONLY ladder writer)
        python3 upgrade-test.py --list
        --soak: how long the memory watch runs after a successful readback (default 600; 0 = off)
Layout: templates under /opt/upg/templates, evidence under /opt/upg/evidence
"""
import importlib.util, json, os, re, shutil, subprocess, sys, time
from datetime import datetime, timezone
from pathlib import Path

HARNESS_VERSION = 3   # 2: the memory watch (R-635); 3: box fixtures on the bench + files_may_change (2026-09-23 night)
ROOT = Path("/opt/upg")
TEMPLATES = ROOT / "templates"
EVIDENCE = ROOT / "evidence"

# On the bench the helpers sit beside this file in /opt/upg; the ladder WRITER (`--write-ladder`) runs
# on DooPlex against a catalog checkout and needs none of them, so a missing one is only fatal to a run.
cvp = fx = boxport = None
if (ROOT / "check-volume-persistence.py").exists():
    sys.path.insert(0, str(ROOT))
    _spec = importlib.util.spec_from_file_location("cvp", str(ROOT / "check-volume-persistence.py"))
    cvp = importlib.util.module_from_spec(_spec)
    _spec.loader.exec_module(cvp)

    _fspec = importlib.util.spec_from_file_location("fx", str(ROOT / "upgrade_fixtures.py"))
    fx = importlib.util.module_from_spec(_fspec)
    _fspec.loader.exec_module(fx)
    if (ROOT / "upgrade_boxport.py").exists():
        import upgrade_boxport as boxport  # noqa: E402 — the box walk's fixtures, on the bench (R-462)


# --- the edges ---------------------------------------------------------------------------------
#
# Every FROM/TO pair below is a transition THE CATALOG ITSELF MADE, read from its own git history,
# except where the comment says otherwise. C3's `alpine:3.20` is a real image that pulls cleanly and
# exits immediately — measured in SPIKE-app-update-2026-09-01 §4.
EDGES = {
    "C2": dict(app="privatebin", note="no-op control: a version to ITSELF",
               frm={"privatebin": "privatebin/pdo:2.0.5"},
               to={"privatebin": "privatebin/pdo:2.0.5"}),
    "C3": dict(app="privatebin", note="NEGATIVE control: the TO image starts and exits immediately",
               frm={"privatebin": "privatebin/pdo:2.0.5"},
               to={"privatebin": "alpine:3.20"}),
    "E1": dict(app="privatebin", note="catalog transition cf8b645, a major",
               frm={"privatebin": "privatebin/pdo:1.7.5"},
               to={"privatebin": "privatebin/pdo:2.0.5"}),
    "E2": dict(app="docmost", note="catalog transition a2115b2; PostgreSQL constant across it",
               frm={"docmost": "docmost/docmost:0.25.3"},
               to={"docmost": "docmost/docmost:0.95.0"}),
    "E3": dict(app="bookstack", note="catalog transition 0b73e5e: app AND engine together",
               frm={"bookstack": "lscr.io/linuxserver/bookstack:25.02.2", "bookstack-db": "mariadb:11.6"},
               to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"}),
    "E3a": dict(app="bookstack", note="AUTHORED step (the catalog never carried it): app half alone",
                frm={"bookstack": "lscr.io/linuxserver/bookstack:25.02.2", "bookstack-db": "mariadb:11.6"},
                to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:11.6"}),
    "E3b": dict(app="bookstack", note="AUTHORED step: engine half alone",
                frm={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:11.6"},
                to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"}),

    # --- added 2026-09-21 by the update night (R-462's widening) -------------------------------
    # Every one of these was run BOX-SIDE FIRST, through the product's own guarded Update on guest
    # 9202, with the data read back through the app's own front door before and after. They are
    # here so the same edge also gets its ABORT step, which the box deliberately does not offer
    # (`09` §6.1: the box never puts the old version back by itself, because whether the old image
    # starts on migrated data is per-app and cannot be predicted).
    #
    # These are REAL UPSTREAM MOVES that existed on 2026-09-21 and that the catalog has NOT made.
    # They are candidates the operator may promote; the harness is where the abort answer for each
    # of them comes from.
    "U1": dict(app="privatebin", note="real upstream move 2.0.5 -> 2.0.6; file-backed, no database",
               frm={"privatebin": "privatebin/pdo:2.0.5"},
               to={"privatebin": "privatebin/pdo:2.0.6"}),
    "U2": dict(app="docmost", note="real upstream move 0.95.0 -> 0.96.0; PostgreSQL CONSTANT across it",
               frm={"docmost": "docmost/docmost:0.95.0", "docmost-postgres": "postgres:16-alpine"},
               to={"docmost": "docmost/docmost:0.96.0", "docmost-postgres": "postgres:16-alpine"}),
    "U3": dict(app="bookstack", note="real upstream move 26.05.2 -> 26.05.5; MariaDB CONSTANT across it",
               frm={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"},
               to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.5", "bookstack-db": "mariadb:12.3"}),
    "U4": dict(app="actualbudget", note="real upstream move 26.7.0 -> 26.9.0; SQLite in its own volume",
               frm={"actualbudget": "actualbudget/actual-server:26.7.0"},
               to={"actualbudget": "actualbudget/actual-server:26.9.0"}),
    "U5": dict(app="navidrome", note="real upstream move 0.63.2 -> 0.64.0; DATABASE HALF ONLY",
               frm={"navidrome": "deluan/navidrome:0.63.2"},
               to={"navidrome": "deluan/navidrome:0.64.0"}),
    "U6": dict(app="audiobookshelf", note="real upstream move 2.35.1 -> 2.36.1; DATABASE HALF ONLY",
               frm={"audiobookshelf": "ghcr.io/advplyr/audiobookshelf:2.35.1"},
               to={"audiobookshelf": "ghcr.io/advplyr/audiobookshelf:2.36.1"}),
    "U7": dict(app="vikunja", note="real upstream move 2.3.0 -> 2.6.0; the app migrates its own SQLite",
               frm={"vikunja": "vikunja/vikunja:2.3.0"},
               to={"vikunja": "vikunja/vikunja:2.6.0"}),

    # --- added 2026-09-23: the memory watch (R-635) --------------------------------------------
    # The edge that passed its walk on 2026-09-21, was promoted, and then OOM-looped on demo-hp for
    # six hours. M1 runs it on the CURRENT template (768M, two workers — the fix); M1old runs it on
    # the template AS PROMOTED (512M, four workers), taken from catalog commit 15f9ebf and placed in
    # the templates dir as `romm@15f9ebf`. M1old is the memory watch's RED-PROOF: it must fail.
    "M1": dict(app="romm", note="catalog move 15f9ebf 5.0.0 -> 5.3.0 on the CURRENT template (768M, 2 workers)",
               frm={"romm": "rommapp/romm:5.0.0"},
               to={"romm": "rommapp/romm:5.3.0"}),
    "M1old": dict(app="romm", template="romm@15f9ebf",
                  note="the same move on the template AS PROMOTED (512M, 4 workers) — must FAIL the memory watch",
                  frm={"romm": "rommapp/romm:5.0.0"},
                  to={"romm": "rommapp/romm:5.3.0"}),
}

# Light load for the memory watch: paths that make the app's own workers do work. An app not listed
# gets its front page only. Unauthenticated API calls still reach a worker (they answer 401/403),
# which is the point — RomM's workers were killed while nginx in front of them answered 200.
LOAD_PATHS = {
    "romm": ["/api/heartbeat", "/api/platforms", "/api/roms?limit=50", "/api/collections",
             "/api/stats", "/api/users/me", "/api/config", "/"],
}
LOAD_CONCURRENCY = 4
MEMORY_TIGHT = 0.80


# --- compose plumbing --------------------------------------------------------------------------

def render(app: str, images: dict, workdir: Path, env: dict, template: str = None) -> Path:
    """Write the catalog template into workdir with `images` substituted per SERVICE.

    `template` names a directory under TEMPLATES other than the app's own — an older revision of
    the same template kept for a red-proof (`romm@15f9ebf`).

    Substitution is per service and only on that service's own `image:` line — never a blind
    string replace, which would also rewrite an image name that appears in a comment or an env var.
    """
    src = (TEMPLATES / (template or app) / "docker-compose.yml").read_text()
    out, cur = [], None
    for line in src.splitlines():
        m = re.match(r"^  ([A-Za-z0-9_-]+):\s*$", line)
        if m:
            cur = m.group(1)
        mi = re.match(r"^(\s+image:\s*)(\S+)\s*$", line)
        if mi and cur in images:
            line = mi.group(1) + images[cur]
        out.append(line)
    workdir.mkdir(parents=True, exist_ok=True)
    (workdir / "docker-compose.yml").write_text("\n".join(out) + "\n")
    (workdir / ".env").write_text("".join(f"{k}={v}\n" for k, v in env.items()))
    return workdir / "docker-compose.yml"


def compose(workdir: Path, project: str, *args, timeout=1800):
    return cvp._sh(["docker", "compose", "-p", project, "-f", str(workdir / "docker-compose.yml"),
                    "--env-file", str(workdir / ".env")] + list(args), timeout=timeout)


def container_ip(name: str) -> str:
    r = cvp._sh(["docker", "inspect", "-f",
                 "{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}", name])
    return (r.stdout or "").split()[0] if r.stdout.strip() else ""


def settle(project: str, workdir: Path, wait=420):
    """Wait until every container is running AND every one that declares a healthcheck is healthy.

    Returns (ok, seconds, per-container state). A container that has NO healthcheck counts as
    settled once it is running — but `running` is never reported as the RESULT, only as a
    precondition for asking the app itself (see the module docstring).
    """
    t0 = time.time()
    deadline = t0 + wait
    states = {}
    while time.time() < deadline:
        r = compose(workdir, project, "ps", "-aq", timeout=120)
        cids = [c for c in r.stdout.split() if c]
        if not cids:
            time.sleep(3)
            continue
        states, pending = {}, False
        for cid in cids:
            info = cvp._inspect(cid)
            if not info:
                pending = True
                continue
            name = info["Name"].lstrip("/")
            st = info.get("State", {})
            health = (st.get("Health") or {}).get("Status")
            states[name] = {"status": st.get("Status"), "health": health,
                            "restarts": st.get("RestartCount", 0), "exit": st.get("ExitCode")}
            if st.get("Status") != "running":
                pending = True
            elif health in ("starting", "unhealthy"):
                pending = True
        if not pending:
            return True, round(time.time() - t0, 1), states
        time.sleep(5)
    return False, round(time.time() - t0, 1), states


# --- the ENGINE's own view of itself (R-459) ------------------------------------------------------
#
# WHY THIS EXISTS. On edge E3b this harness returned `proven` — correctly: the app's data survived,
# which is what it asked. But MariaDB was at that moment logging that the datadir conversion it
# requires had been SKIPPED, and nothing here could see it. The harness watched the app and the
# migration log; neither looks at engine state.
#
# IT IS REPORTED BESIDE THE VERDICT, NEVER FOLDED INTO IT. An unconverted datadir is not known to be
# a failure — SPIKE-r459-mariadb-upgrade-2026-09-06 measured 5 of 5 restarts with no degradation — so
# a verdict that called it `failed` would encode an unproven judgement, which is worse than reporting
# a fact and letting a person read both.
#
# The two engines fail differently and the field carries both: MariaDB starts anyway and skips the
# conversion quietly; PostgreSQL REFUSES to start on a datadir from an older major. So "the engine
# would not start" is as much an engine-state observation as "the engine says it needs a check".
ENGINE_PROBES = {
    # image-name fragment -> (probe command inside the container, what the answer means)
    "mariadb": (
        "cat /var/lib/mysql/mariadb_upgrade_info 2>&1; echo '|'; "
        "mariadb-upgrade --check-if-upgrade-is-needed --user=root "
        "--password=$MYSQL_ROOT_PASSWORD 2>&1; echo \"[exit=$?]\"",
        "datadir version | the engine's own upgrade verdict",
    ),
    "postgres": (
        "cat /var/lib/postgresql/data/PG_VERSION 2>&1", "datadir major version",
    ),
}


def engine_state(images: dict):
    """Ask every database engine in this stack what it thinks of its own datadir.

    Best-effort and never fatal: a probe that cannot run records why, because "we could not ask" and
    "the engine is content" are different facts and only one of them is about the engine.
    """
    out = {}
    for svc, ref in images.items():
        for frag, (cmd, meaning) in ENGINE_PROBES.items():
            if frag not in ref:
                continue
            r = cvp._sh(["docker", "exec", svc, "sh", "-c", cmd], timeout=120)
            out[svc] = {"image": ref, "probe": meaning,
                        "answer": " ".join(((r.stdout or "") + (r.stderr or "")).split())[:600],
                        "probe_rc": r.returncode}
    return out


# --- the memory watch (R-635) ---------------------------------------------------------------------
#
# WHY THIS EXISTS. `romm 5.0.0 -> 5.3.0` was walked on 2026-09-21: seeded, read back, `proven`. It was
# promoted, applied on demo-hp, reached `done` — and two hours later its workers began to be
# OOM-killed, 4,530 times in six hours, while nginx in front of them answered 200. `proven` meant
# "the update applied and the data survived", never "the new version runs". This step is the first
# half of closing that gap: it runs the new version under light load and watches memory.
#
# THE INSTRUMENT, and why not `docker inspect`'s OOMKilled. On an LXC guest OOMKilled has been
# measured FALSE for real kills (2026-09-15, R-528) and TRUE on another box (2026-09-22). The
# kernel's own counter is the cgroup's `memory.events` `oom_kill`, read HOST-SIDE so it works on
# images that ship no shell (vikunja's does not). A worker killed inside a container that keeps
# running moves that counter and moves nothing else — exactly RomM's failure. OOMKilled and the
# restart count are recorded beside it, never instead of it.

def _cgroup_dir(cid: str):
    for p in (f"/sys/fs/cgroup/system.slice/docker-{cid}.scope", f"/sys/fs/cgroup/docker/{cid}"):
        if os.path.isdir(p):
            return p
    return None


def _read_int(path: str):
    try:
        v = open(path).read().strip()
        return None if v == "max" else int(v)
    except (OSError, ValueError):
        return None


def _oom_kills(cg: str):
    try:
        for line in open(os.path.join(cg, "memory.events")):
            k, _, v = line.partition(" ")
            if k == "oom_kill":
                return int(v)
    except (OSError, ValueError):
        pass
    return None


def memory_snapshot(project: str, workdir: Path):
    """One reading per container: usage, peak, limit, kernel OOM kills, restarts, OOMKilled."""
    out = {}
    r = compose(workdir, project, "ps", "-aq", timeout=120)
    for cid in [c for c in r.stdout.split() if c]:
        info = cvp._inspect(cid)
        if not info:
            continue
        full = info["Id"]
        name = info["Name"].lstrip("/")
        cg = _cgroup_dir(full)
        limit = (info.get("HostConfig") or {}).get("Memory") or None
        out[name] = {
            "limit": limit or (_read_int(os.path.join(cg, "memory.max")) if cg else None),
            "current": _read_int(os.path.join(cg, "memory.current")) if cg else None,
            "peak": _read_int(os.path.join(cg, "memory.peak")) if cg else None,
            "oom_kill": _oom_kills(cg) if cg else None,
            "restarts": (info.get("State") or {}).get("RestartCount", 0),
            "oomkilled_flag": (info.get("State") or {}).get("OOMKilled"),
            "status": (info.get("State") or {}).get("Status"),
            "cgroup": bool(cg),
        }
    return out


def memory_watch(app: str, project: str, workdir: Path, seconds: int, say, ev: Path):
    """Run light load for `seconds`, sampling every 15 s. Returns the `memory` record and the marks.

    Stops EARLY at the first kernel OOM kill or restart: the verdict is decided then, and the time
    to the first kill is the number worth having. A container whose cgroup cannot be read is
    reported `unmeasured` — "we could not look" never reads as "nothing happened" (R-96 rule 3).
    """
    import threading, urllib.error, urllib.request
    paths = LOAD_PATHS.get(app, ["/"])
    native = fx.FIXTURES.get(app)
    cname, port = getattr(native, "container", app), getattr(native, "port", 80)
    if native is None and boxport is not None:
        # a box-fixture app: load the container traefik routes `/` to, at its own port
        rts = boxport.routes((workdir / "docker-compose.yml").read_text())
        root = [r for r in rts if not r[0]] or rts
        if root:
            cname, port = root[0][1], root[0][2]
    target = container_ip(cname)
    stop = threading.Event()
    hits = {"n": 0, "codes": {}}
    lock = threading.Lock()

    def worker(i):
        k = 0
        while not stop.is_set():
            url = f"http://{target}:{port}{paths[(i + k) % len(paths)]}"
            k += 1
            try:
                code = urllib.request.urlopen(url, timeout=20).status
            except urllib.error.HTTPError as e:
                code = e.code
            except Exception:
                code = "err"
            with lock:
                hits["n"] += 1
                hits["codes"][str(code)] = hits["codes"].get(str(code), 0) + 1
            time.sleep(0.2)

    base0 = memory_snapshot(project, workdir)
    unmeasured = [n for n, v in base0.items() if not v["cgroup"]]
    say(f"memory watch: {seconds}s, {LOAD_CONCURRENCY} callers on {len(paths)} path(s) at {target}:{port}"
        + (f"; UNMEASURED (no cgroup): {unmeasured}" if unmeasured else ""))
    threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(LOAD_CONCURRENCY)]
    for t in threads:
        t.start()
    t0, samples, first_bad = time.time(), [], None
    try:
        while time.time() - t0 < seconds:
            time.sleep(15)
            snap = memory_snapshot(project, workdir)
            el = round(time.time() - t0, 1)
            samples.append({"t": el, "containers": snap, "requests": hits["n"]})
            line = "  ".join(
                f"{n}={(v['current'] or 0) // 2**20}M/{(v['limit'] or 0) // 2**20}M"
                f" peak={(v['peak'] or 0) // 2**20}M kills={v['oom_kill']} rs={v['restarts']}"
                for n, v in snap.items())
            say(f"  +{el:5.0f}s  {line}  reqs={hits['n']}")
            for n, v in snap.items():
                b = base0.get(n, {})
                dk = (v["oom_kill"] or 0) - (b.get("oom_kill") or 0)
                dr = (v["restarts"] or 0) - (b.get("restarts") or 0)
                if (dk > 0 or dr > 0) and first_bad is None:
                    first_bad = {"t": el, "container": n, "oom_kills": dk, "restarts": dr}
            if first_bad:
                say(f"memory watch: STOPPING EARLY — {first_bad}")
                break
    finally:
        stop.set()
        for t in threads:
            t.join(timeout=25)
    end = memory_snapshot(project, workdir)
    per = {}
    for n, v in end.items():
        b = base0.get(n, {})
        pk, lim = v["peak"], v["limit"]
        per[n] = {"limit": lim, "peak": pk,
                  "peak_pct": round(pk / lim, 3) if (pk and lim) else None,
                  "oom_kills": None if v["oom_kill"] is None else v["oom_kill"] - (b.get("oom_kill") or 0),
                  "restarts": (v["restarts"] or 0) - (b.get("restarts") or 0),
                  "oomkilled_flag": v["oomkilled_flag"], "measured": v["cgroup"]}
    (ev / "memory-samples.json").write_text(json.dumps(samples, indent=2))
    killed = any((p["oom_kills"] or 0) > 0 or p["restarts"] > 0 or p["oomkilled_flag"] for p in per.values())
    tight = [n for n, p in per.items() if p["peak_pct"] is not None and p["peak_pct"] > MEMORY_TIGHT]
    rec = {"soak_s": round(time.time() - t0, 1), "requested_s": seconds, "requests": hits["n"],
           "codes": hits["codes"], "first_kill": first_bad, "containers": per,
           "unmeasured": [n for n, p in per.items() if not p["measured"]]}
    marks = ["memory_tight"] if tight and not killed else []
    say(f"memory watch: killed={killed} tight={tight} requests={hits['n']} codes={hits['codes']}")
    return rec, killed, marks


MIGRATION_RE = re.compile(
    r"migrat|upgrad|schema|alter table|CREATE TABLE|InnoDB: Upgrad|mysql_upgrade|"
    r"mariadb-upgrade|Running .* migration|Applying|db:migrate",
    re.I)


def migration_lines(project: str, workdir: Path, since_iso: str, limit=6):
    """Verbatim log lines that SAY a migration ran. Never inferred from timing — the finding is the
    sentence the app printed, exactly as the Nextcloud refusal was."""
    r = compose(workdir, project, "logs", "--since", since_iso, "--no-color", timeout=180)
    hits = [ln.strip() for ln in (r.stdout + r.stderr).splitlines() if MIGRATION_RE.search(ln)]
    return hits[:limit]


# --- one edge ----------------------------------------------------------------------------------

def bind_tree_hash(project: str, workdir: Path) -> dict:
    """{bind source dir: sha256 over (relpath, size, content sha)} for every BIND mount of the project's
    containers — the household's own files (a named volume holds the app's state, which a migration is
    SUPPOSED to rewrite). Files above 64 MiB are hashed by size and mtime only, and the result says so."""
    import hashlib
    r = compose(workdir, project, "ps", "-aq", timeout=120)
    srcs = set()
    for cid in (r.stdout or "").split():
        info = cvp._inspect(cid) or {}
        for m in info.get("Mounts") or []:
            if m.get("Type") == "bind" and os.path.isdir(m.get("Source") or "") \
                    and not (m.get("Source") or "").startswith(("/var/run", "/run", "/etc", "/proc", "/sys")):
                srcs.add(m["Source"])
    out = {}
    for src in sorted(srcs):
        h = hashlib.sha256()
        for dp, dn, fn in os.walk(src):
            dn.sort()
            for f in sorted(fn):
                fp = os.path.join(dp, f)
                try:
                    st = os.lstat(fp)
                except OSError:
                    continue
                h.update(os.path.relpath(fp, src).encode() + b"\0" + str(st.st_size).encode())
                if st.st_size <= 64 * 1024 * 1024 and os.path.isfile(fp) and not os.path.islink(fp):
                    try:
                        with open(fp, "rb") as fh:
                            h.update(hashlib.sha256(fh.read()).digest())
                    except OSError:
                        h.update(b"unreadable")
                else:
                    h.update(str(int(st.st_mtime)).encode())
        out[src] = h.hexdigest()
    return out


def run_edge(edge_id: str) -> dict:
    e = EDGES[edge_id]
    app = e["app"]
    project = f"upg{edge_id.lower()}"
    workdir = ROOT / "work" / edge_id
    ev = EVIDENCE / edge_id
    ev.mkdir(parents=True, exist_ok=True)
    shutil.rmtree(workdir, ignore_errors=True)

    tdir = e.get("template") or app
    felhom = (TEMPLATES / tdir / ".felhom.yml").read_text()
    compose_text = (TEMPLATES / tdir / "docker-compose.yml").read_text()
    env = cvp.build_env(app, felhom, compose_text)

    rec = {"harness_version": HARNESS_VERSION, "edge": edge_id, "app": app, "note": e["note"],
           "from": e["frm"], "to": e["to"], "verdict": "inconclusive",
           "seed_read_before": False, "seed_read_after": False, "healthy_after": False,
           "migration_observed": None, "abort": "not-attempted", "abort_detail": None,
           # engine_state is a REPORT, not a judgement — see ENGINE_PROBES.
           "engine_state_after": None,
           # memory is a MEASUREMENT beside the verdict; a kill or a restart during it DOES decide
           # the verdict, a tight peak only marks it (R-635, 2026-09-23).
           "memory": None, "marks": [],
           "duration_s": 0, "measured_at": None, "evidence": f"evidence/{edge_id}"}
    t0 = time.time()
    log = []

    def say(msg):
        line = f"[{datetime.now(timezone.utc).strftime('%H:%M:%S')}] {msg}"
        print(line, flush=True)
        log.append(line)

    try:
        # --- 1. FROM ---
        say(f"{edge_id}: deploying {app} at FROM {e['frm']}")
        render(app, e["frm"], workdir, env, e.get("template"))
        up = compose(workdir, project, "up", "-d")
        if up.returncode != 0:
            rec["abort_detail"] = f"FROM deploy failed rc={up.returncode}: {up.stderr[-800:]}"
            say("FROM deploy FAILED — inconclusive, the edge was never reached")
            return rec
        ok, secs, states = settle(project, workdir)
        say(f"FROM settled={ok} in {secs}s :: {json.dumps(states)}")
        if not ok:
            rec["abort_detail"] = f"FROM never settled: {json.dumps(states)}"
            say("FROM never became healthy — inconclusive, not a verdict about the upgrade")
            return rec

        # --- 2/3. seed + C1 ---
        fixture = fx.FIXTURES.get(app)
        if fixture is None and boxport is not None:
            fixture = boxport.get(app, compose_text, env)
            if fixture is not None:
                say(f"fixture: the BOX walk's own ({type(fixture.box).__name__}), through upgrade_boxport")
        if fixture is None:
            rec["abort_detail"] = "no fixture"
            say("no fixture for this app — inconclusive")
            return rec
        seeded = fixture.seed(container_ip, say)
        if seeded is None:
            rec["verdict"] = "inconclusive"
            rec["abort_detail"] = "no non-browser seed route" + (
                f" — tried: {fixture.tried}" if getattr(fixture, "tried", None) else "")
            say("INCONCLUSIVE — no non-browser seed route. Nothing was planted by hand.")
            return rec
        rec["seed_read_before"] = bool(fixture.verify(container_ip, seeded, say))
        say(f"C1 (seed reads back BEFORE): {rec['seed_read_before']}")
        if not rec["seed_read_before"]:
            rec["abort_detail"] = "C1 failed: the fixture could not prove itself before the upgrade"
            say("C1 FAILED — a fixture that cannot prove itself first proves nothing after")
            return rec

        files_before = bind_tree_hash(project, workdir)
        (ev / "files-before.json").write_text(json.dumps(files_before, indent=2))

        # --- 4. TO ---
        swap_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        say(f"{edge_id}: swapping to TO {e['to']}")
        render(app, e["to"], workdir, env, e.get("template"))
        up2 = compose(workdir, project, "up", "-d")
        say(f"TO up -d rc={up2.returncode}")
        ok2, secs2, states2 = settle(project, workdir)
        rec["healthy_after"] = ok2
        rec["duration_s"] = secs2
        say(f"TO settled={ok2} in {secs2}s :: {json.dumps(states2)}")
        (ev / "to-states.json").write_text(json.dumps(states2, indent=2))

        # CAPTURE THE WHOLE TO-STEP LOG *NOW*, not at the end.
        # `docker compose logs` only shows the CONTAINERS THAT EXIST, and the abort below replaces
        # them — so the TO images' own output is GONE from any capture taken afterwards. Measured on
        # E3, where the single most important line of the run ("MariaDB upgrade … required, but
        # skipped") survived only because it had already been extracted. Same class as R-320: the
        # intermediate teardown is the one that loses the evidence.
        full_to = compose(workdir, project, "logs", "--no-color", timeout=180)
        (ev / "to-full.log").write_text((full_to.stdout + full_to.stderr)[-400000:])

        rec["engine_state_after"] = engine_state(e["to"]) or None
        if rec["engine_state_after"]:
            for svc, st in rec["engine_state_after"].items():
                say(f"engine state {svc}: {st['answer'][:180]}")
        (ev / "engine-state.json").write_text(json.dumps(rec["engine_state_after"], indent=2))

        mig = migration_lines(project, workdir, swap_at)
        rec["migration_observed"] = mig[0] if mig else None
        (ev / "migration-lines.txt").write_text("\n".join(mig))
        say(f"migration lines observed: {len(mig)}")

        # --- 5. THE RESULT ---
        rec["seed_read_after"] = bool(fixture.verify(container_ip, seeded, say))
        say(f"RESULT (seed reads back AFTER): {rec['seed_read_after']}")
        rec["verdict"] = "proven" if (ok2 and rec["seed_read_after"]) else "failed"

        # --- 5a. did the update rewrite the household's FILES? (decision 13's `files may change`) ---
        files_after = bind_tree_hash(project, workdir)
        (ev / "files-after.json").write_text(json.dumps(files_after, indent=2))
        rec["files_changed"] = sorted(k for k in set(files_before) | set(files_after)
                                      if files_before.get(k) != files_after.get(k))
        if rec["files_changed"]:
            rec["marks"] = sorted(set(rec["marks"]) | {"files_may_change"})
            say(f"files_may_change: the bind-mounted tree changed under {rec['files_changed']}")

        # --- 5b. the MEMORY WATCH — only for an edge that just read back; a failed one is decided ---
        if rec["verdict"] == "proven" and SOAK_SECONDS > 0:
            mem, killed, marks = memory_watch(app, project, workdir, SOAK_SECONDS, say, ev)
            rec["memory"], rec["marks"] = mem, sorted(set(rec["marks"]) | set(marks))
            if killed:
                rec["verdict"] = "failed"
                say("VERDICT -> failed: the new version was OOM-killed or restarted under light load")

        # --- 6. the ABORT ---
        say(f"{edge_id}: ABORT — putting the FROM images back")
        render(app, e["frm"], workdir, env, e.get("template"))
        up3 = compose(workdir, project, "up", "-d")
        ok3, secs3, states3 = settle(project, workdir, wait=180)
        (ev / "abort-states.json").write_text(json.dumps(states3, indent=2))
        if not ok3:
            rec["abort"] = "refuses"
            lg = compose(workdir, project, "logs", "--tail", "40", "--no-color", timeout=120)
            tail = (lg.stdout + lg.stderr).strip()
            (ev / "abort-refusal.txt").write_text(tail)
            rec["abort_detail"] = tail[-900:]
            say(f"ABORT: the app did NOT come back (rc={up3.returncode}, {secs3}s)")
        else:
            back = bool(fixture.verify(container_ip, seeded, say))
            rec["abort"] = "starts-and-serves" if back else "starts-data-gone"
            rec["abort_detail"] = None if back else "the app started but the seeded data was gone"
            say(f"ABORT: app came back in {secs3}s; data present={back}")
        return rec
    finally:
        rec["measured_at"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        rec["duration_s"] = rec["duration_s"] or round(time.time() - t0, 1)
        rec["total_s"] = round(time.time() - t0, 1)
        # EVIDENCE FIRST, TEARDOWN SECOND (R-320): the intermediate teardown is the one that gets
        # forgotten, so everything is written before a single container is removed.
        (ev / "run.log").write_text("\n".join(log))
        lg = compose(workdir, project, "logs", "--no-color", timeout=180)
        (ev / "compose-final.log").write_text((lg.stdout + lg.stderr)[-400000:])  # post-abort state only — see to-full.log
        (ev / "verdict.json").write_text(json.dumps(rec, indent=2))
        compose(workdir, project, "down", "-v", "--remove-orphans", timeout=900)


SOAK_SECONDS = 600


def template_images(app: str, template_dir: Path) -> dict:
    """{service: image} of a catalog template — the per-service reading every gate makes."""
    sys.path.insert(0, str(Path(__file__).resolve().parent))
    import ladder
    return ladder.images_in((template_dir / app / "docker-compose.yml").read_text())


def add_move_edge(app: str, moves: list) -> str:
    """`--move <app> svc=ref …` → an edge FROM the template as it stands TO the same with those
    services moved. The FROM side is read, never typed, so it cannot disagree with the catalog."""
    frm = template_images(app, TEMPLATES)
    to = dict(frm)
    for mv in moves:
        svc, _, ref = mv.partition("=")
        if svc not in frm or not ref:
            raise SystemExit(f"--move: {mv!r} — service must be one of {sorted(frm)}")
        to[svc] = ref
    if to == frm:
        raise SystemExit("--move: nothing moves")
    eid = f"MV-{app}"
    EDGES[eid] = dict(app=app, note="night 2026-09-23 within-a-major move: " + ", ".join(moves),
                      frm=frm, to=to)
    return eid


def write_ladder(argv) -> int:
    """`--write-ladder <bench verdict.json> --box <box verdict.json> --catalog <checkout> --evidence <rel>
    [--box-evidence <rel>]` — THE ONLY WRITER of a ladder entry (`09` §6.4 part 4; never by hand).

    It refuses unless BOTH venues say `proven` and the template still stands at the bench's FROM; it
    resolves every TO ref's digest from the registry now; then it moves the compose's image lines, sets
    `catalog_since` to today, and appends the entry. The commit is the operator's (or the session's)."""
    import datetime as _dt
    sys.path.insert(0, str(Path(__file__).resolve().parent))
    import ladder, image_digest

    def arg(name, default=None):
        return argv[argv.index(name) + 1] if name in argv else default
    bench = json.loads(Path(argv[0]).read_text())
    box = json.loads(Path(arg("--box")).read_text())
    cat = Path(arg("--catalog"))
    app = bench["app"]
    if bench.get("verdict") != "proven" or box.get("verdict") != "proven":
        print(f"REFUSED {app}: bench verdict {bench.get('verdict')!r}, box verdict {box.get('verdict')!r} "
              "— only a step proven on BOTH venues is written")
        return 1
    if (bench.get("harness_version") or 0) < 2 or not bench.get("memory"):
        print(f"REFUSED {app}: the bench verdict carries no memory watch (harness v2)")
        return 1
    tdir = cat / "templates" / app
    comp_p, fy_p = tdir / "docker-compose.yml", tdir / ".felhom.yml"
    comp = comp_p.read_text()
    cur = ladder.images_in(comp)
    if cur != bench["from"]:
        print(f"REFUSED {app}: the template is at {cur}, the bench tested FROM {bench['from']}")
        return 1
    if box.get("to") and {k: v for k, v in box["to"].items() if k in bench["to"]} != \
            {k: v for k, v in bench["to"].items() if k in box["to"]}:
        print(f"REFUSED {app}: the box walked TO {box.get('to')}, the bench tested TO {bench['to']}")
        return 1
    digests = {}
    for svc, ref in sorted(bench["to"].items()):
        d, why = image_digest.resolve(ref)
        if not d:
            print(f"INCONCLUSIVE {app}: {svc} {ref}: {why}")
            return 2
        digests[svc] = d
    peaks = [c.get("peak_pct") for c in (bench["memory"].get("containers") or {}).values()
             if isinstance(c.get("peak_pct"), (int, float))]
    # the watch records peak_pct as a FRACTION of the limit (0.81); the ladder carries PERCENT
    peak = round(max(peaks) * 100, 1) if peaks else None
    if peak is None:
        print(f"REFUSED {app}: the memory watch recorded no peak")
        return 1
    marks = set(bench.get("marks") or [])
    entry = {"from": bench["from"], "to": bench["to"], "digest": digests, "verdict": "proven",
             "tested_at": bench["measured_at"], "harness_version": bench["harness_version"],
             "evidence": arg("--evidence"), "box_evidence": arg("--box-evidence"),
             "memory_peak_pct": peak,
             "marks": {"files_may_change": "files_may_change" in marks,
                       "needs_person": None, "memory_tight": peak > ladder.MEMORY_TIGHT_PCT}}
    probs = ladder.check_entry(entry)
    if probs:
        print(f"REFUSED {app}: the entry would not be well-formed: {probs}")
        return 1
    # move the compose, per service, on that service's own image: line
    out, svc = [], None
    for line in comp.splitlines():
        m = ladder.SERVICE_RE.match(line)
        if m:
            svc = m.group(1)
        mi = re.match(r"^(\s+image:\s*)(\S+)\s*$", line)
        if mi and svc in bench["to"] and mi.group(2) == bench["from"][svc]:
            line = mi.group(1) + bench["to"][svc]
        out.append(line)
    comp_p.write_text("\n".join(out) + "\n")
    if ladder.images_in(comp_p.read_text()) != bench["to"]:
        comp_p.write_text(comp)
        print(f"REFUSED {app}: the compose could not be moved line by line — restored")
        return 1
    fy = fy_p.read_text()
    fy = re.sub(r'^catalog_since:.*$', 'catalog_since: "%s"' % _dt.date.today().isoformat(), fy, count=1, flags=re.M)
    fy_p.write_text(ladder.append_entry(fy, entry))
    print(f"WROTE {app}: {bench['from']} -> {bench['to']} peak {peak}% marks {entry['marks']}")
    return 0


def main(argv):
    global SOAK_SECONDS
    if argv and argv[0] == "--write-ladder":
        return write_ladder(argv[1:])
    if argv and argv[0] == "--soak":
        SOAK_SECONDS = int(argv[1])
        argv = argv[2:]
    if argv and argv[0] == "--move":
        argv = [add_move_edge(argv[1], argv[2:])]
    if not argv or argv[0] == "--list":
        for k, v in EDGES.items():
            print(f"{k:5s} {v['app']:12s} {v['note']}")
        return 0
    EVIDENCE.mkdir(parents=True, exist_ok=True)
    results = []
    for edge_id in argv:
        if edge_id not in EDGES:
            print(f"unknown edge {edge_id}", file=sys.stderr)
            return 2
        rec = run_edge(edge_id)
        results.append(rec)
        print(json.dumps(rec, indent=2), flush=True)
    (EVIDENCE / "summary.json").write_text(json.dumps(results, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))