sbb-remote 0.5.0: hint jobs replace mkhint's hint editing

hint-update and hint-new run the helper's own _hint command: version
bumps as mkhint does them (also '-' and dotless URL forms, never inside
MD5SUM values), a DELREQUIRES-only hint pinned from the .info first, md5
of the first URL of each DOWNLOAD pair, --no-dl as DOWNLOAD="no", .bak
before each write and a final SBB-RESULT line that job-result returns.
hint-diff, hint-revert and hint-delete answer at once. neovim and
openvino are refused until phase 4b. Job ids accept hyphenated kinds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 281 insertions, 5 deletions

diff --git a/remote/config.example.toml b/remote/config.example.toml
index d275401..a77ace9 100644
--- a/remote/config.example.toml
+++ b/remote/config.example.toml
@@ -12,3 +12,4 @@ hintdir         = "/etc/slackrepo/SBo-danix/hintfiles"
 default_hintdir = ""                     # searched after hintdir when set
 genreprc        = "/root/.genreprc"      # REPOSOWNERGPG names the signing key
 slackrepo_logdir = "/var/log/slackrepo/SBo-danix"
+manifest_packages = ["neovim", "openvino"] # bundled deps: hint-update refuses them (phase 4b)
diff --git a/remote/sbb-remote b/remote/sbb-remote
index 25427c0..2674ed6 100755
--- a/remote/sbb-remote
+++ b/remote/sbb-remote
@@ -25,6 +25,7 @@ its arguments, this file validates them and builds the argv itself.
 
 import fcntl
 import glob
+import hashlib
 import json
 import os
 import re
@@ -37,8 +38,9 @@ import sys
 import tempfile
 import time
 import tomllib
+import urllib.request
 
-VERSION = "0.4.1"
+VERSION = "0.5.0"
 DEFAULTS = {
     "state_dir": "/var/lib/sb-builder",
     "repo_dir": "/var/lib/sbopkg/SBo-danix",
@@ -51,11 +53,13 @@ DEFAULTS = {
     "default_hintdir": "",
     "genreprc": "/root/.genreprc",
     "slackrepo_logdir": "/var/log/slackrepo/SBo-danix",
+    "manifest_packages": ["neovim", "openvino"],  # bundled deps: left to mkhint until phase 4b
 }
 FLAGS = ("--yes", "--no")
-JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-\d+)?$")
+JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-[a-z]+)*(-\d+)?$")
 PKG = re.compile(r"^[^/]+/([^/]+)/\1\.SlackBuild$")
 PKGNAME = re.compile(r"^[A-Za-z0-9._+-]+$")
+VERSION_ARG = re.compile(r"^[A-Za-z0-9._+-]{1,64}$")
 ACTIONS = ("build", "rebuild", "update", "remove")
 PRESET = "/usr/libexec/gpg-preset-passphrase"
 ITEM = re.compile(r"^[A-Za-z0-9._+-]+/[A-Za-z0-9._+-]+$")  # category/pkg
@@ -85,8 +89,9 @@ def load_config(path=None):
         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")
+    for k in ("overlays", "manifest_packages"):
+        if not all(isinstance(o, str) for o in cfg[k]):
+            raise Error(f"config: {k}: expected list of strings")
     return cfg
 
 
@@ -138,8 +143,67 @@ def kind_publish(cfg, args):
     return ["sb-publish"], {}
 
 
