path: root/docs/superpowers/plans/2026-10-19-sb-builder-phase2.md

# sb-builder Phase 2 (Packages and slackrepo builds) Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Package browser with a structured hint view, slackrepo build/rebuild/update/remove and update-all as previewed jobs, a publish gate with a separate Publish action, and signing key unlock for jobs without a tty.

**Architecture:** slackrepo_setup becomes the home of the slackrepo config (site values moved to a root-only `.local`), gains `sb-publish` and the hook gates. `sbb-remote` 0.2.0 adds `inventory`, `slackrepo` and `publish` kinds (kinds now return argv plus env), `gpg-status` and `gpg-unlock` (passphrase on stdin only). The GUI adds `hints.py` (pure slackrepo hint rules), the Packages tab, Run it with a publish box and the passphrase dialog in the Jobs tab, the key indicator and tile links. Spec: `docs/superpowers/specs/2026-10-19-sb-builder-phase2-design.md`.

**Tech Stack:** as phase 1: Python 3.11+ stdlib, PyQt6, bash; gpg 2.5 on the VM.

**Verified while planning:** all code below ran in a scratch copy: 68 tests pass, offscreen. Against the real VM (helper copied to `/tmp`, temp state dir): `inventory` returns 9957 packages in 2.1s warm / 6.1s cold, 1.5 MB, 599 with a hint, hint key counts match the census; `gpg-status` reports `{"unlocked": false}` for the configured key; the Packages tab rendered that inventory correctly. The config hook gates were exercised with stubs for every `OPT_DRY_RUN`/`OPT_PREVIEW`/`SBB_PUBLISH` combination. `make-local.py` was run against a copy of the live config (5 keys found, mode 600, refuses to overwrite); that copy was shredded.

**Rules for whoever executes this:**
- Commits are GPG-signed; never disable signing. Global hooks scan for personal data; a rejection is right until proven otherwise.
- Never stage `HANDOFF.md`. Never commit or print site values (rsync target, Gotify URL or key, hostnames).
- Anything touching the VM or deleting files asks the user first: Task 1 Step 5 and Task 5.
- GUI tests must run offscreen (the test file forces it). If a dialog ever appears on the desktop, stop: a fake remote is missing an answer.

---

## File map

| file | change |
|---|---|
| slackrepo_setup: `slackrepo_SBo-danix.conf`, `sb-publish`, `SBo-danix.local.example`, `gpg-agent.conf`, `make-local.py`, `README.md`, `AGENTS.md` | Task 1 |
| `remote/sbb-remote` | 0.2.0: new config keys, kinds return (argv, env), `slackrepo`/`publish` kinds, `inventory`, `gpg-status`, `gpg-unlock`, `stdin=DEVNULL` everywhere |
| `remote/config.example.toml` | new keys |
| `tests/test_remote.py` | 37 tests |
| `sb_builder/__init__.py` | `VERSION = "0.2.0"` |
| `sb_builder/config.py`, `config.example.toml` | `[builds] publish` |
| `sb_builder/remote.py` | `call(..., stdin=)` |
| `sb_builder/hints.py` | new: slackrepo hint rules |
| `sb_builder/packages.py` | new: Packages tab |
| `sb_builder/jobs.py` | Run it, publish box, signing gate |
| `sb_builder/dashboard.py` | clickable cards |
| `sb_builder/mainwindow.py` | Packages tab, key indicator, inventory reload after jobs |
| `tests/test_gui.py` | 31 tests |
| `README.md`, `AGENTS.md` | phase 2 docs |

---

### Task 1: slackrepo config, sb-publish and gates (repo slackrepo_setup)

**Files:** create `slackrepo_SBo-danix.conf`, `sb-publish`, `SBo-danix.local.example`, `gpg-agent.conf`, `make-local.py`; modify `README.md`, `AGENTS.md`.

`slackrepo_SBo-danix.conf` is the live VM config of 2026-10-19 with: a header that sources `/etc/slackrepo/SBo-danix.local`; `manage_packages_hook` and `gotify_hook` also checking `OPT_PREVIEW`; `rsync_push_hook` calling `sb-publish` and honouring `SBB_PUBLISH=n` and `OPT_PREVIEW`; Gotify URL and key read from `SB_GOTIFY_URL`/`SB_GOTIFY_KEY`. No site value is in the file.

- [ ] **Step 1: Apply the patch**

Save as `/tmp/slackrepo_setup-p2.patch`, then:

````diff
diff '--color=auto' -ruN a/AGENTS.md b/AGENTS.md
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -43,6 +43,22 @@
 Do **not** add a hintfile generator to this script — it would create a second owner of HINT_DIR and
 fight mkhint's `.bak` handling.
 
+## slackrepo config and publishing
+
+`slackrepo_SBo-danix.conf` is the source of truth for `/etc/slackrepo/slackrepo_SBo-danix.conf`.
+Never commit site values (rsync target, passwords, Gotify URL or key, hostnames): they live in the
+root-only `/etc/slackrepo/SBo-danix.local`, which the config sources; `SBo-danix.local.example`
+documents the keys with placeholders. slackrepo unsets only its own known config names after
+sourcing (`varnames`, `genrepnames`, `initnames`, `OPTIONLIST` in `slackrepo`), so the `SB_*`
+variables survive into the hooks, which run as functions in slackrepo's shell.
+
+Gates in the hooks: `SBB_PUBLISH=n` holds back `rsync_push_hook` (sb-builder sets it for
+unpublished runs; `sb-publish` pushes later). `--preview` does not set `OPT_DRY_RUN`
+(`cmdfunctions.sh`), so hooks check `OPT_PREVIEW` too. To check the gates without a VM, source
+the config with `/usr/local/bin/` rewritten to a stub dir (stub `sb-publish`,
+`manage-packages.sh`, `curl`) and call the hooks under each `OPT_DRY_RUN`/`OPT_PREVIEW`/
+`SBB_PUBLISH` combination.
+
 ## Run
 
 ```bash
diff '--color=auto' -ruN a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -25,6 +25,19 @@
   `--yes` applies without asking, `--no` only shows the plan (for
   non-interactive callers such as sb-builder). `SBO_REPO` overrides the repo
   path.
+- `slackrepo_SBo-danix.conf`: the slackrepo config for this repo
+  (`/etc/slackrepo/`). Site values (rsync target, Gotify URL and key) are not
+  in it: it sources the root-only `/etc/slackrepo/SBo-danix.local`, see
+  `SBo-danix.local.example`. Hooks honour `SBB_PUBLISH=n` (no rsync push, set by
+  sb-builder for unpublished runs), and a `--preview` neither notifies Gotify
+  nor runs `manage-packages.sh` for real.
+- `sb-publish`: pushes the package repository to the mirror. Called by the
+  config's `rsync_push_hook`; run it by hand to publish builds made with
+  `SBB_PUBLISH=n`.
+- `gpg-agent.conf`: root's gpg-agent config, with `allow-preset-passphrase` so
+  sb-builder can unlock the package signing key for jobs that have no tty.
+- `make-local.py`: one-off migration, builds `SBo-danix.local` from the values
+  in an older config that still had them inline.
 - `sbpull.bash`: `sbpull {personal|pentesting} [--yes|--no]` shell function.
   Pulls one overlay subtree and runs `sbshadow` (passing the flag on), to
   carry overlay changes without a full rebuild. Source it from root's
@@ -33,8 +46,11 @@
 ## Install
 
 ```bash
-install -m 755 slackrepo_setup_SBo-danix.sh sbshadow /usr/local/bin/
+install -m 755 slackrepo_setup_SBo-danix.sh sbshadow sb-publish /usr/local/bin/
 echo ". /path/to/sbpull.bash" >> ~/.bashrc
