felhom.eu/documentation/audits/update-arc-gaps-2026-09-21/unattended-caller.py

#!/usr/bin/env python3
"""unattended-caller.py — the Slice 6 spike: an app updates with NOBODY pressing anything.

THIS IS EVIDENCE, NOT PRODUCT. It lives under documentation/audits/ and nothing in the controller
imports it. It exists to MEASURE the mechanism Slice 6 would need before that slice is designed, by
pressing exactly the same guarded Update a person presses — `POST /api/stacks/<n>/update` — and
nothing else. No new endpoint, no new controller code, no privileged path.

WHAT IT DOES NOT DO, deliberately: it does not decide policy. The window is simulated by running it;
the per-app switch of `09` §3b Q2 does not exist yet; it never touches a box that is not the scratch
guest. Run it from DooPlex against guest 9202 only.

THE TWO RULES IT EXISTS TO PROVE
  1. within a major, for EVERY compose service, or it does not press at all (§3 decision 3, §3b Q3);
  2. a refusal's REASON decides whether it ever presses again (R-609):
        TRANSIENT  busy updating deploying migrating self_updating  -> try next pass
        TERMINAL   held downgrade                                   -> never again
        FOR A HUMAN memory disk no_backup                           -> log and leave alone
     Before R-609 the body carried only a Hungarian sentence, so this distinction was unavailable to
     anything that is not a person — which is why a caller like this could not have been written.

Usage:  python3 unattended-caller.py --passes 6 --every 300
"""
import argparse, json, re, subprocess, sys, time

CALL = "/tmp/ctl/c.sh"                       # the helper from 00-api-recipe.md
TRANSIENT = {"busy", "updating", "deploying", "migrating", "self_updating"}
TERMINAL = {"held", "downgrade"}
FOR_A_HUMAN = {"memory", "disk", "no_backup"}
TAG = re.compile(r"^v?(\d+)(?:\.(\d+))?(?:\.(\d+))?(.*)$")


def log(msg):
    print("%s  %s" % (time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), msg), flush=True)


def call(method, path, data=None):
    cmd = [CALL, method, path] + ([data] if data else [])
    out = subprocess.run(cmd, capture_output=True, text=True, timeout=180).stdout
    try:
        return json.loads(out)
    except Exception:
        return {"_raw": out}


def split_ref(ref):
    """(repo, tag) or (None, None) when the reference carries no plain tag."""
    if "@" in ref:
        return None, None
    i = ref.rfind(":")
    if i < 0 or "/" in ref[i + 1:]:
        return None, None
    return ref[:i], ref[i + 1:]


def same_major(a, b):
    """True only when BOTH tags are plain versions, share a suffix, and share a first number.

    Mirrors stacks.CompareImageRefs deliberately: the box's own rule is the one under test, and a
    caller that judged 'within a major' differently would measure its own opinion instead.
    """
    ra, ta = split_ref(a)
    rb, tb = split_ref(b)
    if ra is None or ra != rb:
        return False
    ma, mb = TAG.match(ta or ""), TAG.match(tb or "")
    if not ma or not mb:
        return False
    if ma.group(4) != mb.group(4):          # the suffix must be IDENTICAL (…-apache vs …-apache)
        return False
    if ma.group(2) is None or mb.group(2) is None:
        return False                        # one component is a LINE, not a version
    return ma.group(1) == mb.group(1)


def edge_is_within_a_major(st):
    """ALL services must pass. One unorderable service makes the whole edge 'across' -> a human."""
    installed = {k: v["ref"] for k, v in (st.get("app_config", {}).get("installed_images") or {}).items()}
    catalog = st.get("catalog_images") or {}
    if not installed or not catalog or set(installed) != set(catalog):
        return False, "service sets differ or nothing recorded"
    for svc, want in catalog.items():
        got = installed[svc]
        if got == want:
            continue
        if not same_major(got, want):
            return False, "across: %s %s -> %s" % (svc, got, want)
    return True, "within a major"


def follow(name, timeout=900):
    """Watch one update to its end. Returns done | failed | held | timeout."""
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        st = call("GET", "/api/stacks/%s" % name)
        phase, updating = st.get("update_phase"), st.get("updating")
        if phase != last:
            log("    %s: phase=%s updating=%s" % (name, phase, updating))
            last = phase
        if not updating and phase in ("done", "failed"):
            held = bool(st.get("hold_reason"))
            return "held" if held else phase
        time.sleep(2)
    return "timeout"


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--passes", type=int, default=6)
    ap.add_argument("--every", type=int, default=300)
    args = ap.parse_args()

    never_again, outcomes = set(), {}
    for p in range(1, args.passes + 1):
        log("=== pass %d/%d" % (p, args.passes))
        call("POST", "/api/sync")
        call("POST", "/api/stacks/rescan")        # R-607: a sync can say 'no change' and still move
        stacks = call("GET", "/api/stacks")
        stacks = stacks if isinstance(stacks, list) else stacks.get("data", [])

        for st in stacks:
            name = st.get("name")
            if not st.get("deployed") or st.get("protected") or name in never_again:
                continue
            installed = {k: v["ref"] for k, v in (st.get("app_config", {}).get("installed_images") or {}).items()}
            if not installed or installed == (st.get("catalog_images") or {}):
                continue                          # unknown, or level with the catalog
            ok, why = edge_is_within_a_major(st)
            if not ok:
                log("  %s SKIP — %s" % (name, why))
                continue

            log("  %s BEHIND and within a major — pressing Update" % name)
            r = call("POST", "/api/stacks/%s/update" % name)
            if r.get("ok") is False:
                reason = (r.get("data") or {}).get("reason", "")
                sentence = r.get("error", "")
                if reason in TERMINAL:
                    never_again.add(name)
                    log("  %s REFUSED reason=%s TERMINAL — will not press again. %s" % (name, reason, sentence))
                elif reason in TRANSIENT:
                    log("  %s REFUSED reason=%s transient — retry next pass. %s" % (name, reason, sentence))
                elif reason in FOR_A_HUMAN:
                    never_again.add(name)
                    log("  %s REFUSED reason=%s — needs a person. %s" % (name, reason, sentence))
                else:
                    never_again.add(name)
                    log("  %s REFUSED reason=%r UNKNOWN — stopping on it, safest reading. %s" % (name, reason, sentence))
                continue

            started = time.time()
            end = follow(name)
            outcomes[name] = (end, round(time.time() - started, 1))
            log("  %s ENDED %s after %.1fs" % (name, end, time.time() - started))
            if end in ("held", "timeout"):
                never_again.add(name)

        if p < args.passes:
            time.sleep(args.every)

    log("=== summary  outcomes=%s  never_again=%s" % (outcomes, sorted(never_again)))
    return 0


if __name__ == "__main__":
    sys.exit(main())