generate.py

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""
BAUER GROUP XPD-RPIImage - variant config renderer.

Reads a JSON variant config, resolves ${ENV} references, validates against
schema.json, and renders all artifacts into the CustomPiOS module tree at
src/modules/<module>/filesystem/root/opt/bgrpiimage/<module>/ plus the variant shell config at
src/variants/<name>/config.

Usage:
    python scripts/generate.py config/variants/canbus-plattform.json
"""
from __future__ import annotations

import argparse
import json
import os
import re
import shlex
import shutil
import stat
import sys
import textwrap
from pathlib import Path
from typing import Any

try:
    import jsonschema
    from rich import box
    from rich.console import Console
    from rich.json import JSON
    from rich.panel import Panel
    from rich.table import Table
except ImportError:
    print("error: missing dependencies. run: pip install -r scripts/requirements.txt", file=sys.stderr)
    sys.exit(2)

# On Windows the default stdout encoding is cp1252 which can't render the
# Unicode glyphs rich uses for borders / status. Reconfigure to UTF-8 and
# disable rich's legacy Win32 renderer so ANSI escapes are used instead.
for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8")
    except (AttributeError, Exception):
        pass

console = Console(highlight=False, legacy_windows=False)
# Diagnostics go to STDERR, never stdout. `--json` writes the resolved config to
# stdout for CI to pipe into jq, and _semantic_validate() runs before that
# branch - so a note printed on the shared console lands INSIDE the JSON and
# makes it unparseable. The build step that reads it (`generate.py --json >
# /tmp/resolved.json`, then `jq .variant.version`) fails with no obvious link
# back to the note that caused it.
err_console = Console(highlight=False, legacy_windows=False, stderr=True)


def _error_panel(title: str, body: str, hint: str | None = None) -> None:
    text = body
    if hint:
        text += f"\n\n[dim]hint:[/] {hint}"
    console.print(Panel(text, title=f"[red]{title}[/]", border_style="red", box=box.ROUNDED))

ROOT = Path(__file__).resolve().parent.parent
CONFIG_DIR = ROOT / "config"
SRC_DIR = ROOT / "src"
MODULES_DIR = SRC_DIR / "modules"
VARIANTS_DIR = SRC_DIR / "variants"
SCHEMA_PATH = CONFIG_DIR / "schema.json"

# systemd leaves tx_queue_len to the driver when unset, and the CAN core
# (can_setup() in drivers/net/can/dev/dev.c) picks 10 - far too small to
# absorb a burst. 1024 is the house value across every CAN interface.
# Sizing note: a classic 8-byte frame is ~111 bits on the wire, so at
# 500 kbit/s a full queue is ~227 ms of backlog. The previous 65535 was
# ~14.5 s - latency no control bus can use, and it never applied anyway.
CAN_TXQUEUELEN_DEFAULT = 1024

# A bus-off controller does not come back on its own. can_bus_off() in
# drivers/net/can/dev/dev.c only queues the recovery work "if
# (priv->restart_ms)", and both the kernel and systemd default that to 0 -
# so the shipped image had no auto-recovery at all and `ip -details link
# show` read "can state ERROR-ACTIVE restart-ms 0". On the MCP2515 the miss
# is worse than a slow recovery: mcp251x.c takes the restart_ms == 0 branch
# to set force_quit and call mcp251x_hw_sleep(), i.e. it puts the chip to
# SLEEP and kills its own ISR loop, defeating the controller's own hardware
# bus-off recovery. Nothing short of an ip link down/up revives it - which
# is exactly why the field report needed an on-site engineer.
# 100 ms is the kernel documentation's own example value. On mcp251x the
# number is effectively a boolean (any non-zero value just declines the
# sleep path and lets the chip self-recover in ~2.8 ms of bus time at
# 500 kbit/s); it only becomes load bearing if the fleet moves to a
# controller that uses the generic restart timer.
CAN_RESTART_MS_DEFAULT = 100

# systemd's RuntimeWatchdogSec: the hardware timeout PID 1 programmes into
# /dev/watchdog, kicked once per main-loop iteration and at least every
# timeout/2.
#
# The schema caps it at 15 and that cap is load-bearing, but NOT because larger
# values are clamped - they are not. bcm2835_wdt_start() arms the hardware with
#     SECS_TO_WDOG_TICKS(wdog->timeout) & PM_WDOG_TIME_SET
# i.e. (t << 16) & 0xfffff, so only the low four bits of the seconds survive:
# the effective timeout is t MOD 16. 20 becomes 4 s, 30 becomes 14 s, and 16
# becomes ZERO - an immediate reset loop. Since the 6.8 fix backported into
# rpi-6.1.y and later (f33f5b1fd1be, "Fix WDIOC_SETTIMEOUT handling") the ioctl
# no longer even returns EINVAL for an out-of-range value, so nothing is logged
# at all. Trixie ships exactly those kernels.
#
# The default is the hardware MAXIMUM rather than something comfortably below
# it, which is the opposite of the usual instinct. Nothing else pings: below
# 16 s the kernel starts no keepalive worker (watchdog_need_worker() needs a
# timeout above max_hw_heartbeat_ms = 15999), so PID 1 is the only kicker and
# there is no safety net behind it. systemd issue 7932 measured 4.3 s of PID 1
# blocked in a single SIGCHLD dispatch under a fork storm - which is a near
# miss against the 5 s budget a timeout of 10 gives, and comfortably absorbed
# by the 7.5 s that 15 gives. The extra margin costs nothing.
WATCHDOG_RUNTIME_SEC_DEFAULT = 15
# RebootWatchdogSec does NOT bound the orderly shutdown, which is the natural
# reading and the wrong one. It arms the watchdog only for the SECOND phase of
# a reboot - after PID 1 has been replaced by systemd-shutdown. Stopping Docker
# and its containers happens in phase one, still governed by RuntimeWatchdogSec
# and the units' own TimeoutStopSec.
#
# Phase two then runs sync_with_progress() (which resets its own attempt
# counter while dirty pages shrink, so it is effectively unbounded), then
# SIGTERM to everything with a 90 s timeout, then SIGKILL with another 90 s -
# and only THEN issues its first watchdog_ping(). Worst case before that first
# ping is 210 s and open-ended, which is why upstream defaults to 10 minutes.
# A shorter value hard-resets the board mid-unmount_all(), i.e. it corrupts the
# filesystem it was installed to protect. We take upstream's default.
WATCHDOG_REBOOT_SEC_DEFAULT = 600

# Maps audio.default_output to the ALSA slave PCM. Kept beside the other module
# defaults so the apply script never has to know the mapping.
_ALSA_PCM_BY_OUTPUT = {
    "hdmi0": "sysdefault:CARD=vc4hdmi0",
    "hdmi1": "sysdefault:CARD=vc4hdmi1",
    "headphones": "sysdefault:CARD=Headphones",
    "dac": "sysdefault:CARD=DAC",
}

# Version of the on-device identity contract written to /etc/bgrpiimage-release.
# An in-place updater reads this FIRST: a device whose release file has no
# BGRPIIMAGE_APPLY_CONTRACT predates the contract, carries none of the identity
# an update needs to be safe, and must be told to reflash rather than be
# updated on a best-effort basis. Bump this only when the meaning or the
# required set of keys changes, never for a new optional key.
APPLY_CONTRACT_VERSION = 1

# -----------------------------------------------------------------------------
# Env var resolution
# -----------------------------------------------------------------------------
# Names that look like ${...} but must pass through to downstream tools that
# do their own substitution (notably unattended-upgrades' APT origin patterns).
_PASSTHROUGH_NAMES: set[str] = {"distro_id", "distro_codename"}

_ENV_VAR_RE = re.compile(
    r"\$\{(?P<name>[A-Za-z_][A-Za-z0-9_]*)(?:(?P<op>:-)(?P<default>[^}]*))?\}"
)


def resolve_env_vars(value: str, env: dict[str, str]) -> str:
    """Resolve ${VAR} / ${VAR:-default} references in a string.

    Behaviour:
      - ${VAR}           -> env[VAR]; raises KeyError if unset.
      - ${VAR:-default}  -> env[VAR] if set & non-empty, else default.
      - Names in _PASSTHROUGH_NAMES are left untouched (downstream resolves).
      - Resolution is single-pass; defaults are not re-parsed.

    Rationale:
      BAUER GROUP security standard - fail fast on missing secrets; never
      silently default a secret to empty. Defaults exist exactly for the
      values that are explicitly non-sensitive.
    """
    def replace(m: re.Match[str]) -> str:
        name = m.group("name")
        if name in _PASSTHROUGH_NAMES:
            return m.group(0)
        op = m.group("op")
        default = m.group("default")
        val = env.get(name)
        if op == ":-":
            return val if val else (default or "")
        if val is None:
            raise KeyError(
                f"environment variable '{name}' is required by config but not set "
                f"(use ${{{name}:-default}} to provide a fallback)"
            )
        return val

    return _ENV_VAR_RE.sub(replace, value)


def resolve_tree(node: Any, env: dict[str, str]) -> Any:
    """Recursively resolve ${...} in every string leaf of a JSON-like tree."""
    if isinstance(node, str):
        return resolve_env_vars(node, env)
    if isinstance(node, list):
        return [resolve_tree(x, env) for x in node]
    if isinstance(node, dict):
        return {k: resolve_tree(v, env) for k, v in node.items()}
    return node


# -----------------------------------------------------------------------------
# Variant composition via `extends`
# -----------------------------------------------------------------------------
def load_variant(path: Path, _seen: set[Path] | None = None) -> dict[str, Any]:
    """Load a variant JSON, recursively applying any `extends` reference.

    `extends` is a relative path (from the current file) to a parent JSON.
    The parent is loaded first (recursively), then the child is deep-merged
    onto it. This is BEFORE env-var resolution - so a child can override an
    `${ADMIN_PASSWORD:-...}` default by setting a literal.
    """
    _seen = _seen or set()
    resolved = path.resolve()
    if resolved in _seen:
        chain = " -> ".join(str(p) for p in _seen) + f" -> {resolved}"
        raise ValueError(f"circular extends chain: {chain}")
    _seen.add(resolved)

    with resolved.open("r", encoding="utf-8") as f:
        data = json.load(f)
    data.pop("$schema", None)

    parent_ref = data.pop("extends", None)
    if parent_ref:
        parent_path = (resolved.parent / parent_ref).resolve()
        parent = load_variant(parent_path, _seen=_seen)
        data = deep_merge(parent, data)
    return data


def deep_merge(parent: Any, child: Any) -> Any:
    """Merge child onto parent.

    - dicts          : recursive merge; child keys win on conflict
    - scalar lists   : concat(parent + child) with stable-order dedupe
    - named records  : lists of dicts where every item has a `name` field
                       are merged by name (same name -> deep-merge entries)
    - other lists    : concat(parent + child)
    - scalars        : child overrides parent
    """
    if isinstance(parent, dict) and isinstance(child, dict):
        out: dict[str, Any] = {**parent}
        for k, v in child.items():
            out[k] = deep_merge(out[k], v) if k in out else v
        return out
    if isinstance(parent, list) and isinstance(child, list):
        combined = list(parent) + list(child)
        if not combined:
            return combined
        if all(isinstance(x, (str, int, float, bool)) for x in combined):
            seen: set[Any] = set()
            deduped: list[Any] = []
            for x in combined:
                if x not in seen:
                    seen.add(x)
                    deduped.append(x)
            return deduped
        if all(isinstance(x, dict) and "name" in x for x in combined):
            # `name` is the merge key only when it is actually unique. It is not
            # for dtoverlays: mcp2515 ships one overlay PER CHANNEL
            # (mcp2515-can0 / mcp2515-can1), but mcp251xfd ships ONE overlay for
            # every channel and selects the chip with a boolean spi<n>-<m>
            # parameter - so a two-channel CAN FD HAT needs two entries that are
            # both named "mcp251xfd". Merging those by name collapsed them into a
            # single overlay carrying both spi selectors, with `interrupt`
            # resolved last-wins: one CAN interface instead of two, pointed at
            # the other channel's INT GPIO. Nothing failed - `make validate`
            # passed, the image built, and only the hardware disagreed.
            # `id` is the opt-in disambiguator. It is a merge key ONLY: it is
            # never rendered, so `{"id": "can1", "name": "mcp251xfd"}` still
            # emits `dtoverlay=mcp251xfd,...`. Absent, the key is `name`, so
            # every existing variant merges exactly as before.
            by_name: dict[str, dict[str, Any]] = {}
            order: list[str] = []
            for item in combined:
                n = item.get("id") or item["name"]
                if n in by_name:
                    by_name[n] = deep_merge(by_name[n], item)
                else:
                    by_name[n] = item
                    order.append(n)
            return [by_name[n] for n in order]
        return combined
    return child


# -----------------------------------------------------------------------------
# File writing helpers
# -----------------------------------------------------------------------------
def write(path: Path, content: str, *, executable: bool = False) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    # LF line endings regardless of host OS - these files run on Linux.
    path.write_bytes(content.encode("utf-8"))
    if executable:
        path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)


def shell_var(name: str, value: str | int | bool) -> str:
    if isinstance(value, bool):
        value = "yes" if value else "no"
    return f"{name}={shlex.quote(str(value))}\n"


def shell_array(name: str, values: list[str]) -> str:
    quoted = " ".join(shlex.quote(v) for v in values)
    return f'{name}="{quoted}"\n'


def clean_generated(module_name: str) -> Path:
    # Write to module/filesystem/root/opt/bgrpiimage/<module>/ so the files
    # end up INSIDE the chroot: CustomPiOS copies `module/filesystem/` into
    # the chroot root; our start_chroot_script calls `unpack /filesystem/
    # root / root` which moves the contents to `/`. Result in the chroot:
    # /opt/bgrpiimage/<module>/<generated files>.
    gen = MODULES_DIR / module_name / "filesystem" / "root" / "opt" / "bgrpiimage" / module_name
    if gen.exists():
        shutil.rmtree(gen)
    gen.mkdir(parents=True)
    return gen


def stage_apply(module_name: str, gen: Path) -> None:
    """Copy a module's hand-written apply.sh into its generated payload.

    The apply logic stays hand-written and git-tracked at
    src/modules/<m>/apply.sh - those scripts run ~60% comment density and
    every comment is a scar (rfkill dpkg-conffile semantics, mcp2515 probe
    order, agetty's inotify redraw, the ip6tables -C that left a unit
    permanently failed). Generating shell would bury all of it inside Python
    string literals, where shellcheck cannot see it and `git blame` points at
    the generator instead of the decision.

    But the payload directory must stay 100% derived, or clean_generated()'s
    invariant breaks and a stale apply.sh could outlive a rename. So the file
    is COPIED here on every render: one owner for the logic, one owner for the
    tree it ships in.
    """
    src = MODULES_DIR / module_name / "apply.sh"
    if not src.exists():
        return
    # Normalise to LF: the repo is developed on Windows and scripts/build.sh
    # bind-mounts the working tree straight into the build container, where a
    # CRLF shebang execs as "bad interpreter: No such file or directory" and
    # reads to a user as "the command does not exist".
    body = src.read_text(encoding="utf-8").replace("\r\n", "\n")
    write(gen / "apply.sh", body, executable=True)


def render_common(cfg: dict[str, Any]) -> None:
    """Ship the shared apply library.

    Carries no configuration of its own. It exists so that every module's
    apply.sh has a library to source, and it is first in ACTIVE_MODULES so
    that library is on disk before anything sources it.
    """
    gen = clean_generated("bgrpiimage-common")
    lib = MODULES_DIR / "bgrpiimage-common" / "apply-lib.sh"
    body = lib.read_text(encoding="utf-8").replace("\r\n", "\n")
    write(gen / "apply-lib.sh", body)


# -----------------------------------------------------------------------------
# Renderers - one per feature area
# -----------------------------------------------------------------------------
def render_base(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-base")
    write(gen / "hostname", cfg["hostname"] + "\n")
    locale = cfg.get("locale", {})
    lines = [
        shell_var("BGRPIIMAGE_TIMEZONE", locale.get("timezone", "UTC")),
        shell_var("BGRPIIMAGE_LOCALE", locale.get("locale", "en_US.UTF-8")),
        shell_var("BGRPIIMAGE_KEYBOARD", locale.get("keyboard", "us")),
    ]
    write(gen / "locale.env", "".join(lines))
    packages = list(cfg.get("packages", []))
    bluetooth = cfg.get("bluetooth") or {}
    if bluetooth.get("enabled", True) and "bluez" not in packages:
        # Raspberry Pi OS Lite already ships bluez (pi-gen
        # stage2/01-sys-tweaks/00-packages), so this is a no-op install - but it
        # turns an inherited dependency into a declared one, and the image no
        # longer silently loses Bluetooth if the base image drops it.
        packages.append("bluez")
    write(gen / "packages.list", "\n".join(packages) + ("\n" if packages else ""))

    # /etc/bgrpiimage-release - sourced by the MOTD banner and any ops tooling.
    #
    # The last three keys are for in-place updates, not for display.
    #
    # BASE_IMAGE_SHA256 is the line between a config update and an OS update.
    # A release that rebases onto a new Raspberry Pi OS changes it, and an
    # updater whose bundle records a different value must refuse rather than
    # apply configuration built against a base the device is not running.
    # The schema requires only url and arch, so this can legitimately be
    # empty - which means "unknown", never "matches".
    #
    # BASE_IMAGE_URL is kept beside it because the hash alone identifies
    # nothing to a human reading the file during a support call.
    #
    # APPLY_CONTRACT is the marker an updater tests before touching anything.
    # Absent means the image predates the contract, so the honest answer is
    # "reflash", not a best-effort apply. It is written here rather than
    # derived from BGRPIIMAGE_VERSION because a version number says when an
    # image was built, not what guarantees it makes.
    variant = cfg["variant"]
    base_image = cfg.get("base_image") or {}
    release_lines = [
        'BGRPIIMAGE_DIST="bgrpiimage"\n',
        f'BGRPIIMAGE_VARIANT={shlex.quote(variant["name"])}\n',
        f'BGRPIIMAGE_VERSION={shlex.quote(variant.get("version", "0.0.0"))}\n',
        f'BGRPIIMAGE_DESCRIPTION={shlex.quote(variant.get("description", ""))}\n',
        f'BGRPIIMAGE_BASE_IMAGE_URL={shlex.quote(base_image.get("url", ""))}\n',
        f'BGRPIIMAGE_BASE_IMAGE_SHA256={shlex.quote(base_image.get("sha256", ""))}\n',
        f'BGRPIIMAGE_APPLY_CONTRACT={APPLY_CONTRACT_VERSION}\n',
    ]
    write(gen / "release.env", "".join(release_lines))

    ssh = cfg.get("ssh") or {}
    write(gen / "ssh.env", shell_var("BGRPIIMAGE_SSH_ENABLED", bool(ssh.get("enabled", True))))

    write(
        gen / "bluetooth.env",
        shell_var("BGRPIIMAGE_BLUETOOTH_ENABLED", bool(bluetooth.get("enabled", True))),
    )

    banner = cfg.get("banner") or {}
    if banner.get("enabled", True):
        _render_banner(gen, cfg, banner)


# Width of the rule in /etc/issue and /etc/issue.net, and therefore the width
# everything in those files wraps to. 68 leaves a little room inside an 80
# column terminal for the SSH client's own "| " banner prefix, which is two
# more columns nobody accounts for until the line wraps.
_BANNER_RULE_WIDTH = 68


def _wrap_banner_text(text: str, indent: str = "  ") -> str:
    """Wrap one paragraph to the banner rule, on word boundaries.

    Returns "" for empty input rather than a blank line, so an absent
    description does not leave a hole in the block.
    """
    if not text.strip():
        return ""
    lines = textwrap.wrap(
        text.strip(),
        width=_BANNER_RULE_WIDTH - len(indent),
        break_long_words=False,
        break_on_hyphens=False,
    )
    return "".join(f"{indent}{line}\n" for line in lines)


def _render_banner(gen: Path, cfg: dict[str, Any], banner: dict[str, Any]) -> None:
    """Emit /etc/issue, /etc/issue.net, the MOTD script and the sshd banner
    drop-in. The MOTD script is static - all dynamic info (hostname, IPs, CAN
    state, docker, uptime, pending reboots) is resolved at login time from the
    running system."""
    variant = cfg["variant"]
    note = banner.get("pre_login_note", "")

    # The description is wrapped to the rule, not printed as one long line.
    # /etc/issue.net has no idea how wide the client's terminal is, so an
    # unwrapped line is broken by the SSH client wherever it happens to run
    # out - mid-word, and with its own "> " continuation marker, which reads
    # as a corrupted banner rather than a long one:
    #     |   BAUER GROUP ... dual isolated C
    #     > AN HAT (MCP2515 on SPI)
    # Wrapping at the rule width also keeps the block rectangular, which is
    # the whole point of having a rule.
    rule = "-" * _BANNER_RULE_WIDTH + "\n"
    header = f"bgRPIImage {variant['name']} v{variant.get('version', '0.0.0')}\n"
    header += _wrap_banner_text(variant.get("description", ""))

    # /etc/issue is deliberately STATIC and short.
    #
    # agetty redraws the whole issue file whenever anything calls
    # `agetty --reload` (it watches /run/agetty.reload via inotify) - which
    # happens on network events and on every cloud-init boot stage. A ten
    # line issue with per-interface escapes therefore repaints half the
    # console several times during boot, which reads as a bug.
    #
    # Dynamic state belongs in the MOTD, which renders once per login and
    # can run real commands. The hostname is not lost: agetty already
    # prefixes the prompt with it ("bg-canbus login:").
    issue = f"{header}{rule}"
    if note:
        issue += _wrap_banner_text(note, indent="")
    write(gen / "issue", issue)

    # /etc/issue.net: sshd reads raw (no escapes), so keep it static.
    issue_net = f"{header}{rule}"
    if note:
        issue_net += _wrap_banner_text(note, indent="")
    issue_net += rule
    write(gen / "issue.net", issue_net)

    # sshd drop-in to surface the pre-login banner.
    write(
        gen / "sshd_banner.conf",
        "# bgRPIImage pre-login banner\n"
        "Banner /etc/issue.net\n",
    )

    # Dynamic MOTD - runs on login (pam_motd) and also from console.
    write(gen / "motd-banner.sh", _MOTD_SCRIPT, executable=True)


_MOTD_SCRIPT = r"""#!/bin/bash
# bgRPIImage dynamic MOTD - shown after login.
# Keep this script minimal and tolerant: it must never block a login.
set +e