+install -m 644 slackrepo_SBo-danix.conf /etc/slackrepo/
+install -m 600 SBo-danix.local.example /etc/slackrepo/SBo-danix.local   # then edit
+install -m 600 gpg-agent.conf /root/.gnupg/ && gpgconf --reload gpg-agent
 ```
 
 Optional: `mkhint` (mkhintfile repo) for phantom-dep hintfile
diff '--color=auto' -ruN a/SBo-danix.local.example b/SBo-danix.local.example
--- a/SBo-danix.local.example
+++ b/SBo-danix.local.example
@@ -0,0 +1,8 @@
+# /etc/slackrepo/SBo-danix.local: site settings for slackrepo_SBo-danix.conf
+# and sb-publish. Not in git: it holds the Gotify key. Keep it root-only:
+#   install -m 600 SBo-danix.local.example /etc/slackrepo/SBo-danix.local
+SB_RSYNC_TARGET='rsync://user@mirror.example.org/packages'
+SB_RSYNC_PASSWORD_FILE='/root/.rsync-password'
+SB_RSYNC_EXCLUDES='/root/.repo-excludes.txt'
+SB_GOTIFY_URL='http://gotify.example.org/message'
+SB_GOTIFY_KEY='replace-me'
diff '--color=auto' -ruN a/gpg-agent.conf b/gpg-agent.conf
--- a/gpg-agent.conf
+++ b/gpg-agent.conf
@@ -0,0 +1,10 @@
+pinentry-program        /usr/bin/pinentry-curses
+enable-ssh-support
+# cache passwords for both gpg and ssh for 4h
+max-cache-ttl-ssh       14400
+default-cache-ttl       14400
+# expire passwords after 6h
+max-cache-ttl           21600
+max-cache-ttl-ssh       21600
+# let sb-builder unlock the signing key: jobs have no tty for pinentry
+allow-preset-passphrase
diff '--color=auto' -ruN a/make-local.py b/make-local.py
--- a/make-local.py
+++ b/make-local.py
@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
+#
+# This program is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 2 as published
+# by the Free Software Foundation. See the LICENSE file for details.
+
+# Run on the VM, as root, BEFORE installing the new slackrepo_SBo-danix.conf:
+# copies the site values out of the old config into SBo-danix.local (mode 600).
+# Prints only which keys it found, never their values.
+import os, re, sys
+old, dest = sys.argv[1], sys.argv[2]
+if os.path.exists(dest):
+    sys.exit(f"{dest} exists, left alone")
+s = open(old).read()
+found = {
+    "SB_RSYNC_TARGET": re.search(r"local rsync_target='([^']*)'", s),
+    "SB_RSYNC_EXCLUDES": re.search(r"local rsync_excludes='([^']*)'", s),
+    "SB_RSYNC_PASSWORD_FILE": re.search(r"local rsync_password_file='([^']*)'", s),
+    "SB_GOTIFY_KEY": re.search(r'X-Gotify-Key: ([^"]*)"', s),
+    "SB_GOTIFY_URL": re.search(r'\n\s*"(https?://[^"]*)" >/dev/null', s),
+}
+missing = [k for k, m in found.items() if not m]
+if missing:
+    sys.exit(f"not found in {old}: {', '.join(missing)}")
+fd = os.open(dest, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+with os.fdopen(fd, "w") as f:
+    f.write("# site settings for slackrepo_SBo-danix.conf and sb-publish (root-only)\n")
+    for k, m in found.items():
+        f.write(f"{k}='{m.group(1)}'\n")
+print(f"wrote {dest}: {', '.join(found)}")
diff '--color=auto' -ruN a/sb-publish b/sb-publish
--- a/sb-publish
+++ b/sb-publish
@@ -0,0 +1,35 @@
+#!/bin/bash
+# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
+#
+# This program is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 2 as published
+# by the Free Software Foundation. See the LICENSE file for details.
+
+# sb-publish: push the local package repository to the public mirror.
+#
+# Usage: sb-publish [PKGREPO]      (default /repo)
+#
+# Called by rsync_push_hook in slackrepo_SBo-danix.conf after a build, unless
+# the build ran with SBB_PUBLISH=n; run it by hand (or from sb-builder's
+# Publish action) to push such builds later. Settings come from the root-only
+# SB_LOCAL file (default /etc/slackrepo/SBo-danix.local):
+#   SB_RSYNC_TARGET          rsync destination (required)
+#   SB_RSYNC_PASSWORD_FILE   --password-file, if the file exists
+#   SB_RSYNC_EXCLUDES        --exclude-from, if the file exists
+
+set -euo pipefail
+
+SB_LOCAL=${SB_LOCAL:-/etc/slackrepo/SBo-danix.local}
+[ -r "$SB_LOCAL" ] && . "$SB_LOCAL"
+: "${SB_RSYNC_TARGET:?SB_RSYNC_TARGET is not set in $SB_LOCAL}"
+
+pkgrepo=${1:-/repo}
+args=( -havz --delete-after --stats )
+[ -n "${SB_RSYNC_EXCLUDES:-}" ] && [ -f "$SB_RSYNC_EXCLUDES" ] && \
+  args+=( --exclude-from="$SB_RSYNC_EXCLUDES" )
+[ -n "${SB_RSYNC_PASSWORD_FILE:-}" ] && [ -f "$SB_RSYNC_PASSWORD_FILE" ] && \
+  args+=( --password-file="$SB_RSYNC_PASSWORD_FILE" )
+
+echo "sb-publish: pushing $pkgrepo to $SB_RSYNC_TARGET ..."
+rsync "${args[@]}" "$pkgrepo"/ "$SB_RSYNC_TARGET"
+echo "sb-publish: rsync push complete."
diff '--color=auto' -ruN a/slackrepo_SBo-danix.conf b/slackrepo_SBo-danix.conf
--- a/slackrepo_SBo-danix.conf
+++ b/slackrepo_SBo-danix.conf
@@ -0,0 +1,310 @@
+# slackrepo configuration file for the SBo-danix personal repository.
+#
+# Priority order:
+# (1) Environment variables
+# (2) ~/.slackreporc
+# (3) /etc/slackrepo/slackrepo_SBo-danix.conf
+#
+# %REPO%, %SLACKVER% and %ARCH% are replaced at runtime.
+
+#-------------------------------------------------------------------------------
+# Filestore locations
+
+# Personal SlackBuilds repo (managed as a git fork of SBo):
+# Site settings (rsync target, Gotify URL and key) live outside git in a
+# root-only file, see SBo-danix.local.example in slackrepo_setup.
+SB_LOCAL='/etc/slackrepo/SBo-danix.local'
+[ -r "$SB_LOCAL" ] && . "$SB_LOCAL"
+
+SBREPO="/var/lib/sbopkg/SBo-danix"
+
+# Cached sources:
+SRCREPO="/var/lib/slackrepo/%REPO%/source"
+
+# Built packages:
+#PKGREPO="/var/lib/slackrepo/%REPO%/packages/%SLACKVER%/%ARCH%"
+PKGREPO="/repo"
+
+# Backups (comment out to disable):
+#PKGBACKUP="/var/lib/slackrepo/%REPO%/backups/%SLACKVER%/%ARCH%"
+
+# Local hintfiles (for overriding SBo packages without touching the SlackBuild):
+HINTDIR="/etc/slackrepo/%REPO%/hintfiles"
+
+# Default hintfiles from slackrepo-hints package (optional):
+DEFAULT_HINTDIR="/etc/slackrepo/%REPO%/default_hintfiles/%SLACKVER%"
+
+# Logs:
+LOGDIR="/var/log/slackrepo/%REPO%"
+
+# Database:
+DATABASE="/var/lib/slackrepo/%REPO%/database_%REPO%_%ARCH%.sqlite3"
+
+# Temporary build space (needs plenty of room):
+TMP="/tmp/%REPO%"
+
+#-------------------------------------------------------------------------------
+# Package tag
+# Using _danix instead of _SBo to identify your personal builds.
+TAG='_danix'
+
+# Package compression:
+PKGTYPE='txz'
+
+#-------------------------------------------------------------------------------
+# Build control
+
+SUBSTITUTE=""
+NUMJOBS=''
+ARCH=''
+
+#-------------------------------------------------------------------------------
+# Control argument defaults
+
+VERBOSE='n'
+VERY_VERBOSE='n'
+DRY_RUN='n'
+INSTALL='n'
+CHROOT='/'
+LINT='n'
+KEEP_TMP='n'
+COLOR='auto'
+NICE='5'
+NOWARNING=''
+
+#-------------------------------------------------------------------------------
+# Git branch setup
+#
+# INIT_GITCLONE is only used if SBREPO does not exist yet.
+# INIT_GITBRANCH must match the branch you have checked out in SBREPO.
+# Since we use a personal branch, gitfetch_hook is replaced below with a
+# custom hook that fetches upstream and rebases your branch on top of it.
+
+INIT_GITCLONE='https://github.com/Ponce/slackbuilds.git'
+INIT_GITBRANCH='danix-current'
+
+#-------------------------------------------------------------------------------
+# Hooks
+
+function danix_gitfetch_hook
+{
+  local upstream_branch='current'
+  local personal_branch='danix-current'
+
+  cd "$SR_SBREPO" || return 1
+
+  local currbranch
+  currbranch=$(git rev-parse --abbrev-ref HEAD)
+  if [ "$currbranch" != "$personal_branch" ]; then
+    echo "slackrepo: WARNING: SBREPO is on branch '$currbranch', expected '$personal_branch'. Skipping git update."
+    return 0
+  fi
+
+  local muck
+  muck=$(git status -s .)
+  if [ -n "$muck" ]; then
+    echo "slackrepo: WARNING: SBREPO has uncommitted changes. Skipping git update."
+    return 0
+  fi
+
+  echo "slackrepo: Fetching upstream SBo..."
+  git fetch origin || { echo "slackrepo: WARNING: git fetch failed."; return 1; }
+
+  echo "slackrepo: Rebasing $personal_branch onto origin/$upstream_branch..."
+  git rebase --rebase-merges -X theirs origin/"$upstream_branch" || {
+    echo "slackrepo: ERROR: rebase failed. Resolve conflicts in $SR_SBREPO and run 'git rebase --continue'."
+    return 1
+  }
+
+  return 0
+}
+
+HOOK_START=( danix_gitfetch_hook )
+
+function manage_packages_hook
+{
+  local args=()
+  # --preview does not set OPT_DRY_RUN (slackrepo cmdfunctions.sh), check both
+  if [ "$OPT_DRY_RUN" = 'y' ] || [ "$OPT_PREVIEW" = 'y' ]; then
+    args+=( --dry-run )
+  fi
+  REPO="$SR_PKGREPO" /usr/local/bin/manage-packages.sh "${args[@]}"
+}
+
+#-------------------------------------------------------------------------------
+# generate web files
+
+function gen_web_hook
+{
+  if [ "$OPT_DRY_RUN" = 'y' ] || [ ! -s "$CHANGELOG" ]; then
+    return 0
+  fi
+  SR_PKGREPO="$SR_PKGREPO" /usr/local/bin/gen_web_hook.sh
+  local stat=$?
+  [ $stat -ne 0 ] && log_warning "gen_web_hook failed, status $stat"
+  return $stat
+}
+
+#-------------------------------------------------------------------------------
+# dep template generator
+# Writes <TEMPLATE_DIR>/<prgnam>.template for each successfully built package
+# that has one or more recursive dependencies.
+
+TEMPLATE_DIR='/repo/templates'
+
+function dep_template_hook
+{
+  local template_dir="${TEMPLATE_DIR:-/repo/templates}"
+
+  [ "$OPT_DRY_RUN" = 'y' ] && return 0
+  [ ${#OKLIST[@]} -eq 0 ] && return 0
+
+  mkdir -p "$template_dir" || return 1
+
+  local itemid prgnam dep depname outfile
+  for itemid in "${OKLIST[@]}"; do
+    [ -z "${FULLDEPS[$itemid]}" ] && continue
+
+    prgnam="${ITEMPRGNAM[$itemid]}"
+    outfile="$template_dir/${prgnam}.template"
+
+    {
+      for dep in ${FULLDEPS[$itemid]}; do
+        depname="${ITEMPRGNAM[$dep]}"
+        printf '%s\n' "${depname:-$dep}"
+      done
+      printf '%s\n' "$prgnam"
+    } > "$outfile"
+
+    log_normal "dep_template_hook: wrote $outfile"
+  done
+
+  return 0
+}
+
+#-------------------------------------------------------------------------------
+# rsync push config
+
+function rsync_push_hook
+{
+  # SBB_PUBLISH=n (set by sb-builder for an unpublished run) holds the push
+  # back; sb-publish pushes later. Unset means publish, as from a terminal.
+  if [ "$OPT_DRY_RUN" = 'y' ] || [ "$OPT_PREVIEW" = 'y' ] || [ ! -s "$CHANGELOG" ] \
+     || [ "${SBB_PUBLISH:-y}" = 'n' ]; then
+    return 0
+  fi
+  /usr/local/bin/sb-publish "$SR_PKGREPO"
+  local rsyncstat=$?
+  if [ "$rsyncstat" != 0 ]; then
+    log_warning "rsync_push_hook failed, status $rsyncstat"
+    return 1
+  fi
+  return 0
+}
+
+function recap_hook
+{
+  # TODOLIST is per-item scratch (overwritten on each build_command call), so it
+  # only ever holds the last item. Build the recap from the cumulative status
+  # lists that slackrepo dedupes before running the finish hooks.
+  local recaplist=(
+    "${OKLIST[@]}" "${FAILEDLIST[@]}" "${ABORTEDLIST[@]}"
+    "${SKIPPEDLIST[@]}" "${UNSUPPORTEDLIST[@]}"
+  )
+  recaplist=( $(printf '%s\n' "${recaplist[@]}" | sort -u) )
+  [ ${#recaplist[@]} -eq 0 ] && return 0
+
+  local col_name=4 col_ver=7 col_status=6
+  local itemid prgnam ver stat
+  for itemid in "${recaplist[@]}"; do
+    prgnam="${ITEMPRGNAM[$itemid]}"
+    ver="${HINT_VERSION[$itemid]:-${INFOVERSION[$itemid]:-?}}"
+    stat="${STATUS[$itemid]:-?}"
+    [ ${#prgnam} -gt $col_name ]   && col_name=${#prgnam}
+    [ ${#ver}    -gt $col_ver ]    && col_ver=${#ver}
+    [ ${#stat}   -gt $col_status ] && col_status=${#stat}
+  done
+
+  local sep
+  sep="+-$(printf '%*s' "$col_name"   '' | tr ' ' '-')-+-$(printf '%*s' "$col_ver" '' | tr ' ' '-')-+-$(printf '%*s' "$col_status" '' | tr ' ' '-')-+"
+
+  echo ""
+  echo "Build recap:"
+  echo "$sep"
+  printf "| %-*s | %-*s | %-*s |\n" "$col_name" "Name" "$col_ver" "Version" "$col_status" "Status"
+  echo "$sep"
+  for itemid in "${recaplist[@]}"; do
+    prgnam="${ITEMPRGNAM[$itemid]}"
+    ver="${HINT_VERSION[$itemid]:-${INFOVERSION[$itemid]:-?}}"
+    stat="${STATUS[$itemid]:-?}"
+    printf "| %-*s | %-*s | %-*s |\n" "$col_name" "$prgnam" "$col_ver" "$ver" "$col_status" "$stat"
+  done
+  echo "$sep"
+  echo ""
+
+  return 0
+}
+
+function gotify_hook
+{
+  # --preview does not set OPT_DRY_RUN (slackrepo cmdfunctions.sh), check both
+  if [ "$OPT_DRY_RUN" = 'y' ] || [ "$OPT_PREVIEW" = 'y' ]; then
+    return 0
+  fi
+  if [ -z "${SB_GOTIFY_URL:-}" ] || [ -z "${SB_GOTIFY_KEY:-}" ]; then
+    log_warning "SB_GOTIFY_URL/SB_GOTIFY_KEY not set in $SB_LOCAL, skipping gotify_hook"
+    return 0
+  fi
+
+  local curl_bin
+  curl_bin=$(which curl 2>/dev/null)
+  if [ ! -x "$curl_bin" ]; then
+    log_warning "curl not found, skipping gotify_hook"
+    return 1
+  fi
+
+  local ok="${#OKLIST[@]}"
+  local failed="${#FAILEDLIST[@]}"
+  local aborted="${#ABORTEDLIST[@]}"
+  local skipped="${#SKIPPEDLIST[@]}"
+
+  local title msg priority
+  if [ $(( failed + aborted )) -gt 0 ]; then
+    title="slackrepo [$CMD]: FAILED"
+    priority=8
+  else
+    title="slackrepo [$CMD]: OK"
+    priority=5
+  fi
+
+  msg="OK: $ok  Failed: $failed  Aborted: $aborted  Skipped: $skipped"
+  [ ${#OKLIST[@]}      -gt 0 ] && msg+=$'\nOK: '"$(printf '%s ' "${OKLIST[@]}")"
+  [ ${#FAILEDLIST[@]}  -gt 0 ] && msg+=$'\nFailed: '"$(printf '%s ' "${FAILEDLIST[@]}")"
+  [ ${#ABORTEDLIST[@]} -gt 0 ] && msg+=$'\nAborted: '"$(printf '%s ' "${ABORTEDLIST[@]}")"
+
+  "$curl_bin" -s \
+    -H "X-Gotify-Key: $SB_GOTIFY_KEY" \
+    -F "title=$title" \
+    -F "message=$msg" \
+    -F "priority=$priority" \
+    "$SB_GOTIFY_URL" >/dev/null
+  local stat=$?
+  [ $stat -ne 0 ] && log_warning "gotify_hook failed, status $stat"
+  return $stat
+}
+
+function bell_hook
+{
+  printf '\a'
+  return 0
+}
+
+HOOK_FINISH=( manage_packages_hook bell_hook genrepos_hook gen_web_hook dep_template_hook rsync_push_hook recap_hook gotify_hook )
+
+#-------------------------------------------------------------------------------
+# gen_repos_files.sh (disabled by default)
+
+USE_GENREPOS='1'
+
+#-------------------------------------------------------------------------------
+export XZ_OPT="--threads=0"
````

```bash
cd ~/Programming/GIT/slackrepo_setup && git apply /tmp/slackrepo_setup-p2.patch && chmod 755 sb-publish
bash -n slackrepo_SBo-danix.conf && bash -n sb-publish && python3 -m py_compile make-local.py && echo ok
grep -cE 'X-Gotify-Key: [A-Za-z0-9]{8,}|rsync://[^$]' slackrepo_SBo-danix.conf   # expect 0
```

- [ ] **Step 2: Check the hook gates with stubs**

```bash
T=$(mktemp -d); mkdir $T/stub
for b in sb-publish manage-packages.sh; do printf '#!/bin/sh\necho "%s $*"\n' $b > $T/stub/$b; done
printf '#!/bin/sh\necho "curl $*" >> %s/curl.log\n' $T > $T/stub/curl; chmod 755 $T/stub/*
sed "s|/usr/local/bin/|$T/stub/|g; s|^SB_LOCAL=.*|SB_LOCAL=/nonexistent|" \
    ~/Programming/GIT/slackrepo_setup/slackrepo_SBo-danix.conf > $T/test.conf
for e in "OPT_DRY_RUN=n OPT_PREVIEW=n" "OPT_DRY_RUN=n OPT_PREVIEW=n SBB_PUBLISH=n" \
         "OPT_DRY_RUN=n OPT_PREVIEW=y" "OPT_DRY_RUN=y OPT_PREVIEW=n"; do
  echo "== $e"; : > $T/curl.log
  env $e PATH=$T/stub:$PATH bash -c 'log_warning() { echo "WARN $*"; }; log_normal() { :; }
    . '"$T"'/test.conf; SB_GOTIFY_URL=http://example.org/m SB_GOTIFY_KEY=k
    CHANGELOG='"$T"'/cl; echo x > $CHANGELOG; SR_PKGREPO=/repo; CMD=build; OKLIST=(a)
    rsync_push_hook; manage_packages_hook; gotify_hook'
  grep -o 'X-Gotify-Key: k\|http://example.org/m' $T/curl.log | tr '\n' ' '; echo
done
```

Expected:

```
== OPT_DRY_RUN=n OPT_PREVIEW=n
sb-publish /repo
manage-packages.sh
X-Gotify-Key: k http://example.org/m
== OPT_DRY_RUN=n OPT_PREVIEW=n SBB_PUBLISH=n
manage-packages.sh
X-Gotify-Key: k http://example.org/m
== OPT_DRY_RUN=n OPT_PREVIEW=y
manage-packages.sh --dry-run

== OPT_DRY_RUN=y OPT_PREVIEW=n
manage-packages.sh --dry-run

```

- [ ] **Step 3: Commit and push slackrepo_setup**

```bash
cd ~/Programming/GIT/slackrepo_setup
git add slackrepo_SBo-danix.conf sb-publish SBo-danix.local.example gpg-agent.conf make-local.py README.md AGENTS.md
git commit -F - <<'EOF'
Track the slackrepo config; add sb-publish and publish gates

slackrepo_SBo-danix.conf is now kept here. Site values (rsync target,
Gotify URL and key) moved to the root-only SBo-danix.local it sources;
make-local.py migrates them out of an old inline config. The rsync push
moved into sb-publish; rsync_push_hook calls it and holds back when
SBB_PUBLISH=n, so sb-builder can build without publishing and publish
later. --preview does not set OPT_DRY_RUN, so the hooks now also check
OPT_PREVIEW: previews no longer notify Gotify or run manage-packages.sh
for real. gpg-agent.conf adds allow-preset-passphrase so sb-builder can
unlock the signing key for jobs without a tty.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
EOF
git log --format='%h %G? %s' -1 && git push
```

- [ ] **Step 4: Remove the stale config copy (ask the user first)**

`~/Programming/GIT/repo-html-structure/slackrepo_SBo-danix.conf` is untracked there and may still hold the Gotify key. With the user's OK: `shred -u ~/Programming/GIT/repo-html-structure/slackrepo_SBo-danix.conf`.

- [ ] **Step 5: VM deploy is Task 5.**

---

### Task 2: helper 0.2.0

**Files:** replace `remote/sbb-remote`, `remote/config.example.toml`, `tests/test_remote.py`.

- [ ] **Step 1: Write the tests** (replace `tests/test_remote.py`):

````python
# sb-builder - self-checks for the remote helper
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Run: python3 -m unittest discover -s tests"""

import contextlib
import fcntl
import importlib.machinery
import importlib.util
import io
import json
import os
import shutil
import sqlite3
import tempfile
import threading
import time
import unittest
from unittest import mock

SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "remote", "sbb-remote")
_loader = importlib.machinery.SourceFileLoader("sbb_remote", SCRIPT)
sbb = importlib.util.module_from_spec(importlib.util.spec_from_loader("sbb_remote", _loader))
_loader.exec_module(sbb)


class Base(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.tmp)
        conf = os.path.join(self.tmp, "config.toml")
        with open(conf, "w") as f:
            f.write(f'state_dir = "{self.tmp}"\n'
                    f'repo_dir = "{self.tmp}/repo"\n'
                    f'slackrepo_db = "{self.tmp}/slackrepo.sqlite3"\n'
                    f'hintdir = "{self.tmp}/hints"\n'
                    f'genreprc = "{self.tmp}/genreprc"\n')
        patcher = mock.patch.dict(os.environ, {"SBB_CONFIG": conf})
        patcher.start()
        self.addCleanup(patcher.stop)
        self.cfg = sbb.load_config()
        os.makedirs(os.path.join(self.tmp, "jobs"))

    def make_job(self, argv, **meta):
        jid = "20260101-000000-test"
        d = os.path.join(self.tmp, "jobs", jid)
        os.mkdir(d)
        sbb.write_meta(d, {"id": jid, "kind": "test", "args": [], "argv": argv,
                           "status": "starting", "created": time.time(), **meta})
        open(os.path.join(d, "log"), "wb").close()
        return jid, d

    def make_pkg(self, path, info="", hint=None, desc=None):
        d = os.path.join(self.cfg["repo_dir"], path)
        pkg = os.path.basename(path)
        os.makedirs(d, exist_ok=True)
        open(os.path.join(d, pkg + ".SlackBuild"), "w").close()
        with open(os.path.join(d, pkg + ".info"), "w") as f:
            f.write(info)
        if desc:
            with open(os.path.join(d, "slack-desc"), "w") as f:
                f.write(f"{pkg}: {pkg} ({desc})\n{pkg}:\n")
        if hint is not None:
            os.makedirs(self.cfg["hintdir"], exist_ok=True)
            with open(os.path.join(self.cfg["hintdir"], pkg + ".hint"), "w") as f:
                f.write(hint)

    def fake_bin(self, name, script):
        bindir = os.path.join(self.tmp, "bin")
        os.makedirs(bindir, exist_ok=True)
        path = os.path.join(bindir, name)
        with open(path, "w") as f:
            f.write("#!/bin/sh\n" + script)
        os.chmod(path, 0o755)
        if bindir not in os.environ["PATH"]:
            patcher = mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]})
            patcher.start()
            self.addCleanup(patcher.stop)
        return path

    def hold_lock(self):
        f = open(os.path.join(self.tmp, "lock"), "a")
        fcntl.flock(f, fcntl.LOCK_EX)
        self.addCleanup(f.close)

    def run_main(self, *argv):
        out = io.StringIO()
        with contextlib.redirect_stdout(out):
            rc = sbb.main(list(argv))
        return rc, json.loads(out.getvalue())

    def wait_status(self, d, statuses):
        for _ in range(100):
            if sbb.read_meta(d)["status"] in statuses:
                break
            time.sleep(0.05)
        return sbb.read_meta(d)


class ConfigTest(Base):
    def write(self, text):
        p = os.path.join(self.tmp, "c.toml")
        with open(p, "w") as f:
            f.write(text)
        return p

    def test_missing_file_gives_defaults(self):
        self.assertEqual(sbb.load_config(os.path.join(self.tmp, "nope.toml")), sbb.DEFAULTS)

    def test_override(self):
        self.assertEqual(sbb.load_config(self.write('overlays = ["a"]\n'))["overlays"], ["a"])

    def test_example_matches_defaults(self):
        example = os.path.join(os.path.dirname(SCRIPT), "config.example.toml")
        self.assertEqual(sbb.load_config(example), sbb.DEFAULTS)

    def test_unknown_key(self):
        with self.assertRaisesRegex(sbb.Error, "bogus: unknown key"):
            sbb.load_config(self.write("bogus = 1\n"))

    def test_wrong_type(self):
        with self.assertRaisesRegex(sbb.Error, "state_dir: expected str"):
            sbb.load_config(self.write("state_dir = 1\n"))


class KindsTest(Base):
    def test_sbpull_argv(self):
        self.assertEqual(sbb.KINDS["sbpull"](self.cfg, ["personal", "--no"]),
                         (["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_",
                           self.cfg["sbpull_script"], "personal", "--no"], {}))

    def test_sbshadow_argv(self):
        self.assertEqual(sbb.KINDS["sbshadow"](self.cfg, ["--yes"]), (["sbshadow", "--yes"], {}))

    def test_slackrepo_argv(self):
        self.make_pkg("misc/nuclei")
        self.make_pkg("personal/foo")
        kind = sbb.KINDS["slackrepo"]
        self.assertEqual(kind(self.cfg, ["update", "--preview", "nuclei", "foo", "nuclei"]),
                         (["slackrepo", "--repo=SBo-danix", "update", "--preview", "nuclei", "foo"], {}))
        self.assertEqual(kind(self.cfg, ["build", "--no-publish", "foo"]),
                         (["slackrepo", "--repo=SBo-danix", "build", "foo"], {"SBB_PUBLISH": "n"}))
        self.assertEqual(kind(self.cfg, ["update"]), (["slackrepo", "--repo=SBo-danix", "update"], {}))

    def test_slackrepo_rejects(self):
        self.make_pkg("misc/nuclei")
        for args in (["install", "nuclei"], ["build"], ["remove"], ["build", "nope"],
                     ["build", "../etc"], ["build", "--force", "nuclei"]):
            with self.subTest(args=args), self.assertRaises(sbb.Error):
                sbb.KINDS["slackrepo"](self.cfg, args)

    def test_publish(self):
        self.assertEqual(sbb.KINDS["publish"](self.cfg, []), (["sb-publish"], {}))
        with self.assertRaises(sbb.Error):
            sbb.KINDS["publish"](self.cfg, ["x"])

    def test_rejects_bad_args(self):
        for kind, args in (("sbpull", ["evil", "--no"]), ("sbpull", ["personal"]),
                           ("sbpull", ["personal", "--force"]), ("sbshadow", []),
                           ("sbshadow", ["--yes", "--no"])):
            with self.subTest(kind=kind, args=args), self.assertRaises(sbb.Error):
                sbb.KINDS[kind](self.cfg, args)

    def test_unknown_kind(self):
        with self.assertRaisesRegex(sbb.Error, "unknown kind"):
            sbb.cmd_job_start(self.cfg, "rm", "-rf", "/")


class MainTest(Base):
    def test_unknown_command(self):
        rc, data = self.run_main("nope")
        self.assertEqual(rc, 1)
        self.assertIn("usage", data["error"])

    def test_version(self):
        self.assertEqual(self.run_main("version"),
                         (0, {"version": sbb.VERSION, "overlays": ["personal", "pentesting"]}))


class InventoryTest(Base):
    def test_rows(self):
        self.make_pkg("misc/nuclei", 'VERSION="3.11.0"\nMAINTAINER="Some One"\n'
                      'REQUIRES="go \\\n  git"\n', desc="Fast scanner")
        self.make_pkg("personal/foo", 'VERSION="1.0"\n',
                      hint='VERSION="1.1"  # bumped\nDELREQUIRES="a b"\nOPTIONS="X=$(uname -m)"\n')
        os.makedirs(os.path.join(self.cfg["repo_dir"], "misc", "notapkg"))
        db = sqlite3.connect(self.cfg["slackrepo_db"])
        db.executescript("""
            create table buildresults (itemid text primary key, time text, result text);
            create table revisions (itemid text, dep text, deplist text, version text,
                built text, rev text, os text, hintcksum text, primary key (itemid, dep));
            insert into buildresults values ('misc/nuclei', '5', 'ok'), ('nuclei', '1', 'bad');
            insert into revisions values ('misc/nuclei', '/', '/', '3.11.0', '1788978222', 'r', 'o', '/');
        """)
        db.close()
        inv = sbb.cmd_inventory(self.cfg)
        rows = [dict(zip(inv["columns"], r)) for r in inv["rows"]]
        self.assertEqual([r["path"] for r in rows], ["misc/nuclei", "personal/foo"])
        nuclei, foo = rows
        self.assertEqual((nuclei["version"], nuclei["built_version"], nuclei["built_time"],
                          nuclei["result"], nuclei["maintainer"], nuclei["requires"],
                          nuclei["summary"], nuclei["hint_path"]),
                         ("3.11.0", "3.11.0", 1788978222, "ok", "Some One", ["go", "git"],
                          "Fast scanner", None))
        self.assertEqual((foo["version"], foo["info_version"], foo["hint_version"], foo["result"]),
                         ("1.1", "1.0", "1.1", None))
        self.assertEqual(foo["hints"]["DELREQUIRES"], "a b")
        self.assertEqual(foo["hints"]["OPTIONS"], {"value": '"X=$(uname -m)"', "shell": True})
        json.dumps(inv)

    def test_hint_search_order(self):
        self.make_pkg("misc/foo")
        sub = os.path.join(self.cfg["hintdir"], "misc", "foo")
        os.makedirs(sub)
        with open(os.path.join(sub, "foo.hint"), "w") as f:
            f.write('VERSION="2"\n')
        self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), os.path.join(sub, "foo.hint"))
        with open(os.path.join(self.cfg["hintdir"], "foo.hint"), "w") as f:
            f.write('VERSION="3"\n')
        self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"),
                         os.path.join(self.cfg["hintdir"], "foo.hint"))
        in_repo = os.path.join(self.cfg["repo_dir"], "misc", "foo", "foo.hint")
        open(in_repo, "w").close()
        self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), in_repo)

    def test_missing_db(self):
        self.make_pkg("misc/foo", 'VERSION="1"\n')
        row = sbb.cmd_inventory(self.cfg)["rows"][0]
        self.assertEqual(row[4:7], [None, None, None])


class GpgTest(Base):
    def setup_key(self, cached):
        with open(self.cfg["genreprc"], "w") as f:
            f.write('REPOSOWNERGPG="ABCD1234"\nUSE_GPGAGENT=1\n')
        self.fake_bin("gpg", 'case "$*" in\n'
                      '  *--list-secret-keys*) printf "sec:u:255:22:K:1::::::scESC:\\n'
                      'grp:::::::::GRIPSIGN:\\nssb:u:255:18:S:1::::::e:\\ngrp:::::::::GRIPENC:\\n" ;;\n'
                      f'  *-bas*) [ -f "{self.tmp}/good" ] ;;\n'
                      'esac\n')
        flag = "1" if cached else "-"
        self.fake_bin("gpg-connect-agent", f'echo "S KEYINFO GRIPSIGN D - - {flag} P - - -"; echo OK\n')

    def test_status(self):
        self.assertEqual(sbb.cmd_gpg_status(self.cfg), {"unlocked": None, "key": None})
        self.setup_key(cached=False)
        self.assertEqual(sbb.cmd_gpg_status(self.cfg), {"unlocked": False, "key": "ABCD1234"})
        self.setup_key(cached=True)
        self.assertEqual(sbb.cmd_gpg_status(self.cfg)["unlocked"], True)

    def test_locked_refuses_real_run_only(self):
        self.setup_key(cached=False)
        self.make_pkg("misc/foo")
        with self.assertRaisesRegex(sbb.Error, "signing key locked"):
            sbb.cmd_job_start(self.cfg, "slackrepo", "build", "foo")
        self.fake_bin("slackrepo", "exit 0\n")
        self.assertIn("id", sbb.cmd_job_start(self.cfg, "slackrepo", "build", "--preview", "foo"))

    def test_unlock(self):
        self.setup_key(cached=False)
        log = os.path.join(self.tmp, "preset.log")
        preset = self.fake_bin("preset", f'echo "$* $(cat)" >> {log}\n')
        with mock.patch.object(sbb, "PRESET", preset):
            with self.assertRaisesRegex(sbb.Error, "bad passphrase"):
                sbb.cmd_gpg_unlock(self.cfg, stdin=io.BytesIO(b"wrong\n"))
            open(os.path.join(self.tmp, "good"), "w").close()
            self.assertEqual(sbb.cmd_gpg_unlock(self.cfg, stdin=io.BytesIO(b"right\n")),
                             {"unlocked": True})
        with open(log) as f:
            self.assertEqual(f.read().splitlines(),
                             ["--preset GRIPSIGN wrong", "--forget GRIPSIGN ", "--preset GRIPSIGN right"])


class JobsTest(Base):
    def log(self, d):
        with open(os.path.join(d, "log"), "rb") as f:
            return f.read()

    def test_run_ok(self):
        jid, d = self.make_job(["sh", "-c", "echo hello"])
        sbb.cmd_run(self.cfg, jid)
        m = sbb.read_meta(d)
        self.assertEqual((m["status"], m["exit"]), ("ok", 0))
        self.assertEqual(self.log(d), b"hello\n")

    def test_run_failed(self):
        jid, d = self.make_job(["sh", "-c", "exit 3"])
        sbb.cmd_run(self.cfg, jid)
        m = sbb.read_meta(d)
        self.assertEqual((m["status"], m["exit"]), ("failed", 3))

    def test_run_missing_program(self):
        jid, d = self.make_job(["/nonexistent/prog"])
        sbb.cmd_run(self.cfg, jid)
        self.assertEqual(sbb.read_meta(d)["status"], "failed")
        self.assertIn(b"sbb-remote:", self.log(d))

    def test_kill(self):
        jid, d = self.make_job(["sleep", "30"])
        t = threading.Thread(target=sbb.cmd_run, args=(self.cfg, jid))
        t.start()
        self.wait_status(d, ("running",))
        self.assertEqual(sbb.cmd_job_kill(self.cfg, jid), {"ok": True})
        t.join(5)
        self.assertEqual(sbb.read_meta(d)["status"], "killed")

    def test_kill_not_running(self):
        jid, _ = self.make_job(["true"], status="ok")
        with self.assertRaisesRegex(sbb.Error, "not running"):
            sbb.cmd_job_kill(self.cfg, jid)

    def test_busy(self):
        self.hold_lock()
        jid, d = self.make_job(["true"])
        sbb.cmd_run(self.cfg, jid)
        self.assertEqual(sbb.read_meta(d)["status"], "busy")

    def test_lost_when_lock_free(self):
        self.make_job(["true"], status="running", pid=1)
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost")

    def test_running_kept_while_lock_held(self):
        self.hold_lock()
        self.make_job(["true"], status="running", pid=1)
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "running")

    def test_starting_grace(self):
        _, d = self.make_job(["true"])
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "starting")
        sbb.write_meta(d, dict(sbb.read_meta(d), created=time.time() - 60))
        self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost")

    def test_follow_from_offset(self):
        jid, d = self.make_job(["true"], status="ok")
        with open(os.path.join(d, "log"), "wb") as f:
            f.write(b"hello world")
        out = io.BytesIO()
        sbb.cmd_job_follow(self.cfg, jid, "6", out=out)
        self.assertEqual(out.getvalue(), b"world")

    def test_bad_job_id(self):
        with self.assertRaisesRegex(sbb.Error, "bad job id"):
            sbb.cmd_job_follow(self.cfg, "../../etc", "0")

    def fake_sbshadow(self):
        bindir = os.path.join(self.tmp, "bin")
        os.mkdir(bindir)
        fake = os.path.join(bindir, "sbshadow")
        with open(fake, "w") as f:
            f.write('#!/bin/sh\necho "shadow:   cat/$1"\n')
        os.chmod(fake, 0o755)
        patcher = mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]})
        patcher.start()
        self.addCleanup(patcher.stop)

    def test_env_reaches_child(self):
        jid, d = self.make_job(["sh", "-c", 'echo "publish=$SBB_PUBLISH"'], env={"SBB_PUBLISH": "n"})
        sbb.cmd_run(self.cfg, jid)
        self.assertEqual(self.log(d), b"publish=n\n")

    def test_start_end_to_end(self):
        self.fake_sbshadow()
        rc, data = self.run_main("job-start", "sbshadow", "--no")
        self.assertEqual(rc, 0)
        d = os.path.join(self.tmp, "jobs", data["id"])
        self.assertEqual(self.wait_status(d, ("ok", "failed", "lost"))["status"], "ok")
        self.assertEqual(self.log(d), b"shadow:   cat/--no\n")

    def test_start_busy(self):
        self.fake_sbshadow()
        self.hold_lock()
        jid, _ = self.make_job(["true"], status="running", pid=1)
        self.assertEqual(self.run_main("job-start", "sbshadow", "--no"), (0, {"busy": jid}))
        self.assertEqual(os.listdir(os.path.join(self.tmp, "jobs")), [jid])


class StatsTest(Base):
    def test_missing_sources_are_null(self):
        s = sbb.cmd_stats(self.cfg)
        self.assertIsNotNone(s["load"])
        self.assertIsNone(s["packages"])
        self.assertIsNone(s["slackrepo"])
        self.assertEqual(s["sync"], {"personal": None, "pentesting": None})
        json.dumps(s)

    def test_counts(self):
        r = self.cfg["repo_dir"]
        for p in ("cat/foo", "cat/bar", "personal/foo", "personal/baz", "pentesting/nmap"):
            os.makedirs(os.path.join(r, p))
            open(os.path.join(r, p, os.path.basename(p) + ".SlackBuild"), "w").close()
        os.makedirs(os.path.join(r, "cat", "notapkg"))

        def fake_git(cfg, *args):
            if args[0] == "ls-tree":
                return "cat/foo/foo.SlackBuild\ncat/foo/README\ncat/old/old.SlackBuild\n"
            return "1792351223\n"

        with mock.patch.object(sbb, "_git", fake_git):
            s = sbb.cmd_stats(self.cfg)
        self.assertEqual(s["packages"], {"total": 5, "personal": 2, "pentesting": 1, "shadowed": 1})
        self.assertEqual(s["sync"]["personal"], 1792351223)

    def test_slackrepo(self):
        db = sqlite3.connect(self.cfg["slackrepo_db"])
        db.executescript("""
            create table buildresults (itemid text primary key, time text, result text);
            create table revisions (itemid text, dep text, deplist text, version text,
                built text, rev text, os text, hintcksum text, primary key (itemid, dep));
            insert into buildresults values ('a', '1', 'ok'), ('b', '1', 'ok'), ('c', '1', 'failed');
            insert into revisions values ('cat/a', '/', '/', '1', '1', 'r', 'o', '/'),
                                         ('cat/a', 'dep', '/', '1', '1', 'r', 'o', '/');
        """)
        db.close()
        self.assertEqual(sbb.cmd_stats(self.cfg)["slackrepo"],
                         {"results": {"ok": 2, "failed": 1}, "built": 1})


if __name__ == "__main__":
    unittest.main()
````

- [ ] **Step 2: Run them, expect failures**

Run: `python3 -m unittest tests.test_remote </dev/null`
Expected: errors and failures in the new tests (`KeyError: 'slackrepo'`, tuples expected from kinds, no `cmd_inventory`, no `find_hint`, no `cmd_gpg_status`).

- [ ] **Step 3: Write the helper** (replace `remote/sbb-remote`, keep mode 755):

````python
#!/usr/bin/env python3
# sbb-remote - sb-builder helper on the build VM
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Helper run over ssh by the sb-builder GUI.

Every command prints one JSON object on stdout, except job-follow, which
streams raw log bytes. gpg-unlock reads the passphrase from stdin. Jobs run detached (setsid, stdin=/dev/null), one at a
time under a flock, from a fixed table of kinds: the GUI sends a kind and
its arguments, this file validates them and builds the argv itself.
"""

import fcntl
import glob
import json
import os
import re
import shlex
import shutil
import signal
import sqlite3
import subprocess
import sys
import tempfile
import time
import tomllib

VERSION = "0.2.0"
DEFAULTS = {
    "state_dir": "/var/lib/sb-builder",
    "repo_dir": "/var/lib/sbopkg/SBo-danix",
    "upstream_branch": "current",
    "slackrepo_db": "/var/lib/slackrepo/SBo-danix/database_SBo-danix_x86_64.sqlite3",
    "sbpull_script": "/usr/local/share/slackrepo_setup/sbpull.bash",
    "overlays": ["personal", "pentesting"],
    "slackrepo_repo": "SBo-danix",
    "hintdir": "/etc/slackrepo/SBo-danix/hintfiles",
    "default_hintdir": "",
    "genreprc": "/root/.genreprc",
}
FLAGS = ("--yes", "--no")
JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-\d+)?$")
PKG = re.compile(r"^[^/]+/([^/]+)/\1\.SlackBuild$")
PKGNAME = re.compile(r"^[A-Za-z0-9._+-]+$")
ACTIONS = ("build", "rebuild", "update", "remove")
PRESET = "/usr/libexec/gpg-preset-passphrase"
ACTIVE = ("starting", "running")
START_GRACE = 10  # seconds a job may sit in "starting" before it counts as lost


class Error(Exception):
    pass


def load_config(path=None):
    path = path or os.environ.get("SBB_CONFIG", "/etc/sb-builder/config.toml")
    cfg = dict(DEFAULTS)
    try:
        with open(path, "rb") as f:
            data = tomllib.load(f)
    except FileNotFoundError:
        return cfg
    except tomllib.TOMLDecodeError as e:
        raise Error(f"config: {e}")
    for k, v in data.items():
        if k not in DEFAULTS:
            raise Error(f"config: {k}: unknown key")
        if type(v) is not type(DEFAULTS[k]):
            raise Error(f"config: {k}: expected {type(DEFAULTS[k]).__name__}")
        cfg[k] = v
    if not all(isinstance(o, str) for o in cfg["overlays"]):
        raise Error("config: overlays: expected list of strings")
    return cfg


# ── job kinds: (cfg, args) -> (argv, env), raising Error on bad args ────────

def _flag(args):
    if len(args) != 1 or args[0] not in FLAGS:
        raise Error("expected exactly one of --yes, --no")
    return args[0]


def kind_sbpull(cfg, args):
    if not args or args[0] not in cfg["overlays"]:
        raise Error(f"sbpull: overlay must be one of {cfg['overlays']}")
    return ["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_",
            cfg["sbpull_script"], args[0], _flag(args[1:])], {}


def kind_sbshadow(cfg, args):
    return ["sbshadow", _flag(args)], {}


def kind_slackrepo(cfg, args):
    if not args or args[0] not in ACTIONS:
        raise Error(f"slackrepo: action must be one of {', '.join(ACTIONS)}")
    action, rest = args[0], args[1:]
    opts = [a for a in rest if a.startswith("--")]
    for o in opts:
        if o not in ("--preview", "--no-publish"):
            raise Error(f"slackrepo: unknown option {o}")
    known = {os.path.basename(d) for d in _pkgdirs(cfg["repo_dir"], "*/*")}
    pkgs = []
    for p in (a for a in rest if not a.startswith("--")):
        if not PKGNAME.match(p) or p not in known:
            raise Error(f"slackrepo: no such package: {p}")
        if p not in pkgs:
            pkgs.append(p)
    if not pkgs and action != "update":
        raise Error(f"slackrepo: {action} needs at least one package")
    argv = ["slackrepo", f"--repo={cfg['slackrepo_repo']}", action]
    if "--preview" in opts:
        argv.append("--preview")
    return argv + pkgs, {"SBB_PUBLISH": "n"} if "--no-publish" in opts else {}


