#!/usr/bin/env python3
"""box_walk.py — the BOX venue's client (moved into the catalog 2026-09-30 from the audits' walk.py, so the monthly
re-test `retest-floating.py` has a stable home). ONE app's walk on scratch guest 9202, through the product's own
endpoints. Configuration by environment: SC (a 0600 scratch dir holding `.ctlpw`, the 9202 dashboard password —
never committed), EV (where evidence goes), GUEST/BASE/DOMAIN as before.

EVIDENCE, NOT PRODUCT. It presses exactly the buttons a person presses:
  POST /api/stacks/<n>/deploy · POST /api/sync · POST /api/stacks/rescan
  POST /api/stacks/<n>/update · POST /api/stacks/<n>/remove
and reads GET /api/stacks/<n>. No controller code exists for it.

The walk, per `09` §6.4 and the update-night brief §4:
  1 deploy from the DRILL catalog at the LIVE pin
  2 seed through the app's OWN front door (R-156: never a volume, never SQL)
  3 read the seed back            <- control C1; a fixture that cannot prove itself proves nothing
  4 „Mentés most"
  5 commit the real one-step bump to the DRILL repo, sync, rescan, read the badge in BOTH languages
  6 press the guarded Update, record every phase with timestamps
  7 read the seed back through the front door
  8 the four version observables side by side
  9 write the verdict record in `09`'s JSON shape

`inconclusive` is a first-class verdict and is NEVER collapsed into `failed`.
"""
import argparse, json, os, re, subprocess, sys, time
from datetime import datetime, timezone

SC = os.environ.get('SC', os.path.expanduser('~/.felhom-retest'))
EV = os.environ.get('EV', os.path.join(SC, 'evidence'))
DRILL = "/mnt/5_hdd/felhom.eu/drill/app-catalog-drill"
# GUEST=9201 selects demo-hp's hub-enabled guest (the mail proof); default 9202, the scratch guest.
GUEST = os.environ.get("GUEST", "9202")
BASE = os.environ.get("BASE") or {"9202": "https://192.168.0.114", "9201": "https://192.168.0.155"}[GUEST]
DOMAIN = os.environ.get("DOMAIN", "enkisfelhom.hu")
HOSTHDR = f"Host: felhom.{DOMAIN}"
HP = "demo-hp"

LOG = []


def say(*a):
    line = " ".join(str(x) for x in a)
    ts = datetime.now().strftime("%H:%M:%S")
    print(f"{ts} {line}", flush=True)
    LOG.append(f"{ts} {line}")


def sh(args, timeout=300, inp=None):
    try:
        return subprocess.run(args, capture_output=True, text=True, timeout=timeout, input=inp)
    except (subprocess.TimeoutExpired, OSError) as e:
        return subprocess.CompletedProcess(args, 124, "", f"{e}")


def guest(script, timeout=600):
    """Run a bash script inside guest 9202. Piped as a file — never as an argument (quoting)."""
    # ONE TEMP FILE PER CALL (night 2026-09-23): the shared /tmp/w<guest>.sh swapped scripts under
    # two concurrent walks (memory: guest-helper-shares-one-tmp-file).
    import secrets as _s
    t = f"/tmp/w{GUEST}-{os.getpid()}-{_s.token_hex(4)}.sh"
    r = sh(["ssh", "-o", "ConnectTimeout=20", "-o", "StrictHostKeyChecking=accept-new", HP,
            f"export LC_ALL=C; cat > {t}; pct push {GUEST} {t} {t} >/dev/null 2>&1; "
            f"pct exec {GUEST} -- bash {t}; pct exec {GUEST} -- rm -f {t}; rm -f {t}"],
           timeout=timeout, inp=script)
    return r.stdout or ""


