#!/usr/bin/env python3
"""Hub/spoke WireGuard enrollment over clear WLAN (keys only — tip still via update.sh).

Spoke generates its own WG keypair (private never leaves the device). Hub validates
bank + fleet credential + allowlist (or open-enrollment window), then:
  wg set wg0 peer <pub> allowed-ips 10.99.<bank>.<spoke>/32
  creates deterministic gretap eg-sNN
  persists peer set for reboot restore

Separate UDP listener from progressives_http.py (31234).
"""

from __future__ import annotations

import json
import os
import re
import socket
import subprocess
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable

from pog_edge_bank_wlan import load_fleet_hmac_key, verify_envelope, _sign_body
from pog_edge_identity import latch_cabinet_identity

DEFAULT_ENROLL_PORT = 41235
DEFAULT_STATE_DIR = "/etc/ethergold/hub"
DEFAULT_SPOKE_KEYS_DIR = "/etc/ethergold/spoke"
DEFAULT_OPEN_WINDOW_S = 600  # 10 minutes
# WG keepalive default is 5s (EDGE_WG_KEEPALIVE_S). A dead Pi (or swapped board)
# must fall off fanout without a hub reboot. 180s ≈ 36 missed keepalives —
# WiFi blips of 1–2 min stay.
DEFAULT_SPOKE_STALE_S = 180
REAP_INTERVAL_S = 5.0
# Spoke enroll: Master-service return budget ≤5s once hub IP is known.
SPOKE_ENROLL_NO_HUB_POLL_S = 1.0
SPOKE_ENROLL_RETRY_DELAYS_S = (0.5, 1.0, 2.0, 3.0, 5.0)
# WG "latest handshake" does NOT advance on every keepalive — healthy tunnels
# often sit 30–180s between handshake timestamps. Re-kick must track hub
# stale (EDGE_SPOKE_STALE_S), not a 12s Master-service budget (that thrash
# prevented first handshake and caused hub stale_handshake reap loops / CGS).
SPOKE_HANDSHAKE_STALE_REKICK_S = 150.0
# After enroll/conf install, wait for first WG handshake before "never" re-kick.
SPOKE_FIRST_HANDSHAKE_WAIT_S = 90.0
# Saw a handshake then lost it (hub reboot / peer wipe): re-kick quickly.
SPOKE_HANDSHAKE_LOST_REKICK_S = 5.0
SPOKE_ENROLL_WATCH_HEALTHY_S = 5.0
SPOKE_ENROLL_WATCH_COLD_S = 2.0
# User-writable fallthrough when /etc/ethergold is root-only (no sudo required).
_FALLBACK_STATE_DIRS = (
    lambda: Path.home() / ".local/share/ethergold/hub",
    lambda: Path("/var/tmp/ethergold/hub"),
)
_FALLBACK_SPOKE_KEYS_DIRS = (
    lambda: Path.home() / ".local/share/ethergold/spoke",
    lambda: Path("/var/tmp/ethergold/spoke"),
)


def resolve_enroll_port() -> int:
    try:
        return int(os.environ.get("EDGE_ENROLL_PORT", str(DEFAULT_ENROLL_PORT)))
    except ValueError:
        return DEFAULT_ENROLL_PORT


def _dir_is_writable(path: Path) -> bool:
    try:
        path.mkdir(parents=True, exist_ok=True)
        probe = path / ".eg_write_probe"
        probe.write_text("ok\n", encoding="utf-8")
        probe.unlink(missing_ok=True)
        return True
    except OSError:
        return False


def ensure_hub_state_dir(
    preferred: Path | None = None,
    *,
    emit: Callable[..., None] | None = None,
) -> Path:
    """Return a hub state directory the bridge process can actually write.

    Prefers EDGE_HUB_STATE_DIR / /etc/ethergold/hub; if root-owned or missing
    write permission, automatically falls through to ``~/.local/share/ethergold/hub``
    (no manual chmod, no sudo). Sets EDGE_HUB_STATE_DIR for this process.
    """
    candidates: list[Path] = []
    env = (os.environ.get("EDGE_HUB_STATE_DIR") or "").strip()
    if preferred is not None:
        candidates.append(Path(preferred))
    if env:
        candidates.append(Path(env))
    candidates.append(Path(DEFAULT_STATE_DIR))
    for factory in _FALLBACK_STATE_DIRS:
        try:
            candidates.append(factory())
        except Exception:
            pass
    # De-dupe while preserving order
    seen: set[str] = set()
    ordered: list[Path] = []
    for c in candidates:
        key = str(c)
        if key in seen:
            continue
        seen.add(key)
        ordered.append(c)
    chosen: Path | None = None
    for cand in ordered:
        if _dir_is_writable(cand):
            chosen = cand
            break
    if chosen is None:
        chosen = Path.home() / ".local/share/ethergold/hub"
        chosen.mkdir(parents=True, exist_ok=True)
    os.environ["EDGE_HUB_STATE_DIR"] = str(chosen)
    if emit is not None and str(chosen) != DEFAULT_STATE_DIR:
        try:
            emit(
                "hub_state_dir",
                path=str(chosen),
                preferred=DEFAULT_STATE_DIR,
                reason="auto_writable_fallback"
                if str(chosen) != (env or DEFAULT_STATE_DIR)
                else "configured",
            )
        except Exception:
            pass
    return chosen


def resolve_state_dir() -> Path:
    return ensure_hub_state_dir()


def overlay_ip(bank: int, spoke_id: int) -> str:
    """Bank-scoped overlay: hub .1, spokes .2..254. Bank 0 = legacy 10.99.0.x."""
    b = max(0, min(255, int(bank)))
    s = max(1, min(254, int(spoke_id)))
    return f"10.99.{b}.{s}"


def gretap_name_for_spoke(spoke_id: int) -> str:
    return f"eg-s{int(spoke_id):02d}"


def spoke_stale_s() -> float:
    raw = os.environ.get("EDGE_SPOKE_STALE_S", str(DEFAULT_SPOKE_STALE_S))
    try:
        return float(raw)
    except (TypeError, ValueError):
        return float(DEFAULT_SPOKE_STALE_S)


def parse_wg_dump(text: str) -> dict[str, dict[str, Any]]:
    """Parse ``wg show <if> dump`` (tab-separated). Skip the interface header line."""
    out: dict[str, dict[str, Any]] = {}
    for i, line in enumerate((text or "").splitlines()):
        if i == 0 or not line.strip():
            continue
        parts = line.split("\t")
        if len(parts) < 5:
            continue
        try:
            handshake = int(parts[4] or 0)
        except ValueError:
            handshake = 0
        out[parts[0]] = {
            "endpoint": parts[2],
            "allowed_ips": parts[3],
            "handshake": handshake,
        }
    return out


def overlay_spoke_id(allowed_ips: str) -> int | None:
    for tok in (allowed_ips or "").split(","):
        host = tok.strip().split("/")[0]
        parts = host.split(".")
        if len(parts) == 4 and parts[0] == "10" and parts[1] == "99":
            try:
                sid = int(parts[3])
            except ValueError:
                continue
            if 2 <= sid <= 254:
                return sid
    return None