[ -r /etc/bgrpiimage-release ] && . /etc/bgrpiimage-release
# Written only by bgrpiimage-update. Absent on a device that has never
# been updated, which is why every read below is defaulted.
[ -r /etc/bgrpiimage-applied ] && . /etc/bgrpiimage-applied

if [ -t 1 ]; then
    CY=$'\033[1;36m'; GR=$'\033[1;32m'; DIM=$'\033[2m'
    YE=$'\033[1;33m'; RD=$'\033[1;31m'; NC=$'\033[0m'
else
    CY=''; GR=''; DIM=''; YE=''; RD=''; NC=''
fi

cols=$(tput cols 2>/dev/null || echo 72); [ "$cols" -lt 60 ] && cols=72
sep=$(printf '=%.0s' $(seq 1 "$cols"))

active_color() { [ "$1" = "active" ] && echo "$GR" || echo "$YE"; }

echo "${CY}${sep}${NC}"
# The image version and the configuration version legitimately differ once
# an update has been applied, and both matter: the first says which image was
# flashed - the anchor for deciding whether the next release is a config
# change or a reflash - and the second says what the device is actually
# running. Showing only one of them would make a support call guesswork, so
# the second is appended when, and only when, it differs.
_ver="v${BGRPIIMAGE_VERSION:-0.0.0}"
if [ -n "${BGRPIIMAGE_CONFIG_VERSION:-}" ] \
   && [ "${BGRPIIMAGE_CONFIG_VERSION}" != "${BGRPIIMAGE_VERSION:-}" ]; then
    _ver="${_ver} ${DIM}(config v${BGRPIIMAGE_CONFIG_VERSION})${NC}"
fi
printf "  ${GR}%s${NC}  %s  ${DIM}%s${NC}\n" \
    "${BGRPIIMAGE_DIST:-bgRPIImage}" \
    "${BGRPIIMAGE_VARIANT:-unknown}" \
    "$_ver"
if [ "${BGRPIIMAGE_CONFIG_RESULT:-ok}" = "verify-failed" ]; then
    printf "  ${RD}the last configuration update did not verify${NC} ${DIM}(sudo bgrpiimage-update status)${NC}\n"
fi
# Wrapped to the rule rather than printed as one line. $cols already follows
# the real terminal, so the separator above is exactly as wide as the window -
# but an unwrapped description is broken by the TERMINAL instead, mid-word and
# without the two-space indent, which leaves a ragged line hanging under a
# neat block. `fold -s` breaks on spaces; the sed strips the trailing space it
# leaves behind on each break.
if [ -n "${BGRPIIMAGE_DESCRIPTION:-}" ]; then
    printf '%s\n' "$BGRPIIMAGE_DESCRIPTION" \
        | fold -s -w "$((cols - 2))" \
        | sed 's/[[:space:]]*$//' \
        | while IFS= read -r _dline; do
              printf "  ${DIM}%s${NC}\n" "$_dline"
          done
fi

model=""
if [ -r /sys/firmware/devicetree/base/model ]; then
    model=$(tr -d '\0' < /sys/firmware/devicetree/base/model)
fi
printf "  ${DIM}host:${NC} %-20s  ${DIM}kernel:${NC} %s\n" "$(hostname)" "$(uname -r)"
[ -n "$model" ] && printf "  ${DIM}model:${NC} %s\n" "$model"
up=$(uptime -p 2>/dev/null)
[ -n "$up" ] && printf "  ${DIM}uptime:${NC} %s\n" "$up"
echo "${CY}${sep}${NC}"

# Physical + virtual interfaces we care about.
iface_count=0
for iface in $(ip -o link show 2>/dev/null | \
               awk -F': ' '$2 !~ /^(lo|docker|podman|veth|br-|bond|vlan)/ {print $2}' | \
               cut -d'@' -f1); do
    iface_count=$((iface_count+1))
    state=$(ip -br link show "$iface" 2>/dev/null | awk '{print $2}')
    case "$iface" in
        can*)
            # The netdev flag is the wrong health signal for a CAN link. A
            # bus-off controller keeps it at UP while passing nothing at all,
            # so this line used to read "can1 UP 500 kbit/s" for a bus that had
            # been dead for weeks - the exact failure the banner exists to
            # surface. Read the controller state as well.
            # Match "can .*state", not "can state": iproute2 prints the
            # ctrlmode list first, so "can <BERR-REPORTING> state ..." is a
            # normal reading once listen-only or berr-reporting is enabled.
            # Anchoring on ^[[:space:]]*can keeps it off the header line
            # ("4: can0: ... state UP") and off "link/can promiscuity ...".
            det=$(ip -details link show "$iface" 2>/dev/null)
            bitrate=$(echo "$det" | grep -oE 'bitrate [0-9]+' | awk '{print $2}')
            [ -n "$bitrate" ] && rate_str="$((bitrate/1000)) kbit/s" || rate_str="(no bitrate)"
            can_state=$(echo "$det" | \
                        sed -n 's/^[[:space:]]*can .*state \([A-Z-]\{1,\}\).*/\1/p' | head -1)
            case "$can_state" in
                ""|ERROR-ACTIVE) cstr="" ;;
                BUS-OFF)         cstr="  ${RD}BUS-OFF${NC}" ;;
                *)               cstr="  ${YE}${can_state}${NC}" ;;
            esac
            sc=$([ "$state" = "UP" ] && echo "$GR" || echo "$DIM")
            printf "  ${DIM}%-7s${NC} ${sc}%-6s${NC} %s%s\n" \
                   "$iface" "$state" "$rate_str" "$cstr"
            ;;
        *)
            v4=$(ip -4 -br addr show "$iface" 2>/dev/null | awk '{$1=$2=""; print $0}' | xargs)
            v6=$(ip -6 -br addr show "$iface" 2>/dev/null | \
                 awk '{for(i=3;i<=NF;i++) print $i}' | \
                 grep -v '^fe80' | head -2 | tr '\n' ' ')
            sc=$([ "$state" = "UP" ] && echo "$GR" || echo "$DIM")
            printf "  ${DIM}%-7s${NC} ${sc}%-6s${NC} v4: %s\n" "$iface" "$state" "${v4:-(none)}"
            [ -n "$v6" ] && printf "  %-7s %-6s v6: %s\n" "" "" "$v6"
            ;;
    esac
done
[ "$iface_count" -eq 0 ] && printf "  ${DIM}(no external network interfaces detected)${NC}\n"
echo "${CY}${sep}${NC}"

ssh_s=$(systemctl is-active ssh 2>/dev/null || echo "?")
# Which runtime is installed? podman-docker ships /usr/bin/docker as a
# shim, so probing for `docker` is true under both - `podman` is the only
# honest discriminator. Podman is daemonless, so the unit that means
# "the API is reachable" is the socket, not a service.
if command -v podman >/dev/null 2>&1; then
    rt_name="podman"; rt_unit="podman.socket"
else
    rt_name="docker"; rt_unit="docker"
fi
dk_s=$(systemctl is-active "$rt_unit" 2>/dev/null || echo "?")
uu_s=$(systemctl is-active unattended-upgrades 2>/dev/null || echo "?")
bt_s=$(systemctl is-active bluetooth 2>/dev/null || echo "?")
printf "  ${DIM}ssh:${NC} $(active_color "$ssh_s")%s${NC}" "$ssh_s"
printf "   ${DIM}%s:${NC} $(active_color "$dk_s")%s${NC}" "$rt_name" "$dk_s"
if [ "$dk_s" = "active" ]; then
    n=$("$rt_name" ps -q 2>/dev/null | wc -l)
    printf " ${DIM}(%d running)${NC}" "$n"
fi
printf "   ${DIM}bt:${NC} $(active_color "$bt_s")%s${NC}" "$bt_s"
printf "   ${DIM}unattended-upgrades:${NC} $(active_color "$uu_s")%s${NC}\n" "$uu_s"

if [ -f /var/run/reboot-required ]; then
    pkgs=""
    [ -s /var/run/reboot-required.pkgs ] && pkgs=$(tr '\n' ' ' < /var/run/reboot-required.pkgs)
    printf "  ${YE}reboot pending${NC} %s\n" "${pkgs:+(triggered by: ${pkgs})}"
fi

# Warn while a shipped demo password is still in place.
#
# The marker records user:<first 12 chars of the crypt hash> at build time;
# a still-matching hash means the credential was never rotated. The previous
# test compared sp_lstchg against 0, which can never be true here: PAM's
# account phase forces the expired password to be changed before the session
# phase ever runs pam_motd, so the field is always rewritten first.
#
# Bare-username lines (markers written by images <= 0.5.0) fall back to the
# old test so already-flashed devices keep working.
if [ -f /etc/bgrpiimage-default-password-active ]; then
    _stale=""
    while IFS=: read -r _u _h; do
        [ -n "$_u" ] || continue
        if [ -n "$_h" ]; then
            _cur=$(awk -F: -v u="$_u" '$1==u{print substr($2,1,12)}' /etc/shadow 2>/dev/null)
            [ -n "$_cur" ] && [ "$_cur" = "$_h" ] && _stale="${_stale} ${_u}"
        else
            [ "$(awk -F: -v u="$_u" '$1==u{print $3}' /etc/shadow 2>/dev/null)" = "0" ] \
                && _stale="${_stale} ${_u}"
        fi
    done < /etc/bgrpiimage-default-password-active
    if [ -n "$_stale" ]; then
        printf "  ${RD}SECURITY:${NC} default password still unchanged for:%s\n" "$_stale"
        printf "           rotate it now: ${YE}sudo bgrpiimage-setup password${NC}\n"
    fi