def login():
    pw = open(f"{SC}/.ctlpw").read().strip()
    sh(["curl", "-sk", "-D", f"{SC}/hdr{os.getpid()}.txt", "-o", "/dev/null", "-H", HOSTHDR,
        "-X", "POST", "--data-urlencode", f"password={pw}", f"{BASE}/login"])
    h = open(f"{SC}/hdr{os.getpid()}.txt").read()
    m = re.search(r"felhom_session=[A-Za-z0-9._-]+", h, re.I)
    if not m:
        sys.exit("login failed: no session cookie")
    open(f"{SC}/sess{os.getpid()}.txt", "w").write(m.group(0))
    r = sh(["curl", "-sk", "-L", "-H", HOSTHDR, "-H", f"Cookie: {m.group(0)}", f"{BASE}/"])
    c = re.search(r'<meta name="csrf-token" content="([^"]+)"', r.stdout or "")
    if not c:
        sys.exit("login failed: no csrf token")
    open(f"{SC}/csrf{os.getpid()}.txt", "w").write(c.group(1))


def ctl(method, path, data=None, raw=False, tries=2):
    """One controller API call. Re-logs in once on a 302/401 — the controller's session store is
    in memory, so any controller restart during the night invalidates it silently."""
    for attempt in range(tries):
        sess = open(f"{SC}/sess{os.getpid()}.txt").read().strip()
        csrf = open(f"{SC}/csrf{os.getpid()}.txt").read().strip()
        args = ["curl", "-sk", "-H", HOSTHDR, "-H", f"Cookie: {sess}", "-w", "\n%{http_code}"]
        if method != "GET":
            args += ["-H", f"X-CSRF-Token: {csrf}", "-H", "Content-Type: application/json",
                     "-X", method]
            if data is not None:
                args += ["--data", json.dumps(data)]
        args.append(f"{BASE}{path}")
        r = sh(args)
        body, _, code = (r.stdout or "").rpartition("\n")
        if code.strip() in ("302", "401") and attempt + 1 < tries:
            login()
            continue
        if raw:
            return code.strip(), body
        try:
            return code.strip(), json.loads(body)
        except Exception:
            return code.strip(), {"_raw": body[:600]}
    return code.strip(), {"_raw": body[:600]}


def page(path):
    sess = open(f"{SC}/sess{os.getpid()}.txt").read().strip()
    r = sh(["curl", "-sk", "-H", HOSTHDR, "-H", f"Cookie: {sess}", f"{BASE}{path}"])
    return r.stdout or ""


GATE = {}   # sub -> the felhom_gate cookie the household's browser would hold (setup gate, v0.280.0)


def _loc(out):
    m = re.search(r"(?im)^location:\s*(\S+)", out or "")
    return m.group(1) if m else ""


def gate_cookie(sub):
    """Pass the setup gate (`09` decision 46) the way the HOUSEHOLD does: the app host redirects to the
    dashboard's /__gate/start, which (with the dashboard session) redirects back to the app host's
    /__felhom_gate/cb, which sets `felhom_gate`. Never printed."""
    import urllib.parse
    sess = open(f"{SC}/sess{os.getpid()}.txt").read().strip()
    r = sh(["curl", "-sk", "-D", "-", "-o", "/dev/null", "-H", "Accept: text/html", "-H", f"Host: {sub}.{DOMAIN}", f"{BASE}/"])
    loc = _loc(r.stdout)
    if "/__gate/start" not in loc:
        GATE[sub] = ""
        return ""                      # not gated (open, or no gate for this app)
    u = urllib.parse.urlsplit(loc)
    r = sh(["curl", "-sk", "-D", "-", "-o", "/dev/null", "-H", "Accept: text/html", "-H", f"Host: {u.hostname}", "-H", f"Cookie: {sess}",
            f"{BASE}{u.path}?{u.query}"])
    loc = _loc(r.stdout)
    u = urllib.parse.urlsplit(loc)
    if "/__felhom_gate/cb" not in u.path:
        say(f"    gate: the dashboard did not hand back a callback for {sub} ({loc[:80]})")
        return ""
    r = sh(["curl", "-sk", "-D", "-", "-o", "/dev/null", "-H", "Accept: text/html", "-H", f"Host: {u.hostname}", f"{BASE}{u.path}?{u.query}"])
    m = re.search(r"(?im)^set-cookie:\s*(felhom_gate=[^;\r\n]+)", r.stdout or "")
    GATE[sub] = m.group(1) if m else ""
    say(f"    gate: {sub} is gated — passed as the household (cookie {'set' if GATE[sub] else 'NOT set'})")
    return GATE[sub]


