#!/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 = 4   # 2: the memory watch (R-635); 3: box fixtures on the bench + files_may_change (2026-09-23 night);
                      # 4: a PostgreSQL major is CONVERTED on the bench, and the ladder carries the mark (`09` §6.4 part 10)
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(pg_mounts_for("\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",
    ),
    # $PGDATA, not a fixed path: 18 keeps its datadir at /var/lib/postgresql/18/docker (measured
    # 2026-09-25, audits/night-2026-09-26/A/A1-images.txt).
    "postgres": (
        'cat "$PGDATA/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 _stat_anon(cg: str):
    try:
        for line in open(os.path.join(cg, "memory.stat")):
            k, _, v = line.partition(" ")
            if k == "anon":
                return int(v)
    except OSError:
        return None
    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,
            # the APP's own memory (anon), beside the peak: memory.peak counts the kernel's file cache
            # too, which the kernel reclaims before it kills anything (night 2026-09-23, nextcloud
            # read 100 % with 0 kills). Reported, not yet used for the verdict or the mark.
            "anon": _stat_anon(cg) 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": {}}

    class _NoRedirect(urllib.request.HTTPRedirectHandler):
        def redirect_request(self, *a, **k):
            return None   # a 30x is the app ANSWERING; following it to the unreachable domain is not load
    opener = urllib.request.build_opener(_NoRedirect)
    host = None
    try:
        envtxt = (workdir / ".env").read_text()
        sub = re.search(r"^SUBDOMAIN=(.*)$", envtxt, re.M)
        dom = re.search(r"^DOMAIN=(.*)$", envtxt, re.M)
        if sub and dom:
            host = f"{sub.group(1)}.{dom.group(1)}"
    except OSError:
        pass
    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:
                req = urllib.request.Request(url, headers={"Host": host, "X-Forwarded-Proto": "https"} if host else {})
                code = opener.open(req, 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"]
        anon_max = max([smp["containers"].get(n, {}).get("anon") or 0 for smp in samples] + [v.get("anon") or 0])
        per[n] = {"limit": lim, "peak": pk,
                  "peak_pct": round(pk / lim, 3) if (pk and lim) else None,
                  "anon_peak_sampled": anon_max or None,
                  "anon_peak_pct": round(anon_max / lim, 3) if (anon_max 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())
    # DECIDED 2026-09-23 night (CC, unattended — operator may reverse; `09` §3 decision 22): the mark
    # reads the APP's own memory (anon, sampled) where it was measured. memory.peak counts the file cache,
    # which the kernel drops before it kills anything: nextcloud and immich's postgres read 100 % with 0
    # kills. The cgroup peak stays in the record; a kill or a restart still FAILS the edge either way.
    def _tight_pct(p):
        return p.get("anon_peak_pct") if p.get("anon_peak_pct") is not None else p["peak_pct"]
    tight = [n for n, p in per.items() if _tight_pct(p) is not None and _tight_pct(p) > 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"]],
           "load": load_verdict(hits["n"], hits["codes"])}
    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


def load_verdict(requests: int, codes: dict) -> str:
    """R-653: did the watch's load REACH the app? `reached` when at least half its requests got any HTTP
    answer (a 401 or a 30x is the app answering); `inconclusive` otherwise. Measured 2026-09-23 night:
    ghost's and nextcloud's first ten-minute watches sent 11 797 and 9 427 requests and EVERY one was
    `err` — memory measured, app idle — and the harness wrote `proven` over it."""
    if requests <= 0:
        return "inconclusive"
    answered = sum(n for c, n in codes.items() if c != "err")
    return "reached" if answered * 2 >= requests else "inconclusive"


def clear_scratch_folders(compose_text: str, env: dict, say) -> list:
    """R-656: before FROM, remove the app's OWN scratch drive folders — every `${HDD_PATH}/…`,
    `${USERDATA_PATH}/…` and `${IMPORT_PATH}/…` bind the compose names — so a re-run of the same app does
    not start on the last run's files (nextcloud's second run never installed, 2026-09-23 night: `occ
    status: installed: false` over the first run's config/). `compose down -v` removes named volumes, not
    bind-mounted host folders. NEVER a bare root, never a path outside the scratch roots. Returns what was
    removed, and says so either way."""
    roots = {k: env.get(k) for k in ("HDD_PATH", "USERDATA_PATH", "IMPORT_PATH") if env.get(k)}
    removed = []
    for var, rel in re.findall(r"\$\{(HDD_PATH|USERDATA_PATH|IMPORT_PATH)\}(/[^:\s\"']*)", compose_text):
        root = roots.get(var)
        if not root:
            continue
        rel = rel.strip("/")
        target = os.path.realpath(os.path.join(root, rel))
        safe_root = os.path.realpath(root)
        if not rel or target == safe_root or not target.startswith(safe_root + os.sep):
            say(f"scratch folder NOT cleared (outside or equal to {var}={root}): {rel!r}")
            continue
        if os.path.exists(target):
            shutil.rmtree(target, ignore_errors=False)
            removed.append(target)
    say(f"scratch drive folders cleared before FROM (R-656): {removed or 'none existed'}")
    return removed


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


# --- the PostgreSQL major conversion on the bench (`09` §6.4 part 10, harness v4) ---------------------
#
# The postgres image converts nothing and refuses an older major's datadir (R-463); 18 refuses even an
# EMPTY volume at /var/lib/postgresql/data and wants the mount at /var/lib/postgresql (measured
# 2026-09-25). The bench converts the SAME WAY the box does (felhom-controller stacks/pgconvert.go):
# the old engine alone → the check → pg_dumpall (completion line) → the volume emptied → the new engine
# alone → the entrypoint's empty databases dropped, the existing roles' CREATE lines skipped → the load
# with ON_ERROR_STOP → the check again + PG_VERSION. The BOX venue runs the product's own code.

PG_FAMILY = ("postgres", "postgis", "pgvector", "timescaledb")
PG_DUMPALL_DONE = "PostgreSQL database cluster dump complete"


def is_pg(ref: str) -> bool:
    return any(f in ref.lower() for f in PG_FAMILY)


def pg_major(ref: str):
    tag = ref.split("@", 1)[0].rsplit("/", 1)[-1]
    tag = tag.rsplit(":", 1)[1] if ":" in tag else ""
    m = re.match(r"^(?:pg)?(\d+)", tag.lower())
    return int(m.group(1)) if m else None


def pg_conversion(frm: dict, to: dict):
    """The one PostgreSQL service whose major moves between frm and to, as the mark's shape; None."""
    moved = [s for s, r in to.items() if s in frm and is_pg(r) and pg_major(frm[s]) != pg_major(r)]
    if not moved:
        return None
    if len(moved) > 1:
        raise SystemExit(f"two PostgreSQL majors move at once: {moved} — one conversion per step")
    s = moved[0]
    return {"service": s, "engine": "postgres", "from": pg_major(frm[s]), "to": pg_major(to[s])}


def pg_mounts_for(compose_text: str) -> str:
    """Point each PostgreSQL service's data volume where ITS major wants it: /var/lib/postgresql for 18+,
    /var/lib/postgresql/data below. Only a line that already mounts one of the two is touched."""
    out, cur, major = [], None, None
    for line in compose_text.splitlines():
        m = re.match(r"^  ([A-Za-z0-9_-]+):\s*$", line)
        if m:
            cur, major = m.group(1), None
        mi = re.match(r"^\s+image:\s*(\S+)\s*$", line)
        if mi and is_pg(mi.group(1)):
            major = pg_major(mi.group(1))
        mv = re.match(r"^(\s+-\s+[A-Za-z0-9_.-]+:)/var/lib/postgresql(?:/data)?(\s*)$", line)
        if mv and major:
            line = mv.group(1) + ("/var/lib/postgresql" if major >= 18 else "/var/lib/postgresql/data") + mv.group(2)
        out.append(line)
    return "\n".join(out) + ("\n" if compose_text.endswith("\n") else "")


PG_CHECK_SQL = ("SELECT n.nspname||'.'||c.relname||'='||(xpath('/row/c/text()', query_to_xml(format('select count(*) as c "
                "from %I.%I', n.nspname, c.relname), false, true, '')))[1]::text FROM pg_class c JOIN pg_namespace n ON "
                "n.oid=c.relnamespace WHERE c.relkind IN ('r','p') AND n.nspname NOT IN ('pg_catalog','information_schema') "
                "AND n.nspname NOT LIKE 'pg_toast%' ORDER BY 1")


def _psql(cid, user, db, sql):
    r = cvp._sh(["docker", "exec", cid, "psql", "-h", "127.0.0.1", "-U", user, "-d", db, "-v", "ON_ERROR_STOP=1",
                 "-Atc", sql], timeout=600)
    if r.returncode != 0:
        raise RuntimeError(f"psql {db}: {(r.stderr or '').strip()[:300]}")
    return [l.strip() for l in r.stdout.splitlines() if l.strip()]


def pg_check(cid, user):
    out = ["db:" + l for l in _psql(cid, user, "postgres", "select datname||' owner='||pg_get_userbyid(datdba)||' enc='||"
           "pg_encoding_to_char(encoding)||' coll='||datcollate from pg_database where not datistemplate order by 1")]
    out += ["role:" + l for l in _psql(cid, user, "postgres", "select rolname||' super='||rolsuper||' login='||rolcanlogin||"
            "' pw='||(rolpassword is not null) from pg_roles where rolname !~ '^pg_' order by 1")]
    for db in _psql(cid, user, "postgres", "select datname from pg_database where not datistemplate and datname<>'postgres'"):
        out += [f"ext:{db}:" + l for l in _psql(cid, user, db, "select extname from pg_extension order by 1")]
        out += [f"rows:{db}:" + l for l in _psql(cid, user, db, PG_CHECK_SQL)]
    return sorted(out)


def pg_wait(cid, user, wait=300):
    t0 = time.time()
    while time.time() - t0 < wait:
        r = cvp._sh(["docker", "exec", cid, "psql", "-h", "127.0.0.1", "-U", user, "-d", "postgres", "-Atc", "select 1"], timeout=30)
        if r.returncode == 0 and r.stdout.strip() == "1":
            return round(time.time() - t0, 1)
        time.sleep(2)
    raise RuntimeError(f"{cid} never answered over TCP in {wait}s")


def bench_convert(conv, e, app, project, workdir, env, ev, say):
    """Convert conv['service'] from its FROM major to its TO major, the box's way. Returns the record's
    `engine_conversion` dict; raises on any failure (the edge is then `failed`)."""
    svc, t0, rec = conv["service"], time.time(), dict(conv)
    compose(workdir, project, "stop", timeout=600)
    cid = compose(workdir, project, "ps", "-aq", svc).stdout.strip()
    env_lines = cvp._sh(["docker", "inspect", cid, "--format", "{{range .Config.Env}}{{println .}}{{end}}"]).stdout
    user = next((l.split("=", 1)[1] for l in env_lines.splitlines() if l.startswith("POSTGRES_USER=")), "postgres")
    cvp._sh(["docker", "start", cid])
    pg_wait(cid, user)
    before = pg_check(cid, user)
    (ev / "convert-check-before.txt").write_text("\n".join(before) + "\n")
    # the dump carries the app's rows and its roles' password hashes: it stays in the WORK dir (removed with the
    # edge), never in the evidence that is copied off the bench and committed.
    dump = workdir / "convert-dumpall.sql"
    td = time.time()
    with open(dump, "w") as f:
        r = subprocess.run(["docker", "exec", cid, "pg_dumpall", "-h", "127.0.0.1", "-U", user], stdout=f,
                           stderr=subprocess.PIPE, text=True, timeout=7200)
    if r.returncode != 0 or PG_DUMPALL_DONE not in dump.read_text()[-4096:]:
        raise RuntimeError(f"pg_dumpall rc={r.returncode} / no completion line: {r.stderr[-300:]}")
    rec["dump_s"], rec["dump_bytes"] = round(time.time() - td, 2), dump.stat().st_size
    vols = json.loads(cvp._sh(["docker", "inspect", cid, "--format", "{{json .Mounts}}"]).stdout or "[]")
    vol = next((m["Name"] for m in vols if m.get("Type") == "volume" and m.get("Destination", "").startswith("/var/lib/postgresql")), None)
    if not vol:
        raise RuntimeError(f"no named volume under /var/lib/postgresql on {cid}: {vols}")
    cvp._sh(["docker", "stop", "-t", "60", cid])
    cvp._sh(["docker", "run", "--rm", "-v", vol + ":/v", "alpine", "sh", "-c", "find /v -mindepth 1 -delete"])
    render(app, e["to"], workdir, env, e.get("template"))
    up = compose(workdir, project, "up", "-d", "--no-deps", svc)
    if up.returncode != 0:
        raise RuntimeError(f"the new engine did not start: {up.stderr[-400:]}")
    cid2 = compose(workdir, project, "ps", "-aq", svc).stdout.strip()
    rec["new_engine_ready_s"] = pg_wait(cid2, user)
    dropped = []
    for db in _psql(cid2, user, "postgres", "select datname from pg_database where not datistemplate and datname<>'postgres'"):
        n = _psql(cid2, user, db, "select count(*) from pg_class c join pg_namespace n on n.oid=c.relnamespace "
                  "where c.relkind in ('r','p') and n.nspname not in ('pg_catalog','information_schema')")
        if n != ["0"]:
            raise RuntimeError(f"database {db} on the new engine is not empty ({n})")
        _psql(cid2, user, "postgres", f'DROP DATABASE "{db}"')
        dropped.append(db)
    skip = {f"CREATE ROLE {r};" for r in _psql(cid2, user, "postgres", "select quote_ident(rolname) from pg_roles where rolname !~ '^pg_'")}
    body = "".join(l for l in dump.read_text().splitlines(True) if l.rstrip("\r\n") not in skip)
    tl = time.time()
    r = subprocess.run(["docker", "exec", "-i", cid2, "psql", "-h", "127.0.0.1", "-U", user, "-d", "postgres",
                        "-v", "ON_ERROR_STOP=1", "-q"], input=body, capture_output=True, text=True, timeout=7200)
    (ev / "convert-load.err").write_text(r.stderr)
    if r.returncode != 0:
        raise RuntimeError(f"the load stopped (rc={r.returncode}): {r.stderr.strip()[:400]}")
    rec["load_s"], rec["dropped"], rec["skipped_create_role"] = round(time.time() - tl, 2), dropped, sorted(skip)
    after = pg_check(cid2, user)
    (ev / "convert-check-after.txt").write_text("\n".join(after) + "\n")
    if after != before:
        raise RuntimeError("the check differs after the load: " + "; ".join(sorted(set(before) ^ set(after)))[:400])
    ver = cvp._sh(["docker", "exec", cid2, "sh", "-c", 'cat "$PGDATA/PG_VERSION"']).stdout.strip()
    if ver != str(conv["to"]):
        raise RuntimeError(f"PG_VERSION {ver!r}, want {conv['to']}")
    rec.update(result="converted", check_equal=True, pg_version=ver, tables=sum(1 for l in after if l.startswith("rows:")),
               convert_s=round(time.time() - t0, 1))
    say(f"CONVERTED {svc} PostgreSQL {conv['from']} -> {conv['to']} in {rec['convert_s']}s "
        f"(dump {rec['dump_s']}s / {rec['dump_bytes']} B, load {rec['load_s']}s, check equal over {rec['tables']} tables)")
    return rec


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 ---
        rec["scratch_cleared"] = clear_scratch_folders(compose_text, env, say)
        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")
        conv = pg_conversion(e["frm"], e["to"])
        if conv:  # harness v4: the engine is converted first, the box's way
            say(f"{edge_id}: PostgreSQL major move {conv} — converting on the bench before the swap")
            try:
                rec["engine_conversion"] = bench_convert(conv, e, app, project, workdir, env, ev, say)
            except Exception as ex:  # noqa: BLE001 — every failure is the edge's verdict, stated
                rec["engine_conversion"] = dict(conv, result="failed", detail=str(ex)[:600])
                rec["verdict"], rec["abort_detail"] = "failed", f"the conversion failed: {ex}"
                say(f"CONVERSION FAILED: {ex}")
                return rec
        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")
            elif mem.get("load") != "reached":
                rec["verdict"] = "inconclusive"
                rec["abort_detail"] = (f"memory watch: fewer than half of its {mem.get('requests')} requests reached "
                                       f"the app ({mem.get('codes')}) — memory measured on an idle app (R-653)")
                say("VERDICT -> inconclusive: " + rec["abort_detail"])

        # --- 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
    conts = (bench["memory"].get("containers") or {}).values()
    anon = [c.get("anon_peak_pct") for c in conts if isinstance(c.get("anon_peak_pct"), (int, float))]
    cg = [c.get("peak_pct") for c in conts if isinstance(c.get("peak_pct"), (int, float))]
    # the watch records FRACTIONS of the limit (0.81); the ladder carries PERCENT. memory_peak_pct is the
    # figure the mark is judged on: the app's own memory (anon) when the watch measured it (decision 22),
    # else the cgroup peak; the cgroup peak (file cache included) is carried beside it.
    peak = round(max(anon) * 100, 1) if anon else (round(max(cg) * 100, 1) if cg else None)
    cg_peak = round(max(cg) * 100, 1) if cg else None
    if peak is None:
        print(f"REFUSED {app}: the memory watch recorded no peak")
        return 1
    # harness v4 (`09` §6.4 part 10): a PostgreSQL major is written ONLY when BOTH venues converted it —
    # the bench by its own conversion, the box by the PRODUCT's (the walk records the controller's
    # `CONVERTED` line). The mark is what lets the box convert at all, and the catalog gate reads it.
    conv = pg_conversion(bench["from"], bench["to"])
    if conv:
        bc, xc = bench.get("engine_conversion") or {}, box.get("engine_conversion") or {}
        want = {k: conv[k] for k in ("service", "engine", "from", "to")}
        if bc.get("result") != "converted" or {k: bc.get(k) for k in want} != want:
            print(f"REFUSED {app}: the bench did not convert {want} (its record: {bc})")
            return 1
        if xc.get("result") != "converted" or {k: xc.get(k) for k in want} != want:
            print(f"REFUSED {app}: the box did not convert {want} through the product (its record: {xc})")
            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, "memory_basis": "anon" if anon else "cgroup_peak",
             "memory_cgroup_peak_pct": cg_peak,
             "marks": {"files_may_change": "files_may_change" in marks,
                       "needs_person": None, "memory_tight": peak > ladder.MEMORY_TIGHT_PCT}}
    if conv:
        entry["engine_conversion"] = {k: conv[k] for k in ("service", "engine", "from", "to")}
    probs = ladder.check_entry(entry)
    if probs:
        print(f"REFUSED {app}: the entry would not be well-formed: {probs}")
        return 1
    # `09` §6.4 part 5: the step being SUPERSEDED keeps its own definition. When the ladder's head is the
    # compose as it stands, that compose — the head's images with every fix that flowed since — becomes
    # steps/<step_key(head.to)>.yml, the file a box one step behind will pin (check-test-record.py rule 4).
    prior, _, _ = ladder.parse(fy_p.read_text())
    if prior and prior[-1].get("to") == cur:
        sp = tdir / ladder.step_file(cur)
        if not sp.exists():
            sp.parent.mkdir(parents=True, exist_ok=True)
            sp.write_text(comp)
            print(f"STEP {app}: the superseded step {cur} keeps its definition at {ladder.step_file(cur)}")
        smp = tdir / ladder.step_meta_file(cur)
        if not smp.exists():  # R-664: and its own .felhom.yml, without the ladder
            smp.write_text(ladder.strip_ladder_block(fy_p.read_text()))
            print(f"STEP {app}: … and its .felhom.yml at {ladder.step_meta_file(cur)}")
    # 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(pg_mounts_for("\n".join(out) + "\n") if conv else "\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:]))