def kind_publish(cfg, args):
    if args:
        raise Error("publish takes no arguments")
    return ["sb-publish"], {}


KINDS = {"sbpull": kind_sbpull, "sbshadow": kind_sbshadow,
         "slackrepo": kind_slackrepo, "publish": kind_publish}


def signs(kind, args):
    """A real slackrepo run ends in gen_repos_files.sh, which signs."""
    return kind == "slackrepo" and "--preview" not in args


# ── job state ───────────────────────────────────────────────────────────────

def job_dir(cfg, jid):
    if not JOB_ID.match(jid or ""):
        raise Error(f"bad job id: {jid}")
    d = os.path.join(cfg["state_dir"], "jobs", jid)
    if not os.path.isdir(d):
        raise Error(f"no such job: {jid}")
    return d


def read_meta(d):
    with open(os.path.join(d, "meta.json")) as f:
        return json.load(f)


def write_meta(d, meta):
    tmp = os.path.join(d, "meta.json.tmp")
    with open(tmp, "w") as f:
        json.dump(meta, f)
    os.replace(tmp, os.path.join(d, "meta.json"))


def lock_file(cfg):
    return open(os.path.join(cfg["state_dir"], "lock"), "a")


def refresh(cfg, d, meta):
    """Mark an active job lost when no runner holds the lock.

    _run writes the final status before it releases the lock, so a free lock
    plus a status that is still active (re-read under the lock) means the
    runner died: killed, crashed, or the VM rebooted.
    """
    if meta["status"] not in ACTIVE:
        return meta
    if meta["status"] == "starting" and time.time() - meta["created"] < START_GRACE:
        return meta
    with lock_file(cfg) as lk:
        try:
            fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return meta
        meta = read_meta(d)
        if meta["status"] in ACTIVE:
            meta["status"] = "lost"
            write_meta(d, meta)
    return meta


def all_jobs(cfg):
    out = []
    for d in sorted(glob.glob(os.path.join(cfg["state_dir"], "jobs", "*")), reverse=True):
        try:
            out.append(refresh(cfg, d, read_meta(d)))
        except (OSError, ValueError):
            pass  # a dir being created or removed right now
    return out


# ── commands ────────────────────────────────────────────────────────────────

def cmd_version(cfg):
    return {"version": VERSION, "overlays": cfg["overlays"]}


def cmd_jobs(cfg):
    # ponytail: logs and job dirs are kept forever; prune once disk usage shows on the dashboard
    return {"jobs": all_jobs(cfg)}


def cmd_job_start(cfg, kind=None, *args):
    if kind not in KINDS:
        raise Error(f"unknown kind: {kind}")
    argv, env = KINDS[kind](cfg, list(args))
    if signs(kind, args) and gpg_unlocked(cfg) is False:
        raise Error("signing key locked")
    base = time.strftime("%Y%m%d-%H%M%S-") + kind
    jid, n = base, 1
    while True:
        d = os.path.join(cfg["state_dir"], "jobs", jid)
        try:
            os.mkdir(d)
            break
        except FileExistsError:
            n += 1
            jid = f"{base}-{n}"
    write_meta(d, {"id": jid, "kind": kind, "args": list(args), "argv": argv,
                   "env": env, "status": "starting", "created": time.time()})
    open(os.path.join(d, "log"), "wb").close()
    subprocess.Popen([sys.executable, os.path.abspath(__file__), "_run", jid],
                     stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL, start_new_session=True)
    for _ in range(60):
        status = read_meta(d)["status"]
        if status == "busy":
            shutil.rmtree(d)
            return {"busy": next((m["id"] for m in all_jobs(cfg) if m["status"] == "running"), None)}
        if status != "starting":
            break
        time.sleep(0.05)
    return {"id": jid}


def cmd_run(cfg, jid=None):
    d = job_dir(cfg, jid)
    meta = read_meta(d)
    with lock_file(cfg) as lk:
        try:
            fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            meta["status"] = "busy"
            write_meta(d, meta)
            return None
        with open(os.path.join(d, "log"), "ab") as log:
            meta["start"] = time.time()
            try:
                p = subprocess.Popen(meta["argv"], stdin=subprocess.DEVNULL, stdout=log,
                                     stderr=subprocess.STDOUT, start_new_session=True,
                                     env={**os.environ, **meta.get("env", {})})
            except OSError as e:
                log.write(f"sbb-remote: {e}\n".encode())
                rc = 127
            else:
                meta.update(pid=p.pid, status="running")
                write_meta(d, meta)
                rc = p.wait()
        meta.update(end=time.time(), exit=rc,
                    status="ok" if rc == 0 else "killed" if rc == -signal.SIGTERM else "failed")
        write_meta(d, meta)
    return None


def cmd_job_follow(cfg, jid=None, offset="0", out=None):
    d = job_dir(cfg, jid)
    out = out or sys.stdout.buffer
    with open(os.path.join(d, "log"), "rb") as f:
        f.seek(int(offset))
        while True:
            # status first, then read: the last read after the job ends drains the log
            done = refresh(cfg, d, read_meta(d))["status"] not in ACTIVE
            chunk = f.read()
            if chunk:
                out.write(chunk)
                out.flush()
            elif done:
                return None
            else:
                time.sleep(0.5)