def app_curl(sub, path, *extra, method=None, data=None, timeout=45, _retry=True):
    """A call to the APP's own front door on 9202 — the household's route, not ours. Carries the setup
    gate's cookie when the app is gated, merged into a fixture's own Cookie header (never a second one)."""
    raw = list(extra)
    gc = GATE[sub] if sub in GATE else gate_cookie(sub)
    ext = list(raw)
    if gc:
        merged = False
        for i, a in enumerate(ext):
            if isinstance(a, str) and a.lower().startswith("cookie:") and i > 0 and ext[i - 1] == "-H":
                ext[i] = a + "; " + gc
                merged = True
        if not merged:
            ext = ["-H", f"Cookie: {gc}"] + ext
    args = ["curl", "-sSk", "--max-time", str(timeout), "-H", f"Host: {sub}.{DOMAIN}",
            "-w", "\n%{http_code} %{redirect_url}"]
    if method:
        args += ["-X", method]
    if data is not None:
        args += ["--data-binary", "@-"]
    args += ext + [f"{BASE}{path}"]
    r = sh(args, timeout=timeout + 30, inp=data)
    body, _, tail = (r.stdout or "").rpartition("\n")
    code, _, redir = tail.strip().partition(" ")
    if _retry and "/__gate/start" in redir:
        GATE.pop(sub, None)            # the gate cookie expired or was never taken — log in as the household again
        return app_curl(sub, path, *raw, method=method, data=data, timeout=timeout, _retry=False)
    return r.returncode, code.strip(), body


def stack(name):
    _, d = ctl("GET", f"/api/stacks/{name}")
    return (d.get("data") or {}) if isinstance(d, dict) else {}


def wait_app(sub, path="/", want=("200", "302", "303", "401", "403"), tries=60, delay=5):
    """Settling says the container runs; this says the APP answers. Not the same thing."""
    last = None
    for _ in range(tries):
        rc, code, _ = app_curl(sub, path, timeout=15)
        last = (rc, code)
        if rc == 0 and code in want:
            return True
        time.sleep(delay)
    say(f"    app never answered on {sub}{path} (last rc={last[0]} code={last[1]})")
    return False


# ------------------------------------------------------------------ the walk


DRIVE = "/mnt/felhom-drives/scratch_hdd/userdata"

# What THIS run generated for a deploy, per app. Deploy secrets are ENCRYPTED AT REST in
# `app.yaml` (`ENC:…`), which is right and which means a fixture cannot read an app's admin
# password back off the box — the household sees it once. So the value the harness itself
# generated is kept here for the life of the run, and nowhere else.
GENERATED = {}


def deploy_values(name, sub):
    """Fill EVERY required deploy field the way the wizard would, by asking the box what this app
    asks for — `GET /api/stacks/<n>/deploy-fields` — instead of assuming DOMAIN+SUBDOMAIN.

    Measured 2026-09-21: three apps in one batch refused at the deploy with a correct 400 because
    a required field was absent — `HDD_PATH` (navidrome, audiobookshelf) and an admin password
    (grafana). The refusals happen BEFORE anything is created (`deploy.go:324`), which is the only
    reason this was safe to discover by running it (live-probes rule).

    A `path` field must name a directory that ALREADY EXISTS (`deploy.go:330`), so one is made on
    the scratch drive first — the same act the drive browser performs for a household.
    """
    code, d = ctl("GET", f"/api/stacks/{name}/deploy-fields")
    fields = (((d.get("data") or {}).get("metadata") or {}).get("deploy_fields")) or []
    values = {"DOMAIN": DOMAIN, "SUBDOMAIN": sub}
    made = []
    for f in fields:
        ev, ty = f.get("env_var"), f.get("type")
        if ev in values:
            continue
        # `type: password` is MANDATORY whatever `required` says — `deploy.go:305-312` refuses
        # when the caller sends none, deliberately ("the user needs to know their password"),
        # while `.felhom.yml` declares `required: false` and the API serves that verbatim. A
        # caller that trusts the contract gets a 400. Measured tonight on grafana; filed.
        if not f.get("required") and ty != "password":
            continue                      # the controller generates the optional secrets itself
        if ty == "path":
            p = f"{DRIVE}/{name}"
            values[ev] = p
            made.append(p)
        elif ty in ("secret", "password"):
            import secrets as _s
            values[ev] = "Drill-" + _s.token_hex(12)
            GENERATED.setdefault(name, {})[ev] = values[ev]
        elif f.get("default"):
            values[ev] = f["default"]
        else:
            values[ev] = f"drill-{name}"
    if made:
        guest("mkdir -p " + " ".join(made) + "; ls -ld " + " ".join(made))
        say(f"  [1] made the drive paths this app requires: {made}")
    extra = [k for k in values if k not in ("DOMAIN", "SUBDOMAIN")]
    if extra:
        say(f"  [1] required fields filled beyond DOMAIN/SUBDOMAIN: {extra}")
    return values