+def _pkgmap(cfg):
+    """{package: category/package} of the repo."""
+    return {os.path.basename(d): os.path.relpath(d, cfg["repo_dir"])
+            for d in _pkgdirs(cfg["repo_dir"], "*/*")}
+
+
+def _hint_target(cfg, pkg, pkgs=None):
+    """(category/pkg, <hintdir>/<pkg>.hint); refuses a hint slackrepo finds elsewhere first."""
+    pkgs = pkgs or _pkgmap(cfg)
+    if not PKGNAME.match(pkg or "") or pkg not in pkgs:
+        raise Error(f"no such package: {pkg}")
+    target = os.path.join(cfg["hintdir"], pkg + ".hint")
+    found = find_hint(cfg, pkgs[pkg])
+    if found and found != target:
+        raise Error(f"hint lives at {found}, not managed")
+    return pkgs[pkg], target
+
+
+def hint_args(cfg, kind, args):
+    """Validated ([(pkg, version or None)], no_dl) of a hint-update or hint-new job."""
+    no_dl = "--no-dl" in args
+    for o in (a for a in args if a.startswith("--") and a != "--no-dl"):
+        raise Error(f"{kind}: unknown option {o}")
+    rest = [a for a in args if a != "--no-dl"]
+    if kind == "hint-new":
+        if not 1 <= len(rest) <= 2:
+            raise Error("usage: hint-new <pkg> [<version>] [--no-dl]")
+        pairs = [(rest[0], rest[1] if len(rest) == 2 else None)]
+    else:
+        if not rest:
+            raise Error("hint-update needs at least one pkg=version")
+        if not all("=" in a for a in rest):
+            raise Error("hint-update: expected pkg=version")
+        pairs = [tuple(a.split("=", 1)) for a in rest]
+    pkgs = _pkgmap(cfg)
+    seen = set()
+    for pkg, ver in pairs:
+        if ver is not None and not VERSION_ARG.match(ver):
+            raise Error(f"bad version: {ver}")
+        _, target = _hint_target(cfg, pkg, pkgs)
+        if pkg in seen:
+            raise Error(f"{pkg} named twice")
+        seen.add(pkg)
+        if ver is not None and pkg in cfg["manifest_packages"]:
+            raise Error(f"{pkg}: bundled deps: use mkhint until phase 4b")
+        if kind == "hint-new" and os.path.exists(target):
+            raise Error(f"{pkg}: hint exists")
+    return pairs, no_dl
+
+
+def _self_job(kind):
+    """A job kind run by this file itself (_hint), argv-free like the others."""
+    def build(cfg, args):
+        hint_args(cfg, kind, args)
+        return [sys.executable, os.path.abspath(__file__), "_hint", kind, *args], {}
+    return build
+
+
 KINDS = {"sbpull": kind_sbpull, "sbshadow": kind_sbshadow,
-         "slackrepo": kind_slackrepo, "publish": kind_publish}
+         "slackrepo": kind_slackrepo, "publish": kind_publish,
+         **{k: _self_job(k) for k in ("hint-update", "hint-new")}}
 
 
 def signs(kind, args):
@@ -549,6 +613,212 @@ def cmd_inventory(cfg):
     return {"columns": COLUMNS, "rows": rows}
 
 