def cmd_job_kill(cfg, jid=None):
    meta = read_meta(job_dir(cfg, jid))
    if meta["status"] != "running":
        raise Error(f"job not running: {jid}")
    try:
        os.killpg(meta["pid"], signal.SIGTERM)
    except ProcessLookupError:
        pass  # ended on its own meanwhile
    return {"ok": True}


# ── stats ───────────────────────────────────────────────────────────────────

def _try(fn):
    try:
        return fn()
    except Exception:
        return None


def _git(cfg, *args):
    return subprocess.run(["git", "-C", cfg["repo_dir"], *args], stdin=subprocess.DEVNULL,
                          capture_output=True, text=True, check=True).stdout


def _pkgdirs(root, pattern):
    """Paths of the <pkg>/<pkg>.SlackBuild dirs matching pattern under root."""
    out = []
    for d in glob.glob(os.path.join(root, pattern, "")):
        d = os.path.dirname(d)
        if os.path.isfile(os.path.join(d, os.path.basename(d) + ".SlackBuild")):
            out.append(d)
    return out


def _pkgs(root, pattern):
    return [os.path.basename(d) for d in _pkgdirs(root, pattern)]


def _packages(cfg):
    r = cfg["repo_dir"]
    mine = {o: _pkgs(r, o + "/*") for o in cfg["overlays"]}
    tree = _git(cfg, "ls-tree", "-r", "--name-only", cfg["upstream_branch"]).splitlines()
    upstream = {m.group(1) for m in map(PKG.match, tree) if m}
    shadowed = upstream & {p for names in mine.values() for p in names}
    return {"total": len(_pkgs(r, "*/*")), **{o: len(n) for o, n in mine.items()},
            "shadowed": len(shadowed)}


def _slackrepo(cfg):
    db = sqlite3.connect(f"file:{cfg['slackrepo_db']}?mode=ro", uri=True)
    try:
        results = dict(db.execute("select result, count(*) from buildresults group by result"))
        built = db.execute("select count(distinct itemid) from revisions where dep = '/'").fetchone()[0]
    finally:
        db.close()
    return {"results": results, "built": built}


def _uptime():
    with open("/proc/uptime") as f:
        return float(f.read().split()[0])


def cmd_stats(cfg):
    dirs = {"repo": cfg["repo_dir"], "state": cfg["state_dir"]}
    return {
        "load": _try(os.getloadavg),
        "uptime": _try(_uptime),
        "disk_free": {k: _try(lambda p=p: shutil.disk_usage(p).free) for k, p in dirs.items()},
        "packages": _try(lambda: _packages(cfg)),
        "sync": {o: _try(lambda o=o: int(_git(cfg, "log", "-1", "--format=%ct", "--", o + "/")))
                 for o in cfg["overlays"]},
        "slackrepo": _try(lambda: _slackrepo(cfg)),
    }


# ── inventory ───────────────────────────────────────────────────────────────

def parse_assignments(path):
    """KEY=value assignments of a shell-style file, without running it.

    Values with a $ are returned as {"value": raw, "shell": True}: the shell
    would expand them, this parser does not.
    """
    with open(path, errors="replace") as f:
        text = f.read().replace("\\\n", " ")
    out = {}
    for line in text.splitlines():
        try:
            tokens = shlex.split(line, comments=True)
        except ValueError:
            continue  # unbalanced quotes: slackrepo would choke too, skip the line
        raw = line.split("=", 1)[1] if "=" in line else ""
        for t in tokens:
            name, eq, value = t.partition("=")
            if eq and re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name):
                out[name] = {"value": raw.strip(), "shell": True} if "$" in raw else value
    return out


def _value(v):
    return v["value"] if isinstance(v, dict) else v


def find_hint(cfg, path):
    """slackrepo's search order (parsefunctions.sh): first <pkg>.hint found."""
    pkg = os.path.basename(path)
    dirs = [os.path.join(cfg["repo_dir"], path), cfg["hintdir"],
            os.path.join(cfg["hintdir"], path)]
    if cfg["default_hintdir"]:
        dirs += [cfg["default_hintdir"], os.path.join(cfg["default_hintdir"], path)]
    for d in dirs:
        f = os.path.join(d, pkg + ".hint")
        if os.path.isfile(f):
            return f
    return None


def _summary(pkgdir, pkg):
    try:
        with open(os.path.join(pkgdir, "slack-desc"), errors="replace") as f:
            for line in f:
                m = re.match(rf"^{re.escape(pkg)}: {re.escape(pkg)} \((.*)\)\s*$", line)
                if m:
                    return m.group(1)
    except OSError:
        pass
    return None


def _builds(cfg):
    """{path: (built_version, built_time)}, {path: result} from the slackrepo DB."""
    try:
        db = sqlite3.connect(f"file:{cfg['slackrepo_db']}?mode=ro", uri=True)
        try:
            built = {i: (v, int(t)) for i, v, t in
                     db.execute("select itemid, version, built from revisions where dep = '/'")}
            results = dict(db.execute("select itemid, result from buildresults"))
        finally:
            db.close()
    except sqlite3.Error:
        return {}, {}
    return built, results


COLUMNS = ["path", "version", "info_version", "hint_version", "built_version", "built_time",
           "result", "maintainer", "requires", "summary", "hint_path", "hints"]


def cmd_inventory(cfg):
    built, results = _builds(cfg)
    rows = []
    for d in sorted(_pkgdirs(cfg["repo_dir"], "*/*")):
        path = os.path.relpath(d, cfg["repo_dir"])
        pkg = os.path.basename(d)
        try:
            info = parse_assignments(os.path.join(d, pkg + ".info"))
        except OSError:
            info = {}
        hint_path = find_hint(cfg, path)
        hints = parse_assignments(hint_path) if hint_path else None
        info_version = _value(info.get("VERSION"))
        hint_version = _value(hints.get("VERSION")) if hints else None
        bv, bt = built.get(path, (None, None))
        rows.append([path, hint_version or info_version, info_version, hint_version, bv, bt,
                     results.get(path), _value(info.get("MAINTAINER")),
                     (_value(info.get("REQUIRES")) or "").split(), _summary(d, pkg),
                     hint_path, hints])
    return {"columns": COLUMNS, "rows": rows}


# ── signing key ─────────────────────────────────────────────────────────────

def _signing_key(cfg):
    try:
        with open(cfg["genreprc"]) as f:
            m = re.search(r'^\s*REPOSOWNERGPG=["\']?([^"\'\s]+)', f.read(), re.M)
    except OSError:
        return None
    return m and m.group(1)


def _keygrips(key):
    """Keygrips of the signing-capable (sub)keys of key."""
    out = subprocess.run(["gpg", "--batch", "--with-colons", "--with-keygrip",
                          "--list-secret-keys", key], stdin=subprocess.DEVNULL,
                         capture_output=True, text=True).stdout
    grips, can_sign = [], False
    for line in out.splitlines():
        f = line.split(":")
        if f[0] in ("sec", "ssb"):
            can_sign = "s" in f[11]
        elif f[0] == "grp" and can_sign:
            grips.append(f[9])
    return grips


def gpg_unlocked(cfg):
    """True/False, or None when no signing key is configured or found."""
    key = _signing_key(cfg)
    grips = _keygrips(key) if key else []
    if not grips:
        return None
    for g in grips:
        out = subprocess.run(["gpg-connect-agent", f"keyinfo {g}", "/bye"],
                             stdin=subprocess.DEVNULL, capture_output=True, text=True).stdout
        info = next((ln.split() for ln in out.splitlines() if ln.startswith("S KEYINFO")), [])
        if len(info) < 7 or info[6] != "1":
            return False
    return True


def cmd_gpg_status(cfg):
    return {"unlocked": gpg_unlocked(cfg), "key": _signing_key(cfg)}


def cmd_gpg_unlock(cfg, stdin=None):
    key = _signing_key(cfg)
    grips = _keygrips(key) if key else []
    if not grips:
        raise Error("no signing key configured")
    passphrase = (stdin or sys.stdin.buffer).read().rstrip(b"\n")
    for g in grips:
        subprocess.run([PRESET, "--preset", g], input=passphrase, check=True,
                       capture_output=True)
    with tempfile.NamedTemporaryFile() as t:
        ok = subprocess.run(["gpg", "--batch", "--pinentry-mode", "error", "-u", key,
                             "-bas", "-o", "-", t.name], stdin=subprocess.DEVNULL,
                            capture_output=True).returncode == 0
    if not ok:
        for g in grips:
            subprocess.run([PRESET, "--forget", g], stdin=subprocess.DEVNULL, capture_output=True)
        raise Error("bad passphrase")
    return {"unlocked": True}


COMMANDS = {
    "version": cmd_version,
    "stats": cmd_stats,
    "jobs": cmd_jobs,
    "job-start": cmd_job_start,
    "job-follow": cmd_job_follow,
    "job-kill": cmd_job_kill,
    "inventory": cmd_inventory,
    "gpg-status": cmd_gpg_status,
    "gpg-unlock": cmd_gpg_unlock,
    "_run": cmd_run,
}


def main(argv):
    try:
        if not argv or argv[0] not in COMMANDS:
            names = ",".join(c for c in COMMANDS if not c.startswith("_"))
            raise Error(f"usage: sbb-remote {{{names}}} [args]")
        cfg = load_config()
        os.makedirs(os.path.join(cfg["state_dir"], "jobs"), exist_ok=True)
        result = COMMANDS[argv[0]](cfg, *argv[1:])
    except BrokenPipeError:
        # the GUI hung up on job-follow; silence the flush at exit too
        os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
        return 0
    except Exception as e:
        print(json.dumps({"error": str(e) if isinstance(e, Error) else f"{type(e).__name__}: {e}"}))
        return 1
    if result is not None:
        print(json.dumps(result))
    return 0


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

Replace `remote/config.example.toml`:

````toml
# sbb-remote config on the VM: /etc/sb-builder/config.toml
# Every key is optional; these are the defaults.

state_dir       = "/var/lib/sb-builder"
repo_dir        = "/var/lib/sbopkg/SBo-danix"
upstream_branch = "current"
slackrepo_db    = "/var/lib/slackrepo/SBo-danix/database_SBo-danix_x86_64.sqlite3"
sbpull_script   = "/usr/local/share/slackrepo_setup/sbpull.bash"
overlays        = ["personal", "pentesting"]
slackrepo_repo  = "SBo-danix"
hintdir         = "/etc/slackrepo/SBo-danix/hintfiles"
default_hintdir = ""                     # searched after hintdir when set
genreprc        = "/root/.genreprc"      # REPOSOWNERGPG names the signing key
````

- [ ] **Step 4: Run the tests**

Run: `python3 -m unittest tests.test_remote </dev/null`
Expected: `Ran 37 tests`, `OK` (plus the known ResourceWarnings). `</dev/null` matters: without it a regression that lets a subprocess read stdin hangs instead of failing.

- [ ] **Step 5: Commit**

```bash
git add remote/sbb-remote remote/config.example.toml tests/test_remote.py
git commit -F - <<'EOF'
sbb-remote 0.2.0: inventory, slackrepo and publish jobs, signing key

inventory lists every package with its effective version (hint over
.info), build state from the slackrepo DB, summary, and the parsed
hint found by slackrepo's search order; hints are parsed with shlex,
never sourced. Job kinds now return argv plus env: slackrepo
build/rebuild/update/remove (packages validated against the tree,
--preview, --no-publish sets SBB_PUBLISH=n) and publish (sb-publish).
gpg-status and gpg-unlock let the GUI unlock the package signing key
in gpg-agent; the passphrase comes on stdin only, every other
subprocess gets stdin=DEVNULL. A real slackrepo run is refused while
the key is locked.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
EOF
git log --format='%h %G? %s' -1
```

---

### Task 3: GUI

**Files:** replace `sb_builder/__init__.py`, `sb_builder/config.py`, `config.example.toml`, `sb_builder/remote.py`, `sb_builder/jobs.py`, `sb_builder/dashboard.py`, `sb_builder/mainwindow.py`, `tests/test_gui.py`; create `sb_builder/hints.py`, `sb_builder/packages.py`.

- [ ] **Step 1: Write the tests** (replace `tests/test_gui.py`):