fi
echo "${CY}${sep}${NC}"
"""


# Known-weak credentials that must never reach a device unannounced.
# The CI placeholders matter as much as the config default: build.yml falls
# back to them when secrets.ADMIN_PASSWORD / secrets.WIFI_PSK are unset, so
# without them listed here a published release asset would carry a password
# that is written in plain text in a public repository - and would get
# neither the build warning nor the on-device MOTD nag.
_KNOWN_DEMO_PASSWORDS = {"12345678", "ci-placeholder-pw", "ci-placeholder-psk"}


def render_users(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-users")
    users = cfg.get("users", [])
    remove_users = cfg.get("remove_users", [])
    root = cfg.get("root", {})
    script = ["#!/bin/bash", "# Auto-generated by scripts/generate.py", "set -euo pipefail", ""]

    # Remove the stock accounts FIRST. `useradd` hands out the lowest free
    # UID, so creating admin while `pi` still holds 1000 pushed admin to 1001
    # and left the image with no UID-1000 user at all - which breaks
    # raspi-config's UID-1000 fallback and gives the raspios first-boot
    # wizard a reason to ask which account to rename.
    for victim in remove_users:
        script.append(f"if id -u {shlex.quote(victim)} >/dev/null 2>&1; then")
        script.append(f"  deluser --remove-home {shlex.quote(victim)} || true")
        script.append(f"  delgroup {shlex.quote(victim)} 2>/dev/null || true")
        script.append("fi")
    if remove_users:
        script.append("")

    # Build side: shout when a ship-default password survived env resolution.
    # Without this the only signal is on the device itself, so an image can be
    # built, published and flashed before anyone learns it carries a known
    # credential (e.g. when ADMIN_PASSWORD is dropped by sudo's env_reset in CI).
    weak = [u["name"] for u in users if u.get("password") in _KNOWN_DEMO_PASSWORDS]
    if weak:
        console.print(
            f"[bold red]SECURITY:[/] user(s) {', '.join(weak)} carry a known "
            f"default password - set ADMIN_PASSWORD to build a hardened image"
        )

    for user in users:
        name = user["name"]
        pw = user["password"]
        shell = user.get("shell", "/bin/bash")
        groups = ",".join(user.get("groups", []))
        script.append(f"# === user: {name} ===")
        script.append(f"if ! id -u {shlex.quote(name)} >/dev/null 2>&1; then")
        script.append(f"  useradd -m -s {shlex.quote(shell)} {shlex.quote(name)}")
        script.append("fi")
        if groups:
            # `usermod -aG` fails hard on any non-existent group. Pre-create
            # each one with `groupadd -f` so user-add works even when module
            # order puts users before the packages (docker, i2c-tools) that
            # would otherwise install the group.
            for g in user.get("groups", []):
                script.append(
                    f"getent group {shlex.quote(g)} >/dev/null || groupadd {shlex.quote(g)}"
                )
            script.append(f"usermod -aG {shlex.quote(groups)} {shlex.quote(name)}")
        # chpasswd via stdin keeps the password out of argv / process lists.
        script.append(f"echo {shlex.quote(f'{name}:{pw}')} | chpasswd")
        if pw in _KNOWN_DEMO_PASSWORDS:
            # These are PUBLIC images, so the default credential stays
            # documented and discoverable - that is what keeps onboarding
            # fast. Expiring it only means the operator is prompted to set a
            # new password at the first login they were going to do anyway;
            # login(1) on the console and sshd (UsePAM yes) both enforce it.
            script.append(f"chage -d 0 {shlex.quote(name)}")
        if user.get("sudo_nopasswd"):
            sudoers_line = f"{name} ALL=(ALL) NOPASSWD:ALL"
            script.append(
                f"echo {shlex.quote(sudoers_line)} > /etc/sudoers.d/010-bgrpiimage-{name}"
            )
            script.append(f"chmod 440 /etc/sudoers.d/010-bgrpiimage-{name}")
        keys = user.get("ssh_authorized_keys") or []
        if keys:
            script.append(f"install -d -m 700 -o {shlex.quote(name)} -g {shlex.quote(name)} /home/{name}/.ssh")
            authfile = f"/home/{name}/.ssh/authorized_keys"
            script.append(f"cat > {authfile} <<'__BGRPIIMAGE_EOF__'")
            script.extend(keys)
            script.append("__BGRPIIMAGE_EOF__")
            script.append(f"chown {shlex.quote(name)}:{shlex.quote(name)} {authfile}")
            script.append(f"chmod 600 {authfile}")
        script.append("")

    if weak:
        # Runtime side: record user:<first 12 chars of the crypt hash>, and do
        # it AFTER the accounts exist so the hash is real. The MOTD compares the
        # live hash against this instead of testing sp_lstchg==0, which can
        # never be true by the time pam_motd runs - PAM's account phase has
        # already forced the expired password to be changed.
        script.append("# --- default-password marker (read by the MOTD) ---")
        script.append(": > /etc/bgrpiimage-default-password-active")
        for u in weak:
            q = shlex.quote(u)
            awk = (
                "awk -F: -v u=" + q + " '$1==u{print substr($2,1,12)}' /etc/shadow"
            )
            script.append(
                "printf '%s:%s\\n' " + q + ' "$(' + awk + ')"'
                " >> /etc/bgrpiimage-default-password-active"
            )
        script.append("chmod 644 /etc/bgrpiimage-default-password-active")
        script.append("")

    # su without password for listed users -> pam_wheel.so group trust.
    su_users = root.get("su_nopasswd_users") or []
    if su_users:
        script.append("# su without password for trusted users -> group 'wheel'")
        script.append("getent group wheel >/dev/null || groupadd wheel")
        for u in su_users:
            script.append(f"usermod -aG wheel {shlex.quote(u)}")
        script.append("install -m 644 /tmp/_bgrpiimage_su_pam /etc/pam.d/su")

    ssh_pw = root.get("ssh_password_auth", True)
    ssh_root = root.get("ssh_permit_root_login", False)
    script.append("")
    script.append("# sshd hardening")
    script.append("mkdir -p /etc/ssh/sshd_config.d")
    sshd = []
    sshd.append(f"PasswordAuthentication {'yes' if ssh_pw else 'no'}")
    sshd.append(f"PermitRootLogin {'yes' if ssh_root else 'no'}")
    sshd.append("ChallengeResponseAuthentication no")
    sshd.append("UsePAM yes")
    script.append("cat > /etc/ssh/sshd_config.d/10-bgrpiimage.conf <<'__BGRPIIMAGE_EOF__'")
    script.extend(sshd)
    script.append("__BGRPIIMAGE_EOF__")
    script.append("chmod 644 /etc/ssh/sshd_config.d/10-bgrpiimage.conf")

    write(gen / "create-users.sh", "\n".join(script) + "\n", executable=True)

    # /etc/pam.d/su drop-in enabling pam_wheel trust.
    pam_su = (
        "# /etc/pam.d/su - generated by bgRPIImage\n"
        "auth       sufficient pam_rootok.so\n"
        "auth       [success=ignore default=1] pam_succeed_if.so user = root\n"
        "auth       sufficient pam_wheel.so trust use_uid\n"
        "auth       required   pam_wheel.so use_uid\n"
        "auth       required   pam_unix.so\n"
        "account    required   pam_unix.so\n"
        "session    required   pam_unix.so\n"
        "session    optional   pam_xauth.so\n"
    )
    write(gen / "pam_su", pam_su)


def render_network(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-network")
    net = cfg.get("network", {})
    nwd = gen / "systemd-networkd"
    nwd.mkdir(parents=True, exist_ok=True)

    def iface_network_file(idx: int, iface: dict[str, Any], match: str) -> str:
        mode = iface.get("mode", "dhcp")
        ipv6 = iface.get("ipv6", True)
        lines = [f"[Match]", f"Name={match}", "", "[Network]"]
        if mode == "dhcp":
            lines.append("DHCP=ipv4")
            if ipv6:
                lines.append("IPv6AcceptRA=yes")
                lines.append("LinkLocalAddressing=ipv6")
            else:
                lines.append("LinkLocalAddressing=no")
        elif mode == "static":
            addr = iface.get("address")
            prefix = iface.get("prefix", 24)
            gw = iface.get("gateway")
            if addr:
                lines.append(f"Address={addr}/{prefix}")
            if gw:
                lines.append(f"Gateway={gw}")
            for dns in iface.get("dns", []):
                lines.append(f"DNS={dns}")
            if ipv6:
                addr6 = iface.get("address_v6")
                prefix6 = iface.get("prefix_v6", 64)
                gw6 = iface.get("gateway_v6")
                if addr6:
                    lines.append(f"Address={addr6}/{prefix6}")
                if gw6:
                    lines.append(f"Gateway={gw6}")
        # RequiredForOnline gates systemd-networkd-wait-online, which
        # `systemctl enable systemd-networkd` pulls in through the Also= in its
        # [Install] section. A managed link defaults to "yes", so a wlan0 that
        # never associates (no AP, or rfkill-blocked) holds up
        # network-online.target - and docker.service Wants=/After= that target,
        # so Docker and the Portainer first-boot install queue behind it.
        # Wireless is therefore never required. Ethernet keeps systemd's own
        # default of "degraded"; "routable" would be STRICTER than today and
        # would stall for 120 s on a LAN with no DHCP.
        required = "no" if match.startswith("wlan") else "degraded"
        lines += ["", "[Link]", f"RequiredForOnline={required}"]
        return "\n".join(lines) + "\n"

    eth = net.get("ethernet")
    if eth and eth.get("mode") != "disabled":
        write(nwd / "10-eth.network", iface_network_file(10, eth, eth.get("interface", "eth0")))

    wifi = net.get("wifi")
    if wifi and wifi.get("mode") != "disabled":
        write(nwd / "20-wlan.network", iface_network_file(20, wifi, wifi.get("interface", "wlan0")))

        wpa_dir = gen / "wpa_supplicant"
        wpa_dir.mkdir(parents=True, exist_ok=True)
        country = wifi.get("country", "DE")
        wpa = [
            "ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev",
            "update_config=1",
            f"country={country}",
            "",
        ]
        for net_entry in wifi.get("networks", []):
            wpa.append("network={")
            wpa.append(f'    ssid="{net_entry["ssid"]}"')
            wpa.append(f'    psk="{net_entry["psk"]}"')
            if "priority" in net_entry:
                wpa.append(f'    priority={net_entry["priority"]}')
            if net_entry.get("hidden"):
                wpa.append("    scan_ssid=1")
            wpa.append("    key_mgmt=WPA-PSK")
            wpa.append("}")
            wpa.append("")
        iface_name = wifi.get("interface", "wlan0")
        write(wpa_dir / f"wpa_supplicant-{iface_name}.conf", "\n".join(wpa))

    # raspberrypi-sys-mods ships /etc/modprobe.d/rfkill_default.conf containing
    # `options rfkill default_state=0`. Because CONFIG_RFKILL=m on Pi that takes
    # effect at module init, where net/rfkill/core.c calls
    # rfkill_update_global_state(RFKILL_TYPE_ALL, ...): EVERY radio type is
    # soft-blocked before any switch registers. It exists to stop a device
    # radiating before a WLAN regulatory domain is known, and it is what prints
    # "Wi-Fi is currently blocked by rfkill" on every login. It blocks Bluetooth
    # too - that only works today because pi-gen whitelists a handful of known
    # BT device ids under /var/lib/systemd/rfkill, and only if one of them
    # happens to match the board.
    #
    # We pin the domain ourselves, so restore the kernel default. Emitted
    # regardless of wifi.mode: an image shipped with WiFi off still needs a
    # working Bluetooth radio, and would otherwise keep showing a nag pointing
    # at raspi-config, which this image does not use.
    #
    # modprobe concatenates /etc/modprobe.d/*.conf in lexicographic order and
    # the kernel's parse_args takes the LAST occurrence of an option - hence the
    # zz- prefix. The vendor file is a dpkg conffile of raspberrypi-sys-mods and
    # is deliberately left untouched so upgrades never hit a conffile prompt.
    regdom = (net.get("wifi") or {}).get("country", "DE")
    modprobe_dir = gen / "modprobe.d"
    modprobe_dir.mkdir(parents=True, exist_ok=True)
    write(
        modprobe_dir / "zz-bgrpiimage-rfkill.conf",
        "# bgRPIImage - overrides raspberrypi-sys-mods' rfkill_default.conf.\n"
        "# That file is a dpkg conffile; editing it turns every upgrade into a\n"
        "# conffile conflict, so we win on load order instead.\n"
        "options rfkill default_state=1\n"
        f"options cfg80211 ieee80211_regdom={regdom}\n",
    )


def _overlay_line(name: str, params: dict[str, Any] | None = None) -> str:
    """Render a `dtoverlay=...` line with optional comma-joined params.

    A JSON `true` renders as a BARE token, not `k=true`. The firmware treats a
    boolean overlay parameter as present-is-true / absent-is-false (dtoverlay.c,
    DTOVERRIDE_BOOLEAN: "The target is a boolean parameter (present->true,
    absent->false)"), and upstream's own example is the bare form:
    `dtoverlay=mcp251xfd,spi0-0,interrupt=25`. The old renderer interpolated
    every value with f"{k}={v}", so a JSON `true` arrived as Python's repr and
    emitted `spi0-0=True` - not a value the parser accepts, so the chip-select
    selector was dropped and the overlay fell back to its default target.
    `false` is therefore rendered by OMITTING the key: writing `k=false` would
    still make the parameter present.
    """
    params = params or {}
    if not params:
        return f"dtoverlay={name}"
    parts = [name]
    for k, v in params.items():
        if isinstance(v, bool):
            if v:
                parts.append(k)
            continue
        parts.append(f"{k}={v}")
    return "dtoverlay=" + ",".join(parts)


_MCP2515_OVERLAY_RE = re.compile(r"^mcp2515-can(\d+)$")


def _dtoverlay_lines(overlays: list[dict[str, Any]]) -> list[str]:
    """Render `dtoverlay=` lines, forcing a canonical order for MCP2515 CAN.

    The netdev name is NOT chosen by the overlay name: mcp251x calls
    alloc_candev(..., "can%d") and the index is handed out by dev_alloc_name()
    at register_netdevice() time, i.e. in probe order. Probe order follows the
    device-tree child order of &spi0, and the firmware merges each dtoverlay
    with libfdt's fdt_add_subnode(), which inserts the new node *before* the
    target's existing children. Net effect: the overlay applied LAST probes
    FIRST and takes "can0".

    So to make can0 the CS0 chip - which is what every label, doc and .network
    file assumes - the highest chip-select must be emitted first. Waveshare's
    own config.txt for the 2-CH CAN HAT does exactly this (mcp2515-can1 before
    mcp2515-can0); shipping them in the "natural" order silently swaps the two
    physical connectors.

    Sorting here rather than in the variant JSON keeps the JSON readable and
    means a later tidy-up of that array cannot re-introduce the swap.
    """
    mcp: list[tuple[int, dict[str, Any]]] = []
    rest: list[dict[str, Any]] = []
    for ovl in overlays:
        m = _MCP2515_OVERLAY_RE.match(ovl["name"])
        if m:
            mcp.append((int(m.group(1)), ovl))
        else:
            rest.append(ovl)

    out = [_overlay_line(o["name"], o.get("params")) for o in rest]
    if mcp:
        if len(mcp) > 1:
            out.append(
                "# order is load-bearing: the LAST mcp2515 overlay probes FIRST "
                "and takes can0"
            )
        mcp.sort(key=lambda item: item[0], reverse=True)
        out.extend(_overlay_line(o["name"], o.get("params")) for _, o in mcp)
    return out


def _emit_boot_section(lines: list[str], heading: str, body: list[str]) -> None:
    """Append a labelled section to config.txt, skipping empty sections."""
    if not body:
        return
    if lines and lines[-1] != "":
        lines.append("")
    lines.append(f"# --- {heading} ---")
    lines.extend(body)


def render_boot(cfg: dict[str, Any]) -> None:
    """Render the consolidated config-bgrpiimage.txt fragment.

    Covers every JSON block that ends up as Raspberry Pi boot config: the
    coarse toggles in `boot_config` (I2C/SPI/UART/BT/WiFi + raw dtoverlays),
    camera / hdmi / display / audio / gpio / leds / rtc / fan / overclock /
    memory / pcie / usb. Runtime-only blocks (watchdog, bootloader, and the
    fan / rtc userspace services) are handled by bgrpiimage-hardware.
    """
    gen = clean_generated("bgrpiimage-boot")
    boot = cfg.get("boot_config") or {}

    # This fragment is APPENDED to the end of config.txt, and conditional
    # filters are sticky: everything after a [cm4]/[pi5]/... header applies
    # only to that board until the next filter. Stock Raspberry Pi OS happens
    # to end its config.txt with [all], but nothing guarantees that a
    # hand-edited or Imager-customised file does. Opening with [all] resets
    # any inherited scope so our settings really do apply to all hardware -
    # the reset the firmware docs prescribe for exactly this case.
    lines = [
        "# === BAUER GROUP auto-generated boot config ===",
        "",
        "[all]",
        "",
    ]

    # --- boot_config (the original toggles) -----------------------------------
    core: list[str] = []
    # Stop the core clock scaling. spi-bcm2835 reads the core rate ONCE, in
    # probe (clk_get_rate), and registers no clock notifier - so the SPI
    # divisor is computed against whatever the core happened to be running
    # at that moment and is never recomputed. A CM4 core scales 200-500 MHz,
    # so probing at the low end and boosting later silently multiplies the
    # real SCK by up to 2.5x - enough to shove an MCP2515 past its 10 MHz
    # FCLK ceiling, which shows up as probe failures or frame corruption
    # that look like a wiring fault.
    # core_freq_fixed=1 rather than a per-model core_freq_min: upstream says
    # "Use this in preference to setting specific core_clock frequencies as
    # it provides portability of config files between platforms" - it pins
    # to each board's own turbo frequency, so one line is correct on Pi 4,
    # CM4, Pi 5 and CM5 alike and no [pi4]/[cm4]/[pi5] sections are needed.
    # A fixed core_freq_min would have been wrong per board: 500 pins a CM4
    # (core_freq 500) but is merely the stock minimum on a Pi 5, whose core
    # runs at 910 - i.e. a silent no-op exactly where it was meant to help.
    # Not overclocking, hence not under the warranty-gated `overclock` block.
    if boot.get("core_freq_fixed"):
        core.append("core_freq_fixed=1")
    if boot.get("enable_i2c"):
        core.append("dtparam=i2c_arm=on")
    if boot.get("enable_spi"):
        core.append("dtparam=spi=on")
    if boot.get("enable_i2s"):
        core.append("dtparam=i2s=on")
    if boot.get("enable_uart"):
        core.append("enable_uart=1")
    # `bluetooth.enabled` is the single source of truth for the radio; the old
    # boot_config.disable_bluetooth toggle described the same thing from the
    # other side and nothing kept the two in sync.
    if not (cfg.get("bluetooth") or {}).get("enabled", True):
        core.append("dtoverlay=disable-bt")
    if boot.get("disable_wifi"):
        core.append("dtoverlay=disable-wifi")
    core.extend(_dtoverlay_lines(boot.get("dtoverlays", [])))
    _emit_boot_section(lines, "core", core)

    # --- camera ---------------------------------------------------------------
    cam = cfg.get("camera") or {}
    if cam.get("enabled"):
        body: list[str] = []
        if cam.get("legacy"):
            # Pi4 deprecated stack. Pi5 ignores these but does no harm.
            body.append("start_x=1")
            body.append("gpu_mem=128")
        # autodetect defaults to True when camera is enabled but the field is
        # missing - matches Raspberry Pi OS stock behaviour.
        autodetect = cam.get("autodetect", True)
        body.append(f"camera_auto_detect={1 if autodetect else 0}")
        for sensor in cam.get("sensors") or []:
            body.append(_overlay_line(sensor))
        _emit_boot_section(lines, "camera", body)

    # --- HDMI per-output ------------------------------------------------------
    hdmi = cfg.get("hdmi") or {}
    hdmi_body: list[str] = []
    for out in hdmi.get("outputs") or []:
        p = out["port"]
        # On Pi4+, options take an explicit :port suffix.
        def _opt(key: str, value: Any) -> str:
            return f"{key}:{p}={value}"
        if out.get("force_hotplug"):
            hdmi_body.append(_opt("hdmi_force_hotplug", 1))
        if "group" in out:
            hdmi_body.append(_opt("hdmi_group", out["group"]))
        if "mode" in out:
            hdmi_body.append(_opt("hdmi_mode", out["mode"]))
        drive = out.get("drive")
        audio_force = out.get("audio")
        if drive is not None or audio_force:
            # audio=True implies hdmi, even if drive field is missing; the
            # two options collapse into one line so we don't emit duplicates.
            resolved_drive = 1 if drive == "dvi" else 2
            hdmi_body.append(_opt("hdmi_drive", resolved_drive))
        if audio_force:
            hdmi_body.append(_opt("hdmi_ignore_edid_audio", 0))
        if "rotate" in out and out["rotate"]:
            # Pi4 legacy KMS path.  Pi5 supports the same option for
            # vc4-kms-v3d; anything else needs `video=` on the kernel cmdline.
            hdmi_body.append(_opt("display_hdmi_rotate", out["rotate"] // 90))
        if "boost" in out:
            # config_hdmi_boost is port-agnostic; only honour the last one.
            hdmi_body.append(f"config_hdmi_boost={out['boost']}")
    _emit_boot_section(lines, "hdmi", hdmi_body)

    # --- Display / console rotation -------------------------------------------
    disp = cfg.get("display") or {}
    disp_body: list[str] = []
    if "console_rotate" in disp and disp["console_rotate"]:
        # fbcon rotation is expressed as steps of 90 deg (0..3).
        disp_body.append(f"fbcon=rotate:{disp['console_rotate'] // 90}")
    if "lcd_rotate" in disp and disp["lcd_rotate"]:
        disp_body.append(f"display_lcd_rotate={disp['lcd_rotate'] // 90}")
    _emit_boot_section(lines, "display", disp_body)

    # --- Audio ----------------------------------------------------------------
    audio = cfg.get("audio") or {}
    audio_body: list[str] = []
    if "enabled" in audio:
        audio_body.append(f"dtparam=audio={'on' if audio['enabled'] else 'off'}")
    _emit_boot_section(lines, "audio", audio_body)
    # NOTE: default_output is a runtime ALSA setting, handled by bgrpiimage-hardware.

    # --- GPIO (1-wire) --------------------------------------------------------
    gpio = cfg.get("gpio") or {}
    onew = gpio.get("one_wire") or {}
    if onew.get("enabled"):
        pin = onew.get("pin", 4)
        _emit_boot_section(lines, "1-wire",
                           [_overlay_line("w1-gpio", {"gpiopin": pin})])

    # --- LEDs -----------------------------------------------------------------
    leds = cfg.get("leds") or {}
    led_body: list[str] = []
    # Pi4/5 trigger names that the rpi-eeprom bootloader plus kernel honour
    # via dtparam=act_led_trigger / pwr_led_trigger.
    led_trigger_map = {
        "on": "default-on",
        "off": "none",
        "heartbeat": "heartbeat",
        "mmc0": "mmc0",
        "default": None,
    }
    for which, dtparam_prefix in (("power", "pwr_led"), ("activity", "act_led")):
        val = leds.get(which)
        if val is None or val == "default":
            continue
        trig = led_trigger_map[val]
        if trig is None:
            continue
        led_body.append(f"dtparam={dtparam_prefix}_trigger={trig}")
        if val == "off":
            led_body.append(f"dtparam={dtparam_prefix}_activelow=off")
    _emit_boot_section(lines, "leds", led_body)

    # --- RTC (dt overlay; userspace side is bgrpiimage-hardware) --------------
    rtc = cfg.get("rtc") or {}
    if rtc.get("enabled"):
        model = rtc.get("model")
        if model:
            # dtoverlay=i2c-rtc,<model> needs the bare param, not key=value -
            # rendered by hand so we do not emit `ds3231=` with a trailing eq.
            overlay = f"dtoverlay=i2c-rtc,{model}"
            # Bus selection comes from i2c-buses.dtsi, which i2c-rtc includes:
            # it exposes i2c0..i2c6 as bare flags, with i2c1 the default target.
            # So only a non-default bus needs a flag. rtc.i2c_bus used to be
            # schema-validated and shown in the docs while never reaching
            # config.txt - a non-default bus was silently ignored.
            bus = rtc.get("i2c_bus", 1)
            if bus != 1:
                overlay += f",i2c{bus}"
            _emit_boot_section(lines, "rtc", [overlay])

    # --- Fan (dtoverlay for gpio-fan / rpi-fan / emc2301) ---------------------
    fan = cfg.get("fan") or {}
    if fan.get("enabled"):
        fan_body: list[str] = []
        mode = fan.get("mode", "gpio")
        if mode == "gpio":
            p: dict[str, Any] = {"gpiopin": fan.get("gpio", 14)}
            if "temp_on" in fan:
                p["temp"] = fan["temp_on"]
            # The overlay takes a hysteresis SPAN (hyst), while the config
            # declares an absolute switch-off temperature - so emit the
            # difference. Without this, temp_off validated against the schema
            # and was then silently dropped, leaving the overlay default.
            if "temp_on" in fan and "temp_off" in fan:
                hyst = fan["temp_on"] - fan["temp_off"]
                if hyst > 0:
                    p["hyst"] = hyst
                else:
                    console.print(
                        f"[yellow]WARNING:[/] fan.temp_off ({fan['temp_off']}) is not "
                        f"below fan.temp_on ({fan['temp_on']}) - hysteresis not emitted"
                    )
            fan_body.append(_overlay_line("gpio-fan", p))
        elif mode == "pwm":
            fan_body.append(_overlay_line("pwm-fan"))
        elif mode == "emc2301":
            # Pi5 Active Cooler / official cooling HAT is on by default; the
            # overlay forces detection when autoprobe fails (e.g. CM5 IO board).
            fan_body.append(_overlay_line("rpi-fan"))
        _emit_boot_section(lines, "fan", fan_body)

    # --- Overclock ------------------------------------------------------------
    ovc = cfg.get("overclock") or {}
    if ovc.get("enabled"):
        ovc_body: list[str] = []
        for k in ("arm_freq", "gpu_freq", "sdram_freq", "over_voltage"):
            if k in ovc:
                ovc_body.append(f"{k}={ovc[k]}")
        _emit_boot_section(lines, "overclock", ovc_body)

    # --- Memory split ---------------------------------------------------------
    mem = cfg.get("memory") or {}
    mem_body: list[str] = []
    for k in ("gpu_mem", "gpu_mem_256", "gpu_mem_512", "gpu_mem_1024", "cma"):
        if k in mem:
            if k == "cma":
                mem_body.append(f"dtoverlay=vc4-kms-v3d,cma-{mem[k]}")
            else:
                mem_body.append(f"{k}={mem[k]}")
    _emit_boot_section(lines, "memory", mem_body)

    # --- PCIe (Pi5/CM4/CM5) ---------------------------------------------------
    pcie = cfg.get("pcie") or {}
    if pcie.get("enabled"):
        pcie_body = ["dtparam=pciex1"]
        if "gen" in pcie:
            pcie_body.append(f"dtparam=pciex1_gen={pcie['gen']}")
        _emit_boot_section(lines, "pcie", pcie_body)

    # --- USB (Pi4 current cap) ------------------------------------------------
    usb = cfg.get("usb") or {}
    if usb.get("max_usb_current"):
        _emit_boot_section(lines, "usb", ["max_usb_current=1"])

    # --- extra_lines (escape hatch) -------------------------------------------
    extras = boot.get("extra_lines") or []
    _emit_boot_section(lines, "extra_lines", list(extras))

    # Trailing newline for a clean config.txt fragment.
    if lines and lines[-1] != "":
        lines.append("")
    write(gen / "config-bgrpiimage.txt", "\n".join(lines))


def render_hardware(cfg: dict[str, Any]) -> None:
    """Render runtime-side hardware artefacts.

    Config.txt lines live in render_boot(); this function produces the userspace
    pieces: hardware.env (consumed by the chroot script), packages.list, the
    optional EEPROM oneshot service + script. When no hardware block is active
    the module stays empty - _module_enabled() filters it out downstream.
    """
    gen = clean_generated("bgrpiimage-hardware")
    if not _module_enabled("bgrpiimage-hardware", cfg):
        # clean_generated leaves an empty dir; that's fine - _module_status
        # will count zero files and mark the module as "no artifacts".
        return
    # shell_var already terminates with "\n"; join without adding extra blanks.
    env_chunks: list[str] = [
        "# Auto-generated hardware.env - sourced by apply.sh\n",
    ]
    packages: set[str] = set()

    rtc = cfg.get("rtc") or {}
    env_chunks.append(shell_var("BGRPIIMAGE_RTC_ENABLED", "yes" if rtc.get("enabled") else "no"))
    if rtc.get("enabled"):
        packages.add("util-linux")  # hwclock lives in util-linux on Debian
    # NOT nested under rtc.enabled. fake-hwclock is the fallback for boards
    # with NO RTC HAT - it persists time across reboots by stamping a file on
    # shutdown. Gating it on rtc.enabled meant the documented "works even
    # without a HAT" case (enabled=false, fake_hwclock=true) installed nothing
    # at all, which is the one configuration it exists for.
    if rtc.get("fake_hwclock"):
        packages.add("fake-hwclock")
    env_chunks.append(shell_var(
        "BGRPIIMAGE_RTC_FAKE_HWCLOCK", "yes" if rtc.get("fake_hwclock") else "no"
    ))

    wd = cfg.get("watchdog") or {}
    env_chunks.append(shell_var("BGRPIIMAGE_WATCHDOG_ENABLED", "yes" if wd.get("enabled") else "no"))
    if wd.get("enabled"):
        # Both fallbacks come from the constants, not from literals repeated
        # here. They used to read 10 and 120, while the drop-in below already
        # used the constants - so hardware.env could carry a reboot_sec of 120,
        # a value the schema's own minimum of 240 forbids. Nothing reads these
        # two any more (apply.sh installs the rendered file), which is exactly
        # why the drift went unnoticed and why it is worth removing rather than
        # leaving as a second, quietly wrong source of truth.
        env_chunks.append(shell_var("BGRPIIMAGE_WATCHDOG_RUNTIME_SEC",
                                    wd.get("runtime_sec", WATCHDOG_RUNTIME_SEC_DEFAULT)))
        env_chunks.append(shell_var("BGRPIIMAGE_WATCHDOG_REBOOT_SEC",
                                    wd.get("reboot_sec", WATCHDOG_REBOOT_SEC_DEFAULT)))
        # Rendered as a FILE rather than expanded from hardware.env by a
        # heredoc in the apply script. bg_install compares content and only
        # writes on a real difference, which is what makes re-applying a
        # release silent and what lets the module queue an activation intent
        # from observed change instead of from having been run. A heredoc
        # rewrites the file every time and would make both impossible.
        write(
            gen / "system.conf.d" / "10-bgrpiimage-watchdog.conf",
            "# Auto-generated by bgrpiimage-hardware - do not edit by hand.\n"
            "[Manager]\n"
            f"RuntimeWatchdogSec={wd.get('runtime_sec', WATCHDOG_RUNTIME_SEC_DEFAULT)}\n"
            f"RebootWatchdogSec={wd.get('reboot_sec', WATCHDOG_REBOOT_SEC_DEFAULT)}\n",
        )

    audio = cfg.get("audio") or {}
    if audio.get("default_output"):
        env_chunks.append(shell_var("BGRPIIMAGE_AUDIO_DEFAULT_OUTPUT", audio["default_output"]))
        # Same reasoning as the watchdog drop-in: a generated file, so the
        # apply script installs it by content instead of re-writing it.
        pcm = _ALSA_PCM_BY_OUTPUT.get(audio["default_output"])
        if pcm:
            write(
                gen / "alsa" / "99-bgrpiimage-default.conf",
                "# Auto-generated by bgrpiimage-hardware.\n"
                f'pcm.!default {{ type plug; slave.pcm "{pcm}"; }}\n'
                "ctl.!default { type hw;   card 0; }\n",
            )

    fan = cfg.get("fan") or {}
    if fan.get("enabled") and fan.get("mode") in ("gpio", "pwm"):
        # gpio-fan and pwm-fan need a couple of sysfs knobs that the overlay
        # itself exposes; no additional packages strictly required. Kept here
        # for future trip-curve script drops.
        pass

    boot = cfg.get("bootloader") or {}
    if boot:
        # Userspace oneshot that applies EEPROM settings once on first boot.
        eeprom_env_lines: list[str] = [
            "# Auto-generated eeprom.env - consumed by bgrpiimage-eeprom-apply",
        ]
        for key, env_key in (
            ("boot_order",          "BOOT_ORDER"),
            ("wake_on_gpio",        "WAKE_ON_GPIO"),
            ("power_off_on_halt",   "POWER_OFF_ON_HALT"),
        ):
            if key not in boot:
                continue
            value = boot[key]
            if isinstance(value, bool):
                value = 1 if value else 0
            eeprom_env_lines.append(f"{env_key}={value}")
        write(gen / "eeprom.env", "\n".join(eeprom_env_lines) + "\n")

        # Shell script: read eeprom.env, merge into a temp config, apply once.
        apply_script = """#!/usr/bin/env bash
