app-catalog-felhom.eu/scripts/onboarding_gaps.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""onboarding_gaps.py — what the catalog's FILES already show, per checklist group, for the 53 exempt apps.

Writes onboarding/EXISTING-APPS-GAPS.md. READ ONLY: no network, no containers, no re-testing. It answers "which of
the NEW-APP-CHECKLIST.md groups does the catalog already hold a record for, per old app" from what is committed:
the templates, the ladder entries, the fixture tables, FIRST-ADMIN.md, README.md and the 2026-08-02 persistence
sweep. A cell is a signal the files carry, not a measurement made today — "shown" means a file says it, never
that it is still true. This is information, not work (operator default 2026-10-01).

Run from the repo root (PyYAML needed — this is a local report, not a gate):
    python3 scripts/onboarding_gaps.py            # rewrite onboarding/EXISTING-APPS-GAPS.md
    python3 scripts/onboarding_gaps.py --check    # exit 1 if the committed page differs from a fresh run
"""
import datetime
import io
import json
import os
import re
import subprocess
import sys

import yaml

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
OUT = os.path.join(ROOT, "onboarding", "EXISTING-APPS-GAPS.md")
SWEEP = os.path.join(ROOT, "audits", "persistence-sweep-2026-08-02", "state", "gate.log")
sys.path.insert(0, os.path.join(ROOT, "scripts"))


def exempt():
    src = io.open(os.path.join(ROOT, "scripts", "check-onboarding.py"), encoding="utf-8").read()
    return sorted(re.search(r'EXEMPT = frozenset\("""(.*?)"""', src, re.S).group(1).split())


def mb(v):
    m = re.match(r"^\s*(\d+(?:\.\d+)?)\s*([MG])", str(v or ""))
    if not m:
        return None
    return int(float(m.group(1)) * (1024 if m.group(2) == "G" else 1))


def fixture_apps():
    names = set()
    for f in ("upgrade_fixtures.py", "upgrade_fixtures_box.py", "upgrade_fixtures_box28.py"):
        src = io.open(os.path.join(ROOT, "scripts", f), encoding="utf-8").read()
        for block in re.findall(r"FIXTURES(?:28)?(?: = |\.update\()\{(.*?)\n\}", src, re.S):
            names |= set(re.findall(r'^\s+"([a-z0-9-]+)":', block, re.M))
    return names


def sweep_verdicts():
    """The FIRST run's lists in the 2026-08-02 sweep (53 in scope). BROKEN and UNDETERMINED are named; the rest
    of the 53 were CLEAN."""
    out = {}
    if not os.path.isfile(SWEEP):
        return out
    text = io.open(SWEEP, encoding="utf-8").read()
    first = text.split("of 53 in scope")[0]
    sec = None
    for line in first.splitlines():
        if line.startswith("BROKEN"):
            sec = "broken"
        elif line.startswith("UNDETERMINED"):
            sec = "undetermined"
        elif sec and re.match(r"^  ([a-z0-9-]+)(:|$)", line):
            out[re.match(r"^  ([a-z0-9-]+)", line).group(1)] = sec
    return out


def first_admin_rows():
    rows = {}
    for line in io.open(os.path.join(ROOT, "FIRST-ADMIN.md"), encoding="utf-8"):
        cells = [c.strip() for c in line.strip().strip("|").split("|")]
        if len(cells) >= 6 and re.match(r"^\**[a-z0-9-]+\**$", cells[0]) and cells[1][:1].isdigit():
            rows[cells[0].strip("*")] = {"class": cells[1], "measured": "**M" in cells[5] or cells[5].startswith("M")}
    return rows


def main(argv):
    apps = exempt()
    fx = fixture_apps()
    sweep = sweep_verdicts()
    fa = first_admin_rows()
    readme = io.open(os.path.join(ROOT, "README.md"), encoding="utf-8").read()
    rows, tally = [], {g: 0 for g in range(9)}
    notes = {"mem_mismatch": [], "no_limit": [], "maria": [], "pg18": []}
    for app in apps:
        d = os.path.join(ROOT, "templates", app)
        fy_text = io.open(os.path.join(d, ".felhom.yml"), encoding="utf-8").read()
        fy = yaml.safe_load(fy_text) or {}
        dc = yaml.safe_load(io.open(os.path.join(d, "docker-compose.yml"), encoding="utf-8")) or {}
        svcs = dc.get("services") or {}
        ai = fy.get("app_info") or {}
        res = fy.get("resources") or {}
        cell = {}

        # 0 Fit
        life = (fy.get("lifecycle") or "available")
        g0 = life == "available" and bool(ai.get("use_cases")) and "pi_compatible" in res
        cell[0] = ("yes" if g0 else "—") + ("" if life == "available" else " (%s)" % life)

        # 1 Images, start command, database — the files show the ENGINE rules only
        engines, ok1 = [], True
        for sn, s in svcs.items():
            img = str(s.get("image", ""))
            env = s.get("environment") or []
            env = env if isinstance(env, list) else ["%s=%s" % kv for kv in env.items()]
            if img.startswith("mariadb:"):
                engines.append("mariadb")
                if not any(str(e).replace(" ", "") in ("MARIADB_AUTO_UPGRADE=1", "MARIADB_AUTO_UPGRADE=\"1\"") for e in env):
                    ok1 = False
                    notes["maria"].append(app)
            if re.match(r"^(postgres|postgis/postgis|ghcr\.io/immich-app/postgres):", img):
                engines.append("pg")
                if re.match(r"^postgres:18", img) and any(":/var/lib/postgresql/data" in str(v) for v in s.get("volumes") or []):
                    ok1 = False
                    notes["pg18"].append(app)
            if ":latest" in img or ":" not in img.split("/")[-1]:
                ok1 = False
        cell[1] = ("engine rules hold" if engines else "no DB sidecar") if ok1 else "ENGINE RULE BROKEN"
        g1 = ok1

        # 2 Storage and backup
        sv = sweep.get(app, "clean" if sweep else None)
        hdd = bool(res.get("needs_hdd"))
        g2 = sv == "clean" and (not hdd or "backup" in fy)
        cell[2] = "%s%s" % ("sweep %s" % sv if sv else "no sweep",
                            (", backup classes" if "backup" in fy else ", HDD w/o classes") if hdd else "")

        # 3 Accounts and strangers
        r = fa.get(app)
        mech = [k for k in ("after_install", "setup_gate", "signup_block", "after_setup") if fy.get(k)]
        g3 = bool(r and r["measured"])
        cell[3] = ("class %s, %s" % (r["class"], "measured" if r["measured"] else "read only") if r else "no row") + \
                  ((" + " + "/".join(mech)) if mech else "")

        # 4 Health
        all_hc = all(s.get("healthcheck") for s in svcs.values())
        probe = bool((fy.get("healthcheck") or {}).get("checks"))
        # the probe dials the container named like the stack, or the one its `container:` names (R-630)
        targets = {c.get("container") for c in (fy.get("healthcheck") or {}).get("checks") or [] if c.get("container")}
        names = {s.get("container_name") for s in svcs.values()}
        named = (app in names) if not targets else targets <= names
        g4 = all_hc and probe and named
        cell[4] = "yes" if g4 else ", ".join(x for x, ok in (("hc missing", all_hc), ("no probe", probe),
                                                              ("probe container not in compose", named)) if not ok)

        # 5 Resources
        lims = [mb(((s.get("deploy") or {}).get("resources") or {}).get("limits", {}).get("memory")) for s in svcs.values()]
        every = all(lims)
        total = sum(x for x in lims if x)
        ml = mb(res.get("mem_limit"))
        if ml != total:
            notes["mem_mismatch"].append("%s (%s vs %d)" % (app, res.get("mem_limit"), total))
        if not every:
            notes["no_limit"].append(app)
        ladder = [json.loads(l.strip()[2:]) for l in fy_text.splitlines() if l.strip().startswith('- {"from"')]
        watched = [e for e in ladder if e.get("memory_peak_pct") is not None]
        g5 = every and ml == total and bool(watched)
        cell[5] = ("watched %.0f%%" % watched[-1]["memory_peak_pct"] if watched else "no watch") + \
                  ("" if ml == total else ", mem_limit≠sum") + ("" if every else ", a service w/o limit")

        # 6 Updates
        proven = [e for e in ladder if e.get("verdict") == "proven" and not e.get("backfilled")]
        g6 = bool(proven) and app in fx
        cell[6] = "%d proven step(s)%s" % (len(proven), ", fixture" if app in fx else ", no fixture")

        # 7 Mail — the files cannot say whether an app WANTS mail; only whether it is mapped
        cell[7] = "smtp mapped" if fy.get("smtp_mapping") else "—"

        # 8 Text and listing
        en = bool((fy.get("i18n") or {}).get("en"))
        txt = all(ai.get(k) for k in ("tagline", "use_cases", "first_steps"))
        # README's App Catalog table names an app by display name, so the row is found by its subdomain cell
        listed = re.search(r"\|\s*%s\.\*\s*\|" % re.escape(str(fy.get("subdomain", "\0"))), readme) is not None
        g8 = en and txt and listed and bool(r)
        cell[8] = "yes" if g8 else ", ".join(x for x, ok in (("no en", en), ("app_info gap", txt),
                                                              ("not in README", listed), ("no FIRST-ADMIN row", bool(r))) if not ok)
        for g, v in enumerate((g0, g1, g2, g3, g4, g5, g6, None, g8)):
            if v:
                tally[g] += 1
        rows.append((app, cell))

    n = len(apps)
    sha = subprocess.run(["git", "rev-parse", "--short", "HEAD"], cwd=ROOT, capture_output=True, text=True).stdout.strip()
    L = []
    L.append("# EXISTING APPS — what the catalog already shows, per checklist group")
    L.append("")
    L.append("> Generated by `scripts/onboarding_gaps.py` from committed files (catalog `%s`). **Do not edit by hand.**" % sha)
    L.append("> Read only: nothing was re-tested. A cell is what a FILE says, not a measurement made today. The 53 apps")
    L.append("> published before the checklist (2026-10-01) are exempt from the onboarding gate; this page is information,")
    L.append("> not work (operator default 2026-10-01, may be reversed).")
    L.append("")
    L.append("## Headline — apps whose files show the group covered (of %d)" % n)
    L.append("")
    L.append("| group | covered | what \"covered\" means here (the signal read) |")
    L.append("|---|---|---|")
    defs = [
        ("0 Fit", "`lifecycle` available, `use_cases` and `pi_compatible` present — licence, telemetry, internet need and phone apps are recorded nowhere"),
        ("1 Images, start command, DB", "pins clean and the engine rules hold (MariaDB auto-upgrade, PG 18 mount) — entrypoint switches, the production server, migrations and secrets read (1.4–1.9) are recorded for NO app"),
        ("2 Storage and backup", "the 2026-08-02 persistence sweep read the app CLEAN, and an HDD app carries `backup:` classes — no restore round trip is recorded per app"),
        ("3 Accounts and strangers", "a FIRST-ADMIN.md row whose source is MEASURED on a box — lock-out (3.6) is recorded only for the R-752 apps"),
        ("4 Health", "every service has a compose healthcheck, a controller probe exists, the exposed container is named like the stack — no negative control is recorded"),
        ("5 Resources", "every service limited, `mem_limit` = the sum, and a ladder entry carries a measured memory watch — no first-start-from-birth watch is recorded except immich's"),
        ("6 Updates", "a proven (not backfilled) ladder step AND an upgrade fixture"),
        ("7 Mail", "not countable from files: whether an app WANTS mail is not recorded; the mapped count is below"),
        ("8 Text and listing", "English block, tagline + use_cases + first_steps, listed in README, a FIRST-ADMIN row"),
    ]
    for g, (name, d) in enumerate(defs):
        L.append("| %s | %s | %s |" % (name, "—" if g == 7 else "%d / %d" % (tally[g], n), d))
    L.append("")
    L.append("Mail: %d app(s) carry `smtp_mapping`." % sum(1 for _a, c in rows if c[7] == "smtp mapped"))
    L.append("")
    L.append("## Found while computing this page")
    L.append("")
    L.append("- `mem_limit` differs from the sum of the compose limits (REUSE.md §2 says equal): %d — %s."
             % (len(notes["mem_mismatch"]), ", ".join(notes["mem_mismatch"]) or "none"))
    L.append("- A service with no memory limit: %s." % (", ".join(notes["no_limit"]) or "none"))
    L.append("- A MariaDB sidecar without `MARIADB_AUTO_UPGRADE=1`: %s." % (", ".join(notes["maria"]) or "none"))
    L.append("- A PostgreSQL 18 data mount at the old path: %s." % (", ".join(notes["pg18"]) or "none"))
    L.append("")
    L.append("## Per app")
    L.append("")
    L.append("| app | 0 fit | 1 images/DB | 2 storage | 3 accounts | 4 health | 5 resources | 6 updates | 7 mail | 8 text |")
    L.append("|---|---|---|---|---|---|---|---|---|---|")
    for app, c in rows:
        L.append("| %s | %s |" % (app, " | ".join(c[g] for g in range(9))))
    L.append("")
    body = "\n".join(L)

    if "--check" in argv:
        cur = io.open(OUT, encoding="utf-8").read() if os.path.isfile(OUT) else ""
        strip = lambda t: re.sub(r"catalog `[0-9a-f]+`", "catalog `X`", t)
        if strip(cur) != strip(body):
            print("EXISTING-APPS-GAPS.md is stale — run python3 scripts/onboarding_gaps.py")
            return 1
        print("EXISTING-APPS-GAPS.md is current")
        return 0
    io.open(OUT, "w", encoding="utf-8").write(body)
    print("wrote %s — %d apps; covered per group: %s" % (os.path.relpath(OUT, ROOT), n,
                                                        ", ".join("%d:%s" % (g, "-" if g == 7 else tally[g]) for g in range(9))))
    return 0


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