````python
# sb-builder - self-checks for the GUI
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Run: python3 -m unittest discover -s tests (headless, no VM needed)"""

import os
os.environ["QT_QPA_PLATFORM"] = "offscreen"  # never the real display, even if the session sets one

import importlib.machinery  # noqa: E402
import importlib.util  # noqa: E402
import sys  # noqa: E402
import tempfile  # noqa: E402
import time  # noqa: E402
import unittest  # noqa: E402

ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
sys.path.insert(0, ROOT)

from PyQt6.QtCore import QCoreApplication, QEvent, QObject, QProcess, pyqtSignal  # noqa: E402
from PyQt6.QtWidgets import QApplication  # noqa: E402

from sb_builder import VERSION, config  # noqa: E402
from sb_builder import hints as hintlib  # noqa: E402
from sb_builder import packages as pk  # noqa: E402
from sb_builder.mainwindow import MainWindow  # noqa: E402
from sb_builder.remote import DOWN, MISSING, parse  # noqa: E402

app = QApplication.instance() or QApplication([])


class ConfigTest(unittest.TestCase):
    def load(self, text):
        with tempfile.NamedTemporaryFile("w", suffix=".toml", delete=False) as f:
            f.write(text)
        self.addCleanup(os.unlink, f.name)
        return config.load(f.name)

    def test_missing_file(self):
        self.assertEqual(config.load("/nonexistent/config.toml"), (config.DEFAULTS, None))

    def test_example_matches_defaults(self):
        self.assertEqual(config.load(os.path.join(ROOT, "config.example.toml")), (config.DEFAULTS, None))

    def test_override(self):
        cfg, err = self.load("[refresh]\njobs = 2\n")
        self.assertIsNone(err)
        self.assertEqual(cfg["refresh"], {"dashboard": 30, "jobs": 2})

    def test_unknown_key(self):
        cfg, err = self.load("[remote]\nbogus = 1\n")
        self.assertIn("remote.bogus", err)
        self.assertEqual(cfg, config.DEFAULTS)

    def test_wrong_type(self):
        cfg, err = self.load('[vm]\nautostart = "yes"\n')
        self.assertIn("vm.autostart: expected bool", err)
        self.assertEqual(cfg, config.DEFAULTS)


class ParseTest(unittest.TestCase):
    def test_cases(self):
        self.assertEqual(parse(255, b"", b""), (None, DOWN))
        self.assertEqual(parse(127, b"", b""), (None, MISSING))
        self.assertEqual(parse(0, b'{"a": 1}', b""), ({"a": 1}, None))
        self.assertEqual(parse(1, b'{"error": "boom"}', b""), (None, "boom"))
        self.assertEqual(parse(1, b"", b"bash: x\n"), (None, "bash: x"))


class VersionTest(unittest.TestCase):
    def test_gui_matches_helper(self):
        script = os.path.join(ROOT, "remote", "sbb-remote")
        loader = importlib.machinery.SourceFileLoader("sbb_remote", script)
        sbb = importlib.util.module_from_spec(importlib.util.spec_from_loader("sbb_remote", loader))
        loader.exec_module(sbb)
        self.assertEqual(sbb.VERSION, VERSION)


class FakeRemote:
    def __init__(self, answers):
        self.answers = answers
        self.calls = []
        self.virsh_answer = (1, "")

    def call(self, cb, *args, stdin=None):
        self.calls.append(args if stdin is None else (*args, stdin))
        answer = self.answers.get(args[0], (None, DOWN))
        cb(*(answer(args, stdin) if callable(answer) else answer))

    def virsh(self, cb, *args):
        self.calls.append(("virsh", *args))
        cb(*self.virsh_answer)

    def follow(self, jid, offset):
        p = QProcess(app)  # parented, like Remote's, so it outlives this call
        p.start("true", [])
        return p


class FakeNotifier(QObject):
    clicked = pyqtSignal(str)

    def __init__(self):
        super().__init__()
        self.sent = []

    def send(self, title, body, jid=""):
        self.sent.append((title, body, jid))


STATS = {"load": [0.5, 0.4, 0.3], "uptime": 3600.0,
         "disk_free": {"repo": 10 * 2**30, "state": None},
         "packages": {"total": 10, "personal": 2, "pentesting": 1, "shadowed": 1},
         "sync": {"personal": None, "pentesting": 1792351223}, "slackrepo": None}
JOB = {"id": "20260101-000000-sbshadow", "kind": "sbshadow", "args": ["--no"], "argv": [],
       "status": "running", "created": 0, "start": 0, "pid": 5}


def inv_row(path, version="1.0", built=None, result=None, hints=None, requires=(),
            summary=None, maintainer="Some One", info_version=None):
    return [path, version, info_version or version, None, built, 1788978222 if built else None,
            result, maintainer, list(requires), summary, "/h/x.hint" if hints else None, hints]


COLS = ["path", "version", "info_version", "hint_version", "built_version", "built_time",
        "result", "maintainer", "requires", "summary", "hint_path", "hints"]
INVENTORY = {"columns": COLS, "rows": [
    inv_row("misc/nuclei", "3.11.1", built="3.11.0", result="ok", summary="Fast scanner",
            requires=["go"]),
    inv_row("personal/foo", built="1.0", result="ok", hints={"DELREQUIRES": "x"}, requires=["x", "y"]),
    inv_row("pentesting/nmap", built="7.9", result="bad"),
    inv_row("network/curl"),
]}
PREVIEW = {"id": "20260101-000000-slackrepo", "kind": "slackrepo", "status": "ok",
           "args": ["update", "--preview", "nuclei"], "argv": [], "created": 0, "start": 0, "end": 1}


class HintsTest(unittest.TestCase):
    def view(self, hints, requires=(), info_version="1.0"):
        return hintlib.interpret(hints, "pkg", info_version, requires)

    def section(self, view, title):
        return dict(view["sections"]).get(title)

    def test_no_hint(self):
        v = self.view(None, requires=["a"])
        self.assertEqual((v["skip"], v["sections"], v["deps"], v["ignored"]),
                         (None, [], [("a", "")], []))

    def test_version_arch_and_forced_checksum_skip(self):
        v = self.view({"VERSION": "2.0", "ARCH": "x86_64", "DOWNLOAD": "http://a/src.tgz",
                       "DOWNLOAD_x86_64": "http://a/bin-x64.tgz http://a/extra.tgz"})
        self.assertEqual(self.section(v, "Version override"), [
            ("version", "1.0 \u2192 2.0"), ("arch", "x86_64"), ("download", "http://a/bin-x64.tgz"),
            ("download", "http://a/extra.tgz"), ("md5sum", "not checked")])

    def test_arch_variant_empty_falls_back(self):
        v = self.view({"ARCH": "x86_64", "DOWNLOAD": "http://a/s", "MD5SUM": "abc", "MD5SUM_x86_64": ""})
        self.assertEqual(self.section(v, "Version override")[1:],
                         [("download", "http://a/s"), ("md5sum", "abc")])

    def test_no_download_forms(self):
        for hints in ({"DOWNLOAD": "no"}, {"NODOWNLOAD": "yes"}):
            with self.subTest(hints=hints):
                self.assertEqual(self.section(self.view(hints), "Version override"),
                                 [("download", "no download, the source must already be in the cache")])
                self.assertEqual(self.view(hints)["ignored"], [])

    def test_dependencies(self):
        v = self.view({"DELREQUIRES": "b %README%", "ADDREQUIRES": "z", "BUILDTIME": "cmake",
                       "PRAGMA": "python3 need_net", "CONFLICTS": "old"},
                      requires=["a", "b", "%README%"])
        self.assertEqual(v["deps"], [("a", ""), ("b", "del"), ("%README%", "del"),
                                     ("python3", "pragma"), ("z", "add"), ("cmake", "build")])
        self.assertEqual(v["conflicts"], ["old"])
        self.assertEqual(self.section(v, "Pragmas")[1], ("need_net", "network not blocked during the build"))

    def test_skip_forms(self):
        self.assertIsNone(self.view({"SKIP": "no"})["skip"])
        self.assertEqual(self.view({"SKIP": "yes"})["skip"], "always skipped")
        self.assertEqual(self.view({"SKIP": "remove nc first"})["skip"], "remove nc first")

    def test_build_settings(self):
        v = self.view({"OPTIONS": "A=yes B=no", "NUMJOBS": "-j1", "INSTALL": "N",
                       "ANSWER": "ok\\nyes", "NOWARNINGS": "Nonstandard"})
        self.assertEqual(self.section(v, "Build settings"), [
            ("option", "A=yes"), ("option", "B=no"), ("make jobs", "-j1"),
            ("install after build", "no"), ("answers to the build", "ok \u23ce yes"),
            ("hide warnings matching", "Nonstandard")])
        self.assertEqual(dict(self.section(self.view({"INSTALL": "1"}), "Build settings")),
                         {"install after build": "yes"})

    def test_users_and_groups(self):
        v = self.view({"GROUPADD": "209:postgres 300", "USERADD": "209:postgres:-d/var/lib/pgsql:-r"})
        self.assertEqual(self.section(v, "Host changes"), [
            ("group", "postgres (gid 209)"), ("group", "pkg (gid 300)"),
            ("user", "postgres (uid 209), group postgres, home /var/lib/pgsql, shell /bin/false, -r")])

    def test_pragma_forms(self):
        self.assertEqual(hintlib.pragma_meaning("x86arch=i686"), "on 32 bit x86, build with ARCH=i686")
        self.assertEqual(hintlib.pragma_meaning("unset_DISTCC_HOSTS"), "unset DISTCC_HOSTS for the build")
        self.assertEqual(hintlib.pragma_meaning("bogus"), "unknown pragma")

    def test_ignored_and_shell(self):
        v = self.view({"ANSWERS": "x", "FOO": "1", "DOWNLOAD_arm": "u",
                       "OPTIONS": {"value": '"X=$(uname -m)"', "shell": True}})
        self.assertEqual(v["ignored"], ["ANSWERS", "FOO"])
        self.assertEqual(v["shell"], ["OPTIONS"])

    def test_cleanup(self):
        self.assertEqual(self.section(self.view({"CLEANUP": "depmod -a"}), "On uninstall"),
                         [("run", "depmod -a")])


class FilterTest(unittest.TestCase):
    def setUp(self):
        self.rows = {r["name"]: r for r in pk.load_rows(INVENTORY)}

    def test_status(self):
        self.assertEqual({n: r["status"] for n, r in self.rows.items()},
                         {"nuclei": "outdated", "foo": "ok", "nmap": "failed", "curl": "never"})

    def test_query(self):
        self.assertEqual(pk.parse_query("Nuc cat:misc bogus:x status:"),
                         [(None, "nuc"), ("category", "misc"), (None, "bogus:x"), (None, "status:")])

    def names(self, text="", overlays=(), statuses=()):
        terms = pk.parse_query(text)
        return sorted(n for n, r in self.rows.items() if pk.matches(r, terms, overlays, statuses))

    def test_matching(self):
        self.assertEqual(self.names("scanner"), ["nuclei"])
        self.assertEqual(self.names("req:go"), ["nuclei"])
        self.assertEqual(self.names("maint:some status:o"), ["foo", "nuclei"])
        self.assertEqual(self.names(overlays={"personal", "pentesting"}), ["foo", "nmap"])
        self.assertEqual(self.names(overlays={"upstream"}, statuses={"never", "failed"}), ["curl"])
        self.assertEqual(self.names("cat:misc", statuses={"ok"}), [])


class SmokeTest(unittest.TestCase):
    def window(self, **answers):
        remote = FakeRemote({"version": ({"version": VERSION, "overlays": ["personal"]}, None),
                             "stats": (STATS, None), "jobs": ({"jobs": []}, None),
                             "gpg-status": ({"unlocked": True, "key": "K"}, None),
                             "inventory": (INVENTORY, None), **answers})
        w = MainWindow(config.load("/nonexistent")[0], remote, FakeNotifier())
        self.addCleanup(w.deleteLater)
        self.addCleanup(lambda: w.jobs_tab and w.jobs_tab.stop())
        return w, remote

    def test_tabs_and_dashboard(self):
        w, _ = self.window()
        w.poll_jobs()
        self.assertEqual(w.conn.text(), "● connected")
        self.assertEqual(w.tabs.tabText(0), "Dashboard")
        tiles = w.dashboard.tiles
        self.assertEqual(tiles["Packages"].value.text(), "10")
        self.assertEqual(tiles["Packages"].detail.text(), "2\u00a0personal · 1\u00a0pentesting · 1\u00a0shadowed")
        self.assertEqual(tiles["Built"].value.text(), "n/a")
        self.assertIn("10.0 GiB free", tiles["VM load"].detail.text())
        self.assertEqual(w.dashboard.legend.text(), "n/a")
        jobs = w.open_jobs()
        self.assertIs(w.open_jobs(), jobs)
        self.assertEqual(w.tabs.count(), 2)
        self.assertEqual(jobs.start_menu.actions()[0].text(), "sbpull personal (preview)")

    def test_end_notification_once(self):
        w, remote = self.window(jobs=({"jobs": [JOB]}, None))
        w.poll_jobs()
        remote.answers["jobs"] = ({"jobs": [dict(JOB, status="failed", end=5, exit=1)]}, None)
        w.poll_jobs()
        w.poll_jobs()
        self.assertEqual(w.notifier.sent, [("sb-builder: failed", "sbshadow --no", JOB["id"])])

    def test_down_offers_vm_start(self):
        w, remote = self.window(jobs=(None, DOWN))
        remote.virsh_answer = (0, "shut off")
        w.poll_jobs()
        self.assertIn(("virsh", "domstate", "buildsystem"), remote.calls)
        self.assertEqual(w.conn.text(), "● VM off")
        self.assertFalse(w.vm_btn.isHidden())

    def test_results_bar(self):
        w, _ = self.window(stats=(dict(STATS, slackrepo={"results": {"failed": 1, "ok": 3, "removed": 1},
                                                          "built": 4}), None))
        w.poll_jobs()
        self.assertEqual([(label, n) for label, n, _ in w.dashboard.bar.parts],
                         [("ok", 3), ("failed", 1), ("removed", 1)])
        self.assertEqual(w.dashboard.tiles["Built"].detail.text(), "of 10 · 6 never built")

    def test_close_jobs_tab_while_following(self):
        # regression: the follow process ended after the tab was deleted, its
        # finished slot touched dead widgets and PyQt aborted the app
        w, remote = self.window(jobs=({"jobs": [JOB]}, None))
        procs = QObject()
        self.addCleanup(procs.deleteLater)

        def follow(jid, offset):
            p = QProcess(procs)
            p.start("sleep", ["5"])
            return p

        remote.follow = follow
        w.poll_jobs()
        jobs = w.open_jobs()
        jobs.select(JOB["id"])
        self.assertIsNotNone(jobs.proc)
        w._close_tab(w.tabs.indexOf(jobs))
        QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete.value)
        end = time.monotonic() + 1
        while time.monotonic() < end:  # let the killed process report in
            app.processEvents()
        self.assertIsNone(w.jobs_tab)

    def test_packages_tab(self):
        w, remote = self.window(**{"job-start": ({"id": "20260101-000001-slackrepo"}, None)})
        w.poll_jobs()
        tab = w.open_packages()
        self.assertEqual(tab.proxy.rowCount(), 4)
        tab.chips["failed"].setChecked(True)
        self.assertEqual(tab.proxy.rowCount(), 1)
        tab.chips["failed"].setChecked(False)
        started = []
        tab.start_requested.connect(started.append)
        tab.act("update")
        self.assertEqual(started, [])  # nothing checked, no current row
        tab.table.setCurrentIndex(tab.proxy.index(1, 1))
        self.assertIn("<s>x</s>", tab.detail.text())
        tab.act("update")
        tab.model.checked.update({"nuclei", "curl"})
        tab.act("build")
        tab.update_all()
        self.assertEqual(started, [["slackrepo", "update", "--preview", "foo"],
                                   ["slackrepo", "build", "--preview", "curl", "nuclei"],
                                   ["slackrepo", "update", "--preview"]])

    def test_run_preview_with_signing(self):
        w, remote = self.window(jobs=({"jobs": [PREVIEW]}, None),
                                **{"gpg-status": ({"unlocked": False, "key": "K"}, None),
                                   "gpg-unlock": lambda args, stdin: (
                                       (None, "bad passphrase") if stdin == b"wrong\n" else ({}, None)),
                                   "job-start": ({"id": "20260101-000001-slackrepo"}, None)})
        w.poll_jobs()
        jobs = w.open_jobs()
        jobs.select(PREVIEW["id"])
        self.assertTrue(jobs.run_btn.isVisibleTo(jobs))
        answers = iter(["wrong", "right"])
        jobs.ask_passphrase = lambda retry: next(answers)
        jobs.publish_box.setChecked(False)
        jobs.run_preview()
        self.assertIn(("gpg-unlock", b"wrong\n"), remote.calls)
        self.assertIn(("gpg-unlock", b"right\n"), remote.calls)
        self.assertEqual(remote.calls[-2], ("job-start", "slackrepo", "update", "--no-publish", "nuclei"))

    def test_signing_cancel_starts_nothing(self):
        w, remote = self.window(jobs=({"jobs": [PREVIEW]}, None),
                                **{"gpg-status": ({"unlocked": False, "key": "K"}, None)})
        w.poll_jobs()
        jobs = w.open_jobs()
        jobs.select(PREVIEW["id"])
        jobs.ask_passphrase = lambda retry: None
        jobs.run_preview()
        self.assertFalse(any(c[0] == "job-start" for c in remote.calls))

    def test_key_indicator_and_tile_links(self):
        w, _ = self.window(**{"gpg-status": ({"unlocked": False, "key": "K"}, None)})
        w.poll_jobs()
        self.assertEqual(w.key.text(), "🔒 key locked")
        w.dashboard.tiles["Packages"].clicked.emit()
        self.assertIs(w.tabs.currentWidget(), w.packages_tab)
        w.dashboard.jobs.clicked.emit()
        self.assertIs(w.tabs.currentWidget(), w.jobs_tab)

    def test_log_carriage_return(self):
        w, _ = self.window()
        jobs = w.open_jobs()
        jobs.append_log("10%\r50%\r100%\ndone\r\n")
        self.assertEqual(jobs.log.toPlainText(), "100%\ndone\n")


if __name__ == "__main__":
    unittest.main()
````

- [ ] **Step 2: Run them, expect failures**

Run: `python3 -m unittest tests.test_gui </dev/null`
Expected: `ImportError` for `sb_builder.hints` / `sb_builder.packages`. Nothing may appear on the desktop.

- [ ] **Step 3: Hint rules** (create `sb_builder/hints.py`):

````python
# sb-builder - slackrepo hint interpretation
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""What a slackrepo hint file does, the way slackrepo reads it.

Rules from slackrepo's libexec/functions.d/parsefunctions.sh (hint and fixup
departments) and slackrepo.hint(5). Pure functions, no Qt.
"""

import os
import re

KNOWN = {"SKIP", "VERSION", "OPTIONS", "GROUPADD", "USERADD", "NOWARNING", "NOWARNINGS",
         "ADDREQUIRES", "DELREQUIRES", "BUILDTIME", "CONFLICTS", "INSTALL", "NUMJOBS",
         "ANSWER", "CLEANUP", "PRAGMA", "SPECIAL", "ARCH", "DOWNLOAD", "MD5SUM", "SHA256SUM",
         "NODOWNLOAD"}  # NODOWNLOAD: written by mkhint, shown as DOWNLOAD="no" by user decision
PER_ARCH = re.compile(r"^(DOWNLOAD|MD5SUM|SHA256SUM)_\w+$")
SHELLS = {"sh", "bash", "zsh", "ksh", "csh", "tcsh", "false", "nologin", "true"}

PRAGMAS = {
    "python3": "force python3 support: adds python3 to the deps, sets PYTHON3=yes",
    "kernelmodule": "rebuild when the host kernel changes; depmod on uninstall",
    "kernel": "rebuild when the host kernel changes",
    "kernelversion": "uname reports $KERNEL as the kernel version",
    "multilib_ldflags": "patch the SlackBuild to set LDFLAGS on a multilib host",
    "wget": "download with a fake wget user agent",
    "curl": "download with a fake curl user agent",
    "download_basename": "symlink downloads to the URL's basename",
    "download_urldecode": "URL-decode DOWNLOAD values before use",
    "need_net": "network not blocked during the build",
    "need_X": "X display not blocked during the build",
    "noexport_ARCH": "set ARCH by patching the SlackBuild, not via the environment",
    "noexport_BUILD": "set BUILD by patching the SlackBuild, not via the environment",
    "noexport_TAG": "set TAG by patching the SlackBuild, not via the environment",
    "noremove": "do not uninstall packages before building",
    "nofakeroot": "build as root with sudo instead of fakeroot",
    "no_make_test": "comment out 'make test' in the SlackBuild",
    "stubs-32": "provide /usr/include/gnu/stubs-32.h during the build",
    "abstar": "patch the SlackBuild to use tar --absolute-names",
}


def _raw(v):
    return v["value"] if isinstance(v, dict) else v


def pragma_meaning(p):
    if p in PRAGMAS:
        return PRAGMAS[p]
    if p.startswith("x86arch="):
        return f"on 32 bit x86, build with ARCH={p[8:]}"
    if p.startswith("unset_"):
        return f"unset {p[6:]} for the build"
    return "unknown pragma"


def parse_groupadd(value, pkg):
    """[(gid, name)] like slackrepo: numeric field is the gid, other the name."""
    out = []
    for spec in value.split():
        gid, name = None, pkg
        for field in spec.split(":"):
            if field[:1].isdigit():
                gid = field
            elif field:
                name = field
        out.append((gid, name))
    return out


def parse_useradd(value, pkg):
    """[dict(uid, name, group, home, shell, extra)] with slackrepo's defaults."""
    out = []
    for spec in value.split():
        u = {"uid": None, "name": pkg, "group": "", "home": "/dev/null",
             "shell": "/bin/false", "extra": []}
        for field in spec.split(":"):
            if field.startswith("-g"):
                u["group"] = field[2:]
            elif field.startswith("-d"):
                u["home"] = field[2:]
            elif field.startswith("-s"):
                u["shell"] = field[2:]
            elif field.startswith("-"):
                u["extra"].append(f"{field[:2]} {field[2:]}".strip())
            elif field.startswith("/"):
                # ponytail: slackrepo tests -x on the build host; guess by name instead
                key = "shell" if os.path.basename(field) in SHELLS else "home"
                u[key] = field
            elif field[:1].isdigit():
                u["uid"] = field
            elif field:
                u["name"] = field
        u["group"] = u["group"] or u["name"]
        out.append(u)
    return out


def dependencies(hints, requires):
    """[(dep, mark)] as slackrepo computes them; mark in '', del, pragma, add, build."""
    deleted = set(hints.get("DELREQUIRES", "").split()) | {"%README%"}
    out = [(r, "del" if r in deleted else "") for r in requires]
    if "python3" in hints.get("PRAGMA", hints.get("SPECIAL", "")).split():
        out.append(("python3", "pragma"))
    out += [(r, "add") for r in hints.get("ADDREQUIRES", "").split()]
    out += [(r, "build") for r in hints.get("BUILDTIME", "").split()]
    return out


def interpret(hints, pkg, info_version=None, requires=()):
    """Structured view of one package's hints.

    hints: {key: value or {"value": raw, "shell": True}} from the helper, or None.
    Returns {"skip", "sections": [(title, [(label, value)])], "deps",
    "conflicts", "ignored", "shell"}.
    """
    raw = hints or {}
    shell = sorted(k for k, v in raw.items() if isinstance(v, dict))
    h = {k: _raw(v) for k, v in raw.items()}
    view = {"skip": None, "sections": [], "deps": dependencies(h, list(requires)),
            "conflicts": h.get("CONFLICTS", "").split(), "shell": shell,
            "ignored": sorted(k for k in h if k not in KNOWN and not PER_ARCH.match(k))}

    skip = h.get("SKIP")
    if skip is not None and skip.lower() != "no":
        view["skip"] = "always skipped" if skip.lower() == "yes" else skip

    # version, downloads and checksums, resolved together as slackrepo does
    arch = h.get("ARCH")
    resolved = {}
    for var in ("DOWNLOAD", "MD5SUM", "SHA256SUM"):
        resolved[var] = h.get(f"{var}_{arch}") if arch and h.get(f"{var}_{arch}") else h.get(var)
    if h.get("VERSION"):
        resolved["MD5SUM"] = resolved["MD5SUM"] or "no"
        resolved["SHA256SUM"] = resolved["SHA256SUM"] or "no"
    rows = []
    if h.get("VERSION"):
        rows.append(("version", f"{info_version or '?'} → {h['VERSION']}"))
    if arch is not None:
        rows.append(("arch", arch or "(empty)"))
    if resolved["DOWNLOAD"] == "no" or h.get("NODOWNLOAD", "").lower() == "yes":
        rows.append(("download", "no download, the source must already be in the cache"))
    elif resolved["DOWNLOAD"]:
        rows += [("download", url) for url in resolved["DOWNLOAD"].split()]
    if resolved["MD5SUM"] == "no":
        rows.append(("md5sum", "not checked"))
    elif resolved["MD5SUM"]:
        rows.append(("md5sum", resolved["MD5SUM"]))
    if any(k.startswith("SHA256SUM") for k in h):
        rows.append(("sha256sum", "not checked" if resolved["SHA256SUM"] == "no"
                     else resolved["SHA256SUM"]))
    if rows:
        view["sections"].append(("Version override", rows))

    rows = [("option", o) for o in h.get("OPTIONS", "").split()]
    if "NUMJOBS" in h:
        rows.append(("make jobs", h["NUMJOBS"]))
    if h.get("INSTALL"):
        rows.append(("install after build", "no" if h["INSTALL"][:1] in "Nn0" else "yes"))
    if "ANSWER" in h:
        rows.append(("answers to the build", h["ANSWER"].replace("\\n", " ⏎ ")))
    nowarning = h.get("NOWARNING") or h.get("NOWARNINGS")
    if nowarning:
        rows.append(("hide warnings matching", nowarning))
    if rows:
        view["sections"].append(("Build settings", rows))

    pragmas = h.get("PRAGMA") or h.get("SPECIAL") or ""
    if pragmas:
        view["sections"].append(("Pragmas", [(p, pragma_meaning(p)) for p in pragmas.split()]))

    rows = [("group", f"{name} (gid {gid or '?'})") for gid, name in parse_groupadd(h.get("GROUPADD", ""), pkg)]
    for u in parse_useradd(h.get("USERADD", ""), pkg):
        extra = f", {' '.join(u['extra'])}" if u["extra"] else ""
        rows.append(("user", f"{u['name']} (uid {u['uid'] or '?'}), group {u['group']}, "
                             f"home {u['home']}, shell {u['shell']}{extra}"))
    if rows:
        view["sections"].append(("Host changes", rows))

    if h.get("CLEANUP"):
        view["sections"].append(("On uninstall", [("run", h["CLEANUP"])]))
    return view