def deploy(name, sub, extra_values=None):
    st = stack(name)
    if st.get("deployed"):
        say(f"  [1] {name} already deployed — reusing")
        return True
    values = deploy_values(name, sub)
    if extra_values:
        values.update(extra_values)
    body = {"values": values}
    if os.environ.get("KEPT"):  # decision 36: the household's answer when the drive holds old data ("fresh" moves it aside, deletes nothing)
        body["kept_data"] = os.environ["KEPT"]
    code, d = ctl("POST", f"/api/stacks/{name}/deploy", body)
    say(f"  [1] deploy -> {code} {str(d)[:120]}")
    if code != "202":
        return False
    # WAIT FOR `deployed`, NOT FOR `running`. Measured 2026-09-21 on tandoor: docker reported the
    # container `healthy` while the controller's own state read `unhealthy` — a gate on `running`
    # alone therefore times out on an app that is up. The state is RECORDED rather than required;
    # the real gate is the fixture's own `wait_app`, which asks whether the APP answers.
    seen = None
    for _ in range(90):
        time.sleep(5)
        st = stack(name)
        seen = st.get("state")
        # `deployed` alone is NOT enough and `state` alone is NOT right. Measured 2026-09-21:
        # tandoor reads `unhealthy` while serving (R-618), so gating on "running" hangs; and romm
        # read `deployed=True, state=degraded, pinned_images=None` twenty seconds in, i.e. the
        # deploy had not finished writing app.yaml. The PIN is the deploy's own completion mark
        # (`runComposeDeploy` writes it), so that is what to wait for.
        pins = (st.get("app_config") or {}).get("pinned_images")
        if st.get("deployed") and pins and seen in ("running", "unhealthy", "degraded"):
            say(f"  [1] deployed, controller state={seen}, "
                f"pinned={(st.get('app_config') or {}).get('pinned_images')}")
            if seen != "running":
                say(f"  [1] NOTE: the controller's own state is {seen!r}, not 'running' — recorded, "
                    f"not treated as a failure; the fixture's front-door wait is the real gate")
            return True
    say(f"  [1] never became deployed (last controller state={seen!r})")
    return False


def backup_now(name):
    """R-648 (2026-09-23): NO whole-box „Mentés most" from a drill, ever.

    `POST /api/backup/run` is the only backup endpoint and it is WHOLE-BOX: on 9201 it stopped and
    restarted 9 of 10 standing apps twice, and on 9202 it broke a deploy in flight (R-634). The product
    has NO per-app backup endpoint (router.go: /backup/run, /backup/tier2 only); the per-app backup
    exists only inside the guarded update, whose `backing-up` phase calls RunAppBackupNow for the one
    app. So this presses nothing: the update takes the throwaway app's own backup, and says so in its
    phase list. A seed written "after the backup" is therefore written before the update's own backup
    — the undo's last-second copy is still the one that must bring it back."""
    say(f"  [4] backup press SKIPPED for {name} (R-648: whole-box only; the update's backing-up phase backs up {name} alone)")
    return None

def drill_bump(app, frm, to, service_hint=None):
    """Serialised across concurrent walks: one git working tree, one lock."""
    import fcntl
    with open(f"{SC}/drill.lock", "w") as lk:
        fcntl.flock(lk, fcntl.LOCK_EX)
        sh(["git", "-C", DRILL, "pull", "-q", "--rebase", "origin", "main"], timeout=120)
        return _drill_bump(app, frm, to, service_hint)


