#!/usr/bin/env python3 """ka-install — compute install plan for a kernel-agent host. Third of the three writing verbs (ka-promote → ka-build → ka-install). Read-only: never touches SSH, scp, pacman, or reboot. Usage: ka-install ka-install --render | --dry-run ka-install --version ka-install -h | --help Exit codes: 0 plan printed successfully (no drift) 2 missing input (no manifest.lock, no fleet manifest) 3 drift detected (build receipt missing or empty) 4 manifest parse / schema error --render / --dry-run: Compute and print the install plan that ka-install would execute, without touching any host state. This is the default behaviour — the script is always read-only. --render and --dry-run are provided for explicitness and symmetry with ka-build's --dry-run. Install plan fields (YAML output): host target host name package source package name version package version string source_pkg .pkg.tar.zst filename in build///pkgs/ backup_path hertz backup destination for the replaced package kernel_image detected kernel filename (Image or vmlinuz) post_install_hook mkinitcpio trigger command drift true if build receipt is missing or empty steps structured list of planned operations """ import argparse import os import re import sys import yaml VERSION = 1 def die(msg, code=1): print(f"ka-install: error: {msg}", file=sys.stderr) sys.exit(code) def find_repo_root(): here = os.path.dirname(os.path.abspath(__file__)) root = os.path.dirname(here) if not os.path.isdir(os.path.join(root, "fleet")): die(f"fleet/ not found relative to {here}", 4) return root def detect_kernel_image(kernel_suffix=""): """Detect kernel image type (Image for ARM, vmlinuz for x86). All kernel-agent targets are currently ARM64 → Image. """ suffix = (kernel_suffix or "").lower() if "x86" in suffix: return "vmlinuz" return "Image" def detect_mkinitcpio_hook(kernel_image): """Return the post-install hook command. mkinitcpio's stock hook watches vmlinuz by default. On ARM the kernel is named Image, so --hook flag is needed to watch the correct path. """ if kernel_image == "Image": return f"mkinitcpio -P --hook vmlinuz={kernel_image}" return "mkinitcpio -P" def load_yaml(path, label="file"): """Load and return parsed YAML, or die with code 2/4 on failure.""" try: with open(path) as f: data = yaml.safe_load(f) except FileNotFoundError: die(f"{label} not found: {path}", 2) except yaml.YAMLError as e: die(f"{label} parse error: {e}", 4) if not isinstance(data, dict): die(f"{label} root must be a mapping: {path}", 4) return data def compute_plan(host, repo_root, build_dir=None): """Compute the install plan for *host*. Returns a dict with all plan fields. Raises SystemExit on fatal errors (missing inputs, parse failures). """ build_dir = build_dir or os.path.join(repo_root, "build") fleet_dir = os.path.join(repo_root, "fleet") manifest_path = os.path.join(fleet_dir, f"{host}.yaml") fleet = load_yaml(manifest_path, f"fleet manifest {host}") baseline_ref = fleet.get("baseline", {}).get("ref") if not baseline_ref: die(f"fleet manifest missing baseline.ref: {manifest_path}", 4) lock_path = os.path.join(build_dir, host, baseline_ref, "manifest.lock") lock = load_yaml(lock_path, f"manifest.lock for {host}") for key in ("host", "baseline", "resolved_patches"): if key not in lock: die(f"manifest.lock missing required key '{key}': {lock_path}", 4) pkg_info = fleet.get("package", {}) pkg_name = pkg_info.get("name", "unknown") versioning = pkg_info.get("versioning", "${baseline_ref}") kernel_suffix = pkg_info.get("kernel_suffix", "") boot_path = pkg_info.get("boot_path", "/boot/") bootloader = pkg_info.get("bootloader", "extlinux") bootloader_path = pkg_info.get("bootloader_path", "/boot/extlinux/extlinux.conf") install_mode = pkg_info.get("install_mode", "alongside") # --- Drift detection: is there a build receipt? --- build_receipt = lock.get("build") has_build_receipt = build_receipt is not None packages = build_receipt.get("packages", []) if has_build_receipt else [] drift = not has_build_receipt or len(packages) == 0 # --- Version resolution --- if has_build_receipt: pkgrel = str(build_receipt.get("ka_build_version", "1")) else: pkgrel = "?" version = versioning.replace("${baseline_ref}", baseline_ref).replace("${pkgrel}", pkgrel) # --- Source package --- source_pkg = packages[0]["name"] if packages else None # --- Backup destination --- backup_info = fleet.get("backup", {}).get("pre_install", "") if backup_info: backup_path = backup_info.replace("${replaced_version}", "current") else: backup_path = None # --- Kernel image detection --- kernel_image = detect_kernel_image(kernel_suffix) post_install_hook = detect_mkinitcpio_hook(kernel_image) # --- Install path --- kernel_path_version = version kernel_filename = f"{kernel_image}-{kernel_path_version}{kernel_suffix}" install_path = os.path.join(boot_path.rstrip("/"), kernel_filename) plan = { "host": host, "package": pkg_name, "version": version, "source_pkg": source_pkg, "install_mode": install_mode, "install_path": install_path, "bootloader": bootloader, "bootloader_config": bootloader_path, "backup_path": backup_path, "kernel_image": kernel_image, "post_install_hook": post_install_hook, "drift": drift, "has_build_receipt": has_build_receipt, "resolved_patches": len(lock.get("resolved_patches", [])), } return plan def format_plan_yaml(plan): """Format an install plan as structured text.""" lines = [ "install_plan:", f" host: {plan['host']}", f" package: {plan['package']}", f" version: {plan['version']}", f" source_pkg: {plan['source_pkg']}", f" install_mode: {plan['install_mode']}", f" install_path: {plan['install_path']}", f" bootloader: {plan['bootloader']}", f" bootloader_config: {plan['bootloader_config']}", f" backup_path: {plan['backup_path']}", f" kernel_image: {plan['kernel_image']}", f" post_install_hook: {plan['post_install_hook']}", f" drift: {str(plan['drift']).lower()}", f" has_build_receipt: {str(plan['has_build_receipt']).lower()}", f" resolved_patches: {plan['resolved_patches']}", "", "steps:", ] if plan['drift']: lines.append(" - warning: drift detected — manifest.lock has no build receipt") lines.append(" - action: run 'ka-build ' first to produce a build receipt") else: lines.append(f" - action: backup current kernel on {plan['host']}") lines.append(f" target: {plan['backup_path']}") lines.append(f" - action: copy {plan['source_pkg']} to {plan['host']}") lines.append(" method: scp") lines.append(f" - action: install package on {plan['host']}") lines.append(" method: pacman -U") lines.append(f" - action: update bootloader config ({plan['bootloader']})") lines.append(f" config: {plan['bootloader_config']}") if plan['post_install_hook']: lines.append(f" - action: run post-install hook") lines.append(f" command: {plan['post_install_hook']}") return "\n".join(lines) def main(): p = argparse.ArgumentParser(prog="ka-install", add_help=True) p.add_argument("host", nargs="?", help="fleet host name (omit with --version)") p.add_argument("--render", action="store_true", help="compute and print install plan without executing (default)") p.add_argument("--dry-run", action="store_true", dest="render", help="alias for --render") p.add_argument("--version", action="store_true", help="print version + exit") args = p.parse_args() if args.version: print(f"ka-install version {VERSION}") return 0 if not args.host: p.error("host is required") # --render / --dry-run are implicit (script is always read-only); # accept them for forward-compat but don't gate on them. repo_root = find_repo_root() plan = compute_plan(args.host, repo_root) print(format_plan_yaml(plan)) return 3 if plan["drift"] else 0 if __name__ == "__main__": sys.exit(main())