test record: an image move must carry its proof (09 decision 13, part 4)

update_ladder: in .felhom.yml, one JSON entry per line (spiked live on
controller v0.266.0 and v0.267.0 first). Two gates: check-test-record.py
(static, CI too) and check-test-record-move.py (history + registry for
moved refs only). 16 decoys, 3 red-proofs. The ONLY writer is
upgrade-test.py --write-ladder (bench AND box proven, digests resolved).
Harness v3: box fixtures on the bench, files_may_change.
Backfill: the 21 moves of 2026-09-22, 21 proven from their records.
No image: line moved.

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

scripts/upgrade-test.py

@@ -36,6 +36,9 @@ felhom.eu/documentation/architecture/09-update-architecture.md §4: once a migra
 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
@@ -44,18 +47,25 @@ import importlib.util, json, os, re, shutil, subprocess, sys, time
 from datetime import datetime, timezone
 from pathlib import Path
 
-HARNESS_VERSION = 2   # 2: the memory watch after the readback (R-635, 2026-09-23)
+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"
 
-_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)
+# 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)
+    _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 ---------------------------------------------------------------------------------
@@ -340,8 +350,15 @@ def memory_watch(app: str, project: str, workdir: Path, seconds: int, say, ev: P
     """
     import threading, urllib.error, urllib.request
     paths = LOAD_PATHS.get(app, ["/"])
-    target = container_ip(getattr(fx.FIXTURES.get(app), "container", app))
-    port = getattr(fx.FIXTURES.get(app), "port", 80)
+    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()
@@ -431,6 +448,43 @@ def migration_lines(project: str, workdir: Path, since_iso: str, limit=6):
 
 # --- 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"]
@@ -481,6 +535,10 @@ def run_edge(edge_id: str) -> dict:
 
         # --- 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")
@@ -488,7 +546,8 @@ def run_edge(edge_id: str) -> dict:
         seeded = fixture.seed(container_ip, say)
         if seeded is None:
             rec["verdict"] = "inconclusive"
-            rec["abort_detail"] = "no non-browser seed route"
+            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))
@@ -498,6 +557,9 @@ def run_edge(edge_id: str) -> dict:
             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']}")
@@ -535,10 +597,19 @@ def run_edge(edge_id: str) -> dict:
         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, marks
+            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")
@@ -578,11 +649,122 @@ def run_edge(edge_id: str) -> dict:
 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']}")