def _drill_bump(app, frm, to, service_hint=None):
    """Commit the edge to the DRILL repo. catalog_since set by hand (the drill repo has no gates).

    `frm`/`to` may be comma-separated lists of the SAME length: an app whose own version lives in
    two images (adventurelog's backend and frontend) moves both in one edge, while its engine
    sidecar stays where it is — `09` §3b Q3's rule is per SERVICE, and an app-half edge must move
    every service that carries the app's own version and no others.
    """
    comp = f"{DRILL}/templates/{app}/docker-compose.yml"
    fy = f"{DRILL}/templates/{app}/.felhom.yml"
    s = open(comp).read()
    froms = [x.strip() for x in frm.split(",") if x.strip()]
    tos = [x.strip() for x in to.split(",") if x.strip()]
    if len(froms) != len(tos):
        say(f"  [5] from/to lists differ in length: {froms} vs {tos}")
        return None
    for f1, t1 in zip(froms, tos):
        if f"image: {f1}" not in s:
            say(f"  [5] FROM ref not found in compose: {f1}")
            return None
        s = s.replace(f"image: {f1}", f"image: {t1}")
    open(comp, "w").write(s)
    f = open(fy).read()
    today = datetime.now().strftime("%Y-%m-%d")
    f = re.sub(r'^catalog_since:.*$', f'catalog_since: "{today}"', f, count=1, flags=re.M)
    open(fy, "w").write(f)
    sh(["git", "-C", DRILL, "add", "-A"])
    sh(["git", "-C", DRILL, "commit", "-q", "-m", f"DRILL {app}: {frm} -> {to}"])
    r = sh(["git", "-C", DRILL, "push", "-q", "origin", "main"], timeout=120)
    h = sh(["git", "-C", DRILL, "rev-parse", "--short=12", "HEAD"]).stdout.strip()
    say(f"  [5] drill commit {h}: {app} {frm} -> {to} (push rc={r.returncode})")
    return h


def sync_rescan(expect_app=None, expect_ref=None, tries=12, delay=5):
    """Sync, rescan, and — when told what to expect — WAIT FOR THE BADGE TO CATCH UP.

    R-607: `POST /api/sync` answers "nincs valtozas" while the catalog HAS moved, and
    `catalog_images` stays stale until a separate rescan. Tonight showed the rescan alone is not
    enough either: mealie's badge read "Naprakesz" seconds after its bump was pushed, and the
    Update that followed moved nothing and still reported "Frissitve". So when the caller knows
    which reference should appear, this polls for it and SAYS HOW LONG IT TOOK — which is the
    NUMBER R-607 asks for and has never had.
    """
    t0 = time.time()
    ctl("POST", "/api/sync")
    time.sleep(2)
    ctl("POST", "/api/stacks/rescan")
    time.sleep(2)
    if not expect_app or not expect_ref:
        return None
    for i in range(tries):
        cat = stack(expect_app).get("catalog_images") or {}
        if expect_ref in cat.values():
            waited = round(time.time() - t0, 1)
            if i:
                say(f"  [sync] the badge needed {waited}s and {i+1} sync+rescan rounds to catch up "
                    f"to {expect_ref} — R-607's window, measured")
            return waited
        time.sleep(delay)
        ctl("POST", "/api/sync")
        time.sleep(1)
        ctl("POST", "/api/stacks/rescan")
    say(f"  [sync] the badge NEVER caught up to {expect_ref} in {round(time.time()-t0,1)}s — "
        f"catalog_images = {stack(expect_app).get('catalog_images')}")
    return None


def badges(name):
    out = {}
    for lang, suffix in (("hu", ""), ("en", "?lang=en")):
        h = page(f"/apps/{name}{suffix}")
        m = re.findall(r'<span class="tag tag-[^"]*"[^>]*title="([^"]*)"[^>]*>([^<]*)<', h)
        out[lang] = [{"title": a.strip(), "text": b.strip()} for a, b in m][:3]
    return out