# Idempotent EEPROM config application. Runs once on first boot; the sentinel
# file keeps us from touching the bootloader on every reboot.
set -euo pipefail
SENTINEL=/var/lib/bgrpiimage/eeprom-applied
STAMP_DIR=/var/lib/bgrpiimage
mkdir -p "$STAMP_DIR"
[[ -f "$SENTINEL" ]] && exit 0

if ! command -v rpi-eeprom-config >/dev/null 2>&1; then
    echo "rpi-eeprom-config not installed - skipping" >&2
    touch "$SENTINEL"
    exit 0
fi

src=/etc/bgrpiimage/eeprom.env
[[ -f "$src" ]] || { echo "no eeprom.env"; exit 0; }

tmp=$(mktemp)
rpi-eeprom-config > "$tmp"
while IFS='=' read -r key value; do
    [[ -z "$key" || "$key" =~ ^# ]] && continue
    if grep -q "^${key}=" "$tmp"; then
        sed -i "s|^${key}=.*|${key}=${value}|" "$tmp"
    else
        echo "${key}=${value}" >> "$tmp"
    fi
done < "$src"

rpi-eeprom-config --apply "$tmp" || true
rm -f "$tmp"
touch "$SENTINEL"
"""
        write(gen / "bgrpiimage-eeprom-apply.sh", apply_script, executable=True)

        eeprom_unit = """[Unit]
Description=Apply BAUER GROUP EEPROM bootloader config (first boot)
After=network.target
ConditionPathExists=!/var/lib/bgrpiimage/eeprom-applied

[Service]
Type=oneshot
ExecStart=/usr/local/sbin/bgrpiimage-eeprom-apply
RemainAfterExit=yes

[Install]
WantedBy=multi-user.target
"""
        write(gen / "bgrpiimage-eeprom-apply.service", eeprom_unit)
        # rpi-eeprom ships preinstalled on Raspberry Pi OS trixie (arm64);
        # listing it would just no-op, so we skip adding it to packages.list.

    if packages:
        write(gen / "packages.list", "\n".join(sorted(packages)) + "\n")
    write(gen / "hardware.env", "".join(env_chunks))


def render_can(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-can")
    can = cfg.get("can", {})
    ifaces = can.get("interfaces", [])
    if not ifaces:
        return
    nwd = gen / "systemd-networkd"
    nwd.mkdir(parents=True, exist_ok=True)

    pkg_list = ["can-utils"]
    write(gen / "packages.list", "\n".join(pkg_list) + "\n")

    for iface in ifaces:
        name = iface["name"]
        bitrate = iface["bitrate"]
        txqlen = iface.get("txqueuelen", CAN_TXQUEUELEN_DEFAULT)
        auto_up = iface.get("auto_up", True)
        restart_ms = iface.get("restart_ms", CAN_RESTART_MS_DEFAULT)
        content = [
            "[Match]",
            f"Name={name}",
            "",
            "[CAN]",
            f"BitRate={bitrate}",
        ]
        # The "%" suffix is load bearing and so is the single decimal place.
        # networkd parses SamplePoint= with parse_permille() (percent-util.c),
        # which returns -EINVAL for any string without a "%" or permille suffix
        # AND for more than one digit after the dot - "0.875" and "87.55%" are
        # both rejected. The old renderer emitted a bare float, so the key could
        # never apply: networkd logged "Failed to parse SamplePoint=..., ignoring"
        # at warning level, dropped the key and applied the rest of the file, so
        # nothing surfaced except in the journal.
        # The schema keeps this in PERCENT (50..95). That lower bound catches
        # the fraction 0.875 that `ip -details link show` prints, which would
        # otherwise render as "SamplePoint=0.9%". systemd accepts that string
        # as 9 permille, but can_calc_bittiming() then finds no candidate
        # satisfying sample_point <= 9 and returns -EINVAL, so the link never
        # takes the timing - a device that boots with a dead CAN bus. Failing
        # in `make validate` is strictly better than failing in the field.
        # Note the kernel treats the value as a TARGET and rounds DOWN to the
        # nearest achievable point, silently: on this HAT at 500 kbit/s the
        # reachable set is 50.0/56.2/62.5/68.7/75.0/81.2/87.5, so a requested
        # 80 quietly becomes 75 and anything above 87.5 becomes 87.5.
        if "sample_point" in iface:
            content.append(f"SamplePoint={iface['sample_point']:.1f}%")
        # FDMode=yes and DataBitRate= are emitted together or not at all, and
        # that is not tidiness - it is the difference between a configured link
        # and an unconfigured one. can_validate() in
        # drivers/net/can/dev/netlink.c reads:
        #     if (data[IFLA_CAN_DATA_BITTIMING] || data[IFLA_CAN_TDC]) {
        #             if (!is_can_fd)
        #                     return -EOPNOTSUPP;
        #     }
        # and the converse for is_can_fd without both bittimings. That -EOPNOTSUPP
        # rejects the ENTIRE RTM_NEWLINK message, so a DataBitRate= written
        # without FDMode=yes does not merely skip the data phase: BitRate=,
        # SamplePoint= and RestartSec= travel in the same message and are lost
        # with it. The link then keeps whatever the driver powered up with.
        # systemd will not catch this for us - it has no coupling between the
        # two keys (CAN_CTRLMODE_FD is set only by FDMode=), so the mistake is
        # only visible as a netlink error in the journal.
        if "dbitrate" in iface:
            content.append("FDMode=yes")
            # Written as a plain integer on purpose. networkd parses this with
            # parse_size(rvalue, 1000), whose suffixes are UPPERCASE only - "2M"
            # works, "2m" is rejected with a warning and the key is dropped,
            # leaving an FD link that negotiated no data phase. Digits avoid the
            # whole class of mistake.
            content.append(f"DataBitRate={iface['dbitrate']}")
            if "data_sample_point" in iface:
                content.append(
                    f"DataSamplePoint={iface['data_sample_point']:.1f}%"
                )
            # Absent means "kernel default" (ISO 11898-1:2015). Emitted only when
            # explicitly asked for, because FDNonISO=no is NOT the same as
            # omitting it: networkd sets the mask bit either way, so writing it
            # unconditionally would force ISO mode over a driver that had a
            # reason to default otherwise.
            if iface.get("fd_non_iso"):
                content.append("FDNonISO=yes")
        # The "ms" suffix is load bearing. networkd parses RestartSec= with
        # parse_sec(), whose default unit is SECONDS - a bare "100" would mean
        # 100 s, i.e. restart-ms 100000, and the bus would stay dead for a
        # minute and a half per recovery. Nothing warns: the file parses fine.
        # 0 is omitted rather than emitted, because RestartSec=0 makes networkd
        # skip the netlink attribute entirely (leaving whatever the driver
        # already has) instead of meaning "off" - so writing it would be a lie.
        # The explicit off switch in systemd is "infinity", which we do not
        # expose: dropping the key achieves the same on a freshly booted link.
        if restart_ms:
            content.append(f"RestartSec={restart_ms}ms")
        # [Link] is emitted unconditionally, and RequiredForOnline=no is never
        # optional: it defaults to YES, and a CAN link never becomes routable,
        # so omitting it lets systemd-networkd-wait-online hold
        # network-online.target until it times out - with docker.service and
        # the Portainer first-boot install queued behind that target.
        # auto_up is not expressed by dropping the section either: under the
        # default ActivationPolicy=up networkd brings the link up regardless.
        # ActivationPolicy is the key that actually governs activation.
        content += ["", "[Link]", "RequiredForOnline=no"]
        if not auto_up:
            content.append("ActivationPolicy=manual")
        content.append("")
        write(nwd / f"40-{name}.network", "\n".join(content))

        # TransmitQueueLength is NOT a .network key - it belongs to a .link
        # file, which systemd-udevd (not networkd) applies when the netdev
        # appears. Both file types carry a section literally named [Link] with
        # disjoint key sets, so networkd accepted the file, logged the key as
        # unknown, dropped it, and left the CAN core default of 10 in place -
        # silently, because [CAN] BitRate= in the same file kept working.
        # The [Match] keys differ too: .network matches on Name=, .link has no
        # Name= at all and spells it OriginalName=.
        # 70- sorts ahead of /usr/lib/systemd/network/99-default.link, and the
        # FIRST matching .link file wins - so this one decides.
        link = [
            "[Match]",
            f"OriginalName={name}",
            "",
            "[Link]",
            f"TransmitQueueLength={txqlen}",
            "",
        ]
        write(nwd / f"70-{name}.link", "\n".join(link))


def render_docker(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-docker")
    docker = cfg.get("docker") or {}
    if not docker.get("enabled"):
        write(gen / ".disabled", "")
        return
    # daemon.json
    daemon = docker.get("daemon", {})
    write(gen / "daemon.json", json.dumps(daemon, indent=2) + "\n")
    # sysctl drop-in
    sysctl = docker.get("sysctl", {})
    if sysctl:
        lines = ["# Auto-generated by bgRPIImage"]
        for k, v in sysctl.items():
            lines.append(f"{k}={v}")
        write(gen / "98-docker.conf", "\n".join(lines) + "\n")
    # Networks to create post-install (one-shot service)
    networks = docker.get("networks", [])
    create_lines = ["#!/bin/bash", "# Auto-generated docker network creation", "set -euo pipefail", ""]
    for n in networks:
        args = ["docker network create"]
        args.append(f"--driver={n.get('driver', 'bridge')}")
        if n.get("subnet"):
            args.append(f"--subnet={n['subnet']}")
        if n.get("gateway"):
            args.append(f"--gateway={n['gateway']}")
        if n.get("ipv6"):
            args.append("--ipv6")
            if n.get("subnet_v6"):
                args.append(f"--subnet={n['subnet_v6']}")
            if n.get("gateway_v6"):
                args.append(f"--gateway={n['gateway_v6']}")
        for k, v in (n.get("options") or {}).items():
            args.append(f'-o "{k}={v}"')
        args.append(shlex.quote(n["name"]))
        create_lines.append(
            f"docker network inspect {shlex.quote(n['name'])} >/dev/null 2>&1 || \\"
        )
        create_lines.append("  " + " ".join(args))
    write(gen / "create-networks.sh", "\n".join(create_lines) + "\n", executable=True)
    # ipv6 masquerade helper unit (replaces stock docker-support)
    #
    # The prefix is read from daemon.json rather than hardcoded: the two used
    # to be edited independently and could silently drift apart.
    #
    # Only the default bridge's fixed-cidr-v6 is masqueraded. The IPv6
    # default-address-pools behind user-defined networks are deliberately left
    # out: their traffic leaves via br-* interfaces, and the `! -o docker0`
    # exclusion that keeps intra-bridge traffic un-NATed has no correct
    # equivalent there without enumerating bridges at runtime.
    fixed_v6 = (daemon.get("fixed-cidr-v6") or "").strip()
    if fixed_v6:
        # `-C` is the idempotence check and it FAILS when the rule is absent -
        # i.e. on every clean boot. Under Type=oneshot a bare ExecStart that
        # exits non-zero aborts the unit, so the previous form (an unprefixed
        # -C followed by a `-`-prefixed -A) could never reach the -A: the rule
        # was never installed and the unit sat permanently failed. Chaining
        # both in one shell lets the check do the job it was written for.
        _t = f"-t nat POSTROUTING -s {fixed_v6} ! -o docker0 -j MASQUERADE"
        _chk = "/usr/sbin/ip6tables " + _t.replace("POSTROUTING", "-C POSTROUTING", 1)
        _add = "/usr/sbin/ip6tables " + _t.replace("POSTROUTING", "-A POSTROUTING", 1)
        exec_line = f'ExecStart=/bin/sh -c "{_chk} 2>/dev/null || {_add}"\n'
    else:
        exec_line = ""

    unit = (
        "[Unit]\n"
        "Description=BAUER GROUP IPv6 NAT for Docker\n"
        "BindsTo=docker.service\n"
        "After=docker.service\n"
        "ReloadPropagatedFrom=docker.service\n"
        "\n"
        "[Service]\n"
        "Type=oneshot\n"
        + exec_line +
        "RemainAfterExit=yes\n"
        "\n"
        "[Install]\n"
        "WantedBy=multi-user.target\n"
    )
    write(gen / "docker-support.service", unit)


# netavark's own id for the built-in default network. Reusing it keeps
# `podman network inspect podman` byte-identical to a stock install; a
# different id is legal but makes every diff against a stock box noise.
_NETAVARK_DEFAULT_NETWORK_ID = (
    "2f259bab93aaaaa2542ba43ef33eb990d0999ee1b9924b557b7be53c0b7a1bb9"
)


def render_podman(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-podman")
    p = cfg.get("podman") or {}
    if not p.get("enabled"):
        write(gen / ".disabled", "")
        return

    net = p.get("network") or {}

    # containers.conf carries the IPv4 half of the address plan. The IPv6
    # half cannot live here: containers.conf has no key that enables IPv6
    # on the default network - the only declarative path is the network
    # definition file written below.
    cc = ["# Auto-generated by scripts/generate.py", "", "[network]"]
    if net.get("default_subnet"):
        cc.append(f'default_subnet = "{net["default_subnet"]}"')
    pools = net.get("default_subnet_pools") or []
    if pools:
        cc.append("default_subnet_pools = [")
        for pool in pools:
            cc.append(f'  {{"base" = "{pool["base"]}", "size" = {pool["size"]}}},')
        cc.append("]")
    if net.get("firewall_driver"):
        cc.append(f'firewall_driver = "{net["firewall_driver"]}"')
    write(gen / "containers.conf", "\n".join(cc) + "\n")

    # /etc/containers/networks/podman.json - loaded by netavark on the next
    # podman invocation, no daemon to restart. Two silent failure modes:
    # the filename must match the `name` field, and `id` must be 64 hex
    # chars. Violate either and netavark skips the file with nothing but a
    # log line, falling back to a stock IPv4-only default network.
    subnets: list[dict[str, Any]] = []
    if net.get("default_subnet"):
        subnets.append({"subnet": net["default_subnet"]})
    if net.get("ipv6") and net.get("subnet_v6"):
        subnets.append({"subnet": net["subnet_v6"]})
    network = {
        "name": "podman",
        "id": _NETAVARK_DEFAULT_NETWORK_ID,
        "driver": "bridge",
        "network_interface": "podman0",
        "subnets": subnets,
        "ipv6_enabled": bool(net.get("ipv6")),
        "internal": False,
        "dns_enabled": True,
    }
    write(gen / "podman-network.json", json.dumps(network, indent=2) + "\n")

    # Runtime-agnostic kernel setting. It lived under `docker` until the
    # runtime switch; any container workload that mmaps heavily needs it
    # regardless of which engine starts the container.
    sysctl = p.get("sysctl") or {}
    lines = ["# Auto-generated by scripts/generate.py"]
    for key, value in sysctl.items():
        lines.append(f"{key}={value}")
    write(gen / "98-podman.conf", "\n".join(lines) + "\n")

    # Podman logs containers to journald, which this image did not bound
    # before: Docker's json-file driver capped them at 10m x 3 and that cap
    # leaves with Docker. Unbounded container logs on an SD card is a
    # wear-out, not an inconvenience.
    jd = p.get("journald") or {}
    write(
        gen / "99-bgrpiimage-containers.conf",
        "# Auto-generated by scripts/generate.py\n"
        "[Journal]\n"
        f"SystemMaxUse={jd.get('system_max_use', '200M')}\n"
        f"SystemMaxFileSize={jd.get('system_max_file_size', '20M')}\n",
    )

    # Presence alone suppresses "Emulate Docker CLI using podman" on every
    # docker(1) invocation. Content is irrelevant.
    write(gen / "nodocker", "")

    au = p.get("auto_update") or {}
    write(
        gen / "podman.env",
        shell_var("BGRPIIMAGE_PODMAN_DOCKER_EMULATION", p.get("docker_emulation", True))
        + shell_var("BGRPIIMAGE_PODMAN_AUTO_UPDATE", au.get("enabled", False)),
    )

    if not au.get("enabled"):
        return

    sched = au.get("schedule") or {}
    start = sched.get("start", "05:30")
    delay_min = int(sched.get("randomized_delay_minutes", 30))
    persistent = "true" if sched.get("persistent", True) else "false"

    # The empty OnCalendar= is load-bearing: without it the shipped
    # `OnCalendar=daily` stays active alongside this one and the timer fires
    # twice. Same shape as the apt-daily-upgrade override.
    write(
        gen / "podman-auto-update.timer.d/override.conf",
        "# Auto-generated by scripts/generate.py\n"
        "[Timer]\n"
        "OnCalendar=\n"
        f"OnCalendar=*-*-* {start}:00\n"
        f"RandomizedDelaySec={delay_min * 60}\n"
        f"Persistent={persistent}\n",
    )

    backup = ((cfg.get("portainer") or {}).get("backup_before_update") or {})
    if not backup.get("enabled", True):
        return

    keep = int(backup.get("keep", 5))
    write(
        gen / "podman-auto-update.service.d/10-backup.conf",
        "# Auto-generated by scripts/generate.py\n"
        "[Service]\n"
        "ExecStartPre=/usr/local/sbin/bgrpiimage-portainer-backup\n",
    )
    write(
        gen / "bgrpiimage-portainer-backup",
        "#!/usr/bin/env bash\n"
        "# Auto-generated by scripts/generate.py\n"
        "#\n"
        "# Snapshot the Portainer volume before podman-auto-update runs.\n"
        "# Portainer migrates its database one-way on startup: a portainer.db\n"
        "# written by a newer version will not open on an older one, so\n"
        "# podman's own rollback cannot undo a bad update on its own.\n"
        "#\n"
        "# Runs on EVERY timer fire, not only when an update is available,\n"
        "# so pruning to the last KEEP archives is what bounds disk use.\n"
        "set -uo pipefail\n"
        "\n"
        "# The archive is a dump of portainer_data: Portainer's admin password\n"
        "# hash, its JWT signing secret and any stored registry/environment\n"
        "# credentials. umask + an explicit 700 dir keep it unreadable by any\n"
        "# other local account under systemd's default UMask=0022.\n"
        "umask 077\n"
        "\n"
        f"KEEP={keep}\n"
        'DEST=/var/backups/bgrpiimage\n'
        "\n"
        "# Never fail: ExecStartPre= aborts the unit on a non-zero exit, so a\n"
        "# full disk here would block every future update instead of just\n"
        "# skipping one backup.\n"
        "install -d -m 700 \"$DEST\" || exit 0\n"
        "command -v podman >/dev/null 2>&1 || exit 0\n"
        "podman volume exists portainer_data >/dev/null 2>&1 || exit 0\n"
        "\n"
        'stamp=$(date +%Y%m%dT%H%M%S)\n'
        'if ! podman volume export portainer_data 2>/dev/null '
        '| gzip -c > "$DEST/portainer-$stamp.tar.gz"; then\n'
        '    echo "bgrpiimage: portainer volume backup failed, continuing" >&2\n'
        '    rm -f "$DEST/portainer-$stamp.tar.gz"\n'
        "    exit 0\n"
        "fi\n"
        "\n"
        '# shellcheck disable=SC2012  # names are generated, no odd characters\n'
        'ls -1t "$DEST"/portainer-*.tar.gz 2>/dev/null '
        '| tail -n +$((KEEP + 1)) | while read -r old; do rm -f "$old"; done\n'
        "\n"
        "exit 0\n",
        executable=True,
    )


def render_portainer(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-portainer")
    p = cfg.get("portainer") or {}
    if not p.get("enabled"):
        return
    bind = p.get("bind", "0.0.0.0")
    edition = p.get("edition", "ce")
    # Fully qualified: podman resolves unqualified names through
    # unqualified-search-registries, which this image does not ship, and
    # AutoUpdate=registry requires a fully qualified reference regardless.
    image = p.get("image") or (
        "docker.io/portainer/portainer-ce:2.45.0"
        if edition == "ce"
        else "docker.io/portainer/portainer-ee:2.45.0"
    )
    ports = p.get("ports") or {}
    edge = ports.get("edge", 8000)
    http = ports.get("http", 9000)
    https = ports.get("https", 9443)

    # Port mapping with dual-stack awareness:
    #   bind == "0.0.0.0" (default)  -> bind without host IP, so Docker
    #                                   listens on both 0.0.0.0:PORT and
    #                                   [::]:PORT (IPv6) when daemon.json
    #                                   has ipv6:true. "0.0.0.0:PORT:PORT"
    #                                   would restrict to IPv4 only.
    #   any other value              -> pin to that specific address (e.g.
    #                                   "127.0.0.1" for loopback only,
    #                                   "::1" for v6 loopback).
    def port(host_port: int, container_port: int) -> str:
        if bind in ("0.0.0.0", "", None):
            return f'"{host_port}:{container_port}"'
        return f'"{bind}:{host_port}:{container_port}"'

    if (cfg.get("podman") or {}).get("enabled"):
        # Quadlet. No compose file, no first-boot oneshot: systemd's
        # generator turns these two files into units at boot and applies
        # their [Install] section itself, so there is nothing to enable in
        # the chroot.
        write(
            gen / "portainer.image",
            "# Auto-generated by bgRPIImage\n"
            "[Image]\n"
            f"Image={image}\n",
        )

        # PublishPort takes the same [IP:]host:container syntax as
        # `podman run --publish`, so the dual-stack rule is unchanged: with
        # bind=0.0.0.0 the host IP is omitted and the listener covers v4
        # and v6; naming 0.0.0.0 explicitly would pin it to v4 only.
        def qport(host_port: int, container_port: int) -> str:
            if bind in ("0.0.0.0", "", None):
                return f"PublishPort={host_port}:{container_port}"
            return f"PublishPort={bind}:{host_port}:{container_port}"

        auto_update = "AutoUpdate=registry\n" if p.get("auto_update") else ""
        # [Install] is what actually wires the generated unit into
        # multi-user.target - Quadlet applies it itself at boot, "in the same
        # way systemctl enable does when run later" (podman-systemd.unit(5)).
        # Omitting the whole section is therefore how `auto_start: false` is
        # honoured under Podman: Quadlet still generates portainer.service,
        # but nothing pulls it in at boot, so it stays inert until started by
        # hand. Shipping [Install] unconditionally (the pre-fix behaviour)
        # started Portainer regardless of this setting.
        install_section = (
            "\n[Install]\nWantedBy=multi-user.target\n"
            if p.get("auto_start", True)
            else ""
        )
        unit = (
            "# Auto-generated by bgRPIImage\n"
            "[Unit]\n"
            "Description=Portainer CE\n"
            "Requires=podman.socket\n"
            "After=podman.socket\n"
            "\n"
            "[Container]\n"
            # Resolves to the .image unit above, which makes the generated
            # service depend on portainer-image.service. Without it the
            # first pull counts against this unit's start timeout.
            "Image=portainer.image\n"
            "ContainerName=portainer\n"
            f"{auto_update}"
            f"{qport(edge, 8000)}\n"
            f"{qport(http, 9000)}\n"
            f"{qport(https, 9443)}\n"
            # Portainer's own Podman install docs bind the podman socket at
            # the path the Docker client defaults to. No :z - Debian ships
            # no SELinux policy and the suffix would be a no-op at best.
            "Volume=/run/podman/podman.sock:/var/run/docker.sock\n"
            "Volume=portainer_data:/data\n"
            # No dedicated privileged key exists in [Container] at 5.4.2.
            "PodmanArgs=--privileged\n"
            "\n"
            "[Service]\n"
            # Quadlet emits no Restart= for .container units, and
            # --restart=always inside PodmanArgs would be ignored because
            # systemd owns the lifecycle.
            "Restart=always\n"
            "TimeoutStartSec=900\n"
            f"{install_section}"
        )
        write(gen / "portainer.container", unit)
        write(
            gen / "portainer.env",
            shell_var("BGRPIIMAGE_PORTAINER_AUTOSTART", p.get("auto_start", True))
            + shell_var("BGRPIIMAGE_PORTAINER_IMAGE", image)
            + shell_var("BGRPIIMAGE_PORTAINER_RUNTIME", "podman"),
        )
        return

    # Declarative compose file - Docker's restart=unless-stopped policy
    # handles lifecycle after first-boot install. No systemd wrapping around
    # the running container, no --rm, no double supervisor.
    compose = (
        "# Auto-generated by bgRPIImage\n"
        "# Operator workflow:\n"
        "#   sudo docker compose -f /etc/bgrpiimage/portainer/docker-compose.yml pull\n"
        "#   sudo docker compose -f /etc/bgrpiimage/portainer/docker-compose.yml up -d\n"
        "\n"
        "services:\n"
        "  portainer:\n"
        f"    image: {image}\n"
        "    container_name: portainer\n"
        "    restart: unless-stopped\n"
        "    ports:\n"
        f"      - {port(edge, 8000)}\n"
        f"      - {port(http, 9000)}\n"
        f"      - {port(https, 9443)}\n"
        "    volumes:\n"
        "      - /var/run/docker.sock:/var/run/docker.sock\n"
        "      - portainer_data:/data\n"
        "\n"
        "volumes:\n"
        "  portainer_data:\n"
        "    name: portainer\n"
    )
    write(gen / "docker-compose.yml", compose)

    # First-boot oneshot. After it runs once, the Docker daemon itself
    # keeps the container alive via the compose file's restart policy.
    #
    # Type=oneshot disables TimeoutStartSec by default, so a registry pull
    # that stalls would hold multi-user.target open indefinitely (targets get
    # an implicit After= on what they Want). Bound it, and wait for the
    # network first so the pull does not race DNS coming up.
    install_unit = (
        "[Unit]\n"
        "Description=bgRPIImage Portainer bootstrap (first boot only)\n"
        "Requires=docker.service\n"
        "Wants=network-online.target\n"
        "After=docker.service docker-support.service bgrpiimage-docker-networks.service network-online.target\n"
        "ConditionPathExists=!/var/lib/bgrpiimage/portainer.installed\n"
        "\n"
        "[Service]\n"
        "Type=oneshot\n"
        "RemainAfterExit=yes\n"
        "TimeoutStartSec=600\n"
        "WorkingDirectory=/etc/bgrpiimage/portainer\n"
        "ExecStart=/usr/bin/docker compose up -d\n"
        "ExecStartPost=/bin/sh -c \"mkdir -p /var/lib/bgrpiimage && touch /var/lib/bgrpiimage/portainer.installed\"\n"
        "\n"
        "[Install]\n"
        "WantedBy=multi-user.target\n"
    )
    write(gen / "bgrpiimage-portainer-install.service", install_unit)
    write(
        gen / "portainer.env",
        "".join(
            [
                shell_var("BGRPIIMAGE_PORTAINER_AUTOSTART", bool(p.get("auto_start", True))),
                shell_var("BGRPIIMAGE_PORTAINER_IMAGE", image),
            ]
        ),
    )


def render_unattended(cfg: dict[str, Any]) -> None:
    gen = clean_generated("bgrpiimage-unattended-upgrades")
    u = cfg.get("unattended_upgrades") or {}
    if not u.get("enabled"):
        write(gen / ".disabled", "")
        return

    origins = u.get("allowed_origins") or []
    blocklist = u.get("package_blocklist") or []
    remove_unused = u.get("remove_unused_dependencies", True)
    mail = u.get("mail") or {}
    reboot = u.get("auto_reboot") or {}
    schedule = u.get("schedule") or {}

    # 50unattended-upgrades
    cfg_lines = ["// Auto-generated by scripts/generate.py", "Unattended-Upgrade::Origins-Pattern {"]
    for origin in origins:
        cfg_lines.append(f'    "{origin}";')
    cfg_lines.append("};")
    cfg_lines.append("")
    cfg_lines.append("Unattended-Upgrade::Package-Blacklist {")
    for pkg in blocklist:
        cfg_lines.append(f'    "{pkg}";')
    cfg_lines.append("};")
    cfg_lines.append("")
    cfg_lines.append(f'Unattended-Upgrade::Remove-Unused-Dependencies "{"true" if remove_unused else "false"}";')
    cfg_lines.append('Unattended-Upgrade::Remove-New-Unused-Dependencies "true";')
    # Reboot is handled by our own window service, not by u-u directly.
    cfg_lines.append('Unattended-Upgrade::Automatic-Reboot "false";')
    if mail.get("address"):
        cfg_lines.append(f'Unattended-Upgrade::Mail "{mail["address"]}";')
        cfg_lines.append(f'Unattended-Upgrade::MailOnlyOnError "{"true" if mail.get("on_error_only", True) else "false"}";')
    cfg_lines.append("")
    write(gen / "50unattended-upgrades", "\n".join(cfg_lines))

    # 20auto-upgrades - enables apt to call u-u.
    write(
        gen / "20auto-upgrades",
        "APT::Periodic::Update-Package-Lists \"1\";\n"
        "APT::Periodic::Download-Upgradeable-Packages \"1\";\n"
        "APT::Periodic::AutocleanInterval \"7\";\n"
        "APT::Periodic::Unattended-Upgrade \"1\";\n",
    )

    # apt-daily-upgrade.timer override -> fires inside [start, end] window.
    start = schedule.get("start", "02:00")
    end = schedule.get("end", "04:00")
    persistent = "true" if schedule.get("persistent", True) else "false"
    window_minutes = _window_minutes(start, end)
    override = (
        "[Timer]\n"
        "OnCalendar=\n"
        f"OnCalendar=*-*-* {start}:00\n"
        f"RandomizedDelaySec={window_minutes * 60}\n"
        f"Persistent={persistent}\n"
    )
    write(gen / "apt-daily-upgrade.timer.d/override.conf", override)
    # same cadence for apt-daily (download) - shift 30 min earlier within bounds
    dl_override = (
        "[Timer]\n"
        "OnCalendar=\n"
        f"OnCalendar=*-*-* {start}:00\n"
        f"RandomizedDelaySec=1800\n"
        f"Persistent={persistent}\n"
    )
    write(gen / "apt-daily.timer.d/override.conf", dl_override)

    # Reboot window service + timer
    if reboot.get("enabled"):
        rwin = reboot.get("window", {})
        r_start = rwin.get("start", "03:00")
        r_end = rwin.get("end", "05:00")
        r_window = _window_minutes(r_start, r_end)
        if_required = reboot.get("if_required_only", True)

        # The check script is invoked from both
        #   (a) the maintenance timer (safety net, runs once per day)
        #   (b) apt-daily-upgrade.service ExecStartPost (event-driven, runs
        #       right after each update attempt)
        # A reboot only happens when:
        #   - /var/run/reboot-required exists (set by kernel / libc / etc.
        #     package post-install hooks) AND
        #   - current local time is inside the configured reboot window.
        check_script = [
            "#!/bin/bash",
            "# Auto-generated: reboot iff a package upgrade set the",
            "# /var/run/reboot-required flag AND we are inside the window.",
            "set -euo pipefail",
            f'WINDOW_START="{r_start}"',
            f'WINDOW_END="{r_end}"',
            f'IF_REQUIRED_ONLY={"1" if if_required else "0"}',
            'TAG="bgrpiimage-reboot"',
            'now=$(date +%H:%M)',
            'in_window() {',
            '  local now=$1 start=$2 end=$3',
            '  if [[ "$start" < "$end" ]]; then',
            '    [[ "$now" > "$start" || "$now" == "$start" ]] && [[ "$now" < "$end" ]]',
            '  else',
            '    [[ "$now" > "$start" || "$now" == "$start" || "$now" < "$end" ]]',
            '  fi',
            '}',
            '# (1) guard: must have a pending reboot request from a package',
            'if [[ $IF_REQUIRED_ONLY -eq 1 ]]; then',
            '  if [[ ! -f /var/run/reboot-required ]]; then',
            '    logger -t "$TAG" "no reboot required - skipping"',
            '    exit 0',
            '  fi',
            'fi',
            '# (2) guard: must be inside the configured reboot window',
            'if ! in_window "$now" "$WINDOW_START" "$WINDOW_END"; then',
            '  logger -t "$TAG" "reboot required but outside window ($now not in $WINDOW_START-$WINDOW_END) - deferring"',
            '  exit 0',
            'fi',
            '# (3) log the packages that triggered the reboot',
            'pkgs=""',
            'if [[ -s /var/run/reboot-required.pkgs ]]; then',
            '  pkgs=$(tr "\\n" " " < /var/run/reboot-required.pkgs)',
            'fi',
            'logger -t "$TAG" "rebooting inside window ${WINDOW_START}-${WINDOW_END} (triggered by: ${pkgs:-unknown})"',
            '/sbin/shutdown -r +1 "bgRPIImage: scheduled reboot after unattended-upgrade (${pkgs:-kernel/system package update})"',
            '',
        ]
        write(gen / "bgrpiimage-reboot-window.sh", "\n".join(check_script), executable=True)

        # Event-driven trigger: run the reboot-window check right after every
        # apt-daily-upgrade.service execution. '-' prefix ignores failures so
        # a broken script never blocks the upgrade service from succeeding.
        apt_upgrade_dropin = (
            "[Service]\n"
            "ExecStartPost=-/usr/local/sbin/bgrpiimage-reboot-window.sh\n"
        )
        write(gen / "apt-daily-upgrade.service.d/override.conf", apt_upgrade_dropin)

        svc = (
            "[Unit]\n"
            "Description=BAUER GROUP conditional reboot after unattended-upgrade\n"
            "After=apt-daily-upgrade.service\n"
            "\n"
            "[Service]\n"
            "Type=oneshot\n"
            "ExecStart=/usr/local/sbin/bgrpiimage-reboot-window.sh\n"
        )
        write(gen / "bgrpiimage-reboot-window.service", svc)

        tmr = (
            "[Unit]\n"
            "Description=BAUER GROUP reboot window check\n"
            "\n"
            "[Timer]\n"
            f"OnCalendar=*-*-* {r_start}:00\n"
            f"RandomizedDelaySec={r_window * 60}\n"
            "Persistent=true\n"
            "Unit=bgrpiimage-reboot-window.service\n"
            "\n"
            "[Install]\n"
            "WantedBy=timers.target\n"
        )
        write(gen / "bgrpiimage-reboot-window.timer", tmr)


# The chip selects mcp251xfd-overlay.dts actually implements, as the eight
# `spi<n>-<m>` booleans in its __overrides__ block. spi0-2 is deliberately
# absent: SPI0 has two hardware chip selects. A typo here is not a build error
# on the device - the dtoverlay parameter matcher ignores names it does not
# know, so `spi0-2` would leave the overlay on its default target (spi0.0) and
# quietly stack a second controller onto the first one's chip select.
_MCP251XFD_CS = {
    "spi0-0": (0, 0), "spi0-1": (0, 1),
    "spi1-0": (1, 0), "spi1-1": (1, 1), "spi1-2": (1, 2),
    "spi2-0": (2, 0), "spi2-1": (2, 1), "spi2-2": (2, 2),
}
# spi1-1cs / spi1-2cs / spi1-3cs (and the spi2 twins) - the overlays that turn
# the AUX SPI controllers on. The digit is how many chip selects they expose.
_SPI_AUX_OVERLAY_RE = re.compile(r"^spi([12])-([123])cs$")


def _validate_mcp251xfd(cfg: dict[str, Any], overlays: list[dict[str, Any]]) -> None:
    """Cross-check mcp251xfd overlays against can.interfaces and each other.

    The MCP251XFD family is configured very differently from the MCP2515 and
    every difference is a fresh way to ship a silently dead channel:

    - ONE overlay named `mcp251xfd` serves every channel, so the entries carry
      no interface name to match against. The only structural check available
      is that there is exactly one overlay per interface.
    - The chip select is a BOOLEAN parameter (`spi0-0`), not part of the overlay
      name. The README marks it "(boolean, required)"; omit it and the overlay
      silently lands on its default target.
    - `speed` sets the SPI clock here. `spimaxfrequency` is the MCP2515
      spelling, and the dtoverlay parameter matcher drops unknown names without
      complaint, so writing it costs the setting and says nothing.
    """
    fd = [o for o in overlays if o["name"] == "mcp251xfd"]
    ifaces = (cfg.get("can") or {}).get("interfaces", [])

    if not fd:
        # dbitrate on a board whose overlays are all MCP2515 is not a slow FD
        # link, it is a dead one: mcp251x has no FD support at all, so the
        # kernel rejects CAN_CTRLMODE_FD and takes the whole netlink message
        # with it - including the nominal bitrate.
        classic = [i["name"] for i in ifaces if "dbitrate" in i]
        if classic and any(_MCP2515_OVERLAY_RE.match(o["name"]) for o in overlays):
            raise ValueError(
                f"can.interfaces {classic} set dbitrate (CAN FD) but the overlays are "
                "mcp2515-*, and the MCP2515 is a Classic-CAN-only controller - the "
                "kernel refuses CAN_CTRLMODE_FD and the whole link configuration "
                "fails with it. Use an mcp251xfd (MCP2517FD/MCP2518FD) board, or "
                "drop dbitrate."
            )
        return

    if len(fd) != len(ifaces):
        raise ValueError(
            f"boot_config.dtoverlays has {len(fd)} mcp251xfd overlay(s) but "
            f"can.interfaces declares {len(ifaces)} interface(s). One overlay "
            "creates exactly one controller, and mcp251xfd carries no interface "
            "name to match on - so the counts are the only cross-check there is. "
            "If two entries were meant to be distinct, give each its own `id`: "
            "without it they merge by name into a single overlay."
        )

    seen_cs: dict[str, str] = {}
    seen_int: dict[str, str] = {}
    buses: dict[int, int] = {}          # spi bus -> highest chip select used
    first_use: dict[int, int] = {}      # spi bus -> index of its first overlay

    for idx, ovl in enumerate(overlays):
        if ovl["name"] != "mcp251xfd":
            continue
        params = ovl.get("params") or {}
        label = f"mcp251xfd#{idx}" + (f" (id={ovl['id']})" if ovl.get("id") else "")

        if "spimaxfrequency" in params:
            raise ValueError(
                f"boot_config.dtoverlays: {label} sets params.spimaxfrequency, which "
                "is the mcp2515-can0/can1 spelling. The mcp251xfd overlay calls it "
                "`speed`, and dtoverlay silently ignores parameter names it does not "
                "recognise - the SPI clock would stay at the overlay default and "
                "nothing would say so."
            )

        selectors = [k for k, v in params.items() if k in _MCP251XFD_CS and v]
        unknown = [
            k for k in params
            if k.startswith("spi") and "-" in k and k not in _MCP251XFD_CS
        ]
        if unknown:
            raise ValueError(
                f"boot_config.dtoverlays: {label} sets {unknown}, which the "
                f"mcp251xfd overlay does not implement. Valid chip selects are "
                f"{sorted(_MCP251XFD_CS)} - note SPI0 has no third chip select."
            )
        if len(selectors) != 1:
            raise ValueError(
                f"boot_config.dtoverlays: {label} must set exactly one spi<n>-<m> "
                f"chip-select parameter (found {selectors or 'none'}). The overlay "
                "README marks it '(boolean, required)'; with none the overlay stays "
                "on its default target and two channels collide on spi0.0."
            )
        cs = selectors[0]
        if cs in seen_cs:
            raise ValueError(
                f"boot_config.dtoverlays: {label} and {seen_cs[cs]} both select {cs}; "
                "two controllers cannot share one chip select"
            )
        seen_cs[cs] = label

        bus, cs_idx = _MCP251XFD_CS[cs]
        buses[bus] = max(buses.get(bus, -1), cs_idx)
        first_use.setdefault(bus, idx)

        pin = params.get("interrupt")
        if pin is None:
            raise ValueError(
                f"boot_config.dtoverlays: {label} must set params.interrupt - the "
                "mcp251xfd overlay defaults to GPIO 25 for every instance, so a "
                "second channel without it collides with the first. A wrong or "
                "missing INT is silent: the driver requests the IRQ late, so the "
                "interface still appears and simply never receives."
            )
        if str(pin) in seen_int:
            raise ValueError(
                f"boot_config.dtoverlays: {label} and {seen_int[str(pin)]} both use "
                f"interrupt={pin}; each controller needs its own INT GPIO"
            )
        seen_int[str(pin)] = label

        # The driver clamps the SPI clock to 0.85 x (osc / 2) to honour silicon
        # errata DS80000789 item 4 (SPI writes corrupt RAM at high SCK while the
        # CAN bus is active):
        #     priv->spi_max_speed_hz_slow = min(spi->max_speed_hz,
        #                                       freq / 2 / 1000 * 850);
        # A value above that ceiling is therefore SAFE - min() discards it, and
        # the upstream overlay itself defaults to 20 MHz, which exceeds the
        # ceiling for the usual 40 MHz crystal (the 20 MHz figure is a leftover
        # from datasheet revision A of April 2019; revision B, December 2020,
        # cut FSCK to 17 MHz). Writing it is a documentation choice, not a
        # hazard, so this notes the discrepancy instead of refusing the build:
        # config.txt then says 20 MHz where the bus runs at 17, and the boot
        # banner prints both ("m:20.00MHz rs:17.00MHz").
        osc = params.get("oscillator")
        speed = params.get("speed")
        if osc is not None and speed is not None:
            ceiling = int(osc) // 2 // 1000 * 850
            if int(speed) > ceiling:
                err_console.print(
                    f"[yellow]note:[/] {label} sets speed={speed}, which the "
                    f"mcp251xfd driver clamps to {ceiling} Hz for a {osc} Hz "
                    "oscillator (0.85 x osc/2, silicon errata DS80000789 item 4). "
                    "Harmless - the bus runs at the clamped rate - but config.txt "
                    "then states a clock nothing honours."
                )

    # An mcp251xfd on spi1/spi2 needs that controller switched on separately -
    # the overlay only enables spi0. The dependency is stronger than a status
    # flag: mcp251xfd disables the conflicting spidev node with
    # `target-path = "spi1/spidev@0"`, and a target-path can only resolve
    # against a node that already exists. Applied before spi1-Ncs has created
    # that node the fragment silently does nothing, and spidev then claims the
    # same chip select as the CAN controller.
    aux: dict[int, int] = {}
    aux_at: dict[int, int] = {}
    for idx, ovl in enumerate(overlays):
        m = _SPI_AUX_OVERLAY_RE.match(ovl["name"])
        if m:
            aux[int(m.group(1))] = int(m.group(2))
            aux_at.setdefault(int(m.group(1)), idx)

    for bus, highest in sorted(buses.items()):
        if bus == 0:
            continue
        need = highest + 1
        have = aux.get(bus)
        if have is None:
            raise ValueError(
                f"boot_config.dtoverlays: an mcp251xfd sits on spi{bus} but no "
                f"spi{bus}-<N>cs overlay enables that controller. Add "
                f"{{\"name\": \"spi{bus}-{need}cs\"}} before it."
            )
        if have < need:
            raise ValueError(
                f"boot_config.dtoverlays: spi{bus}-{have}cs exposes {have} chip "
                f"select(s) but an mcp251xfd uses CS{highest} on spi{bus}; use "
                f"spi{bus}-{need}cs"
            )
        if aux_at[bus] > first_use[bus]:
            raise ValueError(
                f"boot_config.dtoverlays: spi{bus}-{have}cs must be listed BEFORE "
                f"the mcp251xfd entries that use spi{bus}. mcp251xfd disables the "
                f"conflicting spidev with target-path \"spi{bus}/spidev@N\", which "
                "only resolves once that node exists - applied first, the fragment "
                "is a no-op and spidev keeps the chip select."
            )

    # FDMode=yes and DataBitRate= are emitted as a pair, so a data rate below
    # the arbitration rate is always a swapped pair of numbers rather than an
    # intent: CAN FD exists to run the data phase FASTER once arbitration is won.
    for iface in ifaces:
        dbr = iface.get("dbitrate")
        if dbr is not None and dbr < iface["bitrate"]:
            raise ValueError(
                f"can.interfaces {iface['name']}: dbitrate={dbr} is below "
                f"bitrate={iface['bitrate']}. The FD data phase is the fast one - "
                "these two look swapped."
            )


def _hhmm_to_minutes(value: str) -> int:
    """"HH:MM" -> minutes since midnight. Raises ValueError on junk."""
    hh, _, mm = value.partition(":")
    return int(hh) * 60 + int(mm)


def _windows_overlap(a_start: int, a_len: int, b_start: int, b_len: int) -> bool:
    """Do two [start, start+len) minute windows intersect on a 24h clock?

    Both windows are unrolled onto a 48h axis so a window that wraps past
    midnight (reboot 23:00-01:00) still compares correctly against one that
    does not.
    """
    for shift in (0, 1440):
        if a_start + shift < b_start + b_len and b_start < a_start + shift + a_len:
            return True
        if b_start + shift < a_start + a_len and a_start < b_start + shift + b_len:
            return True
    return False


def _semantic_validate(cfg: dict[str, Any]) -> None:
    """Cross-field validation beyond what JSON Schema expresses.

    Fails fast with a human-readable error rather than a confusing dtoverlay
    or unit file later on.
    """
    ovc = cfg.get("overclock") or {}
    if ovc.get("enabled") and not ovc.get("accept_warranty_void"):
        raise ValueError(
            "overclock.enabled=true requires overclock.accept_warranty_void=true "
            "- overclocking permanently flips the Pi's warranty OTP bit."
        )

    fan = cfg.get("fan") or {}
    if fan.get("enabled") and fan.get("mode") not in ("gpio", "pwm", "emc2301"):
        raise ValueError("fan.enabled=true requires fan.mode to be gpio|pwm|emc2301")

    rtc = cfg.get("rtc") or {}
    if rtc.get("enabled") and not rtc.get("model"):
        raise ValueError("rtc.enabled=true requires rtc.model")

    boot = cfg.get("boot_config") or {}
    overlays = boot.get("dtoverlays") or []

    # boot_config.dtoverlays and can.interfaces describe the same hardware but
    # are rendered by two functions that never look at each other. That gap is
    # exactly how the canbus-plattform variant shipped an interface bound to a
    # GPIO the HAT does not connect, and nothing complained.
    mcp: dict[str, dict[str, Any]] = {}
    for ovl in overlays:
        m = _MCP2515_OVERLAY_RE.match(ovl["name"])
        if m:
            mcp[ovl["name"]] = ovl.get("params") or {}

    if mcp:
        wanted = [i["name"] for i in (cfg.get("can") or {}).get("interfaces", [])]
        missing = [n for n in wanted if f"mcp2515-{n}" not in mcp]
        if missing:
            raise ValueError(
                f"can.interfaces {missing} have no matching mcp2515-<name> entry in "
                f"boot_config.dtoverlays (present: {sorted(mcp)}) - the interface "
                "would never be created"
            )

        # Both upstream overlays default to interrupt=25. Two chips sharing one
        # GPIO produce a pinctrl conflict at boot, not a readable error, and a
        # chip pointed at an unconnected GPIO comes up clean and passes no
        # traffic - mcp251x requests its IRQ in ndo_open, so probe still logs
        # "MCP2515 successfully initialized".
        seen: dict[str, str] = {}
        for name, params in sorted(mcp.items()):
            pin = params.get("interrupt")
            if pin is None:
                raise ValueError(
                    f"boot_config.dtoverlays: {name} must set params.interrupt - "
                    "every mcp2515 overlay defaults to GPIO 25, so leaving it out "
                    "collides with the other channel"
                )
            if str(pin) in seen:
                raise ValueError(
                    f"boot_config.dtoverlays: {name} and {seen[str(pin)]} both use "
                    f"interrupt={pin}; each MCP2515 needs its own INT GPIO"
                )
            seen[str(pin)] = name

    _validate_mcp251xfd(cfg, overlays)

    # bluetooth.enabled is the single source of truth; a hand-written
    # disable-bt in extra_lines would silently win over it at boot.
    bt_enabled = (cfg.get("bluetooth") or {}).get("enabled", True)
    if bt_enabled and any(
        "disable-bt" in line for line in boot.get("extra_lines", [])
    ):
        raise ValueError(
            "bluetooth.enabled=true conflicts with a manual dtoverlay=disable-bt in "
            "boot_config.extra_lines - set bluetooth.enabled=false instead"
        )

    # Exactly one container runtime. Both modules install a /usr/bin/docker
    # and both claim the container storage; the image that ships with both
    # is not "belt and braces", it is undefined.
    docker_on = bool((cfg.get("docker") or {}).get("enabled"))
    podman_on = bool((cfg.get("podman") or {}).get("enabled"))
    if docker_on and podman_on:
        raise ValueError(
            "docker.enabled and podman.enabled are mutually exclusive - pick one "
            "container runtime per variant"
        )

    portainer = cfg.get("portainer") or {}
    podman = cfg.get("podman") or {}
    au = podman.get("auto_update") or {}

    # Podman resolves an unqualified image name through
    # unqualified-search-registries, which this image does not ship, so a
    # short name like "portainer/portainer-ce:lts" fails to pull at first
    # boot with the failure buried inside portainer-image.service - a
    # dependency, so Restart=always on the container unit never retries it.
    # Catches the likely operator error of hand-setting a short image name;
    # the code's own fallback (render_portainer()) is always fully qualified.
    if podman_on and portainer.get("enabled") and portainer.get("image"):
        _img = str(portainer["image"])
        _host = _img.split("/", 1)[0]
        if not ("." in _host or ":" in _host or _host == "localhost"):
            raise ValueError(
                f"portainer.image={_img!r} has no registry host - podman "
                "cannot resolve short image names without "
                "unqualified-search-registries, which this image does not "
                "ship. Use a fully qualified reference, e.g. "
                "'docker.io/portainer/portainer-ce:lts'"
            )

    if portainer.get("auto_update"):
        if not podman_on:
            raise ValueError(
                "portainer.auto_update=true requires podman.enabled=true - "
                "auto-update is a podman mechanism and does nothing under Docker"
            )
        if not au.get("enabled"):
            raise ValueError(
                "portainer.auto_update=true requires podman.auto_update.enabled=true "
                "- without the timer the AutoUpdate=registry label is inert"
            )
        if "@sha256:" in str(portainer.get("image", "")):
            raise ValueError(
                "portainer.auto_update=true is incompatible with a digest-pinned "
                "portainer.image - the remote digest can never differ, so the "
                "update would never fire"
            )

    # The device reboots itself inside the unattended-upgrades reboot window,
    # and Portainer migrates its database one-way on startup. An update that
    # takes a scripted reboot mid-migration leaves a portainer.db that no
    # version can open - a truck roll, not a retry.
    if podman_on and au.get("enabled"):
        sched = au.get("schedule") or {}
        start = _hhmm_to_minutes(sched.get("start", "05:30"))
        length = int(sched.get("randomized_delay_minutes", 30))
        uu = cfg.get("unattended_upgrades") or {}
        if uu.get("enabled"):
            uu_sched = uu.get("schedule") or {}
            uu_start = _hhmm_to_minutes(uu_sched.get("start", "02:00"))
            uu_len = _window_minutes(
                uu_sched.get("start", "02:00"), uu_sched.get("end", "04:00")
            )
            if _windows_overlap(start, length, uu_start, uu_len):
                raise ValueError(
                    f"podman.auto_update.schedule ({sched.get('start')} +{length}m) "
                    f"overlaps the unattended_upgrades window "
                    f"({uu_sched.get('start')}-{uu_sched.get('end')})"
                )
            reboot = uu.get("auto_reboot") or {}
            if reboot.get("enabled"):
                win = reboot.get("window") or {}
                r_start = _hhmm_to_minutes(win.get("start", "03:00"))
                r_len = _window_minutes(
                    win.get("start", "03:00"), win.get("end", "05:00")
                )
                if _windows_overlap(start, length, r_start, r_len):
                    raise ValueError(
                        f"podman.auto_update.schedule ({sched.get('start')} "
                        f"+{length}m) overlaps the auto-reboot window "
                        f"({win.get('start')}-{win.get('end')}) - a reboot during "
                        "Portainer's one-way database migration is unrecoverable"
                    )


def _validate_apply_contract(cfg: dict[str, Any]) -> list[str]:
    """Assert every rendered payload file is actually installed by its module.

    The failure this catches is quiet and expensive: a renderer starts
    emitting a file, nobody adds it to apply.sh, and the file rides along in
    /opt/bgrpiimage on every device without ever reaching the path that makes
    it do something. Nothing errors. The setting simply has no effect, and the
    symptom shows up weeks later as "we shipped that fix, why is the bus still
    dead" - which is exactly how the SamplePoint and TransmitQueueLength bugs
    presented before them.

    Deliberately a textual reference check rather than a manifest. A manifest
    would be a second description of the same thing, kept in sync by hand -
    the duplication that produced the CAN constant drift between generate.py
    and bgrpiimage-setup. Here the apply script stays the single description,
    and this only asserts it mentions everything the renderer produced.
    """
    problems: list[str] = []
    for module in ACTIVE_MODULES:
        if not _module_enabled(module, cfg):
            continue
        apply_sh = MODULES_DIR / module / "apply.sh"
        if not apply_sh.exists():
            continue
        body = apply_sh.read_text(encoding="utf-8")
        gen = MODULES_DIR / module / "filesystem" / "root" / "opt" / "bgrpiimage" / module
        if not gen.is_dir():
            continue
        for path in sorted(gen.rglob("*")):
            if not path.is_file():
                continue
            rel = path.relative_to(gen)
            if rel.as_posix() == "apply.sh":
                continue  # staged by the generator, not consumed by the script
            # A file may be named outright, or covered by its directory when
            # the module mirrors a whole tree with bg_install_tree.
            names = {rel.as_posix(), path.name}
            parent = rel.parent.as_posix()
            if parent != ".":
                names.add(parent)
            if not any(n in body for n in names):
                problems.append(f"{module}: renders {rel.as_posix()} but apply.sh never mentions it")
    return problems


def _window_minutes(start_hhmm: str, end_hhmm: str) -> int:
    """Minutes between two HH:MM timestamps, wrapping past midnight if needed."""
    def to_min(s: str) -> int:
        h, m = s.split(":")
        return int(h) * 60 + int(m)
    delta = to_min(end_hhmm) - to_min(start_hhmm)
    if delta <= 0:
        delta += 24 * 60
    return delta


# -----------------------------------------------------------------------------
# Variant shell config & module selection
# -----------------------------------------------------------------------------
ACTIVE_MODULES: list[str] = [
    # First on purpose: it puts apply-lib.sh in the rootfs, and every
    # other module's apply.sh sources it.
    "bgrpiimage-common",
    "bgrpiimage-base",
    "bgrpiimage-users",
    "bgrpiimage-network",
    "bgrpiimage-boot",
    "bgrpiimage-hardware",
    "bgrpiimage-can",
    "bgrpiimage-podman",
    "bgrpiimage-docker",
    "bgrpiimage-portainer",
    "bgrpiimage-unattended-upgrades",
]


def render_variant_config(cfg: dict[str, Any]) -> None:
    name = cfg["variant"]["name"]
    variant_dir = VARIANTS_DIR / name
    variant_dir.mkdir(parents=True, exist_ok=True)

    modules = [m for m in ACTIVE_MODULES if _module_enabled(m, cfg)]

    # CustomPiOS variant config: exports shell variables that refine the
    # distro-level src/config. MODULES controls execution order in the chroot.
    variant_cfg: list[str] = []
    variant_cfg.append(f"# Auto-generated variant config for {name}\n")
    variant_cfg.append(f"export DIST_NAME={shlex.quote(f'bgrpiimage-{name}')}\n")
    variant_cfg.append(f"export DIST_VERSION={shlex.quote(cfg['variant'].get('version', '0.0.0'))}\n")
    # CustomPiOS strips spaces from MODULES and splits on commas; a
    # space-separated list ends up concatenated into one invalid token.
    variant_cfg.append(f"export MODULES={shlex.quote(','.join(modules))}\n")
    variant_cfg.append(f"export BGRPIIMAGE_VARIANT={shlex.quote(name)}\n")
    variant_cfg.append(f"export BGRPIIMAGE_HOSTNAME={shlex.quote(cfg['hostname'])}\n")
    variant_cfg.append(f"export BGRPIIMAGE_TARGETS={shlex.quote(','.join(cfg['targets']))}\n")
    # Override the hostname raspios ships, via CustomPiOS's built-in hook.
    variant_cfg.append(f"export BASE_OVERRIDE_HOSTNAME={shlex.quote(cfg['hostname'])}\n")
    write(variant_dir / "config", "".join(variant_cfg))


def _module_enabled(module: str, cfg: dict[str, Any]) -> bool:
    """Some modules are only included if their section is populated/enabled."""
    if module == "bgrpiimage-can":
        return bool((cfg.get("can") or {}).get("interfaces"))
    if module == "bgrpiimage-docker":
        return bool((cfg.get("docker") or {}).get("enabled"))
    if module == "bgrpiimage-podman":
        return bool((cfg.get("podman") or {}).get("enabled"))
    if module == "bgrpiimage-portainer":
        return bool((cfg.get("portainer") or {}).get("enabled"))
    if module == "bgrpiimage-unattended-upgrades":
        return bool((cfg.get("unattended_upgrades") or {}).get("enabled"))
    if module == "bgrpiimage-hardware":
        # Active iff at least one runtime-touching hardware block is set.
        return any((
            (cfg.get("rtc") or {}).get("enabled"),
            (cfg.get("watchdog") or {}).get("enabled"),
            bool(cfg.get("bootloader")),
            ((cfg.get("audio") or {}).get("default_output") not in (None, "auto")),
        ))
    return True


# -----------------------------------------------------------------------------
# Entry point
# -----------------------------------------------------------------------------
def _module_status(module: str) -> tuple[str, int]:
    """Return (status, file_count) for a module's generated files.

    Files live under module/filesystem/root/opt/bgrpiimage/<module>/ as well
    as module/filesystem/root/... for files that land at fixed chroot paths
    (e.g. the static bgrpiimage-setup helper).
    """
    gen_tree = MODULES_DIR / module / "filesystem" / "root"
    if not gen_tree.exists():
        return ("empty", 0)
    # A renderer whose config block is switched off bails out after writing a
    # `.disabled` marker (render_docker, render_unattended). That marker is
    # itself a file, so counting blindly reported a DISABLED module as
    # "1 rendered" - and made main()'s "disabled in config" row unreachable.
    if (gen_tree / "opt" / "bgrpiimage" / module / ".disabled").exists():
        return ("disabled", 0)
    count = sum(1 for p in gen_tree.rglob("*") if p.is_file())
    return ("rendered" if count else "empty", count)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("config", type=Path, help="path to variant JSON config")
    parser.add_argument("--env-file", type=Path, help="optional .env file (KEY=VALUE lines)")
    parser.add_argument("--dry-run", action="store_true", help="validate & resolve only (rich output)")
    parser.add_argument("--json", action="store_true", help="validate & print resolved JSON to stdout (no decoration)")
    args = parser.parse_args()

    # Load + follow `extends` chain (deep-merge parents into this variant).
    try:
        raw = load_variant(args.config)
    except FileNotFoundError as e:
        _error_panel("config not found", str(e))
        return 2
    except ValueError as e:
        _error_panel("invalid extends chain", str(e))
        return 2

    env = dict(os.environ)
    if args.env_file and args.env_file.exists():
        for line in args.env_file.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            env.setdefault(k.strip(), v.strip().strip('"').strip("'"))

    try:
        resolved = resolve_tree(raw, env)
    except KeyError as e:
        _error_panel(
            "missing environment variable",
            str(e).strip("'"),
            hint="export the variable or use ${VAR:-default} in the JSON",
        )
        return 2

    with SCHEMA_PATH.open("r", encoding="utf-8") as f:
        schema = json.load(f)
    # The schema $id points at GitHub raw; strip to avoid online fetch.
    schema.pop("$id", None)
    try:
        jsonschema.validate(resolved, schema)
    except jsonschema.ValidationError as e:
        path = " → ".join(str(x) for x in e.absolute_path) or "(root)"
        _error_panel(
            "config validation failed",
            f"[bold]{e.message}[/]\n[dim]at:[/] {path}",
        )
        return 2

    # Cross-field checks the JSON schema can't express.
    try:
        _semantic_validate(resolved)
    except ValueError as e:
        _error_panel("config validation failed", str(e))
        return 2

    variant_name = resolved["variant"]["name"]
    variant_version = resolved["variant"].get("version", "?")

    if args.json:
        # Plain JSON to stdout - for piping into jq in CI scripts.
        sys.stdout.write(json.dumps(resolved, indent=2) + "\n")
        return 0

    if args.dry_run:
        console.print(Panel.fit(
            f"[bold cyan]{variant_name}[/] [dim]v{variant_version}[/] [green]valid[/]",
            border_style="green", box=box.ROUNDED,
        ))
        console.print(JSON.from_data(resolved))
        return 0

    # Header with resolved variant metadata.
    console.print(Panel(
        f"[bold cyan]{variant_name}[/]  [dim]v{variant_version}[/]\n"
        f"[dim]hostname:[/] {resolved['hostname']}\n"
        f"[dim]targets: [/] {', '.join(resolved['targets'])}",
        title="[bold]bgRPIImage render[/]", border_style="cyan", box=box.ROUNDED,
        title_align="left",
    ))

    steps: list[tuple[str, Any]] = [
        ("bgrpiimage-common",               render_common),
        ("bgrpiimage-base",                 render_base),
        ("bgrpiimage-users",                render_users),
        ("bgrpiimage-network",              render_network),
        ("bgrpiimage-boot",                 render_boot),
        ("bgrpiimage-hardware",             render_hardware),
        ("bgrpiimage-can",                  render_can),
        ("bgrpiimage-podman",               render_podman),
        ("bgrpiimage-docker",               render_docker),
        ("bgrpiimage-portainer",            render_portainer),
        ("bgrpiimage-unattended-upgrades",  render_unattended),
    ]

    table = Table(box=box.SIMPLE, show_header=True, header_style="bold dim")
    table.add_column("", width=2)
    table.add_column("module")
    table.add_column("files", justify="right")
    table.add_column("status", style="dim")

    total = 0
    for module, fn in steps:
        fn(resolved)
        # Staged centrally rather than from each renderer: a renderer that
        # forgets to ship its own apply.sh produces a module the updater
        # cannot run, and the failure would only surface on a device.
        gen_dir = MODULES_DIR / module / "filesystem" / "root" / "opt" / "bgrpiimage" / module
        # _module_enabled is the authority on whether this variant runs the
        # module at all. The `.disabled` marker is not: render_can and friends
        # bail out after clean_generated() without writing one, so keying off
        # the marker staged an apply.sh into modules the variant excludes -
        # which then reported as "rendered" in the table below and put a script
        # in a module that is not even in MODULES.
        if _module_enabled(module, resolved) and gen_dir.is_dir():
            stage_apply(module, gen_dir)
        status, count = _module_status(module)
        if status == "rendered":
            table.add_row("[green]✓[/]", module, str(count), "rendered")
            total += count
        elif status == "disabled":
            table.add_row("[yellow]·[/]", f"[dim]{module}[/]", "-", "disabled in config")
        else:
            table.add_row("[dim]·[/]", f"[dim]{module}[/]", "-", "no artifacts (skipped by filter)")

    render_variant_config(resolved)
    console.print(table)

    # After rendering, not before: the check compares what was actually
    # produced against what the apply scripts reference.
    contract_problems = _validate_apply_contract(resolved)
    if contract_problems:
        _error_panel(
            "apply contract violated",
            "\n".join(contract_problems),
            "a rendered file that no apply.sh installs reaches devices but "
            "never takes effect - add it to the module's apply.sh, or stop "
            "rendering it",
        )
        return 1
    console.print(
        f"  [green]{total}[/] artifact{'s' if total != 1 else ''} written to "
        f"[dim]src/modules/*/filesystem/root/[/]\n"
        f"  variant config: [dim]src/variants/{variant_name}/config[/]"
    )
    return 0


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