def peer_enrolled_unix(rec: dict[str, Any] | None) -> float:
    """Unix time of last enroll/reuse for per-peer first-handshake grace."""
    if not rec:
        return 0.0
    raw = rec.get("enrolled_unix")
    if raw is not None and str(raw).strip() != "":
        try:
            return float(raw)
        except (TypeError, ValueError):
            pass
    utc = str(rec.get("enrolled_utc") or "").strip()
    if not utc:
        return 0.0
    try:
        # Accept "YYYY-mm-ddTHH:MM:SSZ"
        if utc.endswith("Z"):
            utc = utc[:-1] + "+00:00"
        from datetime import datetime

        return datetime.fromisoformat(utc).timestamp()
    except ValueError:
        return 0.0


def peer_reap_grace_until(
    rec: dict[str, Any] | None,
    *,
    hub_grace_until: float,
    stale_s: float,
) -> float:
    """Do not reap handshake=0 until hub grace *and* per-peer enroll grace pass.

    Ad-hoc: hub may have been up for hours when a cabinet powers on. Process-start
    grace alone would immediately stale-drop every new peer still at handshake=0.
    """
    if stale_s <= 0:
        return hub_grace_until
    enrolled = peer_enrolled_unix(rec)
    if enrolled <= 0:
        return hub_grace_until
    return max(float(hub_grace_until), float(enrolled) + float(stale_s))


def peer_handshake_stale(
    handshake: int,
    *,
    now: float,
    stale_s: float,
    grace_until: float,
) -> bool:
    """True when the hub should drop this WG peer.

    ``handshake`` is unix seconds from ``wg show dump`` (0 = never).
    ``grace_until`` is unix time: do not reap never-connected peers before this
    (hub just started / peer just enrolled; spokes need time to handshake).
    """
    if stale_s <= 0:
        return False
    if now < grace_until:
        return False
    if handshake <= 0:
        return True
    return (now - float(handshake)) > stale_s


def spoke_handshake_rekick_s() -> float:
    """Seconds of WG latest-handshake age before spoke re-enrolls."""
    raw = os.environ.get("EDGE_SPOKE_HANDSHAKE_REKICK_S", "").strip()
    if raw:
        try:
            return max(30.0, float(raw))
        except ValueError:
            pass
    stale = spoke_stale_s()
    if stale > 0:
        # Re-kick before hub would reap, with headroom for WG timestamp lag.
        return max(60.0, min(float(SPOKE_HANDSHAKE_STALE_REKICK_S), stale * 0.75))
    return float(SPOKE_HANDSHAKE_STALE_REKICK_S)


def drop_hub_peer(
    *,
    wg_if: str,
    pub: str,
    rec: dict[str, Any] | None = None,
    emit: Callable[..., None] | None = None,
    reason: str = "stale",
) -> None:
    """Remove a live WG peer + its gretap. Missing iface/peer is not fatal."""
    spoke_id = 0
    name = ""
    if rec:
        try:
            spoke_id = int(rec.get("spoke_id") or 0)
        except (TypeError, ValueError):
            spoke_id = 0
        name = str(rec.get("gretap") or "")
    if not name and spoke_id:
        name = gretap_name_for_spoke(spoke_id)
    _run(["wg", "set", wg_if, "peer", pub, "remove"], check=False)
    if name:
        _run(["ip", "link", "del", name], check=False)
    if emit:
        emit(
            "hub_peer_dropped",
            spoke_id=spoke_id or None,
            gretap=name or None,
            pub=pub[:16],
            reason=reason,
        )