def press_update(name, poll=1.0, cap_s=1800):
    code, d = ctl("POST", f"/api/stacks/{name}/update")
    say(f"  [6] Update -> {code} {str(d)[:220]}")
    if code not in ("202", "200"):
        return {"accepted": False, "http": code, "refusal": d, "phases": [], "duration_s": 0}
    phases, seen, t0 = [], None, time.time()
    while time.time() - t0 < cap_s:
        st = stack(name)
        ph = st.get("update_phase")
        if ph != seen:
            seen = ph
            rec = {"t": round(time.time() - t0, 1), "phase": ph,
                   "label": st.get("update_phase_label"), "updating": st.get("updating"),
                   "error": st.get("update_error"), "hold": st.get("hold_reason")}
            phases.append(rec)
            say(f"      +{rec['t']:>6.1f}s  phase={ph}  label={rec['label']}  "
                f"err={rec['error']}  hold={rec['hold']}")
        if not st.get("updating") and ph in ("done", "failed", "undone", None) and time.time() - t0 > 3:
            break
        time.sleep(poll)
    st = stack(name)
    return {"accepted": True, "http": code, "phases": phases,
            "duration_s": round(time.time() - t0, 1),
            "final_phase": st.get("update_phase"), "update_error": st.get("update_error"),
            "hold_reason": st.get("hold_reason"), "state": st.get("state")}


def observables(name):
    st = stack(name)
    ac = st.get("app_config") or {}
    live = guest(f"""
grep -E '^\\s+image:' /opt/docker/stacks/{name}/docker-compose.yml 2>/dev/null | sed 's/^ *//'
echo '---inspect---'
for c in $(docker ps -a --filter label=com.docker.compose.project={name} --format '{{{{.Names}}}}'); do
  echo -n "$c  "; docker inspect "$c" --format '{{{{.Config.Image}}}}  running={{{{.State.Running}}}}  restarts={{{{.RestartCount}}}}'
done
""")
    a, _, b = live.partition("---inspect---")
    return {
        "pinned_images": ac.get("pinned_images"),
        "installed_images": {k: (v.get("ref") if isinstance(v, dict) else v)
                             for k, v in (ac.get("installed_images") or {}).items()},
        "catalog_images": st.get("catalog_images"),
        "live_compose_image_lines": [x for x in a.strip().splitlines() if x.strip()],
        "docker_inspect": [x for x in b.strip().splitlines() if x.strip()],
    }


def app_logs(name, lines=400):
    """The app's own container log, DECODED. The endpoint answers a JSON envelope whose `logs` is
    one string with escaped newlines — a scan over the envelope sees a single enormous line and
    finds nothing, which reads exactly like "the app printed no migration line" and is not. R-96
    rule 3 in a new place: an absent line is not evidence when the instrument cannot see lines."""
    code, d = ctl("GET", f"/api/stacks/{name}/logs?lines={lines}")
    if isinstance(d, dict):
        data = d.get("data")
        if isinstance(data, dict) and isinstance(data.get("logs"), str):
            return data["logs"]
        if isinstance(d.get("_raw"), str):
            return d["_raw"]
    return str(d)


def write_verdict(rec, appdir):
    os.makedirs(appdir, exist_ok=True)
    p = os.path.join(appdir, "verdict.json")
    json.dump(rec, open(p, "w"), indent=2, ensure_ascii=False)
    say(f"  [9] verdict {rec['verdict']}  -> {p}")


def remove(name):
    """Remove through the PRODUCT, never `docker rm` (live-probes rule). The remove endpoint
    refuses a running stack — `409 still running` — so the stop is part of the act, not a tidy-up."""
    c1, d1 = ctl("POST", f"/api/stacks/{name}/stop")
    say(f"  [X] stop -> {c1} {str(d1)[:100]}")
    for _ in range(24):
        time.sleep(5)
        if stack(name).get("state") != "running":
            break
    code, d = ctl("POST", f"/api/stacks/{name}/remove",
                  {"remove_hdd_data": True, "remove_backups": True})
    say(f"  [X] remove (with drive data) -> {code} {str(d)[:160]}")
    if code == "409":
        # R-442's fail-closed guard: when the storage subsystem cannot RESOLVE the app's drive
        # path, the removal is REFUSED and the app is kept rather than half-deleted. On guest 9202
        # `/api/disks` answers `agent not configured`, so every app deployed with an HDD_PATH hits
        # this. The household's other choice — remove the app, KEEP the data — is accepted, and the
        # harness takes it, then tidies its own directory by name at teardown.
        say("  [X] refused because the drive path cannot be resolved (R-442, fail-closed and right)"
            " — removing the app and KEEPING the drive data instead")
        code, d = ctl("POST", f"/api/stacks/{name}/remove",
                      {"remove_hdd_data": False, "remove_backups": True})
        say(f"  [X] remove (keeping drive data) -> {code} {str(d)[:160]}")
    time.sleep(5)
    st = stack(name)
    left = guest(f"ls -d /opt/docker/stacks/{name} 2>/dev/null; "
                 f"docker ps -a --filter label=com.docker.compose.project={name} --format '{{{{.Names}}}}'")
    say(f"  [X] after remove: deployed={st.get('deployed')} leftovers={left.strip()!r}")
    return code