````

- [ ] **Step 4: Packages tab** (create `sb_builder/packages.py`):

````python
# sb-builder - Packages tab
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Packages tab: search the inventory, read a package's hints, start slackrepo previews."""

import html

from PyQt6.QtCore import QAbstractTableModel, QModelIndex, QSortFilterProxyModel, Qt, pyqtSignal
from PyQt6.QtGui import QColor
from PyQt6.QtWidgets import (QHBoxLayout, QLabel, QLineEdit, QMessageBox, QPushButton,
                             QScrollArea, QSplitter, QTableView, QVBoxLayout, QWidget)

from . import hints as hintlib
from .dashboard import STATUS
from .fmt import NA, fmt_time

OVERLAYS = ("personal", "pentesting")
STATUSES = ("ok", "failed", "never", "outdated")
COLUMNS = ("", "name", "category", "version", "built", "status", "maintainer")
STATUS_COLOR = {"ok": STATUS["ok"], "failed": STATUS["failed"], "outdated": STATUS["aborted"]}
FIELDS = {"cat": "category", "maint": "maintainer", "req": "requires", "status": "status"}


def status(row):
    # ponytail: outdated compares versions only; slackrepo also rebuilds on
    # SlackBuild or dependency changes, which its preview shows
    if not row["built_version"] and not row["result"]:
        return "never"
    if row["result"] in ("failed", "bad", "aborted"):
        return "failed"
    if row["built_version"] != row["version"]:
        return "outdated"
    return "ok"


def overlay(row):
    cat = row["path"].split("/", 1)[0]
    return cat if cat in OVERLAYS else "upstream"


def parse_query(text):
    """[(field or None, term)], lower-cased; unknown prefixes count as plain text."""
    terms = []
    for word in text.lower().split():
        field, sep, term = word.partition(":")
        if sep and field in FIELDS and term:
            terms.append((FIELDS[field], term))
        else:
            terms.append((None, word))
    return terms


def matches(row, terms, overlays=(), statuses=()):
    """All terms match, and the row is in a toggled chip of each non-empty group."""
    if overlays and overlay(row) not in overlays:
        return False
    if statuses and row["status"] not in statuses:
        return False
    for field, term in terms:
        if field is None:
            hay = f"{row['name']} {row['summary'] or ''}"
        elif field == "requires":
            hay = " ".join(row["requires"])
        else:
            hay = row[field] or ""
        if term not in hay.lower():
            return False
    return True


def load_rows(inventory):
    rows = []
    for values in inventory["rows"]:
        row = dict(zip(inventory["columns"], values))
        row["category"], row["name"] = row["path"].split("/", 1)
        row["status"] = status(row)
        rows.append(row)
    return rows


class Model(QAbstractTableModel):
    def __init__(self):
        super().__init__()
        self.rows = []
        self.checked = set()  # package names

    def set_rows(self, rows):
        self.beginResetModel()
        self.rows = rows
        names = {r["name"] for r in rows}
        self.checked &= names
        self.endResetModel()

    def rowCount(self, parent=QModelIndex()):
        return 0 if parent.isValid() else len(self.rows)

    def columnCount(self, parent=QModelIndex()):
        return len(COLUMNS)

    def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole):
        if orientation == Qt.Orientation.Horizontal and role == Qt.ItemDataRole.DisplayRole:
            return COLUMNS[section]
        return None

    def flags(self, index):
        f = Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable
        return f | Qt.ItemFlag.ItemIsUserCheckable if index.column() == 0 else f

    def data(self, index, role=Qt.ItemDataRole.DisplayRole):
        row, col = self.rows[index.row()], index.column()
        if role == Qt.ItemDataRole.CheckStateRole and col == 0:
            return Qt.CheckState.Checked if row["name"] in self.checked else Qt.CheckState.Unchecked
        if role == Qt.ItemDataRole.DisplayRole:
            return (None, row["name"], row["category"], row["version"],
                    row["built_version"] or "", row["status"], row["maintainer"] or "")[col]
        if role == Qt.ItemDataRole.ForegroundRole and col == 5 and row["status"] in STATUS_COLOR:
            return QColor(STATUS_COLOR[row["status"]])
        if role == Qt.ItemDataRole.UserRole:
            return row
        return None

    def setData(self, index, value, role=Qt.ItemDataRole.EditRole):
        if role != Qt.ItemDataRole.CheckStateRole or index.column() != 0:
            return False
        name = self.rows[index.row()]["name"]
        if Qt.CheckState(value) == Qt.CheckState.Checked:
            self.checked.add(name)
        else:
            self.checked.discard(name)
        self.dataChanged.emit(index, index)
        return True


class Filter(QSortFilterProxyModel):
    def __init__(self):
        super().__init__()
        self.terms, self.overlays, self.statuses = [], set(), set()

    def update(self, text, overlays, statuses):
        self.terms, self.overlays, self.statuses = parse_query(text), overlays, statuses
        self.invalidateFilter()

    def filterAcceptsRow(self, source_row, _parent):
        row = self.sourceModel().rows[source_row]
        return matches(row, self.terms, self.overlays, self.statuses)


def chip(text):
    b = QPushButton(text, checkable=True)
    b.setStyleSheet("QPushButton { border: 1px solid palette(mid); border-radius: 10px;"
                    " padding: 2px 10px; }"
                    "QPushButton:checked { background: palette(highlight);"
                    " color: palette(highlighted-text); }")
    return b


def detail_html(row):
    """The detail pane for one package: facts, dependencies, then its hints."""
    view = hintlib.interpret(row["hints"], row["name"], row["info_version"], row["requires"])
    e = html.escape
    out = [f"<h3>{e(row['name'])}</h3><p>{e(row['path'])}"
           f"{' · ' + e(row['summary']) if row['summary'] else ''}</p>"]
    if view["skip"]:
        out.append(f'<p style="color:{STATUS["aborted"]}"><b>&#9888; Skipped by slackrepo:</b> '
                   f"{e(view['skip'])}</p>")
    facts = [(".info version", row["info_version"]), ("hint version", row["hint_version"]),
             ("built version", row["built_version"]),
             ("built", fmt_time(row["built_time"]) if row["built_time"] else None),
             ("last result", row["result"]), ("maintainer", row["maintainer"])]
    out.append(table([(k, v or NA) for k, v in facts]))
    marks = {"": "", "del": "<s>{}</s> <small>(DELREQUIRES)</small>",
             "add": "+{} <small>(ADDREQUIRES)</small>", "build": "+{} <small>(build time only)</small>",
             "pragma": "+{} <small>(pragma)</small>"}
    deps = [(marks[m] or "{}").format(e(d)) for d, m in view["deps"]]
    out.append(f"<h4>Dependencies</h4><p>{' &nbsp; '.join(deps) or 'none'}</p>")
    if view["conflicts"]:
        out.append(f"<p>Conflicts: {e(' '.join(view['conflicts']))}</p>")
    for title, rows in view["sections"]:
        out.append(f"<h4>{e(title)}</h4>" + table(rows))
    if view["shell"]:
        out.append(f"<p><small>Expanded by the shell at build time, shown as written: "
                   f"{e(', '.join(view['shell']))}</small></p>")
    if view["ignored"]:
        out.append(f'<p style="color:{STATUS["aborted"]}">Not read by slackrepo: '
                   f"{e(', '.join(view['ignored']))}</p>")
    if row["hint_path"]:
        out.append(f"<p><small>hint: {e(row['hint_path'])}</small></p>")
    return "".join(out)


def table(rows):
    cells = "".join(f"<tr><td style='padding-right:12px'><i>{html.escape(str(k))}</i></td>"
                    f"<td>{html.escape(str(v))}</td></tr>" for k, v in rows)
    return f"<table>{cells}</table>"


class PackagesTab(QWidget):
    start_requested = pyqtSignal(list)  # job-start arguments

    def __init__(self, remote, parent=None):
        super().__init__(parent)
        self.remote = remote
        self.model = Model()
        self.proxy = Filter()
        self.proxy.setSourceModel(self.model)

        self.query = QLineEdit(placeholderText="search  (cat:  maint:  req:  status:)")
        self.query.textChanged.connect(self.refilter)
        self.chips = {}
        chips = QHBoxLayout()
        for group in (("personal", "pentesting", "upstream"), STATUSES):
            for name in group:
                self.chips[name] = chip(name)
                self.chips[name].toggled.connect(self.refilter)
                chips.addWidget(self.chips[name])
            chips.addSpacing(16)
        chips.addStretch()
        self.refresh_btn = QPushButton("Refresh")
        self.refresh_btn.clicked.connect(self.reload)
        chips.addWidget(self.refresh_btn)

        self.table = QTableView(sortingEnabled=True)
        self.table.setModel(self.proxy)
        self.table.setSelectionBehavior(QTableView.SelectionBehavior.SelectRows)
        self.table.setSelectionMode(QTableView.SelectionMode.SingleSelection)
        self.table.verticalHeader().hide()
        self.table.horizontalHeader().setStretchLastSection(True)
        for col, width in enumerate((28, 220, 110, 120, 120, 80)):
            self.table.setColumnWidth(col, width)
        self.table.selectionModel().currentRowChanged.connect(self.show_detail)
        self.detail = QLabel(textFormat=Qt.TextFormat.RichText, wordWrap=True,
                             alignment=Qt.AlignmentFlag.AlignTop)
        self.detail.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
        self.detail.setContentsMargins(12, 8, 12, 8)
        scroll = QScrollArea(widgetResizable=True)
        scroll.setWidget(self.detail)
        split = QSplitter()
        split.addWidget(self.table)
        split.addWidget(scroll)
        split.setSizes([700, 400])

        self.count = QLabel()
        bar = QHBoxLayout()
        bar.addWidget(self.count)
        for label, slot in (("Select visible", self.select_visible), ("Clear", self.clear)):
            b = QPushButton(label)
            b.clicked.connect(slot)
            bar.addWidget(b)
        bar.addStretch()
        for action in ("build", "update", "rebuild", "remove"):
            b = QPushButton(action.capitalize())
            b.clicked.connect(lambda _c, a=action: self.act(a))
            bar.addWidget(b)
        for label, slot in (("Update all", self.update_all), ("Publish", self.publish)):
            b = QPushButton(label)
            b.clicked.connect(slot)
            bar.addWidget(b)

        layout = QVBoxLayout(self)
        layout.addWidget(self.query)
        layout.addLayout(chips)
        layout.addWidget(split)
        layout.addLayout(bar)
        self.model.dataChanged.connect(lambda *_: self.update_count())
        self.update_count()

    def reload(self):
        self.count.setText("loading…")
        self.remote.call(self._loaded, "inventory")

    def _loaded(self, data, err):
        if err:
            self.count.setText(f"inventory failed: {err}")
            return
        self.model.set_rows(load_rows(data))
        self.update_count()

    def refilter(self):
        on = {n for n, b in self.chips.items() if b.isChecked()}
        self.proxy.update(self.query.text(), on - set(STATUSES), on & set(STATUSES))
        self.update_count()

    def update_count(self):
        self.count.setText(f"{self.proxy.rowCount():,} shown · {len(self.model.checked)} selected")

    def select_visible(self):
        for i in range(self.proxy.rowCount()):
            self.model.checked.add(self.proxy.index(i, 0).data(Qt.ItemDataRole.UserRole)["name"])
        self.model.layoutChanged.emit()
        self.update_count()

    def clear(self):
        self.model.checked.clear()
        self.model.layoutChanged.emit()
        self.update_count()

    def targets(self):
        """Checked packages, else the current row; sorted names."""
        if self.model.checked:
            return sorted(self.model.checked)
        current = self.table.currentIndex()
        return [current.data(Qt.ItemDataRole.UserRole)["name"]] if current.isValid() else []

    def act(self, action):
        names = self.targets()
        if names:
            self.start_requested.emit(["slackrepo", action, "--preview", *names])

    def update_all(self):
        self.start_requested.emit(["slackrepo", "update", "--preview"])

    def publish(self):
        if QMessageBox.question(self, "Publish", "Push the package repository to the mirror now?") \
                == QMessageBox.StandardButton.Yes:
            self.start_requested.emit(["publish"])

    def show_detail(self, current, _previous):
        row = current.data(Qt.ItemDataRole.UserRole) if current.isValid() else None
        self.detail.setText(detail_html(row) if row else "")
````

- [ ] **Step 5: Version, config, transport**

Replace `sb_builder/__init__.py`:

````python
# sb-builder - GUI package
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""sb-builder: Qt6 GUI driving the SlackBuild tooling on a build VM."""

VERSION = "0.2.0"  # must match remote/sbb-remote VERSION
````

Replace `sb_builder/config.py`:

````python
# sb-builder - GUI configuration
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""GUI configuration: built-in defaults overlaid by the user's TOML file."""

import copy
import os
import tomllib

DEFAULTS = {
    "remote": {"target": "buildsystem", "ssh_options": []},
    "refresh": {"dashboard": 30, "jobs": 5},
    "notify": {"enabled": True, "events": ["finished", "failed", "killed", "lost", "vm-up"]},
    "vm": {"libvirt_uri": "qemu:///system", "domain": "buildsystem", "autostart": False},
    "builds": {"publish": True},
}


def path():
    base = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config")
    return os.path.join(base, "sb-builder", "config.toml")


def load(p=None):
    """Return (config, error). On any problem: the defaults plus the error text."""
    p = p or path()
    try:
        with open(p, "rb") as f:
            data = tomllib.load(f)
    except FileNotFoundError:
        return copy.deepcopy(DEFAULTS), None
    except (OSError, tomllib.TOMLDecodeError) as e:
        return copy.deepcopy(DEFAULTS), f"{p}: {e}"
    cfg = copy.deepcopy(DEFAULTS)
    for sec, keys in data.items():
        if sec not in DEFAULTS or not isinstance(keys, dict):
            return copy.deepcopy(DEFAULTS), f"{p}: unknown section [{sec}]"
        for k, v in keys.items():
            if k not in DEFAULTS[sec]:
                return copy.deepcopy(DEFAULTS), f"{p}: unknown key {sec}.{k}"
            want = type(DEFAULTS[sec][k])
            if type(v) is not want:
                return copy.deepcopy(DEFAULTS), f"{p}: {sec}.{k}: expected {want.__name__}"
            cfg[sec][k] = v
    return cfg, None
````

Replace `config.example.toml`:

````toml
# sb-builder GUI config: ~/.config/sb-builder/config.toml
# Every key is optional; these are the defaults.

[remote]
target = "buildsystem"     # any ssh destination: host alias, user@host
ssh_options = []           # extra ssh args, e.g. ["-p", "2222"]

[refresh]
dashboard = 30             # seconds, while the Dashboard tab is visible
jobs = 5                   # seconds; doubles up to 30 while the VM is down

[notify]
enabled = true
events = ["finished", "failed", "killed", "lost", "vm-up"]

[vm]
libvirt_uri = "qemu:///system"
domain = "buildsystem"     # libvirt domain of the VM; "" disables VM control
autostart = false          # true: start the VM without asking when it is off

[builds]
publish = true             # initial state of the "publish" box next to Run it
````

Replace `sb_builder/remote.py`:

````python
# sb-builder - ssh and virsh transport
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""The only code that talks to the VM: ssh to sbb-remote, and local virsh."""

import json
import os
import shlex

from PyQt6.QtCore import QObject, QProcess, QStandardPaths

DOWN = "ssh: connection failed"
MISSING = "sbb-remote is not installed on the VM"


def parse(rc, out, err):
    """Map one sbb-remote call to (data, error)."""
    if rc == 255:
        return None, DOWN
    if rc == 127:
        return None, MISSING
    try:
        data = json.loads(out)
    except ValueError:
        return None, err.decode(errors="replace").strip() or f"exit {rc}"
    if rc != 0 or "error" in data:
        return None, data.get("error", f"exit {rc}")
    return data, None


class Remote(QObject):
    def __init__(self, cfg, parent=None):
        super().__init__(parent)
        self.cfg = cfg  # shared dict, updated in place on config reload
        run = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.RuntimeLocation)
        self.control = os.path.join(run or "/tmp", "sbb-%C")

    def ssh_args(self, *args):
        r = self.cfg["remote"]
        # ssh joins its args into a remote shell command: quote every one
        return ["-o", "ControlMaster=auto", "-o", f"ControlPath={self.control}",
                "-o", "ControlPersist=10m", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
                *r["ssh_options"], r["target"], "sbb-remote", *map(shlex.quote, args)]

    def call(self, cb, *args, stdin=None):
        """Run sbb-remote <args>, then cb(data, error). stdin: bytes for the helper."""
        self._run("ssh", self.ssh_args(*args), lambda rc, out, err: cb(*parse(rc, out, err)),
                  stdin)

    def virsh(self, cb, *args):
        """Run local virsh against the configured libvirt URI, then cb(rc, text)."""
        argv = ["-c", self.cfg["vm"]["libvirt_uri"], *args]
        self._run("virsh", argv,
                  lambda rc, out, err: cb(rc, (out or err).decode(errors="replace").strip()))

    def follow(self, jid, offset):
        """Start job-follow; the caller wires readyReadStandardOutput and finished."""
        p = QProcess(self)
        p.start("ssh", self.ssh_args("job-follow", jid, str(offset)))
        return p

    def _run(self, prog, argv, cb, stdin=None):
        p = QProcess(self)

        def done(rc, _status=None):
            cb(rc, bytes(p.readAllStandardOutput()), bytes(p.readAllStandardError()))
            p.deleteLater()

        p.finished.connect(done)
        p.errorOccurred.connect(
            lambda e: e == QProcess.ProcessError.FailedToStart and done(-1))
        p.start(prog, argv)
        if stdin is not None:
            p.write(stdin)
            p.closeWriteChannel()
````

- [ ] **Step 6: Jobs tab, Dashboard, main window**

Replace `sb_builder/jobs.py`:

````python
# sb-builder - Jobs tab
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Jobs tab: job list, live log through job-follow, start and kill."""

import codecs
import re

from PyQt6.QtCore import Qt, pyqtSignal
from PyQt6.QtGui import QFontDatabase, QTextCursor
from PyQt6.QtWidgets import (QCheckBox, QHBoxLayout, QInputDialog, QLineEdit, QListWidget,
                             QListWidgetItem, QMenu, QMessageBox, QPlainTextEdit, QPushButton,
                             QSplitter, QToolButton, QVBoxLayout, QWidget)

from .fmt import describe

ANSI = re.compile(rb"\x1b\[[0-9;?]*[A-Za-z]")
SHADOW_PLAN = re.compile(r"^(un)?shadow: ", re.M)
ACTIVE = ("starting", "running")
BAD = ("failed", "killed", "lost")