def _run(cmd: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
    """Run a command; elevate wg/ip via ``sudo -n`` when bridge is not root.

    ``run_ethergold`` elevates the shell for tunnel bring-up, then drops to
    topbox for pog422_bridge. Enroll still needs NET_ADMIN for ``wg set`` /
    gretap — passwordless sudo (netdev/sudo group) is the field path.
    """
    real = list(cmd)
    if (
        real
        and real[0] in ("wg", "ip")
        and hasattr(os, "geteuid")
        and os.geteuid() != 0
    ):
        real = ["sudo", "-n", *real]
    return subprocess.run(real, check=check, text=True, capture_output=True)


def wg_gen_keypair() -> tuple[str, str]:
    """Return (private, public) WireGuard keys as strings."""
    priv = _run(["wg", "genkey"]).stdout.strip()
    proc = subprocess.run(
        ["wg", "pubkey"],
        input=priv + "\n",
        check=True,
        text=True,
        capture_output=True,
    )
    return priv, proc.stdout.strip()


def _path_is_reg_file(path: Path) -> bool:
    """is_file() that treats EACCES like missing (CPython re-raises PermissionError)."""
    try:
        return path.is_file()
    except OSError:
        return False


def _keys_dir_ready_for_spoke(keys_dir: Path) -> bool:
    """True when we can use keys_dir for enroll (read existing or generate).

    EG 3.0.2.9: /etc/ethergold/spoke can look writable (create probe) yet hold
    root:600 ``wg_private.key`` — write then raises Permission denied and enroll
    loops forever (pos5 soak 1953). Prefer the dir only if key files are absent
    or RW for this UID, and the directory itself is writable for minting keys.

    EG 4.0.0.50: pathlib ``is_file``/``exists`` re-raise EACCES on root:700 dirs
    (Bookworm) — wrap so Pass-1 does not abort before ~/.local fallthrough.
    """
    if not _dir_is_writable(keys_dir):
        return False
    priv_path = keys_dir / "wg_private.key"
    pub_path = keys_dir / "wg_public.key"
    try:
        has_priv = priv_path.exists()
        has_pub = pub_path.exists()
    except OSError:
        return False
    if has_priv or has_pub:
        if has_priv and not (
            os.access(priv_path, os.R_OK) and os.access(priv_path, os.W_OK)
        ):
            return False
        if has_pub and not os.access(pub_path, os.R_OK):
            return False
    return True


def ensure_spoke_keys_dir(
    preferred: Path | None = None,
    *,
    emit: Callable[..., None] | None = None,
) -> Path:
    """Return a spoke key directory the bridge process can read/write.

    Prefers EDGE_SPOKE_KEYS_DIR / /etc/ethergold/spoke. When root-owned keys or
    missing write permission (pos5: Permission denied on wg_private.key), fall
    through to ``~/.local/share/ethergold/spoke`` — same pattern as
    ``ensure_hub_state_dir`` (EG 3.0.1.30 / 3.0.1.47 / 3.0.2.9).
    """
    candidates: list[Path] = []
    env = (os.environ.get("EDGE_SPOKE_KEYS_DIR") or "").strip()
    if preferred is not None:
        candidates.append(Path(preferred))
    if env:
        candidates.append(Path(env))
    candidates.append(Path(DEFAULT_SPOKE_KEYS_DIR))
    for factory in _FALLBACK_SPOKE_KEYS_DIRS:
        try:
            candidates.append(factory())
        except Exception:
            pass
    seen: set[str] = set()
    ordered: list[Path] = []
    for c in candidates:
        key = str(c)
        if key in seen:
            continue
        seen.add(key)
        ordered.append(c)
    chosen: Path | None = None
    for cand in ordered:
        if _keys_dir_ready_for_spoke(cand):
            chosen = cand
            break
    if chosen is None:
        chosen = Path.home() / ".local/share/ethergold/spoke"
        chosen.mkdir(parents=True, exist_ok=True)
    os.environ["EDGE_SPOKE_KEYS_DIR"] = str(chosen)
    if emit is not None and str(chosen) != DEFAULT_SPOKE_KEYS_DIR:
        try:
            emit(
                "spoke_keys_dir",
                path=str(chosen),
                preferred=DEFAULT_SPOKE_KEYS_DIR,
                reason="auto_writable_fallback",
            )
        except Exception:
            pass
    return chosen


def ensure_spoke_keypair(
    path: Path | None = None,
    *,
    emit: Callable[..., None] | None = None,
) -> tuple[str, str]:
    """Generate once; private key never leaves the device.

    Tries preferred / EDGE_SPOKE_KEYS_DIR / /etc/ethergold/spoke, then user
    fallthrough. If a system path has readable keys but is not writable for
    generation, still prefer those keys when both files exist and are readable.
    Root-owned unreadable/unwritable files under /etc must not block ~/.local.
    """
    candidates: list[Path] = []
    if path is not None:
        candidates.append(Path(path))
    env = (os.environ.get("EDGE_SPOKE_KEYS_DIR") or "").strip()
    if env:
        candidates.append(Path(env))
    candidates.append(Path(DEFAULT_SPOKE_KEYS_DIR))
    for factory in _FALLBACK_SPOKE_KEYS_DIRS:
        try:
            candidates.append(factory())
        except Exception:
            pass
    seen: set[str] = set()
    ordered: list[Path] = []
    for c in candidates:
        key = str(c)
        if key in seen:
            continue
        seen.add(key)
        ordered.append(c)

    last_err: OSError | None = None
    # Pass 1: read existing keypairs from any candidate (prefer system path order).
    for keys_dir in ordered:
        priv_path = keys_dir / "wg_private.key"
        pub_path = keys_dir / "wg_public.key"
        # Bookworm pathlib: is_file() raises PermissionError on root:700 dirs.
        if not (_path_is_reg_file(priv_path) and _path_is_reg_file(pub_path)):
            continue
        try:
            priv = priv_path.read_text(encoding="utf-8").strip()
            pub = pub_path.read_text(encoding="utf-8").strip()
            if priv and pub:
                os.environ["EDGE_SPOKE_KEYS_DIR"] = str(keys_dir)
                return priv, pub
        except OSError as exc:
            last_err = exc
            continue

    # Pass 2: generate into first usable directory (skip locked root key residue).
    for keys_dir in ordered:
        try:
            if not _keys_dir_ready_for_spoke(keys_dir):
                continue
            priv_path = keys_dir / "wg_private.key"
            pub_path = keys_dir / "wg_public.key"
            # Existing root-owned key: do not clobber — try next candidate.
            try:
                locked = _path_is_reg_file(priv_path) and not os.access(
                    priv_path, os.W_OK
                )
            except OSError as exc:
                last_err = exc
                continue
            if locked:
                last_err = OSError(13, f"Permission denied: {priv_path}")
                continue
            priv, pub = wg_gen_keypair()
            priv_path.write_text(priv + "\n", encoding="utf-8")
            pub_path.write_text(pub + "\n", encoding="utf-8")
            try:
                os.chmod(priv_path, 0o600)
                os.chmod(pub_path, 0o644)
            except OSError:
                pass
            os.environ["EDGE_SPOKE_KEYS_DIR"] = str(keys_dir)
            if emit is not None and str(keys_dir) != DEFAULT_SPOKE_KEYS_DIR:
                try:
                    emit(
                        "spoke_keys_dir",
                        path=str(keys_dir),
                        preferred=DEFAULT_SPOKE_KEYS_DIR,
                        reason="auto_writable_fallback",
                    )
                except Exception:
                    pass
            return priv, pub
        except OSError as exc:
            last_err = exc
            continue

    # Last resort through writable ensure (may create ~/.local/...).
    keys_dir = ensure_spoke_keys_dir(path, emit=emit)
    priv_path = keys_dir / "wg_private.key"
    pub_path = keys_dir / "wg_public.key"
    try:
        priv, pub = wg_gen_keypair()
        priv_path.write_text(priv + "\n", encoding="utf-8")
        pub_path.write_text(pub + "\n", encoding="utf-8")
        try:
            os.chmod(priv_path, 0o600)
            os.chmod(pub_path, 0o644)
        except OSError:
            pass
        return priv, pub
    except OSError as exc:
        if last_err is not None:
            raise RuntimeError(
                f"spoke keypair unavailable under {ordered!r}: {exc}"
            ) from last_err
        raise


@dataclass
class HubPeerStore:
    path: Path
    peers: dict[str, dict[str, Any]] = field(default_factory=dict)  # pub -> record

    @classmethod
    def load(
        cls,
        state_dir: Path | None = None,
        *,
        emit: Callable[..., None] | None = None,
    ) -> "HubPeerStore":
        preferred = Path(
            state_dir
            if state_dir is not None
            else (os.environ.get("EDGE_HUB_STATE_DIR") or DEFAULT_STATE_DIR)
        )
        write_root = ensure_hub_state_dir(preferred, emit=emit)
        peers: dict[str, dict[str, Any]] = {}
        # Read merged roster: system path first (often root-provisioned), then
        # user-writable path (always the save target).
        for root in (preferred, write_root):
            path = Path(root) / "peers.json"
            if not path.is_file():
                continue
            try:
                data = json.loads(path.read_text(encoding="utf-8"))
                if isinstance(data, dict):
                    for k, v in data.items():
                        if isinstance(v, dict):
                            peers[str(k)] = v
            except (OSError, json.JSONDecodeError):
                continue
        return cls(path=write_root / "peers.json", peers=peers)

    def save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(self.peers, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        os.replace(tmp, self.path)
        try:
            os.chmod(self.path, 0o640)
        except OSError:
            pass  # non-fatal on some FS / ownership layouts

    def next_spoke_id(self) -> int:
        used = {int(p.get("spoke_id", 0)) for p in self.peers.values()}
        for cand in range(2, 255):
            if cand not in used:
                return cand
        raise RuntimeError("no free spoke ids (2..254)")

    def allowlist_pubs(self) -> set[str]:
        path = self.path.parent / "allowlist.txt"
        if not path.is_file():
            return set()
        out: set[str] = set()
        for line in path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            out.add(line.split()[0])
        return out


def apply_hub_peer(
    *,
    wg_if: str,
    pub: str,
    overlay: str,
    spoke_id: int,
    hub_overlay: str,
) -> str:
    """Live-add WG peer + deterministic gretap. Returns gretap name."""
    try:
        from wlan_star_apply import resolve_wg_keepalive_s

        ka = str(resolve_wg_keepalive_s())
    except Exception:
        ka = "5"
    _run(
        [
            "wg",
            "set",
            wg_if,
            "peer",
            pub,
            "allowed-ips",
            f"{overlay}/32",
            "persistent-keepalive",
            ka,
        ]
    )
    name = gretap_name_for_spoke(spoke_id)
    # Do not delete sibling gretaps — create only if missing.
    check = _run(["ip", "link", "show", name], check=False)
    if check.returncode != 0:
        _run(
            [
                "ip",
                "link",
                "add",
                name,
                "type",
                "gretap",
                "local",
                hub_overlay,
                "remote",
                overlay,
                "dev",
                wg_if,
            ]
        )
    _run(["ip", "link", "set", name, "up"], check=False)
    return name


def restore_hub_peers_from_disk(
    *,
    wg_if: str = "wg0",
    bank: int | None = None,
    emit: Callable[..., None] | None = None,
) -> list[str]:
    """On hub reboot: re-apply persisted peers + gretaps."""
    store = HubPeerStore.load()
    b = int(bank if bank is not None else os.environ.get("EDGE_BANK", "0") or 0)
    hub_overlay = overlay_ip(b, 1)
    names: list[str] = []
    for pub, rec in store.peers.items():
        spoke_id = int(rec.get("spoke_id", 0))
        overlay = str(rec.get("overlay") or overlay_ip(b, spoke_id))
        try:
            name = apply_hub_peer(
                wg_if=wg_if,
                pub=pub,
                overlay=overlay,
                spoke_id=spoke_id,
                hub_overlay=hub_overlay,
            )
            names.append(name)
            if emit:
                emit("hub_peer_restored", spoke_id=spoke_id, overlay=overlay, gretap=name)
        except Exception as exc:
            if emit:
                emit("hub_peer_restore_error", pub=pub[:16], message=str(exc))
    return names


def eth_ports_from_spokes(
    game_if: str | None,
    spoke_ids: list[int],
) -> str:
    """Build ETH_PORTS from live spoke set.

    Wlan-star / gretap-only: pass game_if=None (or wlan*) → eg-sNN only.
    Single-hop copper may prepend eth*.
    """
    parts: list[str] = []
    g = (game_if or "").strip()
    if g and not g.startswith("wlan"):
        parts.append(g)
    for sid in sorted(spoke_ids):
        parts.append(gretap_name_for_spoke(sid))
    return ",".join(parts)


@dataclass
class EnrollmentHub:
    bank: int
    emit: Callable[..., None] | None = None
    """Optional: called after peer is on WG+gretap (enroll or restore)."""
    on_peer_ready: Callable[..., None] | None = None
    """Optional: called after a stale/replaced peer is dropped from WG+gretap."""
    on_peer_gone: Callable[..., None] | None = None
    port: int = field(default_factory=resolve_enroll_port)
    open_until: float = 0.0
    store: HubPeerStore | None = None
    key: bytes = field(default_factory=load_fleet_hmac_key)
    wg_if: str = "wg0"
    _stop: threading.Event = field(default_factory=threading.Event)
    _thread: threading.Thread | None = None
    _started_unix: float = 0.0
    _last_reap_mono: float = 0.0

    def __post_init__(self) -> None:
        if self.store is None:
            self.store = HubPeerStore.load(emit=self.emit)

    def _notify_peer_ready(self, rec: dict[str, Any], *, reason: str) -> None:
        if self.on_peer_ready is None:
            return
        try:
            self.on_peer_ready(rec, reason=reason)
        except Exception as exc:
            if self.emit:
                self.emit(
                    "hub_peer_ready_cb_error",
                    message=str(exc),
                    reason=reason,
                    gretap=rec.get("gretap"),
                )

    def _notify_peer_gone(self, rec: dict[str, Any], *, reason: str) -> None:
        if self.on_peer_gone is None:
            return
        try:
            self.on_peer_gone(rec, reason=reason)
        except Exception as exc:
            if self.emit:
                self.emit(
                    "hub_peer_gone_cb_error",
                    message=str(exc),
                    reason=reason,
                    gretap=rec.get("gretap"),
                )

    def _drop_pub(self, pub: str, rec: dict[str, Any] | None, *, reason: str) -> None:
        drop_hub_peer(
            wg_if=self.wg_if,
            pub=pub,
            rec=rec,
            emit=self.emit,
            reason=reason,
        )
        if self.store is not None and pub in self.store.peers:
            del self.store.peers[pub]
            self.store.save()
        if rec:
            self._notify_peer_gone(rec, reason=reason)

    def _maybe_reap(self) -> None:
        stale_s = spoke_stale_s()
        if stale_s <= 0:
            return
        now_mono = time.monotonic()
        if now_mono - self._last_reap_mono < REAP_INTERVAL_S:
            return
        self._last_reap_mono = now_mono
        now = time.time()
        grace_until = (self._started_unix or now) + stale_s
        try:
            dump = _run(["wg", "show", self.wg_if, "dump"], check=False).stdout
        except OSError:
            return
        live = parse_wg_dump(dump)
        store_peers = dict(self.store.peers) if self.store is not None else {}
        seen: set[str] = set()
        for pub, info in live.items():
            seen.add(pub)
            rec = store_peers.get(pub)
            if rec is None:
                sid = overlay_spoke_id(str(info.get("allowed_ips") or ""))
                rec = {
                    "spoke_id": sid or 0,
                    "gretap": gretap_name_for_spoke(sid) if sid else "",
                    "overlay": f"10.99.{self.bank}.{sid}" if sid else "",
                }
            if peer_handshake_stale(
                int(info.get("handshake") or 0),
                now=now,
                stale_s=stale_s,
                grace_until=peer_reap_grace_until(
                    rec, hub_grace_until=grace_until, stale_s=stale_s
                ),
            ):
                self._drop_pub(pub, rec, reason="stale_handshake")
        # Disk-only leftovers with no live WG row: drop after grace so a restored
        # peer that never comes back does not stay in ETH_PORTS forever.
        # Use per-peer enroll grace (same as handshake=0) so ad-hoc joiners are
        # not wiped while wg set / first handshake is still in flight.
        if self.store is not None:
            for pub, rec in list(self.store.peers.items()):
                if pub not in seen:
                    peer_grace = peer_reap_grace_until(
                        rec, hub_grace_until=grace_until, stale_s=stale_s
                    )
                    if now >= peer_grace:
                        self._drop_pub(pub, rec, reason="missing_wg")

    def open_enrollment(self, window_s: float | None = None) -> float:
        secs = float(
            window_s
            if window_s is not None
            else os.environ.get("EDGE_OPEN_ENROLL_S", DEFAULT_OPEN_WINDOW_S)
        )
        self.open_until = time.time() + max(0.0, secs)
        if self.emit:
            self.emit("enroll_window_open", until=self.open_until, window_s=secs)
        return self.open_until

    def enrollment_open(self) -> bool:
        """True when new (non-allowlisted) spokes may join.

        Product default (Chris 2026-09-09): **ad hoc anytime** — fleet HMAC +
        bank match are the gates. Time windows are only for EDGE_ENROLL_LOCK=1.
        """
        if _env_enroll_locked():
            return time.time() < self.open_until
        return True

    def start(self) -> None:
        if self._thread is not None and self._thread.is_alive():
            return
        restore_hub_peers_from_disk(wg_if=self.wg_if, bank=self.bank, emit=self.emit)
        # Hot-attach every restored gretap into the live eth router (field reboot
        # already has eth-ports at process start; this covers restore+enroll gaps).
        for _pub, rec in list(self.store.peers.items()):
            self._notify_peer_ready(rec, reason="restore")
        if _env_enroll_locked():
            # Locked floor: optional timed window from EDGE_OPEN_ENROLL*.
            if _env_open_at_start():
                self.open_enrollment()
        elif self.emit:
            self.emit("enroll_adhoc", bank=self.bank, note="anytime")
        self._started_unix = time.time()
        self._last_reap_mono = 0.0
        self._stop.clear()
        self._thread = threading.Thread(target=self._run, name="edge-enroll-hub", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2.0)

    def _run(self) -> None:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(("", self.port))
        sock.settimeout(0.5)
        if self.emit:
            self.emit("enroll_listen", port=self.port, bank=self.bank)
        while not self._stop.is_set():
            try:
                data, addr = sock.recvfrom(65535)
            except socket.timeout:
                self._maybe_reap()
                continue
            except OSError:
                break
            self._maybe_reap()
            try:
                msg = json.loads(data.decode("utf-8"))
            except (UnicodeDecodeError, json.JSONDecodeError):
                continue
            if str(msg.get("type") or "") != "enroll_request":
                continue
            if not verify_envelope(msg, self.key):
                # Silent on crypto failure — probes get no "service exists" clue.
                if self.emit:
                    self.emit("enroll_bad_mac", from_ip=addr[0])
                continue
            # Valid fleet HMAC: reply with accept *or* signed diagnostic deny
            # (wrong bank, window closed, etc.) so commissioning seats know why.
            resp = self._handle_request(msg, addr[0])
            if resp is None:
                continue
            try:
                sock.sendto(json.dumps(resp, separators=(",", ":")).encode("utf-8"), addr)
            except OSError:
                pass
        try:
            sock.close()
        except OSError:
            pass

    def _handle_request(self, msg: dict[str, Any], from_ip: str) -> dict[str, Any] | None:
        pub = str(msg.get("wg_pub") or "").strip()
        bank = int(msg.get("bank", -1))
        if bank != self.bank:
            if self.emit:
                self.emit("enroll_denied", reason="bank_mismatch", from_ip=from_ip)
            return self._reject("bank_mismatch", msg)
        if not pub:
            if self.emit:
                self.emit("enroll_denied", reason="missing_pub", from_ip=from_ip)
            return self._reject("missing_pub", msg)
        allow = self.store.allowlist_pubs()
        open_ok = self.enrollment_open()
        if pub in self.store.peers:
            rec = self.store.peers[pub]
            # Re-apply kernel peer/gretap and hot-add eth in case Mid-session
            # gretap was torn down or bridge ports lagged.
            try:
                spoke_id = int(rec.get("spoke_id") or 0)
                overlay = str(rec.get("overlay") or overlay_ip(self.bank, spoke_id))
                hub_overlay = overlay_ip(self.bank, 1)
                gretap = apply_hub_peer(
                    wg_if=self.wg_if,
                    pub=pub,
                    overlay=overlay,
                    spoke_id=spoke_id,
                    hub_overlay=hub_overlay,
                )
                rec["gretap"] = gretap
                rec["overlay"] = overlay
                # Refresh enroll time so handshake=0 peers are not reaped mid-reuse
                # while WG completes the first crypto handshake.
                rec["enrolled_unix"] = time.time()
                rec["enrolled_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                self.store.peers[pub] = rec
                self.store.save()
            except Exception as exc:
                if self.emit:
                    self.emit("enroll_apply_error", message=str(exc), from_ip=from_ip, reused=True)
            self._notify_peer_ready(rec, reason="reused")
            return self._accept_response(rec, msg, reused=True)
        if allow and pub not in allow and not open_ok:
            if self.emit:
                self.emit("enroll_denied", reason="not_allowlisted", from_ip=from_ip, pub=pub[:16])
            return self._reject("not_allowlisted", msg)
        if not allow and not open_ok:
            # Only reachable when EDGE_ENROLL_LOCK=1 and timed window expired.
            if self.emit:
                self.emit("enroll_denied", reason="window_closed", from_ip=from_ip)
            return self._reject("window_closed", msg)
        try:
            spoke_id = self._allocate_spoke_id(msg)
            overlay = overlay_ip(self.bank, spoke_id)
            hub_overlay = overlay_ip(self.bank, 1)
            gretap = apply_hub_peer(
                wg_if=self.wg_if,
                pub=pub,
                overlay=overlay,
                spoke_id=spoke_id,
                hub_overlay=hub_overlay,
            )
        except Exception as exc:
            if self.emit:
                detail = str(exc)
                err = getattr(exc, "stderr", None)
                if err:
                    detail = f"{detail}: {str(err).strip()}"
                self.emit("enroll_apply_error", message=detail, from_ip=from_ip)
            return self._reject("apply_failed", msg)
        rec = {
            "spoke_id": spoke_id,
            "overlay": overlay,
            "gretap": gretap,
            "term_number": msg.get("term_number"),
            "device_id": msg.get("device_id"),
            "enrolled_unix": time.time(),
            "enrolled_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "from_ip": from_ip,
        }
        self.store.peers[pub] = rec
        self.store.save()
        if self.emit:
            self.emit(
                "enroll_accepted",
                spoke_id=spoke_id,
                overlay=overlay,
                gretap=gretap,
                from_ip=from_ip,
            )
        self._notify_peer_ready(rec, reason="enroll")
        return self._accept_response(rec, msg, reused=False)

    def _allocate_spoke_id(self, msg: dict[str, Any]) -> int:
        """Prefer request/EDGE preferred id when free; else next free 2..254."""
        used = {int(p.get("spoke_id", 0)) for p in self.store.peers.values()}
        raw = msg.get("spoke_id")
        if raw is None or raw == "":
            raw = msg.get("preferred_spoke_id")
        if raw is not None and str(raw).strip() != "":
            try:
                want = int(raw)
            except (TypeError, ValueError):
                want = 0
            if 2 <= want <= 254 and want not in used:
                return want
            if 2 <= want <= 254 and self.store is not None and spoke_stale_s() > 0:
                occupant_pub = None
                occupant: dict[str, Any] | None = None
                for pub, rec in list(self.store.peers.items()):
                    if int(rec.get("spoke_id") or 0) == want:
                        occupant_pub, occupant = pub, rec
                        break
                if occupant_pub is not None:
                    hs = 0
                    try:
                        dump = _run(
                            ["wg", "show", self.wg_if, "dump"], check=False
                        ).stdout
                        hs = int(
                            parse_wg_dump(dump).get(occupant_pub, {}).get("handshake") or 0
                        )
                    except OSError:
                        hs = 0
                    if peer_handshake_stale(
                        hs,
                        now=time.time(),
                        stale_s=spoke_stale_s(),
                        grace_until=peer_reap_grace_until(
                            occupant,
                            hub_grace_until=0.0,
                            stale_s=spoke_stale_s(),
                        ),
                    ):
                        self._drop_pub(occupant_pub, occupant, reason="replaced")
                        return want
        return self.store.next_spoke_id()

    def _accept_response(
        self, rec: dict[str, Any], req: dict[str, Any], *, reused: bool
    ) -> dict[str, Any]:
        hub_ip = resolve_hub_uplink_ip() or _local_uplink_ipv4()
        hub_pub = _hub_wg_public_key()
        port = int(os.environ.get("POG_EDGE_WG_PORT", "51820") or "51820")
        body: dict[str, Any] = {
            "v": 1,
            "type": "enroll_response",
            "ok": True,
            "reused": reused,
            "bank": self.bank,
            "spoke_id": rec["spoke_id"],
            "overlay": rec["overlay"],
            "hub_overlay": overlay_ip(self.bank, 1),
            "gretap": rec.get("gretap"),
            "counter": int(req.get("counter") or 0),
            "ts": int(time.time()),
        }
        if hub_pub:
            body["hub_wg_pub"] = hub_pub
        if hub_ip:
            body["hub_endpoint"] = f"{hub_ip}:{port}"
            body["hub_uplink_ip"] = hub_ip
        body["mac"] = _sign_body({k: v for k, v in body.items() if k != "mac"}, self.key)
        return body

    def _reject(self, reason: str, req: dict[str, Any]) -> dict[str, Any]:
        """Signed diagnostic deny — only reached after fleet HMAC verified."""
        body: dict[str, Any] = {
            "v": 1,
            "type": "enroll_response",
            "ok": False,
            "reason": reason,
            "bank": self.bank,
            "counter": int(req.get("counter") or 0),
            "ts": int(time.time()),
        }
        body["mac"] = _sign_body({k: v for k, v in body.items() if k != "mac"}, self.key)
        return body


def _env_enroll_locked() -> bool:
    """Optional lockdown: allowlist + timed window instead of ad hoc anytime."""
    return os.environ.get("EDGE_ENROLL_LOCK", "").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )


def _env_open_at_start() -> bool:
    """Timed window at hub start — only used when EDGE_ENROLL_LOCK=1."""
    raw = os.environ.get("EDGE_OPEN_ENROLL", "").strip().lower()
    if raw in ("0", "false", "no", "off"):
        return False
    if raw in ("1", "true", "yes", "on"):
        return True
    # Locked hubs default to a start window so commission still works.
    return _env_enroll_locked()


def spoke_enroll_once(
    *,
    hub_ip: str,
    bank: int | None = None,
    beacon_counter: int = 0,
    device_id: str | None = None,
) -> dict[str, Any]:
    """Spoke posts enrollment request to hub; returns signed response dict."""
    ident = latch_cabinet_identity()
    # Prefer explicit arg → EDGE_BANK (Apply / edge.env) → GAMESTATS identity.
    # Bench seats often have empty DyzineFrame config.ini (bank_number=0) while
    # EDGE_BANK=1 — using identity alone caused hub enroll_denied bank_mismatch.
    if bank is not None:
        b = int(bank)
    else:
        env_bank = (os.environ.get("EDGE_BANK") or "").strip()
        if env_bank.isdigit():
            b = int(env_bank)
        else:
            b = int(ident.bank_number or 0)
    _priv, pub = ensure_spoke_keypair()
    key = load_fleet_hmac_key()
    body: dict[str, Any] = {
        "v": 1,
        "type": "enroll_request",
        "wg_pub": pub,
        "bank": b,
        "term_number": ident.term_number,
        "machine_role": ident.machine_role,
        "device_id": device_id or os.environ.get("EDGE_DEVICE_ID", socket.gethostname()),
        "beacon_counter": int(beacon_counter),
        "counter": int(time.time()),
        "ts": int(time.time()),
    }
    # Prefer stable field spoke id (EDGE_SPOKE_ID / LAN octet) when free on hub.
    pref = os.environ.get("EDGE_SPOKE_ID", "").strip()
    if pref.isdigit():
        body["preferred_spoke_id"] = int(pref)
        body["spoke_id"] = int(pref)
    body["mac"] = _sign_body({k: v for k, v in body.items() if k != "mac"}, key)
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    # Fast fail so Master-return enroll can retry inside the ≤5s budget.
    try:
        enroll_to = float(os.environ.get("EDGE_ENROLL_TIMEOUT_S", "2") or "2")
    except ValueError:
        enroll_to = 2.0
    sock.settimeout(max(0.5, min(5.0, enroll_to)))
    port = resolve_enroll_port()
    sock.sendto(json.dumps(body, separators=(",", ":")).encode("utf-8"), (hub_ip, port))
    data, _addr = sock.recvfrom(65535)
    sock.close()
    resp = json.loads(data.decode("utf-8"))
    if not verify_envelope(resp, key):
        raise RuntimeError("enroll response bad mac")
    return resp


def write_thumb_drive_bundle(
    out_dir: str | Path,
    *,
    bank: int,
    hub_overlay: str,
    hub_wg_pub: str,
    hub_endpoint: str,
    spoke_id: int,
    spoke_priv: str,
    spoke_pub: str,
) -> Path:
    """Offline enrollment fallback for seats that cannot reach the hub on WLAN."""
    root = Path(out_dir)
    root.mkdir(parents=True, exist_ok=True)
    overlay = overlay_ip(bank, spoke_id)
    try:
        from wlan_star_apply import resolve_wg_keepalive_s

        ka = resolve_wg_keepalive_s()
    except Exception:
        ka = 5
    conf = f"""# EtherGold hub-spoke thumb-drive enrollment — bank {bank} spoke {spoke_id}
# Tip/software still lands via Dropbox update.sh or SCP — this file is keys only.
[Interface]
PrivateKey = {spoke_priv}
ListenPort = 51820
Table = off

[Peer]
PublicKey = {hub_wg_pub}
AllowedIPs = {hub_overlay}/32
Endpoint = {hub_endpoint}
PersistentKeepalive = {ka}
"""
    conf_path = root / f"wg0.conf.spoke{spoke_id:02d}"
    conf_path.write_text(conf, encoding="utf-8")
    meta = {
        "bank": bank,
        "spoke_id": spoke_id,
        "overlay": overlay,
        "hub_overlay": hub_overlay,
        "gretap": gretap_name_for_spoke(spoke_id),
        "spoke_pub": spoke_pub,
        "note": "Install wg conf; on hub add peer + create gretap eg-sNN; tip via update.sh separately.",
    }
    (root / "enroll_meta.json").write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8")
    return conf_path


_hub: EnrollmentHub | None = None


def resolve_hub_uplink_ip() -> str:
    """Best-effort hub clear-WLAN / underlay IP for spoke enroll (no manual EDGE_HUB_IP).

    Order: explicit env → bank-wlan discovery learn → EDGE_UPLINK_IP → wg0 Endpoint.
    Prefer env over conf so a stale Endpoint cannot poison discovery forever.
    """
    for key in (
        "EDGE_HUB_IP",
        "EDGE_UPLINK_IP_LEARNED",
        "EDGE_UPLINK_IP",
    ):
        raw = (os.environ.get(key) or "").strip()
        if raw:
            return raw.split(":")[0].strip()
    endpoint = (os.environ.get("EDGE_HUB_ENDPOINT") or "").strip()
    if endpoint:
        return endpoint.split(":")[0].strip()
    conf = Path(os.environ.get("POG_EDGE_WG_CONF", "/etc/wireguard/wg0.conf"))
    try:
        if conf.is_file():
            text = conf.read_text(encoding="utf-8", errors="replace")
            # Stale SPOKE conf from old PC provision — ignore Endpoint.
            if "wlan-star SPOKE" in text and "bank 0" in text:
                return ""
            for line in text.splitlines():
                if "Endpoint" in line and "=" in line:
                    part = line.split("=", 1)[-1].strip()
                    host = part.split(":")[0].strip()
                    if host:
                        return host
    except OSError:
        pass
    return ""


def _hub_wg_public_key() -> str:
    for path in (
        Path(os.environ.get("EDGE_HUB_STATE_DIR") or DEFAULT_STATE_DIR) / "wg_public.key",
        Path("/etc/ethergold/hub/wg_public.key"),
    ):
        try:
            pub = path.read_text(encoding="utf-8").strip()
            if pub:
                return pub
        except OSError:
            continue
    env = (os.environ.get("EDGE_HUB_WG_PUB") or "").strip()
    return env


def _local_uplink_ipv4() -> str:
    iface = (os.environ.get("EDGE_UPLINK") or "wlan0").strip()
    if not iface:
        return ""
    try:
        proc = _run(
            ["ip", "-4", "-o", "addr", "show", "dev", iface],
            check=False,
        )
    except OSError:
        return ""
    for line in (proc.stdout or "").splitlines():
        m = re.search(r"\binet\s+(\d+\.\d+\.\d+\.\d+)", line)
        if m:
            return m.group(1)
    return ""


def install_spoke_wg_from_enroll(resp: dict[str, Any], *, hub_ip: str) -> dict[str, Any]:
    """After successful enroll, write /etc/wireguard/wg0.conf (Pi SoT)."""
    if not resp.get("ok"):
        return {"ok": False, "skipped": "enroll_not_ok"}
    hub_pub = str(resp.get("hub_wg_pub") or "").strip()
    endpoint = str(resp.get("hub_endpoint") or "").strip()
    if not endpoint and hub_ip:
        port = int(os.environ.get("POG_EDGE_WG_PORT", "51820") or "51820")
        endpoint = f"{hub_ip}:{port}"
    spoke_id = int(resp.get("spoke_id") or os.environ.get("EDGE_SPOKE_ID") or 0)
    bank = int(resp.get("bank") or os.environ.get("EDGE_BANK") or 0)
    if not hub_pub or not endpoint or spoke_id < 2:
        return {
            "ok": False,
            "error": "missing hub_wg_pub/endpoint/spoke_id in enroll response",
            "spoke_id": spoke_id or None,
        }
    try:
        from wlan_star_apply import write_spoke_wg_conf

        return write_spoke_wg_conf(
            bank=bank,
            spoke_id=spoke_id,
            hub_wg_pub=hub_pub,
            hub_endpoint=endpoint,
        )
    except Exception as exc:
        return {"ok": False, "error": str(exc)}


def start_enrollment_hub_for_bridge(
    bridge: Any,
    stop: threading.Event,
    router: Any | None = None,
) -> EnrollmentHub | None:
    """Explicit hub only (EDGE_TOPOLOGY=hub).

    GAMESTATS MACHINE_ROLE=M must NOT auto-start this listener — copper Masters
    (EDGE_MODE=0) and single-hop EDGE seats stay exactly as before.

    When *router* is a live DirectionalEthRouter, accepted enrolls (and restore)
    hot-add ``eg-sNN`` AF_PACKET ports so Master fanout works without restart.

    Hub peer store auto-falls back to a user-writable path when
    ``/etc/ethergold/hub`` is not writable (EG 3.0.1.30) — no manual chmod.
    """
    global _hub
    topo = os.environ.get("EDGE_TOPOLOGY", "").strip().lower()
    if topo != "hub":
        return None
    if os.environ.get("EDGE_ENROLL", "1").strip().lower() in ("0", "false", "no"):
        return None

    def _on_peer(rec: dict[str, Any], *, reason: str) -> None:
        gretap = str(rec.get("gretap") or "")
        spoke_id = rec.get("spoke_id")
        if not gretap or router is None:
            return
        try:
            ok = bool(router.add_port(gretap))
        except Exception as exc:
            bridge.event_log.emit(
                "hub_eth_port_add_error",
                gretap=gretap,
                spoke_id=spoke_id,
                reason=reason,
                message=str(exc),
            )
            return
        bridge.event_log.emit(
            "hub_eth_port_added" if ok else "hub_eth_port_add_failed",
            gretap=gretap,
            spoke_id=spoke_id,
            reason=reason,
            ports=list(router.port_names()) if hasattr(router, "port_names") else None,
        )
        # Keep launcher restart list in sync with live peers.
        try:
            root = Path(
                os.environ.get(
                    "POG_EDGE_ROOT",
                    os.path.expanduser("~/EtherGold/dev/pog-edge"),
                )
            )
            act = root / ".active_spoke_gretaps"
            if ok and act.parent.is_dir():
                lines: list[str] = []
                if act.is_file():
                    lines = [
                        ln.strip()
                        for ln in act.read_text(encoding="utf-8").splitlines()
                        if ln.strip()
                    ]
                if gretap not in lines:
                    lines.append(gretap)
                    act.write_text("\n".join(lines) + "\n", encoding="utf-8")
        except OSError:
            pass

    def _on_gone(rec: dict[str, Any], *, reason: str) -> None:
        gretap = str(rec.get("gretap") or "")
        spoke_id = rec.get("spoke_id")
        if gretap and router is not None and hasattr(router, "remove_port"):
            try:
                router.remove_port(gretap)
            except Exception as exc:
                bridge.event_log.emit(
                    "hub_eth_port_remove_error",
                    gretap=gretap,
                    spoke_id=spoke_id,
                    reason=reason,
                    message=str(exc),
                )
        try:
            ports = list(router.port_names()) if router is not None and hasattr(router, "port_names") else None
        except Exception:
            ports = None
        bridge.event_log.emit(
            "hub_eth_port_removed",
            gretap=gretap or None,
            spoke_id=spoke_id,
            reason=reason,
            ports=ports,
        )
        try:
            root = Path(
                os.environ.get(
                    "POG_EDGE_ROOT",
                    os.path.expanduser("~/EtherGold/dev/pog-edge"),
                )
            )
            act = root / ".active_spoke_gretaps"
            if gretap and act.is_file():
                lines = [
                    ln.strip()
                    for ln in act.read_text(encoding="utf-8").splitlines()
                    if ln.strip() and ln.strip() != gretap
                ]
                act.write_text(("\n".join(lines) + "\n") if lines else "", encoding="utf-8")
        except OSError:
            pass

    try:
        # Ensure writable state before EnrollmentHub constructs peer store.
        ensure_hub_state_dir(emit=bridge.event_log.emit)
        bank = int(os.environ.get("EDGE_BANK", "0") or latch_cabinet_identity().bank_number)
        _hub = EnrollmentHub(
            bank=bank,
            emit=bridge.event_log.emit,
            on_peer_ready=_on_peer,
            on_peer_gone=_on_gone,
        )
        store = _hub.store
        if store is not None:
            bridge.event_log.emit(
                "hub_peer_store_ready",
                path=str(store.path),
                peer_count=len(store.peers),
            )
        _hub.start()
        try:
            from wlan_star_apply import ensure_live_wg_keepalive

            bumped = ensure_live_wg_keepalive()
            bridge.event_log.emit("hub_wg_keepalive_bump", **bumped)
        except Exception as exc:
            bridge.event_log.emit("hub_wg_keepalive_bump_error", message=str(exc))
    except Exception as exc:
        # Never take the bridge down over provisioning (e.g. /etc/ethergold not writable).
        bridge.event_log.emit("enroll_hub_start_error", message=str(exc))
        _hub = None
        return None

    def _watch() -> None:
        stop.wait()
        if _hub is not None:
            _hub.stop()

    threading.Thread(target=_watch, name="enroll-stop", daemon=True).start()
    return _hub


def _is_hub_overlay_allowed(allowed_ips: str) -> bool:
    for tok in (allowed_ips or "").split(","):
        host = tok.strip().split("/")[0]
        parts = host.split(".")
        if len(parts) == 4 and parts[0] == "10" and parts[1] == "99" and parts[3] == "1":
            return True
    return False


def spoke_hub_handshake_unix(*, wg_if: str = "wg0") -> int:
    """Latest handshake unix seconds for the hub peer (0 = never)."""
    try:
        proc = _run(["wg", "show", wg_if, "dump"], check=False)
    except OSError:
        return 0
    for _pub, info in parse_wg_dump(proc.stdout or "").items():
        if _is_hub_overlay_allowed(str(info.get("allowed_ips") or "")):
            try:
                return int(info.get("handshake") or 0)
            except (TypeError, ValueError):
                return 0
    return 0


def start_spoke_enroll_for_bridge(bridge: Any, stop: threading.Event) -> None:
    """Spoke: UDP enroll retries so a running hub learns this seat live.

    EG 3.0.1.30: never require manual EDGE_HUB_IP. Poll bank-wlan discovery
    (EDGE_UPLINK_IP_LEARNED) and env until the hub appears; keep retrying
    while the bridge runs (hub may start later than spokes).

    EG 4.0.0.61: Master-service return ≤5s — keepalive 5s, fast enroll polls,
    and stay watching after first success so a hub reboot re-kicks quickly.
    """
    topo = os.environ.get("EDGE_TOPOLOGY", "").strip().lower()
    if topo != "spoke":
        return
    if os.environ.get("EDGE_ENROLL", "1").strip().lower() in ("0", "false", "no"):
        return

    def _worker() -> None:
        attempt = 0
        last_wait_log = 0.0
        delays = SPOKE_ENROLL_RETRY_DELAYS_S
        enrolled_ok = False
        enroll_mono = 0.0
        saw_handshake = False
        try:
            from wlan_star_apply import ensure_live_wg_keepalive

            bumped = ensure_live_wg_keepalive()
            bridge.event_log.emit("spoke_wg_keepalive_bump", **bumped)
        except Exception as exc:
            bridge.event_log.emit(
                "spoke_wg_keepalive_bump_error",
                message=str(exc),
            )
        while not stop.is_set():
            # After a good enroll: wait for WG handshake; do not thrash conf.
            if enrolled_ok:
                hs = spoke_hub_handshake_unix()
                now = time.time()
                if hs > 0:
                    saw_handshake = True
                    age = now - float(hs)
                    rekick_s = spoke_handshake_rekick_s()
                    if age < rekick_s:
                        if stop.wait(SPOKE_ENROLL_WATCH_HEALTHY_S):
                            return
                        continue
                    reason = "handshake_stale"
                    age_out = round(age, 1)
                else:
                    waited = time.monotonic() - enroll_mono
                    if not saw_handshake and waited < SPOKE_FIRST_HANDSHAKE_WAIT_S:
                        # First crypto handshake often needs tens of seconds;
                        # re-installing WG conf here guarantees handshake stays 0.
                        if stop.wait(SPOKE_ENROLL_WATCH_COLD_S):
                            return
                        continue
                    if saw_handshake and waited < SPOKE_HANDSHAKE_LOST_REKICK_S:
                        if stop.wait(SPOKE_HANDSHAKE_LOST_REKICK_S):
                            return
                        continue
                    reason = "handshake_lost" if saw_handshake else "handshake_never"
                    age_out = None
                bridge.event_log.emit(
                    "spoke_enroll_reconnect",
                    reason=reason,
                    handshake_age_s=age_out,
                    rekick_s=spoke_handshake_rekick_s(),
                )
            hub_ip = resolve_hub_uplink_ip()
            if not hub_ip:
                now = time.time()
                if now - last_wait_log >= 30.0:
                    last_wait_log = now
                    bridge.event_log.emit(
                        "spoke_enroll_wait",
                        reason="no_hub_ip",
                        note="waiting_bank_wlan_or_env",
                    )
                if stop.wait(SPOKE_ENROLL_NO_HUB_POLL_S):
                    return
                continue
            attempt += 1
            try:
                # Ensure user-writable keys dir before enroll (Permission denied
                # on /etc/ethergold/spoke is handled inside ensure_spoke_keypair).
                ensure_spoke_keys_dir(emit=bridge.event_log.emit)
                resp = spoke_enroll_once(hub_ip=hub_ip)
            except Exception as exc:
                enrolled_ok = False
                bridge.event_log.emit(
                    "spoke_enroll_error",
                    hub_ip=hub_ip,
                    attempt=attempt,
                    message=str(exc),
                    spoke_keys_dir=os.environ.get("EDGE_SPOKE_KEYS_DIR"),
                )
                delay = delays[min(attempt - 1, len(delays) - 1)]
                if stop.wait(delay):
                    return
                continue
            ok = bool(resp.get("ok"))
            bridge.event_log.emit(
                "spoke_enroll_response",
                hub_ip=hub_ip,
                ok=ok,
                reason=resp.get("reason"),
                spoke_id=resp.get("spoke_id"),
                overlay=resp.get("overlay"),
                gretap=resp.get("gretap"),
                reused=resp.get("reused"),
                attempt=attempt,
            )
            if ok:
                installed = install_spoke_wg_from_enroll(resp, hub_ip=hub_ip)
                bridge.event_log.emit(
                    "spoke_wg_conf_install",
                    ok=bool(installed.get("ok")),
                    error=installed.get("error"),
                    spoke_id=installed.get("spoke_id") or resp.get("spoke_id"),
                    hub_endpoint=installed.get("hub_endpoint"),
                )
                enrolled_ok = True
                enroll_mono = time.monotonic()
                saw_handshake = False
                # Reset attempt so a later reconnect uses the short backoff again.
                attempt = 0
                if stop.wait(SPOKE_ENROLL_WATCH_COLD_S):
                    return
                continue
            enrolled_ok = False
            delay = delays[min(attempt - 1, len(delays) - 1)]
            if stop.wait(delay):
                return

    try:
        threading.Thread(target=_worker, name="edge-enroll-spoke", daemon=True).start()
    except Exception as exc:
        bridge.event_log.emit("spoke_enroll_start_error", message=str(exc))


if __name__ == "__main__":
    import argparse

    ap = argparse.ArgumentParser(description="EtherGold hub-spoke enrollment helpers")
    sub = ap.add_subparsers(dest="cmd", required=True)
    p_open = sub.add_parser("open-window", help="Print open-enrollment seconds (hub operator)")
    p_open.add_argument("--seconds", type=float, default=DEFAULT_OPEN_WINDOW_S)
    p_spoke = sub.add_parser("spoke-request", help="Send one enrollment request")
    p_spoke.add_argument("--hub-ip", required=True)
    p_thumb = sub.add_parser("thumb-bundle", help="Write offline thumb-drive enrollment bundle")
    p_thumb.add_argument("--out", required=True)
    p_thumb.add_argument("--bank", type=int, default=0)
    p_thumb.add_argument("--spoke-id", type=int, required=True)
    p_thumb.add_argument("--hub-overlay", default="10.99.0.1")
    p_thumb.add_argument("--hub-wg-pub", required=True)
    p_thumb.add_argument("--hub-endpoint", required=True)
    args = ap.parse_args()
    if args.cmd == "open-window":
        print(f"EDGE_OPEN_ENROLL=1 EDGE_OPEN_ENROLL_S={int(args.seconds)}")
    elif args.cmd == "spoke-request":
        print(json.dumps(spoke_enroll_once(hub_ip=args.hub_ip), indent=2))
    elif args.cmd == "thumb-bundle":
        priv, pub = ensure_spoke_keypair()
        path = write_thumb_drive_bundle(
            args.out,
            bank=args.bank,
            hub_overlay=args.hub_overlay,
            hub_wg_pub=args.hub_wg_pub,
            hub_endpoint=args.hub_endpoint,
            spoke_id=args.spoke_id,
            spoke_priv=priv,
            spoke_pub=pub,
        )
        print(path)