+# ── hint text ───────────────────────────────────────────────────────────────
+# Hints are edited as text so comments, order and other keys stay as written.
+
+PIN_KEYS = ("VERSION", "DOWNLOAD", "MD5SUM", "DOWNLOAD_x86_64", "MD5SUM_x86_64")
+DL_PAIRS = (("DOWNLOAD", "MD5SUM"), ("DOWNLOAD_x86_64", "MD5SUM_x86_64"))
+SKIP_URLS = ("UNSUPPORTED", "UNTESTED")
+DL_TIMEOUT = 60
+
+
+def _assignment(key):
+    """KEY=value at line start; a quoted value may span lines. key is a regex."""
+    return re.compile(rf'^(?:{key})=(?:"([^"]*)"|(\S*))[^\n]*\n?', re.M)
+
+
+def hint_tokens(text, key):
+    """Whitespace-separated words of KEY's value, [] when unset or empty."""
+    m = _assignment(key).search(text)
+    if not m:
+        return []
+    value = m.group(1) if m.group(1) is not None else m.group(2)
+    return value.replace("\\\n", " ").split()
+
+
+def hint_set(text, key, value):
+    """KEY="value" in place of the first KEY line, else appended."""
+    line = f'{key}="{value}"\n'
+    m = _assignment(key).search(text)
+    if m:
+        return text[:m.start()] + line + text[m.end():]
+    return text + ("\n" if text and not text.endswith("\n") else "") + line
+
+
+def hint_drop(text, key):
+    return _assignment(key).sub("", text)
+
+
+def swap_version(text, old, new):
+    """mkhint's bump of old to new, everywhere but in MD5SUM* values.
+
+    Also the '-' form when either has '_' (dated URLs), and the dotless form of
+    a dotted old version inside URLs, as a whole number (build_4200 for 4.2.00).
+    """
+    pairs = [(old, new)]
+    if "_" in old or "_" in new:
+        pairs.append((old.replace("_", "-"), new.replace("_", "-")))
+    o, n = old.replace(".", ""), new.replace(".", "")
+    dotless = "." in old and o != n
+    out, pos = [], 0
+    for m in [*_assignment(r"MD5SUM\w*").finditer(text), None]:
+        part = text[pos:m.start() if m else len(text)]
+        for a, b in pairs:
+            part = part.replace(a, b)
+        if dotless:
+            part = re.sub(rf'(://[^\s"]*?)(?<!\d){re.escape(o)}(?!\d)',
+                          lambda x: x.group(1) + n, part)
+        out.append(part)
+        if m:
+            out.append(m.group(0))
+            pos = m.end()
+    return "".join(out)
+
+
+def _pin(info):
+    """The version pin of an .info: its VERSION, DOWNLOAD* and MD5SUM* lines."""
+    lines = [m.group(0) for k in PIN_KEYS if (m := _assignment(k).search(info))]
+    if not lines or not lines[0].startswith("VERSION="):
+        raise Error("no VERSION in the .info")
+    return "".join(ln.rstrip("\n") + "\n" for ln in lines)
+
+
+def new_hint(info):
+    """mkhint's hint from an .info: the pin, REQUIRES commented, ARCH.
+
+    No phantom deps: a package that has them already has a hint, the
+    DELREQUIRES one sbshadow's mkhint -F writes after every sync.
+    """
+    text = _pin(info)
+    m = _assignment("REQUIRES").search(info)
+    if m:
+        text += "#" + m.group(0).rstrip("\n") + "\n"
+    return text + 'ARCH="x86_64"\n'
+
+
+def _md5_url(url):
+    req = urllib.request.Request(url, headers={"User-Agent": "sbb-remote"})
+    h = hashlib.md5()
+    with urllib.request.urlopen(req, timeout=DL_TIMEOUT) as r:
+        for chunk in iter(lambda: r.read(1 << 16), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def apply_version(text, info, new, no_dl, log):
+    """The hint text at version new (None: keep the version), log(str) per download.
+
+    An unpinned hint (DELREQUIRES only) gets the .info pin first. Downloads the
+    first URL of each DOWNLOAD pair for its md5; continuation URLs keep theirs,
+    as in mkhint. no_dl: no download, DOWNLOAD="no" in place of every
+    DOWNLOAD*/MD5SUM* line, since DOWNLOAD_$ARCH would override it.
+    """
+    if not hint_tokens(text, "VERSION"):
+        text = _pin(info) + ("" if hint_tokens(text, "ARCH") else 'ARCH="x86_64"\n') + text
+    if new is not None:
+        text = swap_version(text, hint_tokens(text, "VERSION")[0], new)
+    if no_dl:
+        return hint_drop(text, r"(?:DOWNLOAD|MD5SUM)\w*") + 'DOWNLOAD="no"\n'
+    if new is None:
+        return text
+    for dl, sums in DL_PAIRS:
+        urls = hint_tokens(text, dl)
+        if not urls or urls[0] in SKIP_URLS:
+            continue
+        log(f"downloading {urls[0]}")
+        text = hint_set(text, sums, " \\\n    ".join([_md5_url(urls[0]), *hint_tokens(text, sums)[1:]]))
+    return text
+
+
+# ── hint jobs and commands ──────────────────────────────────────────────────
+
+def _read(path):
+    try:
+        with open(path, errors="replace") as f:
+            return f.read()
+    except FileNotFoundError:
+        return None
+
+
+def _write(path, text):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    tmp = path + ".tmp"
+    with open(tmp, "w") as f:
+        f.write(text)
+    os.replace(tmp, path)
+
+
+def _save(target, text):
+    """Write a hint, its previous text to .bak; a new hint drops a stale .bak (revert = delete)."""
+    if os.path.exists(target):
+        shutil.copy2(target, target + ".bak")
+    elif os.path.exists(target + ".bak"):
+        os.remove(target + ".bak")
+    _write(target, text)
+
+
+def cmd_hint(cfg, kind=None, *args):
+    """Body of the hint jobs: log lines, then SBB-RESULT {"ok": [...], "failed": [[pkg, why]]}."""
+    if kind not in ("hint-update", "hint-new"):
+        raise Error(f"unknown hint job: {kind}")
+    ok, failed = [], []
+    pkgs = _pkgmap(cfg)
+    pairs, no_dl = hint_args(cfg, kind, list(args))
+    for pkg, ver in pairs:
+        try:
+            path, target = _hint_target(cfg, pkg, pkgs)
+            info = _read(os.path.join(cfg["repo_dir"], path, pkg + ".info")) or ""
+            text = _read(target)
+            if text is None:
+                text = new_hint(info)
+            _save(target, apply_version(text, info, ver, no_dl,
+                                        lambda s, p=pkg: print(f"{p}: {s}", flush=True)))
+            print(f"{pkg}: ok", flush=True)
+            ok.append(pkg)
+        except (Error, OSError) as e:
+            print(f"{pkg}: failed: {e}", flush=True)
+            failed.append([pkg, str(e)])
+    print("SBB-RESULT " + json.dumps({"ok": ok, "failed": failed}), flush=True)
+    if failed:
+        raise Error(f"{len(failed)} failed")
+
+
+def cmd_hint_diff(cfg, pkg=None):
+    _, target = _hint_target(cfg, pkg)
+    bak, current = _read(target + ".bak"), _read(target)
+    if bak is None and current is None:
+        raise Error("no backup")
+    return {"bak": bak, "current": current}
+
+
+def cmd_hint_revert(cfg, pkg=None):
+    _, target = _hint_target(cfg, pkg)
+    if os.path.exists(target + ".bak"):
+        os.replace(target + ".bak", target)
+    elif os.path.exists(target):
+        os.remove(target)  # a new hint: reverting it is deleting it
+    else:
+        raise Error("nothing to revert")
+    return {"ok": True}
+
+
+def cmd_hint_delete(cfg, pkg=None):
+    _, target = _hint_target(cfg, pkg)
+    if not os.path.exists(target):
+        raise Error(f"no hint for {pkg}")
+    os.replace(target, target + ".bak")
+    return {"ok": True}
+
+
+def cmd_job_result(cfg, jid=None):
+    """The SBB-RESULT of a hint job's log."""
+    text, _ = _tail(os.path.join(job_dir(cfg, jid), "log"), LOG_TAIL)
+    for line in reversed(text.splitlines()):
+        if line.startswith("SBB-RESULT "):
+            return json.loads(line.split(" ", 1)[1])
+    raise Error(f"no result in {jid}")
+
+
 # ── signing key ─────────────────────────────────────────────────────────────
 
 def _signing_key(cfg):
@@ -626,7 +896,12 @@ COMMANDS = {
     "build-log": cmd_build_log,
     "gpg-status": cmd_gpg_status,
     "gpg-unlock": cmd_gpg_unlock,
+    "hint-diff": cmd_hint_diff,
+    "hint-revert": cmd_hint_revert,
+    "hint-delete": cmd_hint_delete,
+    "job-result": cmd_job_result,
     "_run": cmd_run,
+    "_hint": cmd_hint,
 }