class JobsTab(QWidget):
    refresh_requested = pyqtSignal()

    def __init__(self, remote, publish=True, parent=None):
        super().__init__(parent)
        self.remote = remote
        self.pending = ()
        self.jobs = {}
        self.current = None
        self.offset = 0
        self.proc = None
        self.decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")

        self.start_menu = QMenu(self)
        start = QToolButton(text="Start", popupMode=QToolButton.ToolButtonPopupMode.InstantPopup)
        start.setMenu(self.start_menu)
        self.apply_btn = QPushButton("Apply shadowing")
        self.apply_btn.clicked.connect(lambda: self.confirm_start(
            "Apply the shadowing plan to the repo (sbshadow --yes)?", "sbshadow", "--yes"))
        self.apply_btn.hide()
        self.publish_box = QCheckBox("publish", checked=publish)
        self.publish_box.setToolTip("Push to the mirror when the build ends; "
                                    "unticked, use Publish later")
        self.run_btn = QPushButton("Run it")
        self.run_btn.clicked.connect(self.run_preview)
        self.kill_btn = QPushButton("Kill")
        self.kill_btn.clicked.connect(self.kill)
        bar = QHBoxLayout()
        for w in (start, self.apply_btn, self.run_btn, self.publish_box):
            bar.addWidget(w)
        bar.addStretch()
        bar.addWidget(self.kill_btn)

        self.list = QListWidget()
        self.list.currentItemChanged.connect(self._on_select)
        self.log = QPlainTextEdit(readOnly=True)
        self.log.setFont(QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont))
        self.log.setMaximumBlockCount(50000)  # ponytail: older lines drop off; the full log stays on the VM
        split = QSplitter()
        split.addWidget(self.list)
        split.addWidget(self.log)
        split.setSizes([300, 700])
        layout = QVBoxLayout(self)
        layout.addLayout(bar)
        layout.addWidget(split)
        self.set_overlays([])

    def set_overlays(self, overlays):
        m = self.start_menu
        m.clear()
        for o in overlays:
            m.addAction(f"sbpull {o} (preview)", lambda o=o: self.start("sbpull", o, "--no"))
        m.addSeparator()
        m.addAction("sbshadow (preview)", lambda: self.start("sbshadow", "--no"))
        m.addAction("sbshadow (apply)…", lambda: self.confirm_start(
            "Run sbshadow --yes on the repo?", "sbshadow", "--yes"))

    def set_jobs(self, jobs):
        self.jobs = {j["id"]: j for j in jobs}
        self.list.blockSignals(True)
        self.list.clear()
        for j in jobs:
            item = QListWidgetItem(describe(j))
            item.setData(Qt.ItemDataRole.UserRole, j["id"])
            if j["status"] in BAD:
                item.setForeground(Qt.GlobalColor.red)
            self.list.addItem(item)
            if j["id"] == self.current:
                self.list.setCurrentItem(item)
        self.list.blockSignals(False)
        j = self.jobs.get(self.current)
        if j and j["status"] in ACTIVE and self.proc is None:
            self._follow()  # stream dropped while the job kept running: resume at offset
        self._update_buttons()

    def select(self, jid):
        for i in range(self.list.count()):
            if self.list.item(i).data(Qt.ItemDataRole.UserRole) == jid:
                self.list.setCurrentItem(self.list.item(i))
                return
        self._show(jid)  # not listed yet: just started

    def append_log(self, text):
        cur = self.log.textCursor()
        cur.movePosition(QTextCursor.MoveOperation.End)
        # ponytail: a \r split across two chunks is printed as-is
        for i, part in enumerate(text.replace("\r\n", "\n").split("\r")):
            if i:  # carriage return: the next text replaces the current line
                cur.movePosition(QTextCursor.MoveOperation.StartOfBlock,
                                 QTextCursor.MoveMode.KeepAnchor)
                cur.removeSelectedText()
            cur.insertText(part)
        self.log.setTextCursor(cur)

    def stop(self):
        if self.proc:
            p, self.proc = self.proc, None
            # disconnect first: finished arrives later, maybe after this tab is deleted
            p.readyReadStandardOutput.disconnect()
            p.finished.disconnect()
            p.kill()
            p.deleteLater()

    def start(self, *args):
        self.pending = args
        self.remote.call(self._started, "job-start", *args)

    def confirm_start(self, question, *args):
        if QMessageBox.question(self, "Confirm", question) == QMessageBox.StandardButton.Yes:
            self.start(*args)

    def kill(self):
        if QMessageBox.question(self, "Kill job", f"Send SIGTERM to {self.current}?") \
                == QMessageBox.StandardButton.Yes:
            self.remote.call(lambda _d, err: err and QMessageBox.warning(self, "Kill failed", err),
                             "job-kill", self.current)

    def run_preview(self):
        """Start the previewed slackrepo run for real."""
        args = [a for a in self.jobs[self.current]["args"] if a != "--preview"]
        if not self.publish_box.isChecked():
            args.insert(1, "--no-publish")
        self.start_signed("slackrepo", *args)

    def ask_passphrase(self, retry):
        text, ok = QInputDialog.getText(
            self, "Signing key", ("Wrong passphrase, try again." if retry else
                                  "Passphrase of the package signing key:"),
            QLineEdit.EchoMode.Password)
        return text if ok else None

    def start_signed(self, *args, retry=False):
        """Unlock the signing key first when it is locked, then start the job."""
        def status(data, err):
            if err:
                QMessageBox.warning(self, "Signing key", err)
            elif data["unlocked"] is False:
                self._unlock(args, retry)
            else:
                self.start(*args)
        self.remote.call(status, "gpg-status")

    def _unlock(self, args, retry):
        passphrase = self.ask_passphrase(retry)
        if passphrase is None:
            return

        def unlocked(_data, err):
            if err == "bad passphrase":
                self._unlock(args, True)
            elif err:
                QMessageBox.warning(self, "Signing key", err)
            else:
                self.start(*args)
        self.remote.call(unlocked, "gpg-unlock", stdin=passphrase.encode() + b"\n")

    def _started(self, data, err):
        if err == "signing key locked":
            self.start_signed(*self.pending)  # locked since the last check: ask again
            return
        if err:
            QMessageBox.warning(self, "Start failed", err)
            return
        if "busy" in data:
            box = QMessageBox(QMessageBox.Icon.Information, "Busy",
                              f"busy: {data['busy']}", parent=self)
            show = box.addButton("Show job", QMessageBox.ButtonRole.AcceptRole)
            box.addButton(QMessageBox.StandardButton.Close)
            box.exec()
            if box.clickedButton() is show and data["busy"]:
                self.select(data["busy"])
            return
        self.select(data["id"])
        self.refresh_requested.emit()

    def _on_select(self, item, _prev):
        jid = item.data(Qt.ItemDataRole.UserRole) if item else None
        if jid != self.current:
            self._show(jid)

    def _show(self, jid):
        self.stop()
        self.current, self.offset = jid, 0
        self.decoder.reset()
        self.log.clear()
        if jid:
            self._follow()
        self._update_buttons()

    def _follow(self):
        jid = self.current
        p = self.proc = self.remote.follow(jid, self.offset)
        p.readyReadStandardOutput.connect(lambda: self._chunk(p, jid))
        p.finished.connect(lambda *_: self._ended(p))

    def _chunk(self, p, jid):
        data = bytes(p.readAllStandardOutput())
        if jid != self.current:
            return
        self.offset += len(data)
        self.append_log(self.decoder.decode(ANSI.sub(b"", data)))

    def _ended(self, p):
        if p is self.proc:
            self.proc = None
        p.deleteLater()
        self._update_buttons()

    def _update_buttons(self):
        j = self.jobs.get(self.current)
        self.kill_btn.setEnabled(bool(j) and j["status"] == "running")
        self.apply_btn.setVisible(
            bool(j) and j["status"] == "ok" and "--no" in j["args"]
            and bool(SHADOW_PLAN.search(self.log.toPlainText())))
        preview = bool(j) and j["kind"] == "slackrepo" and j["status"] == "ok" \
            and "--preview" in j["args"]
        self.run_btn.setVisible(preview)
        self.publish_box.setVisible(preview)
````

Replace `sb_builder/dashboard.py`:

````python
# sb-builder - Dashboard tab
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Dashboard tab: KPI tiles, slackrepo results bar, recent jobs, from stats and jobs."""

import html
import time

from PyQt6.QtCore import QEvent, QRectF, Qt, pyqtSignal
from PyQt6.QtGui import QColor, QFont, QPainter, QPalette
from PyQt6.QtWidgets import (QFrame, QGridLayout, QHBoxLayout, QLabel, QPushButton,
                             QToolTip, QVBoxLayout, QWidget)

from .fmt import ICON, NA, fmt_age, fmt_bytes, fmt_duration, fmt_time, job_title

# reserved status colors (dataviz reference palette); always shown with a text label
STATUS = {"ok": "#0ca30c", "aborted": "#fab219", "bad": "#ec835a", "failed": "#d03b3b",
          "killed": "#fab219", "lost": "#ec835a"}
GAP = 2  # px of surface between bar segments


def num(n):
    return NA if n is None else f"{n:,}"


def muted(label):
    label.setForegroundRole(QPalette.ColorRole.PlaceholderText)
    return label


class Card(QFrame):
    """Rounded panel with a muted title; content goes into self.body."""

    clicked = pyqtSignal()

    def __init__(self, title):
        super().__init__(objectName="card")
        self.body = QVBoxLayout(self)
        self.body.setContentsMargins(16, 12, 16, 14)
        self.body.addWidget(muted(QLabel(title)))

    def link(self, slot):
        """Make the whole card open something."""
        self.clicked.connect(slot)
        self.setCursor(Qt.CursorShape.PointingHandCursor)

    def mouseReleaseEvent(self, event):
        if event.button() == Qt.MouseButton.LeftButton:
            self.clicked.emit()
        super().mouseReleaseEvent(event)


class Tile(Card):
    """Stat tile: title, one large value, a muted detail line."""

    def __init__(self, title):
        super().__init__(title)
        self.value = QLabel(NA)
        f = self.value.font()
        f.setPointSizeF(f.pointSizeF() * 2.2)
        f.setWeight(QFont.Weight.DemiBold)
        self.value.setFont(f)
        self.detail = muted(QLabel())
        self.detail.setWordWrap(True)
        self.body.addWidget(self.value)
        self.body.addWidget(self.detail)
        self.body.addStretch()

    def set(self, value, detail=""):
        self.value.setText(value)
        self.detail.setText(detail)


class StackedBar(QWidget):
    """Horizontal part-to-whole bar; hover a segment for its label and count."""

    def __init__(self):
        super().__init__()
        self.parts = []  # (label, count, QColor)
        self.setFixedHeight(14)

    def set_parts(self, parts):
        self.parts = [p for p in parts if p[1]]
        self.update()

    def rects(self):
        total = sum(c for _, c, _ in self.parts)
        if not total:
            return []
        room = self.width() - GAP * (len(self.parts) - 1)
        out, x = [], 0.0
        for label, count, color in self.parts:
            w = max(2.0, room * count / total)
            out.append((QRectF(x, 0, w, self.height()), label, count, color))
            x += w + GAP
        return out

    def paintEvent(self, _event):
        p = QPainter(self)
        p.setRenderHint(QPainter.RenderHint.Antialiasing)
        p.setPen(Qt.PenStyle.NoPen)
        for rect, _label, _count, color in self.rects():
            p.setBrush(color)
            p.drawRoundedRect(rect, 4, 4)

    def event(self, e):
        if e.type() == QEvent.Type.ToolTip:
            hit = next((f"{label}: {count:,}" for rect, label, count, _ in self.rects()
                        if rect.contains(e.position())), None)
            if hit:
                QToolTip.showText(e.globalPosition().toPoint(), hit, self)
            else:
                QToolTip.hideText()
            return True
        return super().event(e)


class Dashboard(QWidget):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setStyleSheet("QFrame#card { background: palette(base);"
                           " border: 1px solid palette(mid); border-radius: 10px; }")
        self.tiles = {name: Tile(name) for name in ("Packages", "Built", "Last pull", "VM load")}

        self.results = Card("slackrepo results")
        self.bar = StackedBar()
        self.legend = QLabel(NA)
        self.legend.setTextFormat(Qt.TextFormat.RichText)
        self.legend.setWordWrap(True)
        self.results.body.addWidget(self.bar)
        self.results.body.addWidget(self.legend)
        self.results.body.addStretch()

        self.orphans = Card("SBo orphans")
        self.orphans.body.addWidget(muted(QLabel("coming in phase 5")))
        self.orphans.body.addStretch()

        self.jobs = Card("Recent jobs")
        self.jobs_list = QLabel("no jobs yet")
        self.jobs_list.setTextFormat(Qt.TextFormat.RichText)
        self.jobs.body.addWidget(self.jobs_list)
        self.jobs.body.addStretch()

        self.stale = muted(QLabel())
        self.refresh_btn = QPushButton("Refresh")
        top = QHBoxLayout()
        top.addWidget(self.stale)
        top.addStretch()
        top.addWidget(self.refresh_btn)

        grid = QGridLayout()
        grid.setSpacing(12)
        for i, tile in enumerate(self.tiles.values()):
            grid.addWidget(tile, 0, i)
        grid.addWidget(self.results, 1, 0, 1, 3)
        grid.addWidget(self.orphans, 1, 3)
        grid.addWidget(self.jobs, 2, 0, 1, 4)
        grid.setRowStretch(2, 1)
        layout = QVBoxLayout(self)
        layout.setContentsMargins(16, 12, 16, 16)
        layout.addLayout(top)
        layout.addLayout(grid)

    def cards(self):
        return [*self.tiles.values(), self.results, self.orphans, self.jobs]

    def show_stats(self, s):
        p, sr, sync = s.get("packages"), s.get("slackrepo"), s.get("sync") or {}
        t = self.tiles
        if p is None:
            t["Packages"].set(NA)
        else:
            extra = " · ".join(f"{num(v)}\u00a0{k}" for k, v in p.items() if k != "total")
            t["Packages"].set(num(p["total"]), extra)
        if sr is None:
            t["Built"].set(NA, "no slackrepo database")
        else:
            never = "" if p is None else f" · {num(p['total'] - sr['built'])} never built"
            t["Built"].set(num(sr["built"]), f"of {num(p and p['total'])}{never}")
        known = [ts for ts in sync.values() if ts is not None]
        t["Last pull"].set(fmt_age(min(known)) if known else NA,
                           " · ".join(f"{o} {fmt_duration(None if ts is None else time.time() - ts)}"
                                      for o, ts in sync.items()))
        load, disk = s.get("load"), s.get("disk_free") or {}
        t["VM load"].set(NA if load is None else f"{load[0]:.2f}",
                         f"up {fmt_duration(s.get('uptime'))} · {fmt_bytes(disk.get('repo'))} free")
        self._show_results(sr)

    def _show_results(self, sr):
        if sr is None:
            self.bar.set_parts([])
            self.legend.setText(NA)
            return
        gray = self.palette().color(QPalette.ColorRole.PlaceholderText)  # Mid can match the background
        order = sorted(sr["results"].items(),
                       key=lambda kv: (list(STATUS).index(kv[0]) if kv[0] in STATUS else 99, kv[0]))
        self.bar.set_parts([(k, v, QColor(STATUS.get(k)) if k in STATUS else gray) for k, v in order])
        self.legend.setText("&nbsp;&nbsp;&nbsp;".join(
            f'<span style="color:{STATUS.get(k, gray.name())}">&#9632;</span> {html.escape(k)} {num(v)}'
            for k, v in order))

    def show_jobs(self, jobs):
        rows = []
        for j in jobs[:10]:
            color = STATUS.get(j["status"])
            icon = ICON.get(j["status"], "?")
            if color:
                icon = f'<span style="color:{color}">{icon}</span>'
            dur = None if "start" not in j else j.get("end", time.time()) - j["start"]
            rows.append(f"<tr><td>{icon}&nbsp;</td><td>{html.escape(job_title(j))}&nbsp;&nbsp;</td>"
                        f"<td>{j['status']}&nbsp;&nbsp;</td>"
                        f"<td>{fmt_time(j.get('start', j['created']))}&nbsp;&nbsp;</td>"
                        f"<td>{fmt_duration(dur)}</td></tr>")
        self.jobs_list.setText(f"<table>{''.join(rows)}</table>" if rows else "no jobs yet")

    def set_stale(self, since):
        """Grey the cards when the VM stopped answering; None clears it."""
        for card in self.cards():
            card.setEnabled(since is None)
        self.stale.setText("" if since is None else
                           f"stale since {time.strftime('%H:%M', time.localtime(since))}")
````

Replace `sb_builder/mainwindow.py`:

````python
# sb-builder - main window
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Main window: tabs, status bar, connection polling, VM control, notifications."""

import time

from PyQt6.QtCore import QMetaType, QObject, QTimer, pyqtSignal, pyqtSlot
from PyQt6.QtDBus import QDBusArgument, QDBusConnection, QDBusInterface, QDBusMessage
from PyQt6.QtWidgets import QLabel, QMainWindow, QMessageBox, QPushButton, QTabBar, QTabWidget

from . import VERSION, config
from .dashboard import Dashboard
from .fmt import job_title
from .jobs import JobsTab
from .packages import PackagesTab
from .remote import DOWN, MISSING

NOTIFY = "org.freedesktop.Notifications"
NOTIFY_PATH = "/org/freedesktop/Notifications"
END_EVENT = {"ok": "finished", "failed": "failed", "killed": "killed", "lost": "lost"}
MAX_RETRY_MS = 30000


class Notifier(QObject):
    """freedesktop notifications over D-Bus; clicked(job id) on the default action."""

    clicked = pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent)
        bus = QDBusConnection.sessionBus()
        self.iface = QDBusInterface(NOTIFY, NOTIFY_PATH, NOTIFY, bus)
        bus.connect(NOTIFY, NOTIFY_PATH, NOTIFY, "ActionInvoked", self._action)
        self.ids = {}  # notification id -> job id

    def send(self, title, body, jid=""):
        reply = self.iface.call(
            "Notify", "sb-builder", QDBusArgument(0, QMetaType.Type.UInt.value), "", title, body,
            QDBusArgument(["default", "Open"], QMetaType.Type.QStringList.value), {}, -1)
        if reply.type() == QDBusMessage.MessageType.ReplyMessage:
            self.ids[reply.arguments()[0]] = jid

    @pyqtSlot("uint", "QString")
    def _action(self, nid, _key):
        if nid in self.ids:
            self.clicked.emit(self.ids.pop(nid))