def app_env(name, key):
    """Read one deploy value the CUSTOMER was given (e.g. the generated admin password) from the
    app's own `app.yaml`. This is not seeding — it is how the household logs in; the controller
    shows them the same value. Data still goes in through the app's own front door."""
    out = guest(f"grep -E '^\\s*{key}:' /opt/docker/stacks/{name}/app.yaml 2>/dev/null | head -1")
    if ":" in out:
        return out.split(":", 1)[1].strip().strip('"').strip("'")
    return ""


def snapshots(name):
    """The restorable copies the backups page offers for this app."""
    code, d = ctl("GET", f"/api/backup/snapshots?stack={name}")
    data = d.get("data") if isinstance(d, dict) else None
    if isinstance(data, dict):
        for k in ("snapshots", "items", "restore_points"):
            if isinstance(data.get(k), list):
                return data[k]
    return data if isinstance(data, list) else []


def restore(name, snapshot_id=None, wait_s=1200):
    """The household's own way out: the „Visszaállítás a mentésből" button on the backups page.

    A FORM post, not an API call — `POST /backup/restore` with `_csrf`, `stack_name`,
    `snapshot_id` — because that is the button the sentence tells them to press.
    """
    snaps = snapshots(name)
    if snapshot_id is None:
        if not snaps:
            say(f"  [R] no restorable copy offered for {name}")
            return {"ok": False, "why": "no snapshot offered", "snapshots": snaps}
        first = snaps[0]
        snapshot_id = first.get("id") or first.get("snapshot_id") or first.get("short_id")
    say(f"  [R] restoring {name} from snapshot {snapshot_id!r} (of {len(snaps)} offered)")
    sess = open(f"{SC}/sess{os.getpid()}.txt").read().strip()
    csrf = open(f"{SC}/csrf{os.getpid()}.txt").read().strip()
    r = sh(["curl", "-sk", "-D", "-", "-o", "/dev/null", "-H", HOSTHDR, "-H", f"Cookie: {sess}",
            "-X", "POST",
            "--data-urlencode", f"_csrf={csrf}",
            "--data-urlencode", f"stack_name={name}",
            "--data-urlencode", f"snapshot_id={snapshot_id}",
            f"{BASE}/backup/restore"], timeout=180)
    head = (r.stdout or "").split("\n")[0].strip()
    loc = [l for l in (r.stdout or "").split("\n") if l.lower().startswith("location:")]
    say(f"  [R] POST /backup/restore -> {head} {loc[:1]}")
    t0 = time.time()
    last = None
    while time.time() - t0 < wait_s:
        code, d = ctl("GET", "/api/backup/restore-status")
        dd = d.get("data") or {}
        cur = (dd.get("running"), dd.get("phase") or dd.get("state"), dd.get("message"))
        if cur != last:
            say(f"      +{round(time.time()-t0,1):>6.1f}s restore {cur}")
            last = cur
        if not dd.get("running", False) and time.time() - t0 > 5:
            break
        time.sleep(2)
    st = stack(name)
    say(f"  [R] after restore: state={st.get('state')} hold={st.get('hold_reason')!r} "
        f"phase={st.get('update_phase')}")
    return {"ok": True, "snapshot_id": snapshot_id, "snapshots": snaps,
            "http": head, "location": loc[:1], "seconds": round(time.time() - t0, 1),
            "state_after": st.get("state"), "hold_after": st.get("hold_reason"),
            "observables_after": observables(name)}