class MainWindow(QMainWindow):
    def __init__(self, cfg, remote, notifier=None):
        super().__init__()
        self.cfg, self.remote = cfg, remote
        self.notifier = notifier or Notifier(self)
        self.notifier.clicked.connect(self.open_job)
        self.jobs_tab = None
        self.packages_tab = None
        self.seen = None        # job id -> status at the last poll; None before the first
        self.online = None      # None until the first answer
        self.overlays = []
        self.vm_action = None   # "start" or "resume" offered on the VM button
        self.vm_woken = False   # we started the VM: notify once ssh answers
        self.running_id = None
        self.setWindowTitle("sb-builder")

        self.tabs = QTabWidget(tabsClosable=True)
        self.tabs.tabCloseRequested.connect(self._close_tab)
        self.tabs.currentChanged.connect(lambda _i: self.poll_stats())
        self.dashboard = Dashboard()
        self.dashboard.refresh_btn.clicked.connect(self.poll_stats)
        self.dashboard.tiles["Packages"].link(self.open_packages)
        self.dashboard.jobs.link(self.open_jobs)
        self.tabs.addTab(self.dashboard, "Dashboard")
        self.tabs.tabBar().setTabButton(0, QTabBar.ButtonPosition.RightSide, None)
        self.setCentralWidget(self.tabs)

        file_menu = self.menuBar().addMenu("&File")
        file_menu.addAction("&Reload config", self.reload_config)
        file_menu.addAction("&Quit", self.close).setShortcut("Ctrl+Q")
        open_menu = self.menuBar().addMenu("&Open")
        open_menu.addAction("&Packages", self.open_packages)
        open_menu.addAction("&Jobs", self.open_jobs)

        self.conn = QLabel("● connecting")
        self.key = QLabel()
        self.key.hide()
        self.vm_btn = QPushButton()
        self.vm_btn.clicked.connect(lambda: self._vm_do(self.vm_action))
        self.vm_btn.hide()
        self.job_btn = QPushButton(flat=True)
        self.job_btn.clicked.connect(lambda: self.open_job(self.running_id))
        self.job_btn.hide()
        for w in (self.job_btn, self.key, self.vm_btn, self.conn):
            self.statusBar().addPermanentWidget(w)

        self.jobs_timer = QTimer(self, timeout=self.poll_jobs)
        self.stats_timer = QTimer(self, timeout=self.poll_slow)
        self._start_timers()
        QTimer.singleShot(0, self.poll_jobs)

    # ── polling ─────────────────────────────────────────────────────────────

    def _start_timers(self):
        self.jobs_timer.start(self.cfg["refresh"]["jobs"] * 1000)
        self.stats_timer.start(self.cfg["refresh"]["dashboard"] * 1000)

    def poll_jobs(self):
        self.remote.call(self._got_jobs, "jobs")

    def poll_slow(self):
        self.poll_stats()
        if self.online:
            self.remote.call(self._got_key, "gpg-status")

    def _got_key(self, data, err):
        if err:
            return self._failed(err)
        self.key.setVisible(data["unlocked"] is not None)
        self.key.setText("🔓 key unlocked" if data["unlocked"] else "🔒 key locked")
        self.key.setToolTip(f"package signing key {data['key']}")

    def poll_stats(self):
        if self.online and self.tabs.currentWidget() is self.dashboard:
            self.remote.call(self._got_stats, "stats")

    def _got_jobs(self, data, err):
        if err:
            return self._failed(err)
        self._up()
        jobs = data["jobs"]
        self._notify_ends(jobs)
        self.dashboard.show_jobs(jobs)
        if self.jobs_tab:
            self.jobs_tab.set_jobs(jobs)
        run = next((j for j in jobs if j["status"] == "running"), None)
        self.running_id = run and run["id"]
        self.job_btn.setVisible(run is not None)
        if run:
            self.job_btn.setText(f"▶ {job_title(run)}")

    def _got_stats(self, data, err):
        if err:
            return self._failed(err)
        self.dashboard.show_stats(data)

    def _got_version(self, data, err):
        if err:
            return self._failed(err)
        self.overlays = data.get("overlays", [])
        if self.jobs_tab:
            self.jobs_tab.set_overlays(self.overlays)
        if data["version"] != VERSION:
            self.conn.setText(f"● helper {data['version']}, GUI {VERSION}: run ./install.sh --remote")

    def _up(self):
        if self.online:
            return
        self.online = True
        self.conn.setText("● connected")
        self.vm_btn.hide()
        self.dashboard.set_stale(None)
        self.jobs_timer.setInterval(self.cfg["refresh"]["jobs"] * 1000)
        self.remote.call(self._got_version, "version")
        self.poll_slow()
        if self.vm_woken:
            self.vm_woken = False
            self.notify("vm-up", f"{self.cfg['vm']['domain']} is up")

    def _failed(self, err):
        if err == MISSING:
            self.conn.setText("● sbb-remote missing: run ./install.sh --remote")
            return
        if err != DOWN:
            self.statusBar().showMessage(err, 10000)
            return
        if self.online is not False:
            self.online = False
            self.dashboard.set_stale(time.time())
        base = self.cfg["refresh"]["jobs"] * 1000
        self.jobs_timer.setInterval(min(self.jobs_timer.interval() * 2, max(MAX_RETRY_MS, base)))
        self._check_vm()

    # ── VM control ──────────────────────────────────────────────────────────

    def _check_vm(self):
        if not self.cfg["vm"]["domain"]:
            self.conn.setText("● down")
            return
        self.remote.virsh(self._got_domstate, "domstate", self.cfg["vm"]["domain"])

    def _got_domstate(self, rc, state):
        if self.online:
            return  # ssh came back while virsh ran
        self.vm_btn.hide()
        if rc != 0:
            self.conn.setText("● down")
        elif state == "shut off":
            self.conn.setText("● VM off")
            if self.cfg["vm"]["autostart"]:
                self._vm_do("start")
            else:
                self._offer("Start VM", "start")
        elif state == "paused":
            self.conn.setText("● VM paused")
            self._offer("Resume VM", "resume")
        elif state == "running":
            self.conn.setText("● VM running, ssh unreachable")
        else:
            self.conn.setText(f"● VM {state}")

    def _offer(self, label, action):
        self.vm_action = action
        self.vm_btn.setText(label)
        self.vm_btn.show()

    def _vm_do(self, action):
        self.vm_btn.hide()
        self.vm_woken = True
        self.conn.setText(f"● VM {action} requested")
        self.remote.virsh(
            lambda rc, out: rc and self.statusBar().showMessage(f"virsh {action}: {out}", 10000),
            action, self.cfg["vm"]["domain"])

    # ── notifications ───────────────────────────────────────────────────────

    def notify(self, event, body, jid=""):
        n = self.cfg["notify"]
        if n["enabled"] and event in n["events"]:
            self.notifier.send(f"sb-builder: {event}", body, jid)

    def _notify_ends(self, jobs):
        prev, self.seen = self.seen, {j["id"]: j["status"] for j in jobs}
        if prev is None:
            return  # first poll: old jobs are history, not news
        for j in jobs:
            event = END_EVENT.get(j["status"])
            if event and prev.get(j["id"], "running") != j["status"]:
                self.notify(event, job_title(j), j["id"])
                if j["kind"] in ("slackrepo", "publish") and self.packages_tab:
                    self.packages_tab.reload()  # build state changed

    # ── tabs ────────────────────────────────────────────────────────────────

    def open_jobs(self):
        if self.jobs_tab is None:
            self.jobs_tab = JobsTab(self.remote, self.cfg["builds"]["publish"])
            self.jobs_tab.refresh_requested.connect(self.poll_jobs)
            self.jobs_tab.set_overlays(self.overlays)
            self.tabs.addTab(self.jobs_tab, "Jobs")
            self.poll_jobs()
        self.tabs.setCurrentWidget(self.jobs_tab)
        return self.jobs_tab

    def open_packages(self):
        if self.packages_tab is None:
            self.packages_tab = PackagesTab(self.remote)
            self.packages_tab.start_requested.connect(lambda args: self.open_jobs().start(*args))
            self.tabs.addTab(self.packages_tab, "Packages")
            self.packages_tab.reload()
        self.tabs.setCurrentWidget(self.packages_tab)
        return self.packages_tab

    def open_job(self, jid):
        self.show()
        self.raise_()
        self.activateWindow()
        if jid:
            self.open_jobs().select(jid)

    def _close_tab(self, i):
        w = self.tabs.widget(i)
        if w is self.dashboard:
            return
        self.tabs.removeTab(i)
        if w is self.jobs_tab:
            self.jobs_tab = None
            w.stop()
        if w is self.packages_tab:
            self.packages_tab = None
        w.deleteLater()

    def reload_config(self):
        cfg, err = config.load()
        if err:
            QMessageBox.warning(self, "Config error", f"{err}\nUsing defaults.")
        self.cfg.clear()
        self.cfg.update(cfg)  # Remote holds the same dict
        self._start_timers()
````

- [ ] **Step 7: Run the whole suite**

Run: `python3 -m unittest discover -s tests </dev/null`
Expected: `Ran 68 tests`, `OK`. "QProcess: Destroyed while process ("true") is still running" lines are the fake follow processes killed at cleanup.

- [ ] **Step 8: Commit**

```bash
git add sb_builder config.example.toml tests/test_gui.py
git commit -F - <<'EOF'
Add Packages tab, hint view, Run it and signing key unlock

Packages tab over the helper inventory: query bar with cat:/maint:/
req:/status: prefixes, overlay and status chips, check column, and a
detail pane with versions, dependencies after hints and a structured
read-only view of the hint (hints.py mirrors slackrepo's parser).
Actions start slackrepo previews; the Jobs tab offers Run it with a
publish box (default from [builds] publish), asking for the signing
key passphrase first when the key is locked. Status bar shows the key
state; Dashboard tiles open their tabs; inventory reloads after
slackrepo and publish jobs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
EOF
git log --format='%h %G? %s' -1
```

---

### Task 4: docs

- [ ] **Step 1:** Replace `README.md`:

````markdown
# sb-builder

Desktop GUI for managing a Slackware SlackBuilds build VM from one place. It
runs slackrepo builds, overlay syncs (sbpull, sbshadow), hintfile work
(mkhint) and SBo orphan discovery on the VM over ssh, as background jobs that
keep running when the GUI disconnects and can be reattached later. A
dashboard shows VM and repository state, and the GUI can start the VM through
libvirt when it is off.

Work in progress. See `docs/superpowers/specs/` for the design.

## Requirements

- Desktop: Python 3.11 or newer, PyQt6 (with QtDBus), ssh access to the VM,
  optionally `virsh` for starting the VM
- VM: Python 3.11 or newer, root ssh login, the
  [slackrepo_setup](https://forge.slackware.nl/danix/slackrepo_setup) tools (`sbpull`, `sbshadow`)
  and mkhint

## Install

    ./install.sh             # ~/bin/sb-builder -> this repo, GUI config if missing
    ./install.sh --remote    # helper to the VM, its config if missing (asks first)

`BINDIR` changes the launcher directory, `SBB_TARGET` the ssh destination for
`--remote` (default `buildsystem`). Configs:

- GUI: `~/.config/sb-builder/config.toml`, see `config.example.toml`
- VM: `/etc/sb-builder/config.toml`, see `remote/config.example.toml`

Every key is optional. `File > Reload config` rereads the GUI config.

## Usage

    sb-builder

- **Dashboard**: VM load, uptime and free disk, package counts, slackrepo
  build results, last overlay pull, recent jobs. Refreshes while visible.
- **Open > Jobs**: start a job from the Start menu (`sbpull <overlay>`
  preview, `sbshadow` preview or apply), follow its log live, kill it. After a
  preview that lists `shadow:` or `unshadow:` lines, "Apply shadowing" runs
  `sbshadow --yes`.

- **Open > Packages** (or click the Packages tile): every package of the
  repository, with built version and status (`ok`, `failed`, `never`,
  `outdated`). Search by name or summary, or with `cat:`, `maint:`, `req:`,
  `status:`; toggle the overlay and status chips. The detail pane shows the
  package's versions, its dependencies after hints (dropped ones struck
  through), and what its hint file does, section by section.
- Tick packages (or select one row) and press Build, Update, Rebuild or
  Remove; Update all updates everything. Each first runs a slackrepo
  `--preview` job. When the preview is fine, **Run it** in the Jobs tab starts
  the real run; untick **publish** to keep the packages local and press
  **Publish** later to push them.
- Real runs sign the repository. When the package signing key is locked
  (status bar 🔒), the GUI asks for its passphrase first and hands it to the
  VM's gpg-agent; it is never stored.

Jobs run on the VM detached from the ssh session, one at a time. Closing the
GUI or losing the connection does not stop them; reopening the job resumes
the log. Job state and logs live in `/var/lib/sb-builder/jobs/` on the VM.

When ssh fails and `vm.domain` is set, the status bar shows the libvirt state
of the VM and offers to start or resume it. Desktop notifications report
finished, failed, killed and lost jobs, and the VM coming up.

## License

Copyright (C) 2026 Danilo M. <danix@danix.xyz>

GPLv2 only. See [LICENSE](LICENSE) for the full text.

## Development Approach

This project is developed using AI-assisted tools. Code is generated with the help of AI based on human-provided specifications, design decisions, and iterative feedback.

All contributions are reviewed, tested, and curated by the maintainer before being included in the codebase. AI is used as a productivity and exploration tool, while human oversight remains central to all decisions.

The goal is to combine the flexibility of AI-assisted development with standard open-source practices such as transparency, review, and accountability.
````

- [ ] **Step 2:** Replace `AGENTS.md`:

````markdown
# AGENTS.md

Guidance for AI coding agents working in this repository.

## What this is

`sb-builder`: a PyQt6 desktop GUI that drives the SlackBuild tooling on a
Slackware -current build VM (slackrepo, mkhint, sbpull, sbshadow) over ssh.
Two programs: the GUI (`sb_builder/`, PyQt6 incl. QtDBus, stdlib otherwise)
and `remote/sbb-remote`, a single-file stdlib-only Python helper deployed on
the VM. Do not add third-party dependencies to either.

Design and phasing: `docs/superpowers/specs/`. Phase 1 (core) is
`2026-10-19-sb-builder-phase1-design.md`.

## How it works

- Every GUI action is `ssh <target> sbb-remote <cmd> [args]`; the helper
  answers with JSON. `sb_builder/remote.py` is the only code that runs ssh
  (and local `virsh`).
- Trust boundary: the helper runs only a fixed table of job kinds and builds
  argv itself from validated args. Never pass shell strings or paths from the
  GUI to the helper.
- Jobs run detached on the VM (`setsid`, `stdin=/dev/null`), one at a time
  under a flock, logging to `<state_dir>/jobs/<id>/`. The GUI can disconnect
  and reattach with `job-follow <id> <offset>`.
- Exception, by design: `gpg-unlock` reads the signing key passphrase from
  stdin (`remote.call(..., stdin=...)`) and feeds it to
  `gpg-preset-passphrase`. Every other subprocess in the helper runs with
  `stdin=DEVNULL`, so nothing can read the passphrase off the ssh channel.
  Never put the passphrase in argv, env, logs or files.
- Hints are parsed (`shlex`), never sourced. `sb_builder/hints.py` mirrors
  slackrepo's own rules (`../GITHUB/slackrepo/libexec/functions.d/parsefunctions.sh`);
  check that file when changing it.
- Jobs never get input. Any tool run as a job must be non-interactive
  (`--yes`/`--no` style flags). If a tool needs a flag it lacks, add it in
  that tool's own repo (e.g. `slackrepo_setup` for sbshadow, `mkhintfile`
  for mkhint) rather than working around it here.

## Conventions

- GUI config: `$XDG_CONFIG_HOME/sb-builder/config.toml`. Helper config:
  `/etc/sb-builder/config.toml` on the VM. Both read with `tomllib`, both
  have shipped `config.example.toml` files, defaults live in code.
- One file per GUI tab.
- Deploying to the VM is outward-facing: confirm before `install.sh --remote`.
- Placeholders only in code, docs, tests and commits.

## Testing

    python3 -m unittest discover -s tests

`test_gui.py` forces `QT_QPA_PLATFORM=offscreen` (the desktop session sets
its own, and a modal dialog on the real display blocks the run). A fake
remote must answer every call a test triggers: an unanswered call returns
"ssh: connection failed", which opens a warning dialog that blocks forever
offscreen.

Headless (`test_gui.py` sets `QT_QPA_PLATFORM=offscreen`), no VM needed.
`test_remote.py` loads `remote/sbb-remote` with `SourceFileLoader` (no `.py`
extension) and points it at a temp state dir through `SBB_CONFIG`. Two
`ResourceWarning: subprocess ... is still running` lines are expected: the
detached `_run` is never waited for. To try the helper on the VM without
installing it, copy it to `/tmp` there and run it with `SBB_CONFIG` pointing
at a config whose `state_dir` is under `/tmp`.

## Repo facts

- License: GPLv2 only (`LICENSE`, header in each source file).
- Work directly on master.
- `.superpowers/` is brainstorming scratch, ignored.
````

- [ ] **Step 3: Commit and push**

```bash
git add README.md AGENTS.md
git commit -F - <<'EOF'
Document the Packages tab, publishing and signing key unlock

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
EOF
git log --format='%h %G? %s' -1 && git push
```

---

### Task 5: deploy and first real run (with the user)

Ask before each VM step. Order matters: the `.local` file must be built from the old config before the new config replaces it.

- [ ] **Step 1: Inspect targets**

```bash
ssh buildsystem 'ls -l /etc/slackrepo/slackrepo_SBo-danix.conf /etc/slackrepo/SBo-danix.local /usr/local/bin/sb-publish /root/.gnupg/gpg-agent.conf 2>&1'
```

Expected: config and agent conf exist; `.local` and `sb-publish` do not.

- [ ] **Step 2: Build SBo-danix.local on the VM from the old config** (values never leave the VM)

```bash
cd ~/Programming/GIT/slackrepo_setup
scp make-local.py buildsystem:/tmp/make-local.py
ssh buildsystem 'cp -p /etc/slackrepo/slackrepo_SBo-danix.conf /etc/slackrepo/slackrepo_SBo-danix.conf.pre-sb-builder &&
  python3 /tmp/make-local.py /etc/slackrepo/slackrepo_SBo-danix.conf /etc/slackrepo/SBo-danix.local &&
  rm /tmp/make-local.py && stat -c "%a %n" /etc/slackrepo/SBo-danix.local'
```

Expected: `wrote /etc/slackrepo/SBo-danix.local: SB_RSYNC_TARGET, SB_RSYNC_EXCLUDES, SB_RSYNC_PASSWORD_FILE, SB_GOTIFY_KEY, SB_GOTIFY_URL` and `600 /etc/slackrepo/SBo-danix.local`.

- [ ] **Step 3: Install config, sb-publish, agent conf**

```bash
cd ~/Programming/GIT/slackrepo_setup
scp slackrepo_SBo-danix.conf buildsystem:/etc/slackrepo/slackrepo_SBo-danix.conf
scp sb-publish buildsystem:/usr/local/bin/sb-publish
scp gpg-agent.conf buildsystem:/root/.gnupg/gpg-agent.conf
ssh buildsystem 'chmod 755 /usr/local/bin/sb-publish; chmod 600 /root/.gnupg/gpg-agent.conf;
  gpgconf --reload gpg-agent; bash -n /etc/slackrepo/slackrepo_SBo-danix.conf && echo config ok;
  bash -c ". /etc/slackrepo/SBo-danix.local; [ -n \"\$SB_RSYNC_TARGET\" ] && echo local ok"'
```

- [ ] **Step 4: Deploy the helper**: `cd ~/Programming/GIT/sb-builder && ./install.sh --remote`. Expected last line: `{"version": "0.2.0", "overlays": ["personal", "pentesting"]}`. Existing `/etc/sb-builder/config.toml` is kept; its defaults cover the new keys.

- [ ] **Step 5: User runs `sb-builder`**, checks: status bar shows 🔒 key locked; Packages tile opens the tab with ~9957 rows; chips and `cat:personal` filter; a hinted package shows its hint sections; DELREQUIRES deps struck through.

- [ ] **Step 6: Preview, run unpublished, publish**: pick one personal package that shows `outdated`, Update. Preview job ends ✔ with [Run it]; untick publish; Run it asks for the passphrase (a wrong one asks again); the job runs and ends ✔; status bar shows 🔓. On the VM: `ls -l /repo/personal/<pkg>/` shows the new package and the mirror is not updated yet. Then Publish: the `publish` job runs `sb-publish` and ends ✔.

- [ ] **Step 7: Note the preset passphrase lifetime**: check the 🔓 indicator again after more than 4h (default-cache-ttl) and after 6h (max-cache-ttl); record in the spec whether presets expire with the TTLs. Either way the GUI asks again when locked.

- [ ] **Step 8: Terminal check**: from a VM shell, `slackrepo --preview update <pkg>`: no Gotify message arrives (was sent before this change).

Record anything that misbehaves as a follow-up; do not patch it silently.