#!/usr/bin/env python3
"""Igneum bench agent (Linux and macOS). One file, Python 3.10+ standard library only, runs as an ordinary user with
no elevated rights. The job model is GPU engineer A's pow/harness/job.schema.json. See docs/reliability/BENCH-AGENT.md.
On macOS it runs under the founder's user on the mini with the system python3 (or Xcode's); the Metal facts and the
battery state are recorded with the job; the Windows agent is the PowerShell file beside this one.

  igneum-bench-agent.py register --name pc1 [--meter <kind> <url>] [--provider clore --handle <order id>] [--wait]
      makes this machine's own key (never leaves the machine), prints the public line and the one command the
      operator runs on build-1 to authorize it, and exits; with --wait it polls until the authorization works
  igneum-bench-agent.py doctor         what this host has: python, ssh, ssh-keygen -Y, the key, the pin, the connection
  igneum-bench-agent.py run            the loop: heartbeat, pull a signed job, verify, fetch the worker by sha and
                                       the harness tree by commit, run the steps with the meter sampling, push the
                                       results to the Phase Zero store under the job's cell, repeat
  igneum-bench-agent.py stop           stop a running agent by its pid file (the command line is read first)
  igneum-bench-agent.py status         what this machine is doing
  igneum-bench-agent.py meter-test     read the configured wall meter once
  (self-install: run as a job step by the new agent file itself; BENCH-AGENT.md section 8)

Everything the agent does on the network is one ssh forced command on build-1 (igneum-bench-shell), reached with this
machine's key: signer, pull, heartbeat, done, fetch <sha>, tree <commit> <path>, put. There is no shell and no command
execution other than the steps of a job the queue signed. A job names its worker by sha256; the agent runs nothing
whose sha it did not verify, and the harness scripts come from the commit the job names on the mirror.
"""
import base64
import calendar
import ctypes
import glob
import hashlib
import json
import os
import platform
import re
import shutil
import signal
import subprocess
import sys
import tarfile
import threading
import time
import urllib.request

AGENT_VERSION = "0.3.78"
IS_MAC = sys.platform == "darwin"
SERVER = "build@188.40.146.49"
# build-1's ed25519 host key (public), pinned so there is no trust-on-first-use
SERVER_HOSTKEY = "188.40.146.49 ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIBx5DkISQGkL8tm5KQyXrsexaNNMyF0Jh672biCNtHpj"
HOME = os.path.expanduser("~")
DIR = os.environ.get("IGNEUM_BENCH_DIR", os.path.join(HOME, ".igneum-bench"))


def _console_file():
    """Ember Fleet (docs/fleet/DESIGN.md section 7): an operator's rig takes its console host from igneum/console.txt
    on the stick (copied to /var/lib/igneum/stick/console.txt at install, or placed as DIR/console.txt), one line of
    key=value words: console=<user@host> (the queue host the agent's forced command reaches), hostkey=<the host's
    ed25519 public line, "host ssh-ed25519 AAAA...">, label=<the enrolment label, igneum-bench:<rig>>. The host key is
    pinned from that line (no trust on first use); without a hostkey word the file is refused and the agent keeps the
    project's queue host. Nothing of it when the file is absent."""
    for p in (os.path.join(DIR, "console.txt"), "/var/lib/igneum/stick/console.txt"):
        try:
            with open(p) as f:
                line = f.read().strip().splitlines()[0]
        except Exception:
            continue
        words = dict(w.split("=", 1) for w in line.split() if "=" in w)
        host, hk, label = words.get("console", ""), words.get("hostkey", ""), words.get("label", "")
        if not re.fullmatch(r"[a-z_][a-z0-9_-]{0,31}@[A-Za-z0-9.-]{1,120}", host):
            continue
        if not re.fullmatch(r"[A-Za-z0-9.-]{1,120} ssh-ed25519 [A-Za-z0-9+/=]{20,}", hk.replace("_", " ")):
            continue
        return host, hk.replace("_", " "), label
    return None


_CONSOLE = _console_file()
if _CONSOLE:
    SERVER, SERVER_HOSTKEY = _CONSOLE[0], _CONSOLE[1]


def _lane_paths():
    global PIDFILE, STATEFILE
    if LANE:
        PIDFILE = os.path.join(DIR, "agent-%s.pid" % LANE[0])
        STATEFILE = os.path.join(DIR, "state-%s.json" % LANE[0])


def _paths():
    """Every path under the agent directory, recomputed when DIR changes (the self-install infers DIR from its own
    fetched path; the 0.3.7 update on the mini installed the file and then wrote its stop file under the job's work
    directory because these were fixed at import)."""
    global CONFIG, KEY, KNOWN, SIGNERS, PIDFILE, STATEFILE, STOPFILE, JOBS, METERS, RELAY_PID
    CONFIG = os.path.join(DIR, "config.json")
    METERS = os.path.join(DIR, "meters.json")        # the founder's one registration (docs/reliability/METERS.md)
    RELAY_PID = os.path.join(DIR, "relay.pid")
    KEY = os.path.join(DIR, "key")
    KNOWN = os.path.join(DIR, "known_hosts")
    SIGNERS = os.path.join(DIR, "allowed_signers")
    PIDFILE = os.path.join(DIR, "agent.pid")
    STATEFILE = os.path.join(DIR, "state.json")
    STOPFILE = os.path.join(DIR, "stop")
    JOBS = os.path.join(DIR, "jobs")


_paths()
HEARTBEAT_S = 60
IDLE_POLL_S = 30
RELAY_PORT = 7411                       # the plug relay on the mini (METERS.md section 3)
FLEET_SOURCE_FILE = "fleet-source"   # in the agent directory: the fleet file's URL under its token path (FLEET.md), never in the repository
RELAY_DROP = "/Users/Shared/igneum"     # where the founder saves meters.json on the mini; taken in within a heartbeat
WALL_IDLE_S = 120                       # the idle row: the 120 s before the first step of every wall cell
LAST_JOB_END = 0.0                      # when this lane's previous job ended (the idle window must not overlap it)
SIGN_NAMESPACE = "igneum-bench"
SIGN_IDENTITY = "igneum-queue"
STORE = "phase-zero"
STOP = threading.Event()
RUNNING = {}   # the step under way (job, cell, step) while a job runs; empty between jobs; read by the sidecar thread
HOUSEKEEPING = threading.Lock()   # the default agent's housekeeping runs on one thread at a time (the loop or the sidecar)


def housekeeping():
    """The default agent's work between jobs and, from 0.3.25, during them (the sidecar): a lane whose pid is gone is
    started again, the lanes healed, the founder's registration and the rig's network taken in from the drop, the
    plug relay kept alive. The mini, 11 October 2026 18:2x UK: a job's wait step held the loop for an hour and the
    registration in the drop was not taken, so nothing a step does may hold these."""
    if LANE or not HOUSEKEEPING.acquire(blocking=False):
        return
    try:
        start_lanes()
        heal_lanes()
        try:
            meters_intake()   # the founder's registration, when a new file sits in the drop
            wifi_intake()     # the rig's network, the same way
            start_relay()     # the plug relay, kept alive while the registration names a plug
        except Exception as ex:
            log("meters: %s" % str(ex)[:120])
    finally:
        HOUSEKEEPING.release()


def sidecar():
    """While a job runs (RUNNING set), every HEARTBEAT_S: the housekeeping and a running heartbeat, on this thread,
    so no step blocks the heartbeat, the intake or the relay. Between jobs the loop does both itself."""
    while not STOP.is_set():
        STOP.wait(HEARTBEAT_S)
        if not RUNNING:
            continue
        try:
            housekeeping()
            heartbeat(dict(RUNNING, state="running"))
        except Exception as ex:
            log("sidecar: %s" % str(ex)[:120])
try:  # output line by line even when stdout is a pipe (an ssh session without a tty, a log file)
    sys.stdout.reconfigure(line_buffering=True)
    sys.stderr.reconfigure(line_buffering=True)
except Exception:
    pass


def log(msg):
    sys.stdout.write(time.strftime("%Y-%m-%dT%H:%M:%SZ ", time.gmtime()) + msg + "\n")
    sys.stdout.flush()


def utc():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def load_config():
    with open(CONFIG) as f:
        return json.load(f)


def write_state(**kw):
    kw["ts"] = int(time.time())
    tmp = STATEFILE + ".tmp"
    with open(tmp, "w") as f:
        json.dump(kw, f)
    os.replace(tmp, STATEFILE)


def ssh_cmd(verb_args, stdin_bytes=None, timeout=120):
    """One forced command on build-1. Returns (rc, stdout bytes, stderr text)."""
    cmd = ["ssh", "-i", KEY, "-o", "IdentitiesOnly=yes", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", "-o", "PasswordAuthentication=no",
           "-o", "UserKnownHostsFile=" + KNOWN, "-o", "StrictHostKeyChecking=yes", "-o", "ServerAliveInterval=30",
           SERVER, " ".join(verb_args)]
    try:
        p = subprocess.run(cmd, input=(stdin_bytes if stdin_bytes is not None else b""), capture_output=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return 124, b"", "ssh timed out"
    return p.returncode, p.stdout, p.stderr.decode(errors="replace")


_IDENTITY = None


def refresh_identity():
    """The cards read again (a card fitted or removed while the agent runs, 10 October 2026: the RX 9060 XT fitted
    into PC 2 after the agent had read its cards). Returns (added, removed) card strings."""
    global _IDENTITY
    before = list((_IDENTITY or {}).get("cards") or [])
    _IDENTITY = None
    after = list(identity().get("cards") or [])
    return [c for c in after if c not in before], [c for c in before if c not in after]


def identity():
    """What the feed shows for this machine beside its name: the provider and physical handle from the config (an
    owned machine has none; a rented pod carries its provider and order id) and the cards nvidia-smi lists."""
    global _IDENTITY
    if _IDENTITY is None:
        cfg = load_config() if os.path.exists(CONFIG) else {}
        ident = {"provider": cfg.get("provider"), "handle": cfg.get("handle"), "pcie_link": cfg.get("pcie_link")}
        cards = []
        if IS_MAC:
            try:
                out = subprocess.run(["system_profiler", "SPDisplaysDataType", "-detailLevel", "mini"], capture_output=True, timeout=20).stdout.decode(errors="replace")
                cards = [l.split(":", 1)[1].strip() for l in out.splitlines() if "Chipset Model:" in l]
            except Exception:
                pass
        else:
            cards = linux_cards()
        # every display device the host shows, by name and bus id (NVIDIA by nvidia-smi, every vendor by lspci or
        # the Mac's chipset line); a swap-queue hold matches a substring of these strings
        ident["cards"] = cards[:24] or None
        _IDENTITY = ident
    return _IDENTITY


PCI_VENDORS = {"10de": "NVIDIA", "1002": "AMD", "8086": "Intel"}


def linux_cards():
    """Every display device of a Linux host as "<Vendor> <model> [<domain:bus:dev.fn>] <vendor id>:<device id>", one
    entry per device: NVIDIA by the driver's own name (nvidia-smi, matched to the bus), AMD and Intel by lspci -nn's
    marketing bracket, with amdgpu's sysfs product_name when the VBIOS carries one (the lspci bracket of a Polaris or
    Navi part lists the whole family: "Radeon RX 470/480/570/570X/580/580X/590"; the string keeps the bracket whole so a
    job's card substring matches, and the lane's name takes the first model). Rig 1, 11 October 2026: a raw lspci line
    made a lane name of 59 characters (the queue's heartbeat verb allows 32) and an NVIDIA card appeared twice."""
    smi = {}
    try:
        out = subprocess.run(["nvidia-smi", "--query-gpu=name,pci.bus_id", "--format=csv,noheader"], capture_output=True, timeout=20).stdout.decode(errors="replace")
        for l in out.splitlines():
            if "," in l:
                n, b = l.rsplit(",", 1)
                smi[b.strip().lower()[-7:]] = n.strip()      # "00000000:0B:00.0" -> "0b:00.0"
    except Exception:
        pass
    try:
        out = subprocess.run(["sh", "-c", "lspci -D -nn -d ::0300 2>/dev/null; lspci -D -nn -d ::0302 2>/dev/null"], capture_output=True, timeout=20).stdout.decode(errors="replace")
    except Exception:
        out = ""
    cards, seen = [], set()
    for l in out.splitlines():
        m = re.match(r"^(\S+)\s+[^:]+:\s+(.*)\s+\[([0-9a-f]{4}):([0-9a-f]{4})\]", l.strip())
        if not m:
            continue
        bus, desc, ven, dev = m.groups()
        short = bus.lower()[-7:]
        if ven == "10de" and short in smi:
            name = smi[short]
            # a part the driver names only "Graphics Device" (every CMP mining card: the 170HX on rig 1, 12 October 2026)
            # takes lspci's bracket, so a job can name the card by its model
            if name.strip().lower() in ("nvidia graphics device", "graphics device"):
                mb = re.search(r"\[([^\]]+)\]\s*\[10de:", desc + " [10de:")
                if mb and mb.group(1).strip():
                    name = "NVIDIA " + mb.group(1).strip() + " (Graphics Device)"
        else:
            vendor = PCI_VENDORS.get(ven) or desc.split(" ")[0]
            model = None
            if ven == "1002":
                for p in glob.glob("/sys/bus/pci/devices/%s/product_name" % bus):
                    try:
                        pn = open(p).read().strip()
                        if pn and len(pn) < 64:
                            model = pn if pn.lower().startswith("radeon") else "Radeon " + pn
                    except Exception:
                        pass
            if not model:
                for b in re.findall(r"\[([^\]]+)\]", desc):
                    if re.search(r"(Radeon|GeForce|Arc|RTX|RX |Quadro|Tesla)", b):
                        model = b.strip()
                        break
            if not model:
                model = re.sub(r"\[[^\]]*\]", "", desc)
                model = re.sub(r"^(Advanced Micro Devices, Inc\.|NVIDIA Corporation|Intel Corporation)\s*", "", model).strip()
            name = (vendor + " " + model).strip()
        if short in seen:
            continue
        seen.add(short)
        cards.append("%s [%s] %s:%s" % (name, bus, ven, dev))
    if not cards:   # no lspci on the host: nvidia-smi alone
        cards = ["%s [%s]" % (n, b) for b, n in smi.items()]
    return cards


LANE = None   # (name, card) when this process is a card lane (one job per card, 10 October 2026)


def heartbeat(state):
    state.update(agent=AGENT_VERSION, ts=int(time.time()), **identity())
    state["pull_ok"] = int(PULL_OK)
    # the files held by sha in the agent's cache (0.3.71): the queue credits them so a held worker never counts
    # against the founder's line again (main's order, 15 October 2026 15:4x UK)
    try:
        cd = os.path.join(DIR, "cache")
        state["cache"] = sorted(f for f in os.listdir(cd) if len(f) == 64)[:200] if os.path.isdir(cd) else []
    except Exception:
        state["cache"] = []
    if POWER_GUARD and not LANE:
        state["power_guard"] = {"cap_w": POWER_GUARD.get("max_w", 1500), "watts": GUARD_LAST["w"], "read_age_s": int(time.time() - GUARD_LAST["ts"]) if GUARD_LAST["ts"] else None,
                                "paused": [sh for sh, _ in PAUSED], "lanes": [sh for sh, _, _ in LANE_STARTS]}
    if not LANE and os.path.exists(METERS):
        # the registration host (the mini): the public registry and the relay's address ride on every heartbeat, so
        # the queue's plug verb can answer every lane; never a credential, never a plug's address
        try:
            state["plugs"], state["cables"] = public_registry()
            # the relay by this host's mDNS name (the mini's DHCP lease moved at 18:4x UK, 11 October 2026, and every
            # reader that held the address broke); the address rides beside it as the readers' fallback
            state["relay"] = "http://%s:%d" % (mdns_name(), RELAY_PORT)
            state["relay_address"] = "http://%s:%d" % (lan_ip(), RELAY_PORT)
            # each plug's watts of the minute for the fleet file (FLEET.md): the relay's own /plugs answer, keys and
            # watts only, never an address
            state["plug_watts"] = relay_watts()
            try:
                with open(os.path.join(DIR, "plug-cycles.log")) as cf:
                    last = cf.read().strip().splitlines()[-1]
                state["plug_cycle_last"] = json.loads(last)
            except Exception:
                pass
        except Exception as ex:
            state["plugs_error"] = str(ex)[:120]
    if PAUSED_FOR:
        state["paused_for"] = PAUSED_FOR   # the long job a short one pauses at the step boundary (the queue's return rule reads it)
    # the fleet file's 1.1 fields (FLEET.md): every card's fingerprint and its idle telemetry on the default
    # heartbeat; the running step's own rate and telemetry lines on a running one
    if not LANE:
        try:
            state["card_facts"] = card_facts()
        except Exception as ex:
            state["card_facts_error"] = str(ex)[:80]
    if state.get("state") == "running" and RUNNING.get("out") and RUNNING.get("step"):
        state["step_rate"] = step_rate(RUNNING["out"], RUNNING["step"])
    if not LANE:
        state["lane_restarts"] = LANE_RESTARTS
        if LAST_LANE_RESTART:
            state["last_lane_restart"] = LAST_LANE_RESTART
        # Ember Fleet (DESIGN.md 3.1): the ember block, the action receipts and the watchdog's last act, on a fleet host
        try:
            if _CONSOLE:
                state["console_host"] = SERVER.split("@", 1)[-1]
            if fleet_host(load_config()):
                state["ember"] = ember_block()
                if FLEET["receipts"]:
                    state["action_receipts"] = FLEET["receipts"][-10:]
                if FLEET["watchdog"]:
                    state["watchdog"] = FLEET["watchdog"]
                if FLEET["paused"]:
                    state["ember_paused"] = True
                if FLEET.get("resume_wanted"):
                    state["ember_resume_wanted"] = True
                try:
                    pf = json.load(open(PROFILE_FILE))
                    state["profile"] = {k: pf.get(k) for k in ("name", "sha", "applied_utc")}
                except Exception:
                    pass
                state["watchdog_rules"] = watchdog_rules()
        except Exception as ex:
            state["ember_error"] = str(ex)[:80]
    if LANE:
        state.update(lane=LANE[0], card=LANE[1])
        ssh_cmd(["heartbeat", LANE[0]], stdin_bytes=json.dumps(state).encode(), timeout=60)
    else:
        ssh_cmd(["heartbeat"], stdin_bytes=json.dumps(state).encode(), timeout=60)


# ---------------------------------------------------------------- the plug registry and the relay (METERS.md, 11 Oct 2026)
def lan_ip():
    """This host's address on the house network: the source address of a route to build-1 (no packet is sent)."""
    import socket
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect((SERVER.split("@")[-1], 22))
        return s.getsockname()[0]
    finally:
        s.close()


def mdns_name():
    """This host's name on the house network's mDNS: the Mac's LocalHostName plus .local (scutil), else the kernel's
    hostname with .local when it lacks a dot."""
    name = ""
    if IS_MAC:
        try:
            name = subprocess.run(["scutil", "--get", "LocalHostName"], capture_output=True, text=True, timeout=5).stdout.strip()
        except Exception:
            name = ""
    if not name:
        import socket
        name = socket.gethostname().split(".")[0]
    return name + ".local"


def _relay_get(meter, field, fn):
    """A relay read by the meter's named URL (the mDNS name) with the address as the fallback when the name does not
    resolve or answer: fn(url) is tried on meter[field], then on meter[field + "_fallback"]; the fallback's use is
    logged once per meter."""
    try:
        return fn(meter[field])
    except Exception as ex:
        fb = meter.get(field + "_fallback")
        if not fb:
            raise
        if not meter.get("_fallback_logged"):
            log("relay by name failed (%s); reading by address" % str(ex)[:80])
            meter["_fallback_logged"] = True
        return fn(fb)


def relay_watts():
    """{plug key: watts} from the relay's /plug/<key> answers (the last sample when it is under 10 s old), empty when
    the relay is down."""
    out = {}
    try:
        with urllib.request.urlopen("http://127.0.0.1:%d/plugs" % RELAY_PORT, timeout=4) as r:
            keys = list(json.load(r).keys())
        for k in keys:
            try:
                with urllib.request.urlopen("http://127.0.0.1:%d/plug/%s" % (RELAY_PORT, urllib.parse.quote(k, safe="")), timeout=3) as r:
                    d = json.load(r)
                out[k] = round(float(d["watts"]), 1) if d.get("watts") is not None else None
            except Exception:
                out[k] = None
    except Exception:
        pass
    return out


def public_registry():
    """The registration without its secrets: plugs {key: {name, bus}} and cables {key: {meter, serial, firmware,
    orientation, bus, channels}}; the Tapo credentials and the plugs' addresses stay in the file."""
    with open(METERS) as f:
        reg = json.load(f)
    plugs = {k: {kk: v.get(kk) for kk in ("name", "bus") if v.get(kk) is not None}
             for k, v in (reg.get("plugs") or {}).items() if isinstance(v, dict)}
    cables = {k: {kk: v.get(kk) for kk in ("meter", "serial", "firmware", "orientation", "bus", "channels", "lead") if v.get(kk) is not None}
              for k, v in (reg.get("cables") or {}).items() if isinstance(v, dict)}
    return plugs, cables


def meters_intake():
    """The founder's one hand (METERS.md section 2): a meters.json saved in RELAY_DROP is moved into the agent's
    directory (mode 600 where the agent owns it; the directory itself is the agent's, mode 700) and a receipt with the
    time and the keys, never a credential, is left beside the drop. A file that does not parse is left in place with
    a .refused note. Only the registration host (a Mac) takes files in."""
    if not IS_MAC:
        return False
    try:
        if not os.path.isdir(RELAY_DROP):
            os.makedirs(RELAY_DROP, exist_ok=True)
            os.chmod(RELAY_DROP, 0o777)
    except Exception:
        return False
    src = os.path.join(RELAY_DROP, "meters.json")
    if not os.path.exists(src):
        return False
    try:
        with open(src) as f:
            reg = json.load(f)
        keys = sorted((reg.get("plugs") or {}).keys())
        cables = sorted((reg.get("cables") or {}).keys())
        if not isinstance(reg.get("plugs"), dict):
            raise ValueError("no plugs object")
    except Exception as ex:
        with open(src + ".refused", "w") as f:
            f.write("%s not taken: %s\n" % (utc(), str(ex)[:160]))
        return False
    os.replace(src, METERS)
    try:
        os.chmod(METERS, 0o600)
    except Exception:
        pass
    with open(os.path.join(RELAY_DROP, "meters.json.taken"), "w") as f:
        f.write("%s taken into the agent: plugs %s; cables %s\n" % (utc(), ", ".join(keys) or "none", ", ".join(cables) or "none"))
    log("meters.json taken in: plugs %s, cables %s" % (keys, cables))
    return True


def wifi_intake():
    """The founder's wifi.json for the rig image (11 October 2026): the same step as meters_intake for a file with keys
    ssid and password saved in RELAY_DROP: moved into the agent's directory (mode 600), the receipt names the ssid
    only; the password is never logged. The rig image build reads the taken file by path."""
    if not IS_MAC:
        return False
    src = os.path.join(RELAY_DROP, "wifi.json")
    if not os.path.exists(src):
        return False
    try:
        with open(src) as f:
            d = json.load(f)
        ssid, psk = str(d.get("ssid", "")), str(d.get("password", ""))
        if not ssid or not (8 <= len(psk) <= 63):
            raise ValueError("ssid empty or password not 8 to 63 characters")
    except Exception as ex:
        with open(src + ".refused", "w") as f:
            f.write("%s not taken: %s\n" % (utc(), str(ex)[:160]))
        return False
    dest = os.path.join(DIR, "wifi.json")
    os.replace(src, dest)
    try:
        os.chmod(dest, 0o600)
    except Exception:
        pass
    with open(os.path.join(RELAY_DROP, "wifi.json.taken"), "w") as f:
        f.write("%s taken into the agent: ssid %s\n" % (utc(), ssid))
    log("wifi.json taken in: ssid %s" % ssid)
    return True


_RELAY_MTIME = None


class WrongPlug(Exception):
    """the device at the registered address is not the registered plug (F-154: a DHCP lease move after the home router's
    reboot of 17:09 UK, 15 October 2026 put PC 2's enclosure plug at rig 1's address and the relay read it as rig 1)"""


def _norm_mac(m):
    if not m:
        return ""
    parts = re.split(r"[:\-]", str(m).strip().lower())
    if len(parts) != 6:
        return str(m).strip().lower().replace(":", "").replace("-", "")
    return "".join(x.zfill(2) for x in parts)


def _arp_table():
    """ip -> normalised mac from the host's ARP table (macOS and Linux arp -a)"""
    out = {}
    try:
        txt = subprocess.run(["arp", "-a"], capture_output=True, timeout=10).stdout.decode(errors="replace")
        for m in re.finditer(r"\((\d+\.\d+\.\d+\.\d+)\) at ([0-9a-fA-F:]+)", txt):
            out[m.group(1)] = _norm_mac(m.group(2))
    except Exception:
        pass
    return out


def resolve_plug_address(mac, near):
    """the address of the plug with this MAC: the ARP table first; else a one-second port-80 touch of every address in the
    old address's /24 (the plugs answer on 80) to fill the table, then the table again. None when it is not on the LAN."""
    want = _norm_mac(mac)
    if not want:
        return None
    for ip, m in _arp_table().items():
        if m == want:
            return ip
    try:
        base = ".".join(str(near).split(".")[:3])
        import socket
        def touch(i):
            try:
                sk = socket.socket(); sk.settimeout(1.0); sk.connect((base + "." + str(i), 80)); sk.close()
            except Exception:
                pass
        ths = [threading.Thread(target=touch, args=(i,), daemon=True) for i in range(1, 255)]
        for th in ths: th.start()
        for th in ths: th.join(2.0)
    except Exception:
        pass
    for ip, m in _arp_table().items():
        if m == want:
            return ip
    return None


def start_relay():
    """The default agent keeps one plug relay alive while meters.json names a plug: started under RELAY_PID, checked
    every heartbeat by its pid and command line, restarted when gone or when the registration changed."""
    global _RELAY_MTIME
    if not os.path.exists(METERS):
        return
    try:
        mtime = os.path.getmtime(METERS)
        with open(METERS) as f:
            if not (json.load(f).get("plugs") or {}):
                return
    except Exception as ex:
        log("meters.json unreadable: %s" % str(ex)[:120])
        return
    alive = False
    try:
        pid = int(open(RELAY_PID).read().split()[0])
    except Exception:
        pid = 0
    if not (pid and "plug-relay" in (_cmdline(pid) or "")):
        # the file names no live relay: a relay running under another pid (started by an older agent, or the file
        # lost) is adopted by its command line, never doubled
        found = _running_pids("igneum-bench-agent.py plug-relay")
        if found:
            pid = found[0]
            try:
                with open(RELAY_PID, "w") as f:
                    f.write(str(pid))
            except OSError:
                pass
            log("plug relay found running as pid %d without its file; adopted" % pid)
    try:
        if pid and "plug-relay" in (_cmdline(pid) or ""):
            alive = True
            if _RELAY_MTIME is not None and mtime != _RELAY_MTIME:
                log("meters.json changed; the relay restarts on it")
                _kill_pidfile(RELAY_PID, "plug-relay")
                alive = False
            else:
                # a relay from an older agent file (the self-install swaps the file, the relay keeps running the old
                # one: the mini at 18:30 UK, 11 October 2026, agent 0.3.25 with the relay on 0.3.23) restarts on this
                try:
                    with urllib.request.urlopen("http://127.0.0.1:%d/health" % RELAY_PORT, timeout=5) as r:
                        ver = str(json.load(r).get("agent") or "")
                    if ver and ver != AGENT_VERSION:
                        log("the relay runs agent %s, this agent is %s; the relay restarts on it" % (ver, AGENT_VERSION))
                        _kill_pidfile(RELAY_PID, "plug-relay")
                        alive = False
                except Exception:
                    pass
    except Exception:
        pass
    if alive:
        return
    # the port must be free before a start (an ended relay releases it within seconds); a start against a held port
    # would end at once and the next heartbeat would try again
    import socket
    for _ in range(10):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            s.settimeout(1)
            if s.connect_ex(("127.0.0.1", RELAY_PORT)) != 0:
                break
        finally:
            s.close()
        time.sleep(1)
    else:
        held = _running_pids("igneum-bench-agent.py plug-relay")
        log("port %d still held%s; the relay start waits for the next heartbeat" % (RELAY_PORT, (" by pid %s" % held[0]) if held else ""))
        return
    _RELAY_MTIME = mtime
    log("starting the plug relay on port %d" % RELAY_PORT)
    subprocess.Popen([sys.executable, os.path.abspath(__file__), "plug-relay"],
                     stdout=open(os.path.join(DIR, "relay.log"), "ab"), stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
                     start_new_session=True, env=dict(os.environ, IGNEUM_BENCH_DIR=DIR))


def cmd_plug_relay():
    """The relay (METERS.md section 3): one poller per registered plug at 1 Hz over KLAP with the credentials from
    meters.json, a ring of 900 samples per plug, HTTP on 0.0.0.0:RELAY_PORT: /health, /plugs, /plug/<key>,
    /plug/<key>/window?from=&to= (epoch seconds). A housing key's slash may be written plainly (/plug/pc1/rtx-5080)
    or URL-quoted (pc1%2Frtx-5080): the key is the whole path after /plug/ before an optional /window."""
    import http.server
    import urllib.parse
    import collections
    with open(METERS) as f:
        reg = json.load(f)
    creds = reg.get("tapo") or {}
    plugs = {k: v for k, v in (reg.get("plugs") or {}).items() if isinstance(v, dict) and v.get("address")}
    mod = _meters_module({})
    signal.signal(signal.SIGTERM, lambda *a: STOP.set())
    log("plug relay %s up, pid %d, %d plugs" % (AGENT_VERSION, os.getpid(), len(plugs)))
    rings = {k: collections.deque(maxlen=900) for k in plugs}
    # the rings outlive a relay restart (0.3.43): every agent swap restarts the relay, and on 11 October 2026 the four
    # swaps of the evening emptied the only record of rig 1's plug across its two deaths; the relay going down writes
    # its rings to relay-rings.json and the next one takes every sample under twenty minutes old back
    RINGS = os.path.join(DIR, "relay-rings.json")
    try:
        old = json.load(open(RINGS))
        kept = 0
        for k in plugs:
            for ts, w, e in old.get(k) or []:
                if time.time() - ts < 1200:
                    rings[k].append((ts, w, e))
                    kept += 1
        if kept:
            log("plug relay: %d samples taken back from the previous relay" % kept)
    except Exception:
        pass
    info = {k: {"key": k, "name": v.get("name"), "bus": v.get("bus"), "address": v.get("address"), "serial": None, "model": None,
                "firmware": None, "nickname": None, "fault": None, "last_ts": None, "reads": 0, "errors": 0} for k, v in plugs.items()}
    lock = threading.Lock()
    fleet_cache = {}   # the fleet file from build-1 (FLEET.md), fetched every 10 s on demand

    def poll(key):
        cfg = plugs[key]
        reader, next_identity = None, 0
        faults, retry_at, hs_faults = 0, 0.0, 0   # 0.3.71 (the mini, 15 October 2026 14:2x UK): a plug answering 403 Forbidden to every
        # handshake was asked again every second, which keeps the Tapo lockout alive; after a fault the next handshake waits
        # 5, 10, 20 ... up to 300 s (a Forbidden starts at 60 s), and the ring stays empty meanwhile (the 503 names the fault)
        while not STOP.is_set():
            t = time.time()
            if reader is None and t < retry_at:
                STOP.wait(min(1.0, retry_at - t)); continue
            stage = "read"
            try:
                if reader is None:
                    stage = "handshake"
                    reader = mod.TapoPlug({"address": cfg["address"], "username": creds.get("username"), "password": creds.get("password")})
                    stage = "read"
                if t >= next_identity:
                    d = reader.request("get_device_info")
                    nick = d.get("nickname")
                    try:
                        nick = base64.b64decode(nick).decode() if nick else nick
                    except Exception:
                        pass
                    # 0.3.77 (F-154): the plug is bound by its MAC; a device with another MAC at the registered address is not
                    # this plug and is never read as it (a name mismatch is the same verdict when no MAC is registered)
                    want, got = _norm_mac(cfg.get("mac")), _norm_mac(d.get("mac"))
                    if (want and got and want != got) or (not want and cfg.get("name") and nick and nick != cfg.get("name")):
                        raise WrongPlug("the device at %s is %s (%r), not %s (%s %r)" % (cfg["address"], got or "?", nick, key, want or "no mac", cfg.get("name")))
                    with lock:
                        info[key].update(serial=d.get("device_id"), model=d.get("model"), firmware=d.get("fw_ver"), nickname=nick, mac=got or None,
                                         fault=(None if (not cfg.get("name") or nick == cfg.get("name")) else
                                                "the plug's own name is %r, the registration says %r" % (nick, cfg.get("name"))))
                    next_identity = t + 600
                w, e = reader.read()
                faults = 0
                with lock:
                    rings[key].append((round(t, 3), round(w, 3), e))
                    info[key].update(last_ts=round(t, 3), reads=info[key]["reads"] + 1, fault=info[key].get("fault") if info[key].get("fault") and "name is" in str(info[key].get("fault")) else None)
                if key in cycle_req and not cycle_req[key].get("started"):
                    cycle_on_poll(key, reader)   # the power-cycle verb runs on this thread, the plug's one session
            except Exception as ex:
                fault = str(ex)[:160]
                # errno 65 from a host that answers ping is macOS refusing the process the local network, not a
                # route (the mini, 11 October 2026: the relay under the agent read "No route to host" on every plug
                # while the same addresses answered from a shell); the fault names the cause and the grant
                if "Errno 65" in fault or "No route to host" in fault:
                    try:
                        if subprocess.run(["ping", "-c", "1", "-W", "1000", cfg["address"]], capture_output=True, timeout=5).returncode == 0:
                            fault = "local network access refused by macOS: grant it under Privacy and Security, Local Network (the plug answers ping)"
                    except Exception:
                        pass
                with lock:
                    info[key].update(fault=fault, errors=info[key]["errors"] + 1)
                reader = None
                next_identity = 0
                faults += 1
                # 0.3.73 (rig 1's plug, 15 October 2026 15:3x UK): a Forbidden on a READ is the plug dropping a session over
                # its weak wifi link (the KLAP sequence slips), cured by a fresh handshake at the next second; only a
                # Forbidden on the HANDSHAKE itself, three times in a row, is the lockout that backs off (60 s doubling to 300)
                forbidden = ("403" in fault or "Forbidden" in fault)
                hs_faults = (hs_faults + 1) if (forbidden and stage == "handshake") else 0
                wait = min(300.0, 60.0 * (2 ** (hs_faults - 3))) if hs_faults >= 3 else 0.0
                # 0.3.77 (F-154): a wrong device at the address, or a handshake that fails three times, re-resolves the
                # address by the registered MAC (the ARP table, then a port-80 touch of the /24); a new address is taken
                # at once and written back to the registration so a relay restart starts right
                if cfg.get("mac") and (isinstance(ex, WrongPlug) or (stage == "handshake" and faults >= 3 and faults % 3 == 0)):
                    try:
                        new_addr = resolve_plug_address(cfg["mac"], cfg["address"])
                    except Exception:
                        new_addr = None
                    if new_addr and new_addr != cfg["address"]:
                        log("plug %s: the registered MAC %s answers at %s, not %s; the address moved (a lease change)" % (key, cfg["mac"], new_addr, cfg["address"]))
                        cfg["address"] = new_addr
                        with lock:
                            info[key].update(address=new_addr)
                        try:
                            with open(METERS) as f:
                                regw = json.load(f)
                            regw.setdefault("plugs", {}).setdefault(key, {})["address"] = new_addr
                            tmpf = METERS + ".tmp"
                            with open(tmpf, "w") as f:
                                json.dump(regw, f, indent=1)
                            os.replace(tmpf, METERS)
                        except Exception as ex2:
                            log("plug %s: the registration was not rewritten: %s" % (key, ex2))
                        wait = 0.0
                    elif new_addr is None and isinstance(ex, WrongPlug):
                        wait = max(wait, 30.0)   # the plug is not on the LAN; the wrong device is left alone
                retry_at = time.time() + wait
                if faults in (1, 3, 6):
                    log("plug %s: fault %d (%s); the next handshake in %.0f s" % (key, faults, fault[:80], wait))
            STOP.wait(max(0.0, 1.0 - (time.time() - t)))

    for k in plugs:
        threading.Thread(target=poll, args=(k,), daemon=True).start()
    t_up = time.time()
    # the power-cycle verb (Ember Fleet, main's word of 00:48 UK, 14 October 2026, reliability's shape): POST
    # /plug/<key>/cycle with {"owner_token", "off_s", "caller", "reason"}; a rig's own plug only (the key starts with
    # rig and holds no slash; pc1, pc2, mini and every housing key refused by name); the owner token the one in
    # ~/.igneum-plug-owner (made at the relay's first start, mode 600, never printed); one cycle per key per ten
    # minutes; THE GUARD: before any cycle the relay asks build-1's queue (the shell's busy verb) and refuses while a
    # recorded job is taken or queued on that rig; the cycle runs on the plug's own poll thread (it holds the plug's
    # one KLAP session): off, off_s (60 by default: 15 s left rig 1's board in standby, reliability's reading), on;
    # every call appended to plug-cycles.log with the caller and the reason; the heartbeat carries the last as
    # plug_cycle_last
    OWNER = os.path.join(HOME, ".igneum-plug-owner")
    if not os.path.exists(OWNER):
        try:
            import secrets as _sec
            fd = os.open(OWNER, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, "w") as f:
                f.write(_sec.token_hex(16) + "\n")
            log("plug relay: an owner token made at %s (never printed)" % OWNER)
        except Exception as ex:
            log("plug relay: no owner token: %s" % str(ex)[:80])
    CYCLES_LOG = os.path.join(DIR, "plug-cycles.log")
    cycle_req = {}     # key -> {"off_s", "caller", "reason", "done": Event, "result": {...}}
    last_cycle = {}    # key -> epoch of the last cycle

    def cycle_on_poll(key, reader):
        """Run by the poll thread when a request waits for its key: off, wait, on, the watts before and 10 s after."""
        req = cycle_req.get(key)
        if not req or req.get("started"):
            return
        req["started"] = True
        res = {"key": key, "caller": req["caller"], "reason": req["reason"], "off_s": req["off_s"], "ok": False}
        try:
            try:
                res["watts_before"] = round(reader.read()[0], 1)
            except Exception:
                res["watts_before"] = None
            reader.request("set_device_info", {"device_on": False})
            res["off_utc"] = utc()
            time.sleep(req["off_s"])
            reader.request("set_device_info", {"device_on": True})
            res["on_utc"] = utc()
            time.sleep(10)
            try:
                res["watts_after_10s"] = round(reader.read()[0], 1)
            except Exception:
                res["watts_after_10s"] = None
            res["ok"] = True
        except Exception as ex:
            res["error"] = str(ex)[:160]
        req["result"] = res
        try:
            with open(CYCLES_LOG, "a") as f:
                f.write(json.dumps(dict(res, utc=utc())) + "\n")
        except Exception:
            pass
        log("plug cycle %s by %s: %s (%s)" % (key, req["caller"], "ok" if res["ok"] else res.get("error"), req["reason"][:80]))
        req["done"].set()

    def queue_busy(key):
        """The guard: yes when build-1's queue holds a taken or queued job for the rig; unknown when the queue does not answer."""
        rc, out, err = ssh_cmd(["busy", key], timeout=30)
        if rc != 0:
            return "unknown"
        return out.decode(errors="replace").strip()

    class H(http.server.BaseHTTPRequestHandler):
        def log_message(self, *a):
            pass

        def _send(self, code, obj, cors=False):
            body = json.dumps(obj).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            if cors:
                self.send_header("Access-Control-Allow-Origin", "*")
                self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(body)

        def do_POST(self):
            u = urllib.parse.urlsplit(self.path)
            parts = [urllib.parse.unquote(p) for p in u.path.strip("/").split("/")]
            if len(parts) != 3 or parts[0] != "plug" or parts[2] != "cycle":
                return self._send(404, {"error": "unknown path"})
            key = parts[1]
            try:
                n = int(self.headers.get("Content-Length") or 0)
                body = json.loads(self.rfile.read(min(n, 4096)).decode() or "{}")
            except Exception:
                return self._send(400, {"error": "the body is not JSON"})
            try:
                want = open(OWNER).read().strip()
            except Exception:
                want = ""
            import hmac as _hmac
            tok = str(body.get("owner_token") or "")
            if not want or not re.fullmatch(r"[0-9a-f]{32}", tok) or not _hmac.compare_digest(tok, want):
                log("plug cycle refused for %s: owner token (%s)" % (key, self.client_address[0]))
                return self._send(403, {"error": "owner token"})
            if key not in plugs:
                return self._send(404, {"error": "unknown key"})
            if "/" in key or not key.startswith("rig"):
                log("plug cycle refused for %s: not a rig's own plug (%s)" % (key, self.client_address[0]))
                return self._send(403, {"error": "a rig's own plug only; %s is refused by name" % key})
            try:
                off_s = int(body.get("off_s") or 60)
            except Exception:
                off_s = 60
            if not 15 <= off_s <= 90:
                return self._send(400, {"error": "off_s is 15 to 90"})
            caller = str(body.get("caller") or "")[:60]
            reason = str(body.get("reason") or "")[:160]
            if not caller or not reason:
                return self._send(400, {"error": "caller and reason are needed"})
            with lock:
                if time.time() - last_cycle.get(key, 0) < 600:
                    return self._send(409, {"error": "one cycle per key per ten minutes", "since_s": int(time.time() - last_cycle[key])})
                if key in cycle_req and not cycle_req[key]["done"].is_set():
                    return self._send(409, {"error": "a cycle is under way"})
            busy = queue_busy(key)
            if busy != "no":
                log("plug cycle refused for %s by %s: the queue says %s (%s)" % (key, caller, busy, reason[:60]))
                try:
                    with open(CYCLES_LOG, "a") as f:
                        f.write(json.dumps({"key": key, "caller": caller, "reason": reason, "refused": "the queue holds a job" if busy == "yes" else "the queue did not answer", "utc": utc()}) + "\n")
                except Exception:
                    pass
                return self._send(409, {"error": "refused: a recorded job is taken or queued on %s" % key if busy == "yes" else "refused: build-1's queue did not answer the guard"})
            req = {"off_s": off_s, "caller": caller, "reason": reason, "done": threading.Event()}
            with lock:
                cycle_req[key] = req
                last_cycle[key] = time.time()
            if not req["done"].wait(off_s + 60):
                return self._send(503, {"error": "the plug's poller did not run the cycle in time (its KLAP session may be faulted; see /plugs)", "key": key})
            res = req["result"]
            return self._send(200 if res.get("ok") else 502, res)

        def do_GET(self):
            u = urllib.parse.urlsplit(self.path)
            q = urllib.parse.parse_qs(u.query)
            parts = [urllib.parse.unquote(p) for p in u.path.strip("/").split("/")]
            if parts == ["health"]:
                return self._send(200, {"ok": True, "plugs": len(plugs), "up_s": int(time.time() - t_up), "agent": AGENT_VERSION})
            if parts in (["fleet"], ["fleet", ""], ["fleet", "index.html"]):
                # the fleet page on the LAN (the site lead's site/fleet/index.html, placed by the self-install): its own
                # http origin beside /fleet.json, which it fetches relative
                try:
                    with open(os.path.join(DIR, "fleet.html"), "rb") as f:
                        body = f.read()
                except Exception:
                    return self._send(404, {"error": "no fleet page beside this agent yet (the self-install places it)"})
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Cache-Control", "no-store")
                self.send_header("X-Robots-Tag", "noindex, nofollow")
                self.end_headers()
                self.wfile.write(body)
                return
            if parts == ["fleet.json"]:
                # the fleet file on the LAN (FLEET.md): build-1's /feed/fleet-live.json fetched every 10 s, this relay's
                # own watts of the second laid over every watts_wall whose plug it serves; CORS open for the page
                with lock:
                    doc, age = fleet_cache.get("doc"), time.time() - fleet_cache.get("ts", 0)
                if doc is None or age > 10:
                    try:
                        with open(os.path.join(DIR, FLEET_SOURCE_FILE)) as sf:
                            src = sf.read().strip()
                        with urllib.request.urlopen(src, timeout=6) as r:
                            doc = json.load(r)
                        with lock:
                            fleet_cache["doc"], fleet_cache["ts"] = doc, time.time()
                    except Exception as ex:
                        if doc is None:
                            return self._send(503, {"error": "the fleet file is not reachable from this relay: %s" % str(ex)[:80]})
                doc = json.loads(json.dumps(doc))
                with lock:
                    now = {k: (rings[k][-1][1] if rings[k] and time.time() - rings[k][-1][0] < 10 else None) for k in plugs}
                for h in doc.get("hosts") or []:
                    if h.get("plug_key") in now and now[h["plug_key"]] is not None:
                        h["watts_wall"], h["watts_wall_age_s"] = round(now[h["plug_key"]], 1), 0
                    for c in h.get("cards") or []:
                        ww = c.get("watts_wall") or {}
                        if ww.get("plug_key") in now and now[ww["plug_key"]] is not None:
                            ww["watts"] = round(now[ww["plug_key"]], 1)
                doc["relay_overlay_utc"] = utc()
                return self._send(200, doc, cors=True)
            if parts == ["plugs"]:
                with lock:
                    return self._send(200, {k: {kk: vv for kk, vv in v.items() if kk != "address"} for k, v in info.items()})
            # the key is the whole path after /plug/, with the slash of a housing key written plainly
            # (/plug/pc2/rtx-4070) or encoded (/plug/pc2%2Frtx-4070); the window route is the /window suffix
            # (main's bug of 11 October 2026 18:2x UK: the plain form answered "unknown path")
            if len(parts) >= 2 and parts[0] == "plug":
                rest = parts[1:]
                window = len(rest) >= 2 and rest[-1] == "window"
                key = "/".join(rest[:-1] if window else rest)
                if key in plugs and not window:
                    with lock:
                        ring = list(rings[key])
                        i = dict(info[key])
                    # 0.3.72 (rig 1's plug, 15 October 2026 14:5x UK: a far plug at 300 ms round trips answered one read in
                    # three or four seconds, so a 10 s freshness line turned half of every minute into 503 and the rows'
                    # meter files into ERR rows): the last sample serves up to 30 s with its age_s on it, the reader
                    # judging the age; past 30 s the 503 names the fault as before
                    if not ring or time.time() - ring[-1][0] > 30:
                        return self._send(503, {"key": key, "error": "no sample in the last 30 s", "fault": i.get("fault")})
                    ts, w, e = ring[-1]
                    return self._send(200, {"key": key, "name": i["name"], "bus": i["bus"], "watts": w, "energy_wh": e, "ts": ts,
                                            "age_s": round(time.time() - ts, 1), "serial": i["serial"], "model": i["model"],
                                            "firmware": i["firmware"], "fault": i["fault"]})
                if key in plugs and window:
                    try:
                        a, b = float(q.get("from", [0])[0]), float(q.get("to", [time.time()])[0])
                    except ValueError:
                        return self._send(400, {"error": "from and to are epoch seconds"})
                    with lock:
                        s = [(ts, w) for ts, w, e in rings[key] if a <= ts <= b]
                    return self._send(200, {"key": key, "samples": s, "n": len(s), "mean_w": (round(sum(w for _, w in s) / len(s), 2) if s else None),
                                            "first_ts": (s[0][0] if s else None), "last_ts": (s[-1][0] if s else None)})
            return self._send(404, {"error": "unknown path"})

    try:
        srv = http.server.ThreadingHTTPServer(("0.0.0.0", RELAY_PORT), H)
    except OSError as ex:
        # the port is held (an older relay still up): this process ends without touching the pid file (the mini,
        # 11 October 2026 18:45 UK: failed starts overwrote the file and the live relay was lost to the agent)
        log("plug relay: port %d not free (%s); this start ends" % (RELAY_PORT, str(ex)[:60]))
        return 1
    with open(RELAY_PID, "w") as f:   # the pid file only once the port is ours
        f.write(str(os.getpid()))
    srv.daemon_threads = True
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    while not STOP.is_set():
        STOP.wait(5)
    srv.shutdown()
    try:
        with lock:
            snap = {k: list(rings[k]) for k in plugs}
        with open(RINGS + ".new", "w") as f:
            json.dump(snap, f)
        os.replace(RINGS + ".new", RINGS)
    except Exception as ex:
        log("plug relay: rings not written: %s" % str(ex)[:120])
    log("plug relay down")


def queue_plug(lane):
    """The queue's answer for this lane's wall boundary (the plug verb: the housing's plug for machine/lane, else the
    machine's plug): a meter config of kind json pointing at the relay, or {} when none is registered."""
    rc, out, err = ssh_cmd(["plug", lane or "default"], timeout=60)
    if rc != 0:
        raise IOError("plug: " + err.strip()[:120])
    txt = out.decode(errors="replace").strip()
    if not txt or txt == "none":
        return {}
    return json.loads(txt)


def relay_window(meter, t_from, t_to):
    def get(url):
        with urllib.request.urlopen("%s?from=%.3f&to=%.3f" % (url, t_from, t_to), timeout=5) as r:
            return json.loads(r.read().decode())
    return _relay_get(meter, "window", get)


# ---------------------------------------------------------------- the wall meter (local-API meters)
_READERS = {}


def _meters_module(meter):
    """pow/harness/meters.py (plug, pmd2x, wireview-pro-ii): from the harness tree of the running job when there is
    one, else the copy beside the agent (DIR/meters.py, fetched by the self-install job)."""
    import importlib.util
    for cand in (meter.get("_harness_dir"), DIR):
        path = os.path.join(cand, "meters.py") if cand else None
        if path and os.path.exists(path):
            spec = importlib.util.spec_from_file_location("igneum_meters", path)
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
            return mod
    raise RuntimeError("meters.py is not on this machine (the harness tree pow/harness carries it)")


def meter_read(meter):
    """Returns (watts, energy_wh_total or None). kinds: shelly (gen2 RPC), tasmota, json (dotted paths), and through
    pow/harness/meters.py: plug (Tapo P110 over KLAP), pmd2x (ElmorLabs PMD2X serial frame), wireview-pro-ii (Thermal
    Grizzly WireView Pro II serial frame). A reader is opened once per meter config and kept."""
    kind = str(meter.get("kind") or meter.get("type") or "").lower()
    if kind in ("plug", "pmd2x", "pmd2", "wireview-pro-ii", "wireview"):
        key = json.dumps({k: v for k, v in meter.items() if not k.startswith("_")}, sort_keys=True)
        r = _READERS.get(key)
        if r is None:
            r = _READERS[key] = _meters_module(meter).open_meter(meter)
        return r.read()
    url = str(meter.get("address") or meter.get("url") or "").rstrip("/")
    if url and "://" not in url:
        url = "http://" + url
    if kind.startswith("shelly"):
        d = _get_json(url + "/rpc/Switch.GetStatus?id=" + str(meter.get("id", 0)))
        return float(d["apower"]), float(d.get("aenergy", {}).get("total", 0.0))
    if kind.startswith("tasmota"):
        d = _get_json(url + "/cm?cmnd=Status%208")
        e = d["StatusSNS"]["ENERGY"]
        return float(e["Power"]), float(e.get("Total", 0.0)) * 1000.0
    if kind.startswith("json"):
        d = _relay_get(meter, "address", _get_json) if meter.get("address_fallback") else _get_json(url)
        w = _dig(d, meter["watts_path"])
        e = _dig(d, meter["energy_wh_path"]) if meter.get("energy_wh_path") else None
        return float(w), (float(e) if e is not None else None)
    raise ValueError("unknown meter kind " + kind)


def _get_json(url):
    with urllib.request.urlopen(url, timeout=3) as r:
        return json.loads(r.read().decode())


def _dig(d, path):
    for k in path.split("."):
        d = d[int(k)] if isinstance(d, list) else d[k]
    return d


class MeterSampler(threading.Thread):
    """Samples the meter at sample_hz (default 1) into a csv; integrates watt-hours by the trapezoid rule."""

    def __init__(self, meter, path, card_bus=None):
        super().__init__(daemon=True)
        self.meter, self.path, self.stop_evt = meter, path, threading.Event()
        self.period = 1.0 / float(meter.get("sample_hz") or 1.0)
        self.samples, self.errors, self.wh, self.first_e, self.last_e, self.wsum = 0, 0, 0.0, None, None, 0.0
        self._last = None
        # 0.3.76: the held card's temperature beside every plug sample (main's order 17:1x UK, 15 October 2026)
        self.card_bus, self.tsum, self.tn, self.tmax = card_bus, 0, 0, None

    def run(self):
        with open(self.path, "w") as f:
            f.write("ts,watts,energy_wh_meter,card_c\n")
            while not self.stop_evt.is_set():
                t = time.time()
                try:
                    w, e = meter_read(self.meter)
                    if self._last is not None:
                        lt, lw = self._last
                        self.wh += (lw + w) / 2.0 * (t - lt) / 3600.0
                    self._last = (t, w)
                    if e is not None:
                        self.first_e = e if self.first_e is None else self.first_e
                        self.last_e = e
                    c = held_card_temp(self.card_bus) if self.card_bus else None
                    if c is not None:
                        self.tsum += c; self.tn += 1; self.tmax = c if self.tmax is None else max(self.tmax, c)
                    f.write("%.3f,%.3f,%s,%s\n" % (t, w, "" if e is None else "%.3f" % e, "" if c is None else c))
                    self.samples += 1
                    self.wsum += w
                except Exception as ex:  # a missed sample is recorded, never fatal
                    self.errors += 1
                    f.write("%.3f,,ERR %s\n" % (t, str(ex).replace("\n", " ")[:80]))
                f.flush()
                self.stop_evt.wait(max(0.0, self.period - (time.time() - t)))

    def result(self):
        return {"energy_wh_integrated": round(self.wh, 3), "samples": self.samples, "errors": self.errors,
                "mean_w": (round(self.wsum / self.samples, 2) if self.samples else None),
                "energy_wh_meter_delta": (None if self.first_e is None else round(self.last_e - self.first_e, 3)),
                "card_c_mean": (round(self.tsum / self.tn, 1) if self.tn else None), "card_c_max": self.tmax, "card_c_samples": self.tn}


# ---------------------------------------------------------------- fetching and verifying
def sha256_file(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def verify_signature(job_path, sig_path):
    with open(job_path, "rb") as f:
        p = subprocess.run(["ssh-keygen", "-Y", "verify", "-f", SIGNERS, "-I", SIGN_IDENTITY, "-n", SIGN_NAMESPACE,
                            "-s", sig_path], stdin=f, capture_output=True)
    return p.returncode == 0, p.stderr.decode(errors="replace").strip()


def _sha256_file(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def seed_cache_from_jobs():
    """One pass at start (0.3.71): a worker still lying in a finished job's bin is moved into the cache under its sha,
    so a set whose first row ran before the cache existed pulls nothing more (rig 1, 15 October 2026)."""
    try:
        cd = os.path.join(DIR, "cache"); os.makedirs(cd, exist_ok=True)
        for jd in os.listdir(JOBS):
            b = os.path.join(JOBS, jd, "bin")
            if not os.path.isdir(b): continue
            for f in os.listdir(b):
                fp = os.path.join(b, f)
                if os.path.isfile(fp) and os.path.getsize(fp) > 1 << 20:
                    sha = _sha256_file(fp); tgt = os.path.join(cd, sha)
                    if not os.path.exists(tgt): shutil.copyfile(fp, tgt); log("cache: seeded %s from %s/bin/%s" % (sha[:12], jd, f))
    except Exception as ex:
        log("cache seed: %s" % ex)


def fetch_by_sha(sha, dest):
    """0.3.71 (product A's ask, 15 October 2026, the founder's home line): a file fetched by sha is kept under
    .igneum-bench/cache/<sha> and served from there to every later job naming the same sha (the worker is the same
    bytes in every row of a set), verified by sha on every use, pruned of files older than fourteen days or beyond
    2 GB (oldest first)."""
    _cache = os.path.join(DIR, "cache"); os.makedirs(_cache, exist_ok=True)
    _cached = os.path.join(_cache, sha)
    try:
        if os.path.exists(_cached) and _sha256_file(_cached) == sha:
            shutil.copyfile(_cached, dest); os.utime(_cached, None)
            log("fetch %s: served from the cache (%.1f MB)" % (sha[:12], os.path.getsize(_cached) / 1e6)); return
    except Exception:
        pass
    _fetch_by_sha_remote(sha, dest)
    try:
        shutil.copyfile(dest, _cached)
        fs = sorted((os.path.join(_cache, f) for f in os.listdir(_cache)), key=os.path.getmtime)
        for f in fs:
            if time.time() - os.path.getmtime(f) > 14 * 86400: os.remove(f)
        fs = sorted((os.path.join(_cache, f) for f in os.listdir(_cache) if os.path.isfile(os.path.join(_cache, f))), key=os.path.getmtime)
        total = sum(os.path.getsize(f) for f in fs)
        for f in fs:
            if total <= 2 * 1024 ** 3: break
            total -= os.path.getsize(f); os.remove(f)
    except Exception as ex:
        log("cache: %s" % ex)


def _fetch_by_sha_remote(sha, dest):
    rc, out, err = 1, b"", ""
    for attempt in range(1, 4):   # three tries ten seconds apart: build-1's sshd reset a worker fetch at the key exchange under a burst (the mini, 23:30Z, 10 October 2026)
        rc, out, err = ssh_cmd(["fetch", sha], timeout=1800)
        if rc == 0:
            break
        log("fetch %s: try %d failed: %s" % (sha, attempt, err.strip()[:160]))
        if attempt < 3:
            time.sleep(10)
    if rc != 0:
        raise RuntimeError("fetch %s: %s" % (sha, err.strip()))
    with open(dest, "wb") as f:
        f.write(out)
    got = sha256_file(dest)
    if got != sha:
        os.remove(dest)
        raise RuntimeError("sha mismatch for %s: got %s" % (sha, got))
    # a worker stored compressed (xz magic FD 37 7A 58 5A 00): the job names the stored file's sha; the binary lands at dest
    with open(dest, "rb") as f:
        magic = f.read(6)
    if magic == b"\xfd7zXZ\x00":
        import lzma
        with lzma.open(dest, "rb") as src, open(dest + ".bin", "wb") as dst:
            shutil.copyfileobj(src, dst)
        os.replace(dest + ".bin", dest)


def fetch_tree(commit, path, dest, job_id=None):
    """pow/harness (or any path) of a commit on the mirror, as a tar stream, unpacked under dest. 15 October 2026
    (main's order, after or-build-205 and 232 held the mini half an hour in a 524 MB "pow" fetch with no heartbeat):
    the stream goes to a file on disk, not memory; a heartbeat names the fetch before every try; the owner's cancel is
    read between tries and ends the job; three tries of 600 s, ten seconds apart."""
    rc, err = 1, ""
    os.makedirs(dest, exist_ok=True)
    tmp = os.path.join(dest, ".tree.tar")
    for attempt in range(1, 4):
        if job_id:
            heartbeat({"state": "running", "job": job_id, "step": "harness-fetch %s@%s try %d" % (path, commit[:8], attempt)})
            try:
                crc, cout, _ = ssh_cmd(["cancelled", job_id], timeout=30)
                if crc == 0 and cout.decode(errors="replace").strip() == "yes":
                    raise RuntimeError("cancelled during the harness fetch of %s" % path)
            except RuntimeError:
                raise
            except Exception:
                pass
        cmd = ["ssh", "-i", KEY, "-o", "IdentitiesOnly=yes", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", "-o", "PasswordAuthentication=no",
               "-o", "UserKnownHostsFile=" + KNOWN, "-o", "StrictHostKeyChecking=yes", "-o", "ServerAliveInterval=30",
               SERVER, " ".join(["tree", commit, path])]
        try:
            with open(tmp, "wb") as f:
                p = subprocess.run(cmd, input=b"", stdout=f, stderr=subprocess.PIPE, timeout=600)
            rc, err = p.returncode, p.stderr.decode(errors="replace")
        except subprocess.TimeoutExpired:
            rc, err = 124, "ssh timed out after 600 s"
        if rc == 0:
            break
        log("tree %s %s: try %d failed: %s" % (commit, path, attempt, err.strip()[:160]))
        if attempt < 3:
            time.sleep(10)
    if rc != 0:
        try: os.remove(tmp)
        except Exception: pass
        raise RuntimeError("tree %s %s: %s" % (commit, path, err.strip()))
    with tarfile.open(tmp) as t:
        for m in t.getmembers():  # no absolute paths, no parent escapes, no links
            if m.name.startswith("/") or ".." in m.name.split("/") or m.issym() or m.islnk():
                raise RuntimeError("refused tar member " + m.name)
        t.extractall(dest)
    os.remove(tmp)


def _running_pids(needle, exclude_self=True):
    """The pids (oldest first) whose command line contains needle: ps on every platform; this process left out."""
    try:
        out = subprocess.run(["ps", "-axo", "pid=,command="], capture_output=True, text=True, timeout=10).stdout
    except Exception:
        return []
    pids = []
    for line in out.splitlines():
        parts = line.strip().split(None, 1)
        if len(parts) == 2 and needle in parts[1] and parts[0].isdigit():
            p = int(parts[0])
            if not (exclude_self and p == os.getpid()):
                pids.append(p)
    return pids


def _cmdline(pid):
    """The command line of a pid, for the kill rule (read before any signal)."""
    try:
        if IS_MAC:
            return subprocess.run(["ps", "-o", "command=", "-p", str(pid)], capture_output=True, timeout=5).stdout.decode(errors="replace").strip()
        return open("/proc/%d/cmdline" % pid, "rb").read().replace(b"\0", b" ").decode(errors="replace")
    except Exception:
        return ""


def machine_facts():
    facts = {"agent": AGENT_VERSION, "os": platform.platform(), "kernel": platform.release(),
             "python": platform.python_version(), "machine": platform.machine()}
    if IS_MAC:
        probes = (("hardware", ["sysctl", "-n", "machdep.cpu.brand_string", "hw.memsize", "hw.model"]),
                  ("metal", ["system_profiler", "SPDisplaysDataType", "-detailLevel", "mini"]),
                  ("macos", ["sw_vers", "-productVersion"]),
                  ("battery", ["pmset", "-g", "batt"]),
                  ("thermal", ["pmset", "-g", "therm"]))
    else:
        probes = (("nvidia", ["nvidia-smi", "--query-gpu=name,driver_version,memory.total,power.limit", "--format=csv,noheader"]),
                  ("gpus_pci", ["sh", "-c", "lspci -d ::0300 2>/dev/null; lspci -d ::0302 2>/dev/null"]))
    for name, cmd in probes:
        try:
            facts[name] = subprocess.run(cmd, capture_output=True, timeout=20).stdout.decode(errors="replace").strip()[:4000]
        except Exception:
            pass
    return facts


# ---------------------------------------------------------------- running a job
def _kill_pidfile(pidfile, expect):
    """Stops a step through its own pid file after reading the command line (the plan's kill rule)."""
    try:
        pid = int(open(pidfile).read().split()[0])
    except Exception:
        return False
    cmdline = _cmdline(pid)
    if not cmdline or (expect and expect not in cmdline):
        return False
    try:
        os.kill(pid, signal.SIGTERM)
        time.sleep(3)
        os.kill(pid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    return True


def _bus_of_card(card):
    """the pci bus in a host card string ('AMD RX 9060 XT [0000:11:00.0] 1002:7590'), lower case, or ''"""
    m = re.search(r"\[([0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f])\]", str(card).lower())
    return m.group(1) if m else ""


def _render_nodes():
    """/dev/dri/renderDN -> (pci bus, driver) from sysfs"""
    out = {}
    for r in sorted(glob.glob("/sys/class/drm/renderD*")):
        try:
            bus = os.path.basename(os.path.realpath(os.path.join(r, "device"))).lower()
            drv = ""
            for line in open(os.path.join(r, "device", "uevent")):
                if line.startswith("DRIVER="):
                    drv = line.strip().split("=", 1)[1]
            out["/dev/dri/" + os.path.basename(r)] = (bus, drv)
        except Exception:
            pass
    return out


def _nvidia_minors():
    """/dev/nvidiaN -> pci bus from /proc/driver/nvidia/gpus"""
    out = {}
    for g in glob.glob("/proc/driver/nvidia/gpus/*/information"):
        try:
            for line in open(g):
                if line.lower().startswith("device minor"):
                    out["/dev/nvidia" + line.split(":")[1].strip()] = os.path.basename(os.path.dirname(g)).lower()
        except Exception:
            pass
    return out


def card_env(card, env):
    """0.3.74 (GPU B's fault F-147, 15 October 2026 15:41 to 16:11 UK: a job held to the RX 9060 XT ran its GL steps on
    the RTX 5060 Ti beside Ember because the NVIDIA EGL vendor enumerates first and the steps named no card): the held
    card's vendor chooses the EGL vendor library and the visible-device variables, so a step cannot open the other
    vendor's cards at all; the held card's bus and render node are in the environment for a worker that names them."""
    bus = _bus_of_card(card)
    nodes = _render_nodes()
    vendor = ""
    m = re.search(r"\s([0-9a-f]{4}):[0-9a-f]{4}\s*$", str(card).lower())
    if m:
        vendor = m.group(1)
    if bus:
        env["IGNEUM_CARD_BUS"] = bus
        for node, (b, drv) in nodes.items():
            if b == bus:
                env["IGNEUM_CARD_RENDER"] = node
    vend = {"10de": "10_nvidia.json", "1002": "50_mesa.json", "8086": "50_mesa.json"}.get(vendor)
    if vend:
        for d in ("/usr/share/glvnd/egl_vendor.d", "/etc/glvnd/egl_vendor.d"):
            f = os.path.join(d, vend)
            if os.path.exists(f):
                env["__EGL_VENDOR_LIBRARY_FILENAMES"] = f
                break
    if vendor and vendor != "10de":
        env["CUDA_VISIBLE_DEVICES"] = ""   # no CUDA device for a Mesa card's job
        if bus:
            env["DRI_PRIME"] = "pci-" + bus.replace(":", "_").replace(".", "_")
            amd = [b for node, (b, drv) in sorted(nodes.items()) if drv == "amdgpu"]
            if bus in amd:
                env["HIP_VISIBLE_DEVICES"] = env["ROCR_VISIBLE_DEVICES"] = str(amd.index(bus))
    return env


def _pgid_pids(pgid):
    pids = set()
    for st in glob.glob("/proc/[0-9]*/stat"):
        try:
            if int(open(st).read().rsplit(")", 1)[1].split()[2]) == pgid:
                pids.add(int(st.split("/")[2]))
        except Exception:
            continue
    return pids


def _nvidia_apps():
    """pid -> pci bus for every compute or graphics process nvidia-smi lists, from nvidia-smi -q -x (0.3.75: the CUDA
    runtime opens every /dev/nvidiaN at init whatever CUDA_VISIBLE_DEVICES says, so an fd read on the NVIDIA nodes sees
    every card; GPU B's 2060 KO row ended on that false positive at 16:56 UK, 15 October 2026)"""
    out = {}
    try:
        import xml.etree.ElementTree as ET
        x = subprocess.run(["nvidia-smi", "-q", "-x"], capture_output=True, timeout=20).stdout
        for g in ET.fromstring(x).iter("gpu"):
            bus = (g.get("id") or "").lower()
            for pi in g.iter("process_info"):
                try:
                    out[int(pi.findtext("pid"))] = bus
                except Exception:
                    pass
    except Exception:
        pass
    return out


def opened_card_buses(pgid):
    """the pci buses of the cards the step's process group is using: an NVIDIA card by nvidia-smi's per-GPU process
    list, a Mesa card (amdgpu, i915) by the /dev/dri/renderDN files open in the group (the brace behind the
    environment: a worker on a card other than the held one is ended with the verdict card-mismatch)"""
    pids = _pgid_pids(pgid)
    if not pids:
        return set()
    nodes = {n: (b, d) for n, (b, d) in _render_nodes().items() if d != "nvidia"}
    buses = set()
    for pid, bus in _nvidia_apps().items():
        if pid in pids and bus:
            buses.add(bus)
    for pid in pids:
        try:
            for fd in os.listdir("/proc/%d/fd" % pid):
                try:
                    t = os.readlink("/proc/%d/fd/%s" % (pid, fd))
                except Exception:
                    continue
                if t in nodes and _fd_engine_busy(pid, fd):
                    buses.add(nodes[t][0])
        except Exception:
            continue
    return buses


def _fd_engine_busy(pid, fd):
    """0.3.78 (GPU B's 9060 XT bandwidth row, 18:13 UK, 15 October 2026): a device enumeration opens every DRM render
    node on the host for a moment, so an open node is not a use; a Mesa render node counts as used only when the fd's
    fdinfo shows engine time on it (amdgpu and i915 write drm-engine-<name>: <ns> per open file). A kernel without
    fdinfo engine lines reads as not busy, and the brace then rests on the environment alone for Mesa cards."""
    try:
        for line in open("/proc/%d/fdinfo/%s" % (pid, fd)):
            if line.startswith("drm-engine-"):
                v = line.split(":", 1)[1].strip().split()[0]
                if v.isdigit() and int(v) > 0:
                    return True
    except Exception:
        pass
    return False


FANCTL = """import ctypes, sys
# igneum fanctl (0.3.76): full|auto <pci bus> on an NVIDIA card by NVML (root); the agent writes and calls this under sudo -n
mode, bus = sys.argv[1], sys.argv[2].encode()
l = ctypes.CDLL("libnvidia-ml.so.1"); assert l.nvmlInit_v2() == 0, "nvml init"
h = ctypes.c_void_p(); assert l.nvmlDeviceGetHandleByPciBusId_v2(bus, ctypes.byref(h)) == 0, "no card at " + bus.decode()
n = ctypes.c_uint(); l.nvmlDeviceGetNumFans(h, ctypes.byref(n))
rc = []
for i in range(n.value):
    rc.append(l.nvmlDeviceSetFanSpeed_v2(h, i, 100) if mode == "full" else l.nvmlDeviceSetDefaultFanSpeed_v2(h, i))
sp = ctypes.c_uint(); l.nvmlDeviceGetFanSpeed(h, ctypes.byref(sp))
print("%s fans=%d rc=%s speed=%d" % (mode, n.value, rc, sp.value)); sys.exit(0 if all(r == 0 for r in rc) else 1)
"""


def fan_hold(card, mode):
    """0.3.76 (main's order 17:1x UK, 15 October 2026, rig 1 with its chassis fans off): the held card's own fan to 100
    percent for the row's duration and back to the automatic curve at its end and on any failure; the 80 C guard stays.
    An NVIDIA card by NVML under sudo (nvmlDeviceSetFanSpeed_v2 / SetDefaultFanSpeed_v2, proven on the 2060); an AMD
    card by its hwmon pwm1_enable 1 + pwm1 255 (back: pwm1_enable 2), proven on the 6600; a card without pwm1_enable
    (the RX 9060 XT: RDNA4 exposes no manual pwm) stays on its automatic curve and the row says so."""
    bus = _bus_of_card(card)
    m = re.search(r"\s([0-9a-f]{4}):[0-9a-f]{4}\s*$", str(card).lower())
    vendor = m.group(1) if m else ""
    if not bus or IS_MAC:
        return {"card": card, "mode": "none", "reason": "no bus"}
    try:
        if vendor == "10de":
            fp = os.path.join(DIR, "fanctl.py")
            with open(fp, "w") as f:
                f.write(FANCTL)
            r = subprocess.run(["sudo", "-n", "python3", fp, "full" if mode == "full" else "auto", bus], capture_output=True, timeout=30)
            line = (r.stdout + r.stderr).decode(errors="replace").strip()[:160]
            return {"card": card, "bus": bus, "mode": mode if r.returncode == 0 else "failed", "by": "nvml", "read": line}
        hw = glob.glob("/sys/bus/pci/devices/%s/hwmon/hwmon*" % bus)
        if not hw:
            return {"card": card, "bus": bus, "mode": "none", "reason": "no hwmon"}
        if not os.path.exists(os.path.join(hw[0], "pwm1_enable")):
            return {"card": card, "bus": bus, "mode": "auto-only", "reason": "no manual pwm on this card (its own automatic curve)"}
        if mode == "full":
            cmd = "echo 1 > %s/pwm1_enable && echo 255 > %s/pwm1" % (hw[0], hw[0])
        else:
            cmd = "echo 2 > %s/pwm1_enable" % hw[0]
        r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], capture_output=True, timeout=20)
        try:
            rpm = open(os.path.join(hw[0], "fan1_input")).read().strip()
        except Exception:
            rpm = ""
        return {"card": card, "bus": bus, "mode": mode if r.returncode == 0 else "failed", "by": "hwmon-pwm", "read": "rpm %s %s" % (rpm, r.stderr.decode(errors="replace").strip()[:80])}
    except Exception as ex:
        return {"card": card, "bus": bus, "mode": "failed", "error": str(ex)[:160]}


def card_hold_apply(idx, plan):
    """0.3.76: the held NVIDIA card's power cap and clock lock from the job's run_plan, set under sudo on the card's own
    index, read back, restored at the job's end (GPU C 17:2x UK, 15 October 2026: a shell step cannot cap or lock the
    card, the agent's user has no permission and the step named the wrong index). run_plan.power_limits_w: one value per
    cell on a rig (the first of the ladder); run_plan.clock_lock_mhz: nvidia-smi -lgc X,X; absent fields change nothing."""
    out = {"index": idx}
    pl = plan.get("power_limits_w") or []
    lock = plan.get("clock_lock_mhz")
    if not pl and not lock:
        return None
    try:
        before = subprocess.run(["nvidia-smi", "-i", str(idx), "--query-gpu=power.limit,power.default_limit,clocks.gr", "--format=csv,noheader,nounits"],
                                capture_output=True, timeout=20).stdout.decode(errors="replace").strip()
        out["before"] = before
        parts = [x.strip() for x in before.split(",")]
        out["default_limit_w"] = parts[1] if len(parts) > 1 else None
    except Exception as ex:
        out["before"] = "unread: " + str(ex)[:60]
    if pl:
        r = subprocess.run(["sudo", "-n", "/usr/bin/nvidia-smi", "-i", str(idx), "-pl", str(int(float(pl[0])))], capture_output=True, timeout=20)
        out["power_limit_w"] = {"asked": pl[0], "rc": r.returncode, "read": (r.stdout + r.stderr).decode(errors="replace").strip()[:120]}
    if lock:
        r = subprocess.run(["sudo", "-n", "/usr/bin/nvidia-smi", "-i", str(idx), "-lgc", "%d,%d" % (int(lock), int(lock))], capture_output=True, timeout=20)
        out["clock_lock_mhz"] = {"asked": lock, "rc": r.returncode, "read": (r.stdout + r.stderr).decode(errors="replace").strip()[:120]}
    try:
        out["after"] = subprocess.run(["nvidia-smi", "-i", str(idx), "--query-gpu=power.limit,clocks.gr,clocks.max.gr", "--format=csv,noheader,nounits"],
                                      capture_output=True, timeout=20).stdout.decode(errors="replace").strip()
    except Exception:
        pass
    return out


def card_hold_restore(idx, rec):
    """back to the card's default power limit and its own clocks (-rgc); recorded, never fatal"""
    got = {}
    try:
        if rec.get("clock_lock_mhz"):
            r = subprocess.run(["sudo", "-n", "/usr/bin/nvidia-smi", "-i", str(idx), "-rgc"], capture_output=True, timeout=20)
            got["rgc"] = r.returncode
        if rec.get("power_limit_w") and rec.get("default_limit_w"):
            r = subprocess.run(["sudo", "-n", "/usr/bin/nvidia-smi", "-i", str(idx), "-pl", str(int(float(rec["default_limit_w"])))], capture_output=True, timeout=20)
            got["pl"] = r.returncode
        got["after"] = subprocess.run(["nvidia-smi", "-i", str(idx), "--query-gpu=power.limit,clocks.gr", "--format=csv,noheader,nounits"],
                                      capture_output=True, timeout=20).stdout.decode(errors="replace").strip()
    except Exception as ex:
        got["error"] = str(ex)[:80]
    return got


_NVML = {"lib": None}


def held_card_temp(bus):
    """the held card's temperature in C by its pci bus (an AMD card: its hwmon edge; an NVIDIA card: NVML, no root), or None"""
    if not bus or IS_MAC:
        return None
    b = bus.lower()
    if len(b) == 12:
        b = "0000:" + b[-7:] if ":" in b[-8:] else b
    try:
        hw = glob.glob("/sys/bus/pci/devices/%s/hwmon/hwmon*" % b[-12:])
        if hw and open(os.path.join(hw[0], "name")).read().strip() == "amdgpu":
            for t in sorted(glob.glob(os.path.join(hw[0], "temp*_input"))):
                try:
                    label = open(t.replace("_input", "_label")).read().strip()
                except Exception:
                    label = ""
                if label in ("edge", ""):
                    return int(open(t).read().strip()) // 1000
    except Exception:
        pass
    try:
        if _NVML["lib"] is None:
            l = ctypes.CDLL("libnvidia-ml.so.1")
            if l.nvmlInit_v2() != 0:
                return None
            _NVML["lib"] = l
        l = _NVML["lib"]
        h = ctypes.c_void_p()
        if l.nvmlDeviceGetHandleByPciBusId_v2(b[-12:].encode(), ctypes.byref(h)) != 0:
            return None
        t = ctypes.c_uint()
        if l.nvmlDeviceGetTemperature(h, 0, ctypes.byref(t)) == 0:
            return int(t.value)
    except Exception:
        pass
    return None


def _kill_group(pid):
    try:
        os.killpg(os.getpgid(pid), signal.SIGTERM)
        time.sleep(3)
        os.killpg(os.getpgid(pid), signal.SIGKILL)
    except ProcessLookupError:
        pass


def run_step(step, work, out, env, meter, summary, deadline):
    """One step of the job: cmd with {out} substituted, under its timeout and the job's deadline, meter sampling."""
    name = step["name"]
    rec = {"name": name, "start_utc": utc(), "thermal_start": thermal_sample()}
    cmd = [str(a).replace("{out}", out).replace("{work}", work).replace("{bin}", os.path.join(os.path.dirname(work), "bin")) for a in step["cmd"]]
    exe = shutil.which(cmd[0], path=env["PATH"]) or cmd[0]
    sampler = None
    if meter:
        sampler = MeterSampler(meter, os.path.join(out, "meter_%s.csv" % name), card_bus=env.get("IGNEUM_CARD_BUS"))
        sampler.start()
    t0 = time.time()
    with open(os.path.join(out, name + ".stdout"), "wb") as so, open(os.path.join(out, name + ".stderr"), "wb") as se:
        try:
            p = subprocess.Popen([exe] + cmd[1:], cwd=work, env=env, stdout=so, stderr=se, stdin=subprocess.DEVNULL,
                                 start_new_session=True)
        except OSError as ex:
            rec.update(error=str(ex), verdict="error")
            if sampler:
                sampler.stop_evt.set()
            return rec
        with open(os.path.join(work, "step.pid"), "w") as f:
            f.write(str(p.pid))
        write_state(state="running", job=summary["job_id"], step=name, pid=p.pid)
        RUNNING.update(job=summary["job_id"], cell=summary["cell"], step=name, out=out)
        heartbeat({"state": "running", "job": summary["job_id"], "cell": summary["cell"], "step": name})
        limit = min(t0 + int(step["timeout_s"]), deadline)
        rc, why, last_ask = None, "ok", t0
        while True:
            try:
                # one check a second (the founder's word of 15 October 2026 07:3x UK, rig 1 with its chassis fans off:
                # the thermal guard reads the cards at 1 Hz and ends a step by its pid at the line)
                rc = p.wait(timeout=1)
                break
            except subprocess.TimeoutExpired:
                cancelled = False
                if time.time() - last_ask >= 60:
                    # once a minute: has the owner cancelled this job on the queue (igneum-bench cancel)? the step ends,
                    # the job is pushed with the verdict cancelled, the agent stays up
                    last_ask = time.time()
                    try:
                        crc, cout, _ = ssh_cmd(["cancelled", summary["job_id"]], timeout=30)
                        cancelled = crc == 0 and cout.decode(errors="replace").strip() == "yes"
                    except Exception:
                        pass
                breach = thermal_breach() if THERMAL_GUARD == "1" else None
                mismatch = None
                try:
                    held = env.get("IGNEUM_CARD_BUS", "")
                    if held and int(time.time()) % 5 == 0:
                        # the bus compared on its last seven characters: nvidia-smi writes 00000000:02:00.0, sysfs 0000:02:00.0
                        others = sorted(b for b in opened_card_buses(p.pid) if b[-7:] != held[-7:])
                        if others:
                            mismatch = "the step opened %s, the held card is %s" % (",".join(others), held)
                except Exception:
                    pass
                # 0.3.71: the plug relay is supervised during a job as well (the mini, 15 October 2026 14:3x UK: a relay
                # ended for a fresh handshake while a job ran stayed down until the job's end, the idle loop's check)
                try:
                    if int(time.time()) % 30 == 0 and RELAY_PORT and os.path.exists(METERS):
                        start_relay()
                except Exception:
                    pass
                if cancelled:
                    why = "cancelled"
                elif STOP.is_set() or os.path.exists(STOPFILE):
                    why = "stopped"
                elif time.time() > limit:
                    why = "timeout" if time.time() < deadline else "job-deadline"
                elif mismatch:
                    # 0.3.74: the card brace (GPU B's F-147): a worker on a card other than the held one ends at once
                    why = "card-mismatch"
                    rec["card_mismatch"] = {"utc": utc(), "detail": mismatch}
                    log("card brace: %s; step %s ended" % (mismatch, name))
                    heartbeat({"state": "running", "job": summary["job_id"], "step": name, "card_mismatch": mismatch})
                elif breach:
                    # the thermal guard: the step ends by its pid, the row says why, the heartbeat carries it
                    why = "thermal-stop"
                    rec["thermal_stop"] = {"utc": utc(), "card": breach, "cards": card_temps()}
                    log("thermal guard: %s; step %s ended" % (breach, name))
                    heartbeat({"state": "running", "job": summary["job_id"], "step": name, "thermal_stop": breach})
                else:
                    continue
                pf = step.get("pid_file")
                if not (pf and _kill_pidfile(os.path.join(work, pf), os.path.basename(exe))):
                    _kill_group(p.pid)
                rc = p.wait()
                break
    if sampler:
        sampler.stop_evt.set()
        sampler.join(5)
        rec["meter"] = sampler.result()
    rec.update(exit_code=rc, seconds=round(time.time() - t0, 1), end_utc=utc(), thermal_end=thermal_sample(),
               verdict=(why if why != "ok" else ("ok" if rc == 0 else "failed")))
    if rec["verdict"] == "ok":
        # the worker's own word over its exit code (GPU B, 13 October 2026 03:10 UK): an exactness worker that exits 0
        # with a summary line saying mismatch, refused, refused-room or partial-room is not ok; the step takes that word
        wv = worker_verdict(os.path.join(out, name + ".stdout"))
        if wv:
            rec["worker_verdict"] = wv
            if wv not in ("exact", "ok", "pass"):
                rec["verdict"] = wv
    return rec


def worker_verdict(stdout_path):
    """The last JSON line of a step's stdout that carries a verdict (top level or under "summary"), as the shell's done
    verb spells it (lower case, hyphens); None when the stdout carries no such line."""
    try:
        with open(stdout_path, "rb") as f:
            f.seek(0, 2); size = f.tell(); f.seek(max(0, size - 65536)); tail = f.read().decode("utf-8", "replace")
    except OSError:
        return None
    for line in reversed(tail.splitlines()):
        line = line.strip()
        if not line.startswith("{"):
            continue
        try:
            d = json.loads(line)
        except ValueError:
            continue
        v = d.get("verdict") if isinstance(d, dict) else None
        if v is None and isinstance(d, dict) and isinstance(d.get("summary"), dict):
            v = d["summary"].get("verdict")
        if isinstance(v, str) and v:
            v = re.sub(r"[^a-z-]", "-", v.strip().lower().replace("_", "-"))[:21].strip("-")
            return v or None
    return None


def other_lanes():
    """The other lanes of this host and what they run, from their state files (a row says what shared the host)."""
    out = []
    for f in glob.glob(os.path.join(DIR, "state-*.json")):
        ln = os.path.basename(f)[6:-5]
        if LANE and ln == LANE[0]:
            continue
        try:
            s = json.load(open(f))
            if s.get("state") == "running" and time.time() - os.stat(f).st_mtime < 600:
                out.append({"lane": ln, "job": s.get("job"), "step": s.get("step")})
        except Exception:
            continue
    if LANE:
        try:
            s = json.load(open(os.path.join(DIR, "state.json")))
            if s.get("state") == "running":
                out.append({"lane": "default", "job": s.get("job"), "step": s.get("step")})
        except Exception:
            pass
    return out


def _pull_short(cfg, paused=None):
    global PAUSED_FOR
    PAUSED_FOR = paused
    try:
        return _pull_short_inner(cfg)
    finally:
        PAUSED_FOR = None


def _pull_short_inner(cfg):
    """One short job from the queue for this lane (priority short, or a whole budget of 60 s or less), run whole and
    pushed, between two steps of a long job; returns its id or None."""
    pull = ["pull", base64.urlsafe_b64encode(LANE[1].encode()).decode() if LANE else "-", "short"]
    rc, out, err = ssh_cmd(pull, timeout=60)
    if rc != 0 or not out or out.strip() == b"none":
        return None
    try:
        envl = json.loads(out)
        job_bytes = base64.b64decode(envl["job"]); sig_bytes = base64.b64decode(envl["sig"])
        job = json.loads(job_bytes)
        jid = job.get("job_id", "unknown")
        jobdir = os.path.join(JOBS, jid)
        os.makedirs(jobdir, exist_ok=True)
        open(os.path.join(jobdir, "job.json"), "wb").write(job_bytes); open(os.path.join(jobdir, "job.sig"), "wb").write(sig_bytes)
        if not verify_signature(os.path.join(jobdir, "job.json"), os.path.join(jobdir, "job.sig")) or job.get("machine") != cfg["name"]:
            ssh_cmd(["done", jid, "refused"], stdin_bytes=b"{}", timeout=60)
            return None
        log("short job %s between steps" % jid)
        summary = run_job(cfg, job, jobdir)
        pushed = push_results(cfg, job, jobdir, summary)
        send_done(jid, summary["verdict"], pushed, jobdir)
        log("short job %s %s (%ss)" % (jid, summary["verdict"], summary.get("seconds")))
        return jid
    except Exception as ex:
        log("short job: %s" % ex)
        return None


FAN_HOLD = {"card": None}


def run_job(cfg, job, jobdir):
    """0.3.76: the held card's fan goes back to its automatic curve when the job ends on any path (a verdict, an error, a stop)"""
    try:
        return _run_job(cfg, job, jobdir)
    finally:
        if FAN_HOLD.get("card_hold"):
            try:
                idx, ch = FAN_HOLD["card_hold"]
                log("card hold restored: %s" % card_hold_restore(idx, ch))
            except Exception as ex:
                log("card hold restore failed: %s" % ex)
            FAN_HOLD["card_hold"] = None
        if FAN_HOLD.get("card"):
            try:
                r = fan_hold(FAN_HOLD["card"], "auto")
                log("fan hold restored: %s" % r)
            except Exception as ex:
                log("fan hold restore failed: %s" % ex)
            FAN_HOLD["card"] = None


def _run_job(cfg, job, jobdir):
    """Runs one verified job (GPU A's schema); returns the summary dict. Never raises past here."""
    global LAST_JOB_END
    t0 = time.time()
    work, out, binp = os.path.join(jobdir, "work"), os.path.join(jobdir, "out"), os.path.join(jobdir, "bin")
    for d in (work, out, binp):
        os.makedirs(d, exist_ok=True)
    summary = {"job_id": job["job_id"], "machine": cfg["name"], "cell": job["cell"]["name"], "candidate": job["candidate"],
               "backend": job["backend"], "configuration": job["configuration"], "operator": job["operator"],
               "worker_sha256": job["worker"]["sha256"], "harness_commit": job.get("harness_commit"),
               "start_utc": utc(), "facts": machine_facts(), "agent": AGENT_VERSION, "steps": [], "verdict": "error"}
    if cfg.get("pcie_link"):
        summary["pcie_link"] = cfg["pcie_link"]
    summary["lane"] = LANE[0] if LANE else "default"
    summary["lane_card"] = LANE[1] if LANE else None
    summary["concurrent_lanes_at_start"] = other_lanes()
    if job.get("hold_until"):
        summary["hold_until"], summary["cards_at_start"] = job["hold_until"], identity().get("cards")
    inst = job.get("instrument", {})
    meter = None
    if inst.get("grade") in ("wall", "cable", "wall-differential"):
        meter = dict(inst.get("meter") or cfg.get("meter") or {})
        plug_reason = None
        if not meter and inst.get("grade") != "cable":
            # the registry's answer (METERS.md): the housing's plug for machine/lane, else the machine's plug, read
            # through the relay on the mini as a json meter; none registered = grade none with the reason, never a
            # refused job
            try:
                meter = queue_plug(summary["lane"])
                if not meter:
                    plug_reason = "no plug registered for %s/%s (and none for %s): the founder's meters.json on the mini names none" % (cfg["name"], summary["lane"], cfg["name"])
            except Exception as ex:
                plug_reason = "the queue's plug verb failed: %s" % str(ex)[:120]
        if not meter:
            summary["instrument"] = {"grade": "none", "boundary": "wall" if inst.get("grade") != "cable" else "cable",
                                     "reason_if_none": plug_reason or ("instrument grade is %s but no meter is given in the job or this machine's config" % inst["grade"]),
                                     "note": "no meter; the row supports no efficiency claim"}
    if meter:
        meter["_harness_dir"] = os.path.join(work, "harness", "pow", "harness")
        summary["instrument"] = {"grade": inst["grade"], "kind": meter.get("kind"), "address": meter.get("address") or meter.get("url") or meter.get("port"),
                                 "accuracy": meter.get("accuracy"), "boundary": meter.get("boundary") or ("cable" if meter.get("kind") in ("pmd2x", "wireview-pro-ii") else "wall"),
                                 "channel": meter.get("channel") or "+".join(meter.get("channels") or []) or None, "sample_hz": meter.get("sample_hz") or 1, "source": "direct"}
    else:
        summary["instrument"] = {"grade": inst.get("grade", "telemetry"),
                                 "note": "TELEMETRY: no wall meter, this cell supports no efficiency claim"}
    try:
        exe = os.path.join(binp, job["worker"]["name"])
        fetch_by_sha(job["worker"]["sha256"], exe)
        os.chmod(exe, 0o700)
        if job.get("harness_commit"):
            # the harness tree and the vector sets (pow/vectors, the lead's canonical home, 29 MB) by default; a job may
            # name its own list in harness_paths (GPU B's version 4 job on pc1 read pow/vectors as absent, 10 October 2026)
            for hp in (job.get("harness_paths") or ["pow/harness", "pow/vectors"]):
                fetch_tree(job["harness_commit"], hp, os.path.join(work, "harness"), job_id=job.get("job_id"))
            for root, _, files in os.walk(os.path.join(work, "harness")):
                for fn in files:
                    os.chmod(os.path.join(root, fn), 0o700)
        if meter:
            try:
                w, e = meter_read(meter)
                summary["instrument"]["first_reading_w"] = w
                key = json.dumps({k: v for k, v in meter.items() if not k.startswith("_")}, sort_keys=True)
                if key in _READERS and hasattr(_READERS[key], "identity"):
                    summary["instrument"].update(_READERS[key].identity())
                if meter.get("plug"):
                    # a plug through the relay: the plug's own identity on the row and the bus check of a housing
                    d = _relay_get(meter, "address", _get_json) if meter.get("address_fallback") else _get_json(str(meter.get("address") or meter.get("url")))
                    summary["instrument"].update({k: d.get(k) for k in ("name", "serial", "model", "firmware")},
                                                 plug=meter["plug"], bus=meter.get("bus"), source="relay")
                    if d.get("fault"):
                        summary["instrument"]["plug_fault"] = d["fault"]
                    bus = str(meter.get("bus") or "")
                    card = str((LANE[1] if LANE else "") or "")
                    cards = " ".join(identity().get("cards") or [])
                    summary["instrument"]["bus_check"] = ("not registered" if not bus else
                                                          "match" if bus.lower() in cards.lower() else
                                                          "not checked: no card string on this host carries the bus id %s (lane card %r)" % (bus, card))
            except Exception as ex:
                # the meter record of plan revision 2: an unreadable meter is grade none with the reason, never a refused job
                summary["instrument"] = {"grade": "none", "kind": meter.get("kind"), "boundary": summary["instrument"]["boundary"],
                                         "reason_if_none": str(ex)[:160], "note": "the meter did not answer; the row supports no efficiency claim"}
                meter = None
        env = {"PATH": binp + ":" + os.path.join(work, "harness", "pow", "harness") + ":" + os.environ.get("PATH", "/usr/bin:/bin"),
               "HOME": work, "IGNEUM_JOB": job["job_id"], "IGNEUM_CELL": job["cell"]["name"], "IGNEUM_MACHINE": cfg["name"],
               "IGNEUM_OUT": out, "IGNEUM_WORKER": exe, "IGNEUM_CANDIDATE": job["candidate"], "IGNEUM_BACKEND": job["backend"],
               "IGNEUM_CONFIGURATION": job["configuration"], "LANG": "C.UTF-8", "IGNEUM_BENCH_DIR": DIR,
               "IGNEUM_LANE": summary["lane"], "IGNEUM_CONCURRENT": ";".join("%s:%s" % (l["lane"], l["job"]) for l in summary["concurrent_lanes_at_start"])}
        env.update({str(k): str(v) for k, v in job.get("environment", {}).items()})
        # the held card (BENCH-JOB-FORMS.md section 2): pinned by its string and, for NVIDIA, its index
        hold = job.get("hold_until") or {}
        if LANE and not hold.get("card_present"):   # a lane's job is pinned to the lane's card
            hold = dict(hold, card_present=LANE[1])
        if hold.get("card_present"):
            want = str(hold["card_present"]).lower()
            hit = [c for c in (identity().get("cards") or []) if want in c.lower()]
            if not hit:
                summary.update(error="the held card '%s' is not among this host's cards at start" % hold["card_present"], verdict="held-card-missing", end_utc=utc())
                return _finish(summary, jobdir)
            env["IGNEUM_CARD_INSTANCE"] = hit[0]
            try:
                card_env(hit[0], env)
            except Exception as e:
                log("card_env: %s" % e)
            try:
                # the pin by the card's UUID, never by nvidia-smi's index (the CUDA runtime orders devices its own way;
                # the PCs of 11 October 2026 after their reboots); CUDA_DEVICE_ORDER=PCI_BUS_ID makes any index the bus order
                rows = subprocess.run(["nvidia-smi", "--query-gpu=index,name,pci.bus_id,uuid", "--format=csv,noheader"], capture_output=True, timeout=20).stdout.decode(errors="replace")
                for r in rows.splitlines():
                    c = [x.strip() for x in r.split(",")]
                    if len(c) >= 4 and (want in r.lower() or c[2].lower()[-7:] in hit[0].lower()):
                        env["IGNEUM_CARD_INDEX"], env["IGNEUM_CARD_BUS"], env["CUDA_VISIBLE_DEVICES"], env["CUDA_DEVICE_ORDER"] = c[0], c[2], c[3], "PCI_BUS_ID"
                        break
            except Exception:
                pass
        # 0.3.76: the held card's own fan to full for the row (main's order 17:1x UK, 15 October 2026); restored by FAN_HOLD at the job's end
        if hold.get("card_present") and env.get("IGNEUM_CARD_INSTANCE"):
            FAN_HOLD["card"] = env["IGNEUM_CARD_INSTANCE"]
            summary["fan_hold"] = fan_hold(env["IGNEUM_CARD_INSTANCE"], "full")
            log("fan hold: %s" % summary["fan_hold"])
            if env.get("IGNEUM_CARD_INDEX"):
                try:
                    ch = card_hold_apply(env["IGNEUM_CARD_INDEX"], job.get("run_plan") or {})
                    if ch:
                        summary["card_hold"] = ch
                        FAN_HOLD["card_hold"] = (env["IGNEUM_CARD_INDEX"], ch)
                        log("card hold: %s" % ch)
                except Exception as ex:
                    summary["card_hold"] = {"error": str(ex)[:120]}
        # the rig power rule (BENCH-JOB-FORMS.md section 1)
        rp = job.get("rig_power") or {}
        if rp:
            limits = {str(k): float(v) for k, v in (rp.get("limits_w") or {}).items()}
            total = sum(limits.values())
            summary["rig_power"] = {"mode": rp.get("mode"), "socket_w": rp.get("socket_w"), "stated_sum_w": total, "applied": {}}
            if rp.get("socket_w") and total > float(rp["socket_w"]):
                summary.update(error="rig_power: the stated limits sum to %.0f W over the socket's %s W" % (total, rp["socket_w"]), verdict="refused", end_utc=utc())
                return _finish(summary, jobdir)
            for name, w in limits.items():
                try:
                    rows = subprocess.run(["nvidia-smi", "--query-gpu=index,name", "--format=csv,noheader"], capture_output=True, timeout=20).stdout.decode(errors="replace")
                    for r in rows.splitlines():
                        if name.lower() in r.lower():
                            idx = r.split(",")[0].strip()
                            res = subprocess.run(["sudo", "-n", "/usr/bin/nvidia-smi", "-i", idx, "-pl", str(int(w))], capture_output=True, timeout=20)   # 0.3.76: under sudo (the rig's rule)
                            summary["rig_power"]["applied"][name] = (res.stdout + res.stderr).decode(errors="replace").strip()[:120]
                except Exception as ex:
                    summary["rig_power"]["applied"][name] = "not applied: " + str(ex)[:80]
        deadline = t0 + int(job.get("max_total_s", 7 * 86400))
        # the idle row (METERS.md section 4): the 120 s before the first step of every wall cell, from the relay's
        # ring when this lane has been idle that long, else a fresh sample before the first step; a job may name its
        # own idle_run_s. Reported beside the load, never subtracted from a stored reading.
        wall_grade = inst.get("grade") in ("wall", "wall-differential")
        idle_s = int(inst.get("idle_run_s", 0) or 0) or (WALL_IDLE_S if (meter and wall_grade) else 0)
        if meter and idle_s > 0:
            got = None
            gap = time.time() - LAST_JOB_END if LAST_JOB_END else None
            if meter.get("window") and not inst.get("idle_run_s") and (gap is None or gap > idle_s):
                try:
                    win = relay_window(meter, time.time() - idle_s, time.time())
                    if win.get("n", 0) >= idle_s // 2:
                        got = {"mean_w": win.get("mean_w"), "samples": win.get("n"), "seconds": idle_s,
                               "from_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(win["first_ts"])),
                               "to_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(win["last_ts"])),
                               "source": "relay ring: the %d s before the first step" % idle_s,
                               "lane_idle_s": (None if gap is None else int(gap))}
                        with open(os.path.join(out, "meter_idle.csv"), "w") as f:
                            f.write("ts,watts,energy_wh_meter\n")
                            for ts, wv in win.get("samples", []):
                                f.write("%.3f,%.3f,\n" % (ts, wv))
                except Exception as ex:
                    log("idle window from the relay failed: %s; sampling fresh" % str(ex)[:100])
            if got is None:
                t_a = time.time()
                s = MeterSampler(meter, os.path.join(out, "meter_idle.csv"))
                s.start()
                STOP.wait(idle_s)
                s.stop_evt.set()
                s.join(5)
                got = dict(s.result(), seconds=idle_s, from_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t_a)), to_utc=utc(),
                           source=("fresh sample before the first step" + ("" if gap is None or gap > idle_s else
                                                                            ": the lane's previous job ended %d s ago" % int(gap))))
            summary["idle"] = got
        verdict = "ok"
        steps = job["steps"]
        splittable = job.get("split", True) is not False and not job.get("priority") == "short" and len(steps) > 1
        summary["interleaved"] = []
        for i, step in enumerate(steps):
            rec = run_step(step, work, out, env, meter, summary, deadline)
            rec["store_as"] = step.get("store_as", [])
            idle_w = (summary.get("idle") or {}).get("mean_w")
            if rec.get("meter") and idle_w is not None and rec["meter"].get("mean_w") is not None:
                rec["meter"]["differential_w"] = round(rec["meter"]["mean_w"] - idle_w, 2)   # stored beside both, never in place of either
            summary["steps"].append(rec)
            try:
                push_step(cfg, job, jobdir, summary, rec)
            except Exception as ex:
                log("step %s: store failed: %s" % (rec["name"], ex))
            if rec["verdict"] != "ok":
                verdict = rec["verdict"]
                break
            # the step boundary (the founder's ask of 10 October 2026): between two steps of a long job the lane runs
            # one short job if the queue holds one, then resumes; a job that says split false runs whole
            if splittable and i < len(steps) - 1 and not STOP.is_set():
                short = _pull_short(cfg, summary["job_id"])
                if short:
                    summary["interleaved"].append(short)
                    heartbeat({"state": "running", "job": job["job_id"], "cell": job["cell"]["name"], "step": step["name"], "paused_for": short})
        summary["concurrent_lanes_at_end"] = other_lanes()
        summary["verdict"] = verdict
        idle_w = (summary.get("idle") or {}).get("mean_w")
        loads = [st["meter"] for st in summary["steps"] if st.get("meter") and st["meter"].get("mean_w") is not None]
        if meter and idle_w is not None and loads:
            longest = max(loads, key=lambda m: m.get("samples") or 0)
            summary["wall"] = {"plug": meter.get("plug"), "grade": summary["instrument"].get("grade"), "idle_w": idle_w,
                               "load_w": longest["mean_w"], "differential_w": round(longest["mean_w"] - idle_w, 2),
                               "idle_source": summary["idle"].get("source"), "load_source": "the longest step's mean"}
    except Exception as ex:
        summary["error"] = str(ex)
    summary["seconds"] = round(time.time() - t0, 1)
    summary["end_utc"] = utc()
    LAST_JOB_END = time.time()
    return _finish(summary, jobdir)


def send_done(jid, verdict, pushed, jobdir):
    """The done line to the queue, three tries ten seconds apart; a line that still fails is kept as pending-done in the
    job's directory and retried at each loop (12 October 2026: one lost done left a deadline-ended job "running" in
    the table for ten minutes while its summary sat in the store, and the queue returned it to the head)."""
    body = json.dumps(pushed).encode() if not isinstance(pushed, (bytes, bytearray)) else pushed
    for i in range(3):
        rc, out, err = ssh_cmd(["done", jid, verdict], stdin_bytes=body, timeout=60)
        if rc == 0:
            try:
                os.remove(os.path.join(jobdir, "pending-done"))
            except OSError:
                pass
            return True
        log("done line for %s failed (try %d of 3): %s" % (jid, i + 1, (err or "").strip()[:120]))
        time.sleep(10)
    with open(os.path.join(jobdir, "pending-done"), "w") as f:
        json.dump({"verdict": verdict, "pushed": pushed if not isinstance(pushed, (bytes, bytearray)) else {}}, f)
    return False


def retry_pending_done():
    """Every job directory holding a pending-done file gets its done line sent again (one try each per loop)."""
    for jid in os.listdir(JOBS) if os.path.isdir(JOBS) else []:
        pf = os.path.join(JOBS, jid, "pending-done")
        if not os.path.exists(pf):
            continue
        try:
            pend = json.load(open(pf))
            rc, out, err = ssh_cmd(["done", jid, pend["verdict"]], stdin_bytes=json.dumps(pend.get("pushed") or {}).encode(), timeout=60)
            if rc == 0:
                os.remove(pf); log("pending done line for %s sent" % jid)
        except Exception as ex:
            log("pending done for %s: %s" % (jid, str(ex)[:100]))


def _finish(summary, jobdir):
    with open(os.path.join(jobdir, "out", "summary.json"), "w") as f:
        json.dump(summary, f, indent=1, sort_keys=True)
    return summary


def _put_one(job, jobdir, c, rel):
    path = os.path.join(jobdir, rel)
    if not os.path.isfile(path):
        return "MISSING"
    with open(path, "rb") as f:
        data = f.read()
    rc, o, err = ssh_cmd(["put", STORE, c, job["job_id"] + "_" + os.path.basename(rel)], stdin_bytes=data, timeout=1800)
    return o.decode().strip() if rc == 0 else "FAILED " + err.strip()


def push_step(cfg, job, jobdir, summary, rec):
    """Per-step storing (GPU B's ask and main's order, 10 October 2026): the minute a step ends its store_as files,
    stdout, stderr and meter CSV go to the cell under the end-of-job names; the end-of-job push skips them."""
    ps = summary.setdefault("pushed_steps", {})
    cell = job["cell"]["name"]
    targets = [(cell, "out/%s.stdout" % rec["name"]), (cell, "out/%s.stderr" % rec["name"]), (cell, "out/meter_%s.csv" % rec["name"])]
    for entry in rec.get("store_as", []):
        c, _, fn = entry.rpartition(":")
        targets.append((c or cell, "out/" + fn))
    for c, rel in targets:
        k = c + "|" + rel
        if k in ps or not os.path.isfile(os.path.join(jobdir, rel)):
            continue
        ps[k] = _put_one(job, jobdir, c, rel)


def push_results(cfg, job, jobdir, summary):
    """Everything the agent recorded plus every store_as file, into the job's cell (or <cell>:<file> targets)."""
    out = os.path.join(jobdir, "out")
    cell = job["cell"]["name"]
    pushed = {}
    ps = summary.get("pushed_steps") or {}
    names = ["job.json", "job.sig", "out/summary.json"] + ["out/" + n for n in sorted(os.listdir(out))
                                                            if n.startswith("meter_") or n.endswith((".stdout", ".stderr"))]
    targets = [(cell, n) for n in names]
    for rec in summary.get("steps", []):
        for entry in rec.get("store_as", []):
            c, _, fn = entry.rpartition(":")
            targets.append((c or cell, "out/" + fn))
    seen = set()
    for c, rel in targets:
        if (c, rel) in seen:
            continue
        seen.add((c, rel))
        k = c + "|" + rel
        if k in ps and not str(ps[k]).startswith("FAILED"):
            pushed[rel] = ps[k]
            continue
        pushed[rel] = _put_one(job, jobdir, c, rel)
    return pushed


# ---------------------------------------------------------------- the loop and the commands
def _lane_name(card):
    """A lane's name from a card string: the model words, lower case, joined by dashes."""
    name = card.split(" [")[0].strip()
    name = re.sub(r"^(NVIDIA|AMD|Intel\(R\)|Intel|Apple) ", "", name)
    name = re.sub(r"^(GeForce|Radeon|Arc\(TM\)|Arc) ", "", name)
    name = name.split("/")[0].strip()   # a family bracket ("RX 470/480/570/580/590"): the first model names the lane
    return re.sub(r"[^A-Za-z0-9]+", "-", name).strip("-").lower()[:32]


SINGLE_CARD = None   # the one card of a one-card machine: the default agent is its lane (0.3.18, the mini)


def _lane_owns_pid(pid, short):
    """True when the process at pid runs this file as the lane named short (its command line read: /proc on Linux,
    ps on a Mac); False for any other holder of the pid."""
    try:
        if os.path.exists("/proc/%d/cmdline" % pid):
            with open("/proc/%d/cmdline" % pid, "rb") as f:
                args = f.read().split(b"\0")
            cl = [a.decode("utf-8", "replace") for a in args]
        else:
            out = subprocess.run(["ps", "-p", str(pid), "-o", "command="], capture_output=True, text=True, timeout=10).stdout
            cl = out.split()
        return "--lane" in cl and cl[cl.index("--lane") + 1] == short and any("igneum-bench-agent" in a for a in cl)
    except Exception:
        return False


def _is_integrated(card):
    """A board's integrated graphics is no lane: AMD's "Radeon(TM) Graphics" (the PCs), Intel's integrated controller
    (rig 1, 11 October 2026 20:41 UK: the agent started a lane "xeon-e3-1200-v3" for the B85 board's iGPU) and
    anything at the host bridge's own display slot 00:02.0."""
    c = card or ""
    return ("Radeon(TM) Graphics" in c or "Integrated Graphics" in c or "Iris" in c or "UHD Graphics" in c
            or "HD Graphics" in c or "[0000:00:02.0]" in c)


def _running_lane_pid(short):
    """The oldest live process running this file as the lane named short (ps on every platform), or 0."""
    try:
        out = subprocess.run(["ps", "-axo", "pid=,lstart=,command="], capture_output=True, text=True, timeout=15).stdout
        found = []
        for line in out.splitlines():
            cl = line.split()
            if "--lane" in cl and cl[cl.index("--lane") + 1] == short and any("igneum-bench-agent" in a for a in cl):
                pid = int(cl[0])
                if pid != os.getpid():
                    found.append(pid)
        return found[0] if found else 0   # ps lists in pid order; the oldest lane is the one the queue knows
    except Exception:
        return 0


def start_lanes():
    """One lane per card the host shows (the boards' integrated graphics left out): this file run again as
    `run --lane <name> --card <string> --parent <pid>`; a lane whose pid file holds a live pid is left as it is.
    A machine with exactly one card runs no lane: the default agent takes card jobs and card-less jobs itself
    (the mini, 10 October 2026: a default lane and an m6 lane overlapped on the one GPU)."""
    global SINGLE_CARD
    seen = {}
    real = [c for c in (identity().get("cards") or []) if c and not _is_integrated(c)]
    if RIG_SERIAL:
        # a rig runs ONE card at a time (the founder's rule of 11 October 2026 20:0x UK: five cards seated, the chassis
        # fans out): no lane processes; the default agent takes every card job itself, serially, pinned to the job's
        # card by the hold logic; the other cards idle by construction
        if SINGLE_CARD != "*":
            log("a rig: one card at a time; the default agent takes every card job itself (%d cards: %s)" % (len(real), "; ".join(c.split(" [")[0] for c in real)))
        SINGLE_CARD = "*"
        # a lane process left by an older agent (the swap keeps the lanes up) would pull its card's jobs beside this
        # agent: every lane is ended by its pid file after its command line is read, its file removed
        for pf in glob.glob(os.path.join(DIR, "agent-*.pid")):
            short = os.path.basename(pf)[6:-4]
            try:
                lp = int(open(pf).read().split()[0])
            except Exception:
                lp = 0
            if lp and _lane_owns_pid(lp, short):
                log("a rig: lane %s (pid %d) ended; one card at a time" % (short, lp))
                try:
                    os.kill(lp, signal.SIGTERM)
                except OSError:
                    pass
            try:
                os.remove(pf)
            except OSError:
                pass
        return
    if len(real) == 1:
        if SINGLE_CARD != real[0]:
            log("one card (%s): the default agent is its lane; no lane process" % real[0])
        SINGLE_CARD = real[0]
        return
    SINGLE_CARD = None
    started_this_pass = False
    for card in (identity().get("cards") or []):
        if not card or _is_integrated(card):
            continue
        short = _lane_name(card)
        if not short:
            continue
        if short in seen:
            seen[short] += 1
            short = "%s-%d" % (short, seen[short])
        else:
            seen[short] = 1
        pf = os.path.join(DIR, "agent-%s.pid" % short)
        try:
            lp = int(open(pf).read().strip())
            os.kill(lp, 0)
            # a live pid is the lane's only when its command line says so (PC 1's 5090 lane, 11 October 2026: after
            # a power cycle another process wore the pid in the file and the lane was never started); a stale file
            # is removed and the lane started
            if _lane_owns_pid(lp, short):
                continue
            log("lane %s: pid file %d is another process; stale file removed" % (short, lp))
            try:
                os.remove(pf)
            except OSError:
                pass
        except (FileNotFoundError, ValueError, ProcessLookupError, PermissionError):
            pass
        # a lane process already running without its file (the file lost, or holding another pid) is adopted, never
        # doubled: its pid is written back and the lane left as it is (PC 2, 11 October 2026: a repair job removed
        # live lanes' files and the agent started duplicates)
        running = _running_lane_pid(short)
        if running:
            try:
                with open(pf, "w") as f:
                    f.write(str(running))
            except OSError:
                pass
            log("lane %s already runs as pid %d; adopted, pid file rewritten" % (short, running))
            continue
        if POWER_GUARD:
            # one new lane per housekeeping pass (the stagger: the passes are HEARTBEAT_S apart, at least the guard's stagger_s),
            # and only while the plug answers fresh and reads under the cap
            if started_this_pass:
                return
            w, fresh = plug_watts()
            if not fresh or w >= float(POWER_GUARD.get("max_w", 1500)):
                log("lane %s waits: the plug reads %s (the cap %s W)" % (short, ("%.0f W" % w) if fresh else "no fresh answer", POWER_GUARD.get("max_w", 1500)))
                return
            # the CPU rule (rig 1, 12 October 2026 17:5x UK: a two-core board starved four OpenCL enumerations at once): a
            # new lane starts only while the one-minute load is under max_load (config power_guard.max_load, the core count
            # by default) and while fewer than max_workers lanes run a job (power_guard.max_workers, 3 by default)
            try:
                load1 = os.getloadavg()[0]
                max_load = float(POWER_GUARD.get("max_load", os.cpu_count() or 2))
                busy = 0
                for sf in glob.glob(os.path.join(DIR, "state-*.json")):
                    try:
                        sd = json.load(open(sf))
                        if sd.get("state") == "running" and sd.get("job"):
                            busy += 1
                    except Exception:
                        pass
                max_workers = int(POWER_GUARD.get("max_workers", 3))
                if load1 >= max_load or busy >= max_workers:
                    log("lane %s waits: the load reads %.1f (the line %.0f) and %d lane(s) run a job (the line %d)" % (short, load1, max_load, busy, max_workers))
                    return
            except Exception:
                pass
            if LANE_STARTS and time.time() - LANE_STARTS[-1][2] < float(POWER_GUARD.get("stagger_s", 30)):
                return
        log("starting lane %s for '%s'" % (short, card.split(" [")[0]))
        p = subprocess.Popen([sys.executable, os.path.abspath(__file__), "run", "--lane", short, "--card", card.split(" [")[0].strip(), "--parent", str(os.getpid())],
                         stdout=open(os.path.join(DIR, "agent.log"), "ab"), stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, start_new_session=True,
                         env=dict(os.environ, IGNEUM_BENCH_DIR=DIR))
        LANE_STARTS.append((short, p.pid, time.time())); started_this_pass = True


def prune_jobs(keep):
    """A lane prunes the work and bin trees of its OWN finished job directories only (marker file "lane", summary
    present); never another lane's, never a directory without its summary (pc1 and pc2, 10 October 2026: a job
    starting on one lane removed the harness tree under the job running on another); an unmarked directory with a
    summary older than an hour goes as well."""
    mine = LANE[0] if LANE else "default"
    for old in glob.glob(os.path.join(JOBS, "*")):
        if os.path.basename(old) == keep or not os.path.isdir(old):
            continue
        if not os.path.exists(os.path.join(old, "out", "summary.json")):
            continue
        owner = ""
        try:
            owner = open(os.path.join(old, "lane")).read().strip()
        except Exception:
            pass
        if owner and owner != mine:
            continue
        if not owner and time.time() - os.stat(old).st_mtime < 3600:
            continue
        for sub in ("work", "bin"):
            shutil.rmtree(os.path.join(old, sub), ignore_errors=True)


PULL_OK = time.time()
PAUSED_FOR = None
LANE_RESTARTS = 0
LAST_LANE_RESTART = ""


def heal_lanes():
    """The lane self-heal (the founder's standing order, 10 October 2026): a lane idle whose last successful pull is
    older than 180 s while this agent's own last pull answered within 120 s is ended by its pid file (command line
    read first) and started again at the next heartbeat; a lane running a job is never touched; every restart is a
    line in lane-restarts.log and a count on the heartbeat."""
    global LANE_RESTARTS, LAST_LANE_RESTART
    now = time.time()
    if now - PULL_OK > 120:
        return
    for f in glob.glob(os.path.join(DIR, "state-*.json")):
        ln = os.path.basename(f)[6:-5]
        try:
            s = json.load(open(f))
        except Exception:
            continue
        if s.get("state") != "idle":
            continue
        po = s.get("pull_ok") or 0
        if not po or now - po < 180:
            continue
        if _kill_pidfile(os.path.join(DIR, "agent-%s.pid" % ln), "igneum-bench-agent"):
            LANE_RESTARTS += 1
            LAST_LANE_RESTART = "%s at %s: no successful pull for %d s while the queue answered this agent" % (ln, utc(), now - po)
            log("lane %s ended by its pid file: %s; started again at the next heartbeat" % (ln, LAST_LANE_RESTART))
            with open(os.path.join(DIR, "lane-restarts.log"), "a") as lf:
                lf.write("%s %s no successful pull for %d s\n" % (utc(), ln, now - po))


THERMAL_JUNCTION_C = 95   # the thermal guard (main's order of 11 October 2026 19:55 UK, rig 1 with its chassis fans unplugged):
THERMAL_EDGE_C = 85       # a step ends by its pid when any card's junction or edge crosses these; the card's own fan curve stands
THERMAL_GUARD = os.environ.get("IGNEUM_THERMAL_GUARD", "")   # "1" turns it on; a rig's agent turns it on by its hostname
RIG_SERIAL = False   # a machine named rig*: one card at a time, no lane processes (set at start from the config's name)
POWER_GUARD = None   # a rig with per-card lanes (main's rule of 12 October 2026 17:4x UK): config.json "power_guard": {"url": <the
#   plug relay's JSON for this machine's own plug>, "max_w": 1500, "stagger_s": 30}; a new lane starts only while the plug reads
#   under max_w and at most one lane per housekeeping pass (the stagger); above max_w the newest running lane is paused
#   (SIGSTOP on its session) until the reading falls 150 W under the cap for thirty seconds; every pause and resume is a
#   line in power-guard.log and in the heartbeat; a plug that does not answer starts no new lane and pauses nothing
PAUSED = []          # the lanes paused by the power guard, newest last: [(short, pid)]
LANE_STARTS = []     # the lanes this agent started, in order: [(short, pid, when)]
GUARD_LAST = {"w": None, "ts": 0, "under_since": 0}

def plug_watts():
    """The machine's own plug through the relay named in the guard: (watts, fresh); a reading older than 20 s or no answer is not fresh."""
    g = POWER_GUARD or {}
    try:
        with urllib.request.urlopen(g.get("url", ""), timeout=5) as r:
            d = json.loads(r.read().decode())
        w = d.get("watts"); age = float(d.get("age_s", 0) or 0)
        if w is None or age > 20:
            return None, False
        GUARD_LAST["w"] = float(w); GUARD_LAST["ts"] = time.time()
        return float(w), True
    except Exception:
        return None, False

def guard_line(msg):
    log("power guard: " + msg)
    try:
        with open(os.path.join(DIR, "power-guard.log"), "a") as f:
            f.write("%s %s\n" % (utc(), msg))
    except OSError:
        pass

def lane_may_pull():
    """The CPU rule for a lane on a rig (main's order of 12 October 2026 18:0x UK, after a two-core board starved four
    OpenCL enumerations at once): a lane pulls a job only while fewer than power_guard.max_workers lanes run one (3 by
    default) and the one-minute load is under power_guard.max_load (the core count by default); read from config.json
    at every pull, live. A machine with no power_guard in its config pulls as before."""
    try:
        pg = json.load(open(CONFIG)).get("power_guard")
    except Exception:
        pg = None
    if not isinstance(pg, dict):
        return True
    try:
        busy = 0
        for sf in glob.glob(os.path.join(DIR, "state-*.json")):
            try:
                sd = json.load(open(sf))
                if sd.get("state") == "running" and sd.get("job"):
                    busy += 1
            except Exception:
                pass
        if busy >= int(pg.get("max_workers", 3)):
            return False
        if os.getloadavg()[0] >= float(pg.get("max_load", os.cpu_count() or 2)):
            return False
    except Exception:
        return True
    return True


def power_guard_loop():
    """The default agent's guard thread on a rig with per-card lanes: every ten seconds the plug; above the cap the newest
    running lane is stopped (its whole session, the worker inside it); 150 W under the cap for thirty seconds resumes the
    paused lanes one at a time, the earliest paused first, thirty seconds apart."""
    g = POWER_GUARD; cap = float(g.get("max_w", 1500)); last_resume = 0
    while not STOP.is_set():
        STOP.wait(10)
        w, fresh = plug_watts()
        if not fresh:
            continue
        if w >= cap:
            GUARD_LAST["under_since"] = 0
            live = [(sh, pid) for sh, pid, _ in LANE_STARTS if (sh, pid) not in PAUSED and _lane_owns_pid(pid, sh)]
            if live:
                sh, pid = live[-1]
                try:
                    os.killpg(pid, signal.SIGSTOP)
                    PAUSED.append((sh, pid))
                    guard_line("PAUSED lane %s (pid %d): the plug reads %.0f W, the cap is %.0f W" % (sh, pid, w, cap))
                except OSError as e:
                    guard_line("could not pause lane %s: %s" % (sh, e))
            continue
        if w < cap - 150:
            if not GUARD_LAST["under_since"]:
                GUARD_LAST["under_since"] = time.time()
            if PAUSED and time.time() - GUARD_LAST["under_since"] >= 30 and time.time() - last_resume >= 30:
                sh, pid = PAUSED.pop(0)
                try:
                    os.killpg(pid, signal.SIGCONT)
                    guard_line("resumed lane %s (pid %d): the plug reads %.0f W" % (sh, pid, w))
                except OSError as e:
                    guard_line("could not resume lane %s: %s" % (sh, e))
                last_resume = time.time()
        else:
            GUARD_LAST["under_since"] = 0


def card_temps():
    """Every card's edge and junction temperature in C (Linux: amdgpu's hwmon labels; NVIDIA: nvidia-smi's gpu and
    memory temperatures as edge and junction). Empty where nothing is readable."""
    out = []
    if IS_MAC:
        return out
    for h in glob.glob("/sys/class/drm/card*/device/hwmon/hwmon*"):
        try:
            if open(os.path.join(h, "name")).read().strip() != "amdgpu":
                continue
            edge = junction = None
            for t in glob.glob(os.path.join(h, "temp*_input")):
                try:
                    label = open(t.replace("_input", "_label")).read().strip()
                except Exception:
                    label = ""
                v = int(open(t).read().strip()) // 1000
                if label == "junction":
                    junction = v
                elif label == "edge" or not label:
                    edge = v
            out.append({"card": "amdgpu " + h.split("/")[4], "edge_c": edge, "junction_c": junction})
        except Exception:
            pass
    try:
        r = subprocess.run(["nvidia-smi", "--query-gpu=name,temperature.gpu,temperature.memory", "--format=csv,noheader,nounits"],
                           capture_output=True, text=True, timeout=10)
        for line in r.stdout.splitlines():
            p = [x.strip() for x in line.split(",")]
            if len(p) >= 2 and p[1].isdigit():
                out.append({"card": p[0], "edge_c": int(p[1]), "junction_c": int(p[2]) if len(p) > 2 and p[2].isdigit() else None, "nvidia": True})
    except Exception:
        pass
    return out


def card_facts():
    """Every card's fingerprint and idle telemetry for the fleet file (FLEET.md 1.1): NVIDIA by nvidia-smi (the UUID as
    the fingerprint, form nvidia-uuid; power.draw, temperature.gpu and temperature.memory), AMD on Linux by sysfs
    (unique_id as the fingerprint, form amdgpu-unique-id; hwmon's power1_average and the edge and junction labels),
    keyed by the bus id's short form (BB:DD.F). Nothing read is nothing reported."""
    out = {}
    try:
        r = subprocess.run(["nvidia-smi", "--query-gpu=pci.bus_id,uuid,name,power.draw,temperature.gpu,temperature.memory,power.limit", "--format=csv,noheader,nounits"],
                           capture_output=True, text=True, timeout=10)
        for line in r.stdout.splitlines():
            p = [x.strip() for x in line.split(",")]
            if len(p) < 3 or not p[0]:
                continue
            bus = p[0].upper()[-7:]
            f = {"model": p[2], "fingerprint": p[1] if p[1].startswith("GPU-") else None, "fingerprint_form": "nvidia-uuid" if p[1].startswith("GPU-") else None,
                 "watts": float(p[3]) if len(p) > 3 and re.fullmatch(r"[\d.]+", p[3]) else None,
                 "edge_c": int(p[4]) if len(p) > 4 and p[4].isdigit() else None,
                 "junction_c": int(p[5]) if len(p) > 5 and p[5].isdigit() else None,
                 "power_cap_w": float(p[6]) if len(p) > 6 and re.fullmatch(r"[\d.]+", p[6]) else None, "throttle_c": None, "tlimit_margin_c": None}
            out[bus] = f
        # the slowdown temperature (FLEET.md 1.2): nvidia-smi -q prints it per card in bus order
        q = subprocess.run(["nvidia-smi", "-q", "-d", "TEMPERATURE"], capture_output=True, text=True, timeout=10).stdout
        cur, slow = None, {}
        for line in q.splitlines():
            if line.startswith("GPU "):
                cur = line.split()[1].strip().upper()[-7:]
            elif cur and cur in out and re.search(r"GPU Current T\.Limit Temp\s*:\s*\d+ C", line):
                # FLEET.md 1.3: the degrees left before the slowdown point (the newer drivers' reading on every card; the
                # only one for the cards whose target prints N/A)
                out[cur]["tlimit_margin_c"] = int(re.search(r":\s*(\d+) C", line).group(1))
            elif cur and re.search(r"GPU Slowdown Temp\s*:", line):
                # the 550 driver's absolute point ("GPU Slowdown Temp : 90 C")
                m = re.search(r":\s*(\d+) C", line)
                if m:
                    slow[cur] = int(m.group(1))
            elif cur and cur not in slow and "Target Temperature Specification" in line:
                # the newer drivers print T.Limit margins (never turned into a point) and a target temperature per card
                m = re.search(r":\s*(\d+) C", line)
                if m:
                    slow[cur] = int(m.group(1))
        for bus, t in slow.items():
            if bus in out:
                out[bus]["throttle_c"] = t
    except Exception:
        pass
    if not IS_MAC:
        for dev in glob.glob("/sys/class/drm/card[0-9]*/device"):
            try:
                if "amdgpu" not in os.path.basename(os.readlink(os.path.join(dev, "driver"))):
                    continue
                pci = os.path.basename(os.path.realpath(dev))   # 0000:0b:00.0
                bus = pci.upper()[-7:]
                uid = None
                try:
                    uid = open(os.path.join(dev, "unique_id")).read().strip() or None
                except Exception:
                    pass
                f = {"model": None, "fingerprint": uid, "fingerprint_form": "amdgpu-unique-id" if uid else None, "watts": None, "edge_c": None, "junction_c": None,
                     "power_cap_w": None, "throttle_c": None}
                for h in glob.glob(os.path.join(dev, "hwmon", "hwmon*")):
                    for pw in ("power1_average", "power1_input"):   # the newer amdgpu (RDNA 3 and 4) prints power1_input only
                        try:
                            v = open(os.path.join(h, pw)).read().strip()
                            f["watts"] = round(int(v) / 1e6, 1)
                            break
                        except Exception:
                            pass
                    try:
                        f["power_cap_w"] = round(int(open(os.path.join(h, "power1_cap")).read().strip()) / 1e6, 1)
                    except Exception:
                        pass
                    for t in glob.glob(os.path.join(h, "temp*_crit")):
                        try:
                            label = open(t.replace("_crit", "_label")).read().strip()
                        except Exception:
                            label = ""
                        try:
                            v = int(open(t).read().strip()) // 1000
                        except Exception:
                            continue
                        if label == "junction" or (f["throttle_c"] is None and label in ("edge", "")):
                            f["throttle_c"] = v
                    for t in glob.glob(os.path.join(h, "temp*_input")):
                        try:
                            label = open(t.replace("_input", "_label")).read().strip()
                        except Exception:
                            label = ""
                        try:
                            v = int(open(t).read().strip()) // 1000
                        except Exception:
                            continue
                        if label == "junction":
                            f["junction_c"] = v
                        elif label == "edge" or not label:
                            f["edge_c"] = v
                out[bus] = f
            except Exception:
                continue
    return out


def step_rate(out_dir, step):
    """The running step's own last line when the worker prints its JSON progress lines ({"t", "mhs", "telemetry":
    {"power_w", "temp_c", ...}}, the bench workers' form): {step, mhs, watts, temp_c, t, form}; None when the step
    prints no such line (Ember's mine step prints nothing on stdout). Never derived."""
    p = os.path.join(out_dir, step + ".stdout")
    try:
        with open(p, "rb") as f:
            f.seek(0, 2)
            size = f.tell()
            f.seek(max(0, size - 4096))
            tail = f.read().decode(errors="replace").splitlines()
    except Exception:
        return None
    for line in reversed(tail):
        line = line.strip()
        if not line.startswith("{"):
            continue
        try:
            d = json.loads(line)
        except Exception:
            continue
        if isinstance(d.get("mhs"), (int, float)):
            tel = d.get("telemetry") or {}
            return {"step": step, "mhs": d["mhs"], "t": d.get("t"), "watts": tel.get("power_w"), "temp_c": tel.get("temp_c"), "form": "worker-json-line"}
    return None


def thermal_lines():
    """The guard's lines, read at every check from config.json (keys thermal_edge_c and thermal_junction_c; main's
    order of 11 October 2026 23:4x UK sets rig 1's rows at 80 C edge), else the constants; a change in the file is
    live for the next check, no swap."""
    try:
        c = json.load(open(CONFIG))
        return int(c.get("thermal_edge_c") or THERMAL_EDGE_C), int(c.get("thermal_junction_c") or THERMAL_JUNCTION_C)
    except Exception:
        return THERMAL_EDGE_C, THERMAL_JUNCTION_C


def nvidia_margins():
    """NVIDIA's throttle margin per card name (nvidia-smi -q's GPU Current T.Limit Temp, the degrees left before the card
    slows; the 550 driver prints it on every card, consumer boards give no hotspot): {name: margin C}."""
    out = {}
    try:
        r = subprocess.run(["nvidia-smi", "-q", "-d", "TEMPERATURE"], capture_output=True, text=True, timeout=8)
        cur = None
        for line in r.stdout.splitlines():
            m = re.match(r"\s*Product Name\s*:\s*(.+)", line)
            if m:
                cur = m.group(1).strip()
            m = re.search(r"GPU Current T\.Limit Temp\s*:\s*(-?\d+) C", line)
            if m and cur:
                out[cur] = int(m.group(1))
    except Exception:
        pass
    return out


def thermal_breach():
    """The first card over the guard's lines, as a string, else None. The lines: edge (every card), junction (AMD's
    hotspot; NVIDIA consumer boards give none) and, for NVIDIA, the throttle margin (main's order of 12 October 2026
    18:0x UK: a row ends at 5 C of margin, config.json thermal_nvidia_margin_c, 0 turns it off)."""
    edge_line, junction_line = thermal_lines()
    try:
        margin_line = int(json.load(open(CONFIG)).get("thermal_nvidia_margin_c", 5))
    except Exception:
        margin_line = 5
    margins = None
    for c in card_temps():
        if (c.get("junction_c") or 0) >= junction_line or (c.get("edge_c") or 0) >= edge_line:
            return "%s at edge %s C, junction %s C (the lines: edge %d, junction %d)" % (c["card"], c.get("edge_c"), c.get("junction_c"), edge_line, junction_line)
        if c.get("nvidia") and margin_line > 0:
            if margins is None:
                margins = nvidia_margins()
            m = margins.get(c["card"])
            if m is not None and m <= margin_line:
                return "%s at edge %s C with %d C of throttle margin left (the line: %d C of margin)" % (c["card"], c.get("edge_c"), m, margin_line)
    return None


def thermal_sample():
    """The host's thermal state at a step boundary (macOS: pmset -g therm and the CPU thermal level where the host
    exposes it, no root; elsewhere nothing), for the Apple measurement contract's per-step rows."""
    out = {"utc": utc()}
    if IS_MAC:
        for name, cmd in (("pmset_therm", ["pmset", "-g", "therm"]), ("cpu_thermal_level", ["sysctl", "-n", "machdep.xcpm.cpu_thermal_level"]),
                          ("thermal_pressure", ["sysctl", "-n", "kern.thermalpressure"])):
            try:
                r = subprocess.run(cmd, capture_output=True, timeout=5)
                if r.returncode == 0 and r.stdout.strip():
                    out[name] = r.stdout.decode(errors="replace").strip()[:400]
            except Exception:
                pass
    return out


# ---------------------------------------------------------------- Ember Fleet (docs/fleet/DESIGN.md section 3.1, 13 October 2026)
# The fleet agent is this agent with three additions on a host that runs Ember, all off when it does not: the `ember`
# block on the default heartbeat (read-only, from Ember's own --api endpoint, null when it is down); the owner's
# actions (signed by the queue key as a job is, with a nonce and an expiry, acted once and only between jobs,
# receipted on the next heartbeat with the exit code, every one in fleet.log); and the watchdog (no accepted share for
# fifteen minutes with the source connected, or the endpoint down for five: restart Ember; a second restart in the
# hour that does not bring shares: reboot; three restarts and one reboot an hour at most, then it waits and says so).
# Ember runs as the igneum user's own unit (systemctl --user igneum-ember) so restart, stop and start need no
# elevation; reboot is the one sudo line (/usr/bin/systemctl reboot). On a rig the agent stops Ember at a bench job's
# start and starts it again at the job's end (reliability's condition: a row read with Ember beside it is void).
EMBER_API = os.environ.get("IGNEUM_EMBER_API", "http://127.0.0.1:4068")
EMBER_UNIT = os.environ.get("IGNEUM_EMBER_UNIT", "igneum-ember")
ACTION_VERBS = ("restart-ember", "reboot", "tail", "apply-profile", "run", "update-ember", "stop-ember", "start-ember")
WD_MEM_FLOOR_MB = 1500      # the watchdog never restarts Ember into a host with less available memory (rig 1's out-of-memory, 23:13 UK, 13 October 2026)
WD_ROUTE_S = 300            # and acts on nothing when the last successful pull is older than this (no route to the queue: the blind restart of 23:24 UK)
# the remote console (FEATURES.md tier 1): a fixed menu, never a shell; the action names the index
RUN_MENU = (["ember", "devices", "--json"], ["ember", "status"], ["nvidia-smi"], ["sensors"], ["journalctl", "--user-unit", "igneum-ember", "-n", "100", "--no-pager", "-o", "short-iso"],
            ["uptime"], ["df", "-h", "/"], ["ip", "-brief", "addr"], ["systemctl", "--user", "status", "igneum-ember", "--no-pager"])
ACTION_MAX_AGE_S = 600
WD_NO_SHARES_S = 900
WD_DOWN_S = 300
WD_RESTARTS_PER_HOUR = 3
WD_REBOOTS_PER_HOUR = 1
WD_BIND_GRACE_S = 1200    # a fresh Ember may tune for minutes before its endpoint binds; the down rule waits this long after a start


def ember_started_ts():
    """The unit's last start as a unix time (systemctl show ActiveEnterTimestamp), None when unknown."""
    kind = ember_unit_kind()
    if not kind:
        return None
    args = ["systemctl", "--user", "show", EMBER_UNIT, "-p", "ActiveEnterTimestampMonotonic"] if kind == "user" else ["systemctl", "show", EMBER_UNIT, "-p", "ActiveEnterTimestampMonotonic"]
    try:
        v = subprocess.run(args, capture_output=True, text=True, timeout=10).stdout.strip().split("=", 1)[-1]
        mono = int(v) / 1e6
        if mono <= 0:
            return None
        return time.time() - (time.monotonic() - mono) if hasattr(time, "monotonic") else None
    except Exception:
        return None
FLEET = {"last_accepted": None, "last_accepted_ts": None, "down_since": None, "restarts": [], "reboots": [], "receipts": [],
         "watchdog": None, "paused": False, "unit": "unknown", "deferred": set(), "fault_since": None}
FLEET_LOG = os.path.join(DIR, "fleet.log")
ACTIONS_DIR = os.path.join(DIR, "actions")
ACTED = os.path.join(ACTIONS_DIR, "acted.log")


def flog(msg):
    log("fleet: " + msg)
    try:
        with open(FLEET_LOG, "a") as f:
            f.write("%s %s\n" % (utc(), msg))
    except Exception:
        pass


def fleet_host(cfg):
    """A rig (by its name) or a host whose config.json says "fleet": true carries the fleet parts."""
    return str(cfg.get("name", "")).startswith("rig") or cfg.get("fleet") is True


def ember_state():
    try:
        with urllib.request.urlopen(EMBER_API + "/state", timeout=3) as r:
            return json.loads(r.read().decode())
    except Exception:
        return None


def mem_available_mb():
    try:
        for l in open("/proc/meminfo"):
            if l.startswith("MemAvailable:"):
                return int(l.split()[1]) // 1024
    except Exception:
        pass
    return None


def ember_block():
    """The `ember` block of the heartbeat: what Ember's /state says, compacted; {"up": false} when it does not answer;
    the host's available memory beside it either way."""
    s = ember_state()
    now = time.time()
    if not isinstance(s, dict):
        if FLEET["down_since"] is None:
            FLEET["down_since"] = now
        return {"up": False, "down_s": int(now - FLEET["down_since"]), "unit_active": ember_active(), "host_mem_available_mb": mem_available_mb()}
    FLEET["down_since"] = None
    FLEET["seen_up"] = now
    t = s.get("totals") or {}
    src = s.get("source") or {}
    acc = t.get("accepted")
    if isinstance(acc, int) and acc != FLEET["last_accepted"]:
        FLEET["last_accepted"], FLEET["last_accepted_ts"] = acc, now
    devs = {}
    for d in s.get("devices") or []:
        if not isinstance(d, dict):
            continue
        te = d.get("telemetry") if isinstance(d.get("telemetry"), dict) else {}
        key = str(d.get("bus_id") or d.get("id") or d.get("index"))
        m = re.search(r"([0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-9])$", key)
        key = m.group(1).upper() if m else key
        devs[key] = {"name": d.get("name"), "enabled": d.get("enabled"), "state": d.get("state"), "phase": d.get("phase"),
                     "validated_hps": d.get("validated_hps"), "coord": d.get("coord"), "coord_word": d.get("coord_word"),
                     "accepted": d.get("accepted"), "rejected": d.get("rejected"), "stale": d.get("stale"),
                     "telemetry": {"power_w": te.get("power_w"), "temp_c": te.get("temp_c"), "core_mhz": te.get("core_mhz"),
                                   "mem_mhz": te.get("mem_mhz"), "throttle": te.get("throttle") or []},
                     "pinned": d.get("pinned")}
    return {"up": True, "version": s.get("version"), "mining": s.get("mining"), "uptime_s": t.get("uptime_s"),
            "source": {"name": src.get("name"), "connected": src.get("connected"), "difficulty": src.get("difficulty"), "height": src.get("height")},
            "totals": {"accepted": acc, "rejected": t.get("rejected"), "stale": t.get("stale"), "invalid": t.get("invalid"),
                       "validated_hps": t.get("validated_hps"), "device_watts": t.get("device_watts")},
            "last_accepted_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(FLEET["last_accepted_ts"])) if FLEET["last_accepted_ts"] else None,
            "devices": devs, "tune_running": s.get("tune_running"), "pools_status": s.get("pools_status"), "host_mem_available_mb": mem_available_mb()}


def ember_unit_kind():
    """"user" when igneum-ember is the user's own unit, "system" when it is a system unit (sudo), None when neither."""
    if FLEET["unit"] != "unknown":
        return FLEET["unit"]
    kind = None
    for args in (["systemctl", "--user", "cat", EMBER_UNIT], ["systemctl", "cat", EMBER_UNIT]):
        try:
            if subprocess.run(args, capture_output=True, timeout=10).returncode == 0:
                kind = "user" if "--user" in args else "system"
                break
        except Exception:
            continue
    FLEET["unit"] = kind
    return kind


def ember_ctl(verb):
    """start, stop or restart Ember's unit; the exit code (127 when there is no unit)."""
    kind = ember_unit_kind()
    if kind == "user":
        args = ["systemctl", "--user", verb, EMBER_UNIT]
    elif kind == "system":
        args = ["sudo", "-n", "/usr/bin/systemctl", verb, EMBER_UNIT]
    else:
        return 127, "no unit " + EMBER_UNIT
    try:
        p = subprocess.run(args, capture_output=True, text=True, timeout=120)
        return p.returncode, (p.stderr or p.stdout).strip()[:160]
    except Exception as ex:
        return 1, str(ex)[:160]


def ember_enabled():
    """The owner's standing signal (rig 1, 23:50 to 23:59 UK, 13 October 2026: a stop by hand during the agent's own
    pause was undone by the resume hook two minutes later and the watchdog then rebooted the rig): a unit the owner
    disabled is never started by the resume hook or the watchdog; stop-ember disables it, start-ember enables it."""
    kind = ember_unit_kind()
    if not kind:
        return False
    args = ["systemctl", "--user", "is-enabled", EMBER_UNIT] if kind == "user" else ["systemctl", "is-enabled", EMBER_UNIT]
    try:
        return subprocess.run(args, capture_output=True, text=True, timeout=10).stdout.strip() in ("enabled", "enabled-runtime", "static", "linked")
    except Exception:
        return False


def ember_active():
    kind = ember_unit_kind()
    if not kind:
        return False
    args = ["systemctl", "--user", "is-active", EMBER_UNIT] if kind == "user" else ["systemctl", "is-active", EMBER_UNIT]
    try:
        return subprocess.run(args, capture_output=True, text=True, timeout=10).stdout.strip() == "active"
    except Exception:
        return False


PAUSE_FILE = os.path.join(DIR, "ember-paused.json")


def pause_state_save():
    """The pause and the wanted resume persist on disk (rig 1, 01:57 UK, 14 October 2026: the agent's own self-install
    paused Ember, restarted into the new file, and the wanted resume was gone, so Ember stayed down with the queue
    empty): written at the pause, read at the agent's start, cleared at the resume or by an owner's stop."""
    try:
        if FLEET.get("paused") or FLEET.get("resume_wanted"):
            with open(PAUSE_FILE + ".tmp", "w") as f:
                json.dump({"paused": bool(FLEET.get("paused")), "resume_wanted": bool(FLEET.get("resume_wanted")), "utc": utc()}, f)
            os.replace(PAUSE_FILE + ".tmp", PAUSE_FILE)
        elif os.path.exists(PAUSE_FILE):
            os.remove(PAUSE_FILE)
    except Exception as ex:
        log("pause state: %s" % str(ex)[:80])


def pause_state_load(cfg):
    """At the agent's start: a pause left by the previous agent becomes a wanted resume (the job that paused it is
    over or the previous agent's; the resume starts Ember only when the queue answers empty, as always)."""
    if LANE or not fleet_host(cfg):
        return
    st = jload_file(PAUSE_FILE)
    if st and (st.get("paused") or st.get("resume_wanted")):
        FLEET["paused"] = True
        FLEET["resume_wanted"] = True
        flog("a pause from %s found at start (the previous agent's); Ember's resume wanted, it starts when the queue answers empty" % st.get("utc"))


def jload_file(p):
    try:
        with open(p) as f:
            return json.load(f)
    except Exception:
        return None


def ember_pause(cfg):
    """A bench job starts on a rig: Ember stops for it (the row is void with a miner beside it)."""
    if LANE or not fleet_host(cfg) or not ember_unit_kind():
        return
    if ember_active():
        rc, note = ember_ctl("stop")
        FLEET["paused"] = rc == 0
        pause_state_save()
        flog("paused Ember for the job (rc %d%s)" % (rc, (": " + note) if note else ""))


def ember_resume(cfg):
    """A job ended: Ember does not start at that instant (reliability's rule of 23:57 UK, 13 October 2026: the
    out-of-memory of 23:13 was Ember and a bench worker resident together in the gap between one cell's end and the
    next cell's take); the resume is wanted, and ember_resume_if_idle starts it only when the next pull answers
    empty and no lane runs."""
    if LANE or not FLEET["paused"]:
        return
    FLEET["resume_wanted"] = True
    pause_state_save()
    flog("Ember's resume wanted after the job; it starts when the queue answers empty")


def ember_resume_if_idle():
    """Called by the loop after a pull that answered empty: with a resume wanted, no lane running and memory over
    the floor, Ember starts; otherwise it waits for the next empty pull."""
    if LANE or not FLEET.get("resume_wanted"):
        return
    if not ember_enabled():
        FLEET["paused"] = False
        FLEET["resume_wanted"] = False
        pause_state_save()
        flog("Ember's resume dropped: the unit is disabled (the owner's stop stands)")
        return
    if other_lanes() or RUNNING:
        return
    mem = mem_available_mb()
    if mem is not None and mem < WD_MEM_FLOOR_MB:
        if FLEET.get("mem_said") != "resume":
            flog("Ember's resume waits: %d MB available is under the %d MB floor" % (mem, WD_MEM_FLOOR_MB))
            FLEET["mem_said"] = "resume"
        return
    rc, note = ember_ctl("start")
    FLEET["paused"] = False
    FLEET["resume_wanted"] = False
    pause_state_save()
    FLEET["last_accepted_ts"] = time.time()
    flog("Ember started again: the queue answered empty after the job (rc %d%s)" % (rc, (": " + note) if note else ""))


def fleet_busy():
    return bool(RUNNING) or bool(other_lanes())


EMBER_CONF = os.environ.get("IGNEUM_EMBER_CONF", os.path.join(HOME, ".config", "ember", "ember.conf"))
PROFILE_FILE = os.path.join(DIR, "profile.json")
PROFILE_KEYS = ("pool", "pool_backup", "address", "power_cap", "temp_cap", "coord", "no_pool_fee", "tariff", "currency")


def ember_tail(lines=40):
    """The log tail the console asks for (FEATURES.md tier 1): Ember's unit journal, this agent's log and fleet.log."""
    def tail_file(p, n):
        try:
            return open(p, errors="replace").read().splitlines()[-n:]
        except Exception:
            return []
    kind = ember_unit_kind()
    jl = []
    if kind:
        args = ["journalctl", "--no-pager", "-o", "short-iso", "-n", str(lines), ("--user-unit" if kind == "user" else "-u"), EMBER_UNIT]
        try:
            jl = subprocess.run(args, capture_output=True, text=True, timeout=20).stdout.splitlines()[-lines:]
        except Exception as ex:
            jl = ["journal: %s" % str(ex)[:80]]
    return {"utc": utc(), "ember": [l[:400] for l in jl], "agent": [l[:400] for l in tail_file(os.path.join(DIR, "agent.log"), 20)],
            "fleet": [l[:400] for l in tail_file(FLEET_LOG, 10)]}


HWMON_HELPER = "/usr/local/sbin/igneum-fleet-hwmon"
GUARD_RAISE_C = 70      # any card over it: its fan back to the driver's curve
GUARD_RESTORE_C = 80    # any card over it: the full caps restored, the fault said


def card_inventory():
    """The cards the caps and fans act on: NVIDIA by nvidia-smi (index, name, cap, min, max), AMD by the drm tree
    (bus, name from the product_name or the device id, cap, min, max from hwmon)."""
    cards = []
    try:
        r = subprocess.run(["nvidia-smi", "--query-gpu=index,name,pci.bus_id,power.limit,power.min_limit,power.max_limit", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=10)
        for line in r.stdout.splitlines():
            p = [x.strip() for x in line.split(",")]
            if len(p) >= 6 and p[0].isdigit():
                cards.append({"vendor": "nvidia", "index": int(p[0]), "name": p[1], "bus": p[2].upper()[-7:], "cap": float(p[3]), "min": float(p[4]), "max": float(p[5])})
    except Exception:
        pass
    for dev in glob.glob("/sys/class/drm/card[0-9]*/device"):
        try:
            if "amdgpu" not in os.path.basename(os.readlink(os.path.join(dev, "driver"))):
                continue
            pci = os.path.basename(os.path.realpath(dev))
            name = None
            try:
                name = open(os.path.join(dev, "product_name")).read().strip() or None
            except Exception:
                pass
            h = (glob.glob(os.path.join(dev, "hwmon", "hwmon*")) or [None])[0]
            if not h:
                continue
            rd = lambda f: int(open(os.path.join(h, f)).read().strip())
            cards.append({"vendor": "amd", "pci": pci, "bus": pci.upper()[-7:], "name": name or ("amdgpu " + pci), "cap": rd("power1_cap") / 1e6, "min": rd("power1_cap_min") / 1e6, "max": rd("power1_cap_max") / 1e6,
                          "fan_writable": os.access(os.path.join(h, "pwm1"), os.W_OK) or os.path.exists(HWMON_HELPER)})
        except Exception:
            continue
    return cards


def _sudo(args):
    try:
        p = subprocess.run(["sudo", "-n"] + args, capture_output=True, text=True, timeout=30)
        return p.returncode, (p.stdout or p.stderr).strip()[:100]
    except Exception as ex:
        return 1, str(ex)[:100]


def set_card(card, cap_w=None, fan=None):
    """One card's cap (watts, clamped to the driver's floor and ceiling) and, on AMD, its fan (an integer pwm 0 to 255,
    or "auto"); each a root call through the sudoers grant; the result words for the receipt."""
    out = []
    if cap_w is not None:
        w = int(max(card["min"], min(card["max"], float(cap_w))))
        if card["vendor"] == "nvidia":
            rc, note = _sudo(["/usr/bin/nvidia-smi", "-i", str(card["index"]), "-pl", str(w)])
        else:
            rc, note = _sudo([HWMON_HELPER, card["pci"], "cap", str(w)])
        out.append("%s cap %d W rc %d" % (card["name"], w, rc))
    if fan is not None and card["vendor"] == "amd":
        rc, note = _sudo([HWMON_HELPER, card["pci"], "auto"] if fan == "auto" else [HWMON_HELPER, card["pci"], "pwm", str(int(fan))])
        out.append("%s fan %s rc %d" % (card["name"], fan, rc))
    return out


def match_card_rule(card, rules):
    """The profile's cards map: a key is a model substring ("RTX 5060 Ti"), a bus id ("04:00.0") or a vendor word
    ("nvidia", "amd"); the most specific key wins."""
    best, best_len = None, -1
    for k, v in (rules or {}).items():
        kl = str(k).lower()
        hit = kl == card["bus"].lower() or (kl in ("nvidia", "amd") and kl == card["vendor"]) or (kl not in ("nvidia", "amd") and kl in str(card["name"]).lower())
        if hit and isinstance(v, dict) and len(kl) > best_len:
            best, best_len = v, len(kl)
    return best


def apply_caps(profile):
    """The per-card caps and fan floors of a profile, the full caps (the drivers' ceilings) saved for the guard's
    restore; the words for the receipt."""
    rules = profile.get("cards") or {}
    if not rules:
        return [], {}
    words, full = [], {}
    for c in card_inventory():
        full[c["bus"]] = {"vendor": c["vendor"], "max": c["max"], "index": c.get("index"), "pci": c.get("pci"), "name": c["name"]}
        r = match_card_rule(c, rules)
        if not r:
            continue
        words += set_card(c, r.get("power_cap_w"), r.get("fan_pwm") if c["vendor"] == "amd" else None)
    return words, full


def fans_guard(cfg):
    """Every heartbeat under a profile with cards: any card over GUARD_RAISE_C has its fan put back on the driver's
    curve (AMD; NVIDIA's never left it); over GUARD_RESTORE_C the full caps and fans are restored on every card and
    the fault said on the heartbeat and in fleet.log; the profile then reads restored until applied again."""
    if LANE or not fleet_host(cfg):
        return
    pf = jload_file(PROFILE_FILE) or {}
    full = pf.get("full_caps") or {}
    if not full or pf.get("restored"):
        return
    try:
        facts = card_facts()
    except Exception:
        return
    hottest = None
    for bus, f in facts.items():
        for t in (f.get("edge_c"), f.get("junction_c")):
            if isinstance(t, (int, float)) and (hottest is None or t > hottest[1]):
                hottest = (f.get("model") or bus, t, bus)
    if not hottest:
        return
    name, t, bus = hottest
    if t > GUARD_RESTORE_C:
        words = []
        for b, c in full.items():
            card = {"vendor": c["vendor"], "index": c.get("index"), "pci": c.get("pci"), "name": c.get("name"), "min": 0, "max": c["max"]}
            words += set_card(card, c["max"], "auto" if c["vendor"] == "amd" else None)
        pf["restored"] = {"utc": utc(), "card": name, "temp_c": t, "words": words}
        with open(PROFILE_FILE, "w") as f:
            json.dump(pf, f)
        FLEET["watchdog"] = {"verb": "profile-restored", "cause": "%s at %s C over %d" % (name, t, GUARD_RESTORE_C), "utc": utc()}
        flog("fans guard: %s at %s C over %d; the full caps and fans restored (%s)" % (name, t, GUARD_RESTORE_C, "; ".join(words)))
    elif t > GUARD_RAISE_C:
        raised = pf.setdefault("raised", {})
        if bus not in raised and full.get(bus, {}).get("vendor") == "amd":
            c = full[bus]
            set_card({"vendor": "amd", "pci": c.get("pci"), "name": c.get("name"), "min": 0, "max": c["max"]}, None, "auto")
            raised[bus] = utc()
            with open(PROFILE_FILE, "w") as f:
                json.dump(pf, f)
            flog("fans guard: %s at %s C over %d; its fan back on the driver's curve" % (name, t, GUARD_RAISE_C))


def apply_profile(profile):
    """A profile (FEATURES.md tier 1, the flight sheet): the keys Ember 0.2.5 reads written into ember.conf (the
    file's own api, workers_dir and data_dir lines kept), the profile's name, sha and time recorded for the heartbeat,
    then Ember restarted. The per-card cap is not applied here (no grant for nvidia-smi -pl on the rig yet; the receipt
    says so); the per-card knee and coordinate wait on product A's per-card keys."""
    if not isinstance(profile, dict) or not profile.get("name"):
        return 2, "no profile"
    keep = []
    try:
        for l in open(EMBER_CONF):
            k = l.split("=", 1)[0].strip()
            if k in ("api", "workers_dir", "data_dir", "worker_name", "algo"):
                keep.append(l.rstrip("\n"))
    except Exception:
        return 2, "no ember.conf at " + EMBER_CONF
    lines = ["# Ember: the one config file, written by the fleet profile %s at %s" % (profile["name"], utc())]
    for k in PROFILE_KEYS:
        v = profile.get(k)
        if v is None or v == "":
            continue
        if k in ("no_pool_fee",):
            if v is True:
                lines.append("%s = true" % k)
            continue
        if not re.fullmatch(r"[A-Za-z0-9 ._:/@+-]{1,120}", str(v)):
            return 2, "profile value refused for " + k
        lines.append("%s = %s" % (k, v))
    if profile.get("no_tls") is True or (profile.get("pool") and ":3341" in str(profile.get("pool"))):
        lines.append("no_tls = true")
    lines += keep
    tmp = EMBER_CONF + ".tmp"
    with open(tmp, "w") as f:
        f.write("\n".join(lines) + "\n")
    os.chmod(tmp, 0o600)
    os.replace(tmp, EMBER_CONF)
    sha = hashlib.sha256(json.dumps(profile, sort_keys=True).encode()).hexdigest()[:12]
    with open(PROFILE_FILE, "w") as f:
        json.dump({"name": profile["name"], "sha": sha, "applied_utc": utc(), "watchdog": profile.get("watchdog") if isinstance(profile.get("watchdog"), dict) else None}, f)
    rc, note = ember_ctl("restart")
    words, full = apply_caps(profile)
    if full:
        try:
            pf = json.load(open(PROFILE_FILE))
            pf["full_caps"] = full
            with open(PROFILE_FILE, "w") as f:
                json.dump(pf, f)
        except Exception:
            pass
    if words:
        note = (note + "; " if note else "") + "; ".join(words)
        flog("profile %s: %s" % (profile["name"], "; ".join(words)))
    return rc, note


def run_menu(idx):
    try:
        cmd = list(RUN_MENU[int(idx)])
    except Exception:
        return 2, "no such menu item"
    if cmd[0] == "ember":
        eb = os.path.join(HOME, ".local", "opt", "ember", "bin", "ember")
        if not os.path.exists(eb):
            return 2, "no ember binary"
        cmd[0] = eb
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        out = (p.stdout + ("\n" + p.stderr if p.stderr else ""))[-30000:]
        rc = p.returncode
    except Exception as ex:
        out, rc = str(ex)[:200], 1
    doc = {"utc": utc(), "cmd": " ".join(cmd), "rc": rc, "lines": out.splitlines()[-300:]}
    rc2, o, e = ssh_cmd(["put-run"], stdin_bytes=json.dumps(doc).encode(), timeout=60)
    return (rc if rc2 == 0 else rc2), ("%s: %d lines" % (doc["cmd"], len(doc["lines"]))) if rc2 == 0 else e.decode(errors="replace").strip()[:120]


def update_ember(sha):
    """One-click updates (FEATURES.md tier 2): Ember's Linux archive fetched from the artefact store by its sha256
    (verified as a worker is), unpacked beside the installed set, swapped in, the unit restarted; the previous set kept
    as ember.prev and put back when the new bin/ember does not answer --version or the endpoint stays silent 90 s."""
    if not re.fullmatch(r"[0-9a-f]{64}", str(sha or "")):
        return 2, "a sha256 is needed"
    opt = os.path.join(HOME, ".local", "opt", "ember")
    if not os.path.isdir(opt):
        return 2, "no Ember set at " + opt
    tmp = os.path.join(DIR, "ember-update.tgz")
    new, prev = opt + ".new", opt + ".prev"
    try:
        fetch_by_sha(sha, tmp)
        shutil.rmtree(new, ignore_errors=True)
        os.makedirs(new)
        with tarfile.open(tmp) as t:
            for m in t.getmembers():
                if m.name.startswith("/") or ".." in m.name.split("/"):
                    raise RuntimeError("archive path refused: " + m.name)
            t.extractall(new)
        inner = [d for d in os.listdir(new) if os.path.isdir(os.path.join(new, d))]
        root = os.path.join(new, inner[0]) if len(inner) == 1 and not os.path.exists(os.path.join(new, "bin")) else new
        if not os.path.exists(os.path.join(root, "bin", "ember")):
            raise RuntimeError("the archive carries no bin/ember")
        os.chmod(os.path.join(root, "bin", "ember"), 0o755)
        for w in glob.glob(os.path.join(root, "workers", "igneum-worker-*")):
            os.chmod(w, 0o755)
        v = subprocess.run([os.path.join(root, "bin", "ember"), "--version"], capture_output=True, text=True, timeout=30)
        if v.returncode != 0:
            raise RuntimeError("the new ember --version failed: " + (v.stderr or v.stdout).strip()[:120])
        ver = (v.stdout or "").strip()[:40]
        was_active = ember_active()
        enabled = ember_enabled()
        if was_active:
            ember_ctl("stop")
        shutil.rmtree(prev, ignore_errors=True)
        os.replace(opt, prev)
        os.replace(root, opt)
        shutil.rmtree(new, ignore_errors=True)
        flog("update-ember: %s in place (%s), the previous set at ember.prev" % (ver, sha[:12]))
        if not enabled:
            # the owner's stop stands (0.3.63): the set is swapped and verified by --version alone; nothing starts
            return 0, "updated to %s; the unit is disabled, not started (start-ember starts it)" % ver
        rc, note = ember_ctl("start")
        # the proof of the new set: its endpoint answering, waited for up to WD_BIND_GRACE_S (rig 1, 01:22 UK,
        # 14 October 2026: 03fc92e1 was killed at 90 s while still enumerating its cards, which takes this rig two
        # minutes; the same horizon the watchdog's down rule gives a fresh start); a unit that leaves the active
        # state meanwhile (the binary died) rolls back at once
        t0 = time.time()
        last_hb = time.time()
        while time.time() - t0 < WD_BIND_GRACE_S:
            time.sleep(5)
            if time.time() - last_hb >= HEARTBEAT_S:
                # the heartbeat keeps going through the wait (01:34 UK, 14 October 2026: a twenty-minute proof on the
                # main loop would read as a dark rig to the console and its power-cycle watchdog)
                try:
                    heartbeat({"state": "idle", "updating_ember": {"version": ver, "since_s": int(time.time() - t0)}})
                except Exception:
                    pass
                last_hb = time.time()
            if ember_state():
                FLEET["last_accepted_ts"] = time.time()
                FLEET["seen_up"] = time.time()
                flog("update-ember: %s answers on its endpoint after %d s" % (ver, int(time.time() - t0)))
                return 0, "updated to %s (endpoint up after %d s)" % (ver, int(time.time() - t0))
            if not ember_active():
                flog("update-ember: %s's unit left the active state after %d s" % (ver, int(time.time() - t0)))
                break
        ember_ctl("stop")
        os.replace(opt, opt + ".failed")
        os.replace(prev, opt)
        shutil.rmtree(opt + ".failed", ignore_errors=True)
        ember_ctl("start")
        why = "its unit died" if not ember_active() and time.time() - t0 < WD_BIND_GRACE_S else "no endpoint in %d s" % WD_BIND_GRACE_S
        flog("update-ember: %s rolled back (%s); the previous set is back" % (ver, why))
        return 1, "rolled back: %s, %s" % (ver, why)
    except Exception as ex:
        shutil.rmtree(new, ignore_errors=True)
        flog("update-ember failed: %s" % str(ex)[:160])
        return 1, str(ex)[:160]
    finally:
        try:
            os.remove(tmp)
        except Exception:
            pass


def do_action(verb, a=None):
    a = a or {}
    if verb == "update-ember":
        return update_ember(a.get("sha256"))
    if verb == "stop-ember":
        rc, note = ember_ctl("stop")
        ember_ctl("disable")
        FLEET["paused"] = False
        FLEET["resume_wanted"] = False
        pause_state_save()
        return rc, note or "stopped and disabled (the owner's stop stands until start-ember)"
    if verb == "start-ember":
        mem = mem_available_mb()
        if mem is not None and mem < WD_MEM_FLOOR_MB:
            return 1, "refused: %d MB available is under the %d MB floor" % (mem, WD_MEM_FLOOR_MB)
        ember_ctl("enable")
        rc, note = ember_ctl("start")
        FLEET["paused"] = False
        FLEET["resume_wanted"] = False
        pause_state_save()
        FLEET["last_accepted_ts"] = time.time()
        return rc, note or "enabled and started"
    if verb == "run":
        return run_menu(a.get("item"))
    if verb == "tail":
        try:
            n = max(5, min(200, int(a.get("lines") or 40)))
        except Exception:
            n = 40
        t = ember_tail(n)
        rc, out, err = ssh_cmd(["put-tail"], stdin_bytes=json.dumps(t).encode(), timeout=60)
        return rc, ("%d ember, %d agent lines" % (len(t["ember"]), len(t["agent"]))) if rc == 0 else err.decode(errors="replace").strip()[:120]
    if verb == "apply-profile":
        return apply_profile(a.get("profile"))
    if verb == "restart-ember":
        return ember_ctl("restart")
    if verb == "reboot":
        try:
            p = subprocess.run(["sudo", "-n", "/usr/bin/systemctl", "reboot"], capture_output=True, text=True, timeout=60)
            return p.returncode, (p.stderr or p.stdout).strip()[:160]
        except Exception as ex:
            return 1, str(ex)[:160]
    return 2, "unknown verb"


def _acted_ids():
    try:
        return set(l.split()[0] for l in open(ACTED) if l.strip())
    except Exception:
        return set()


def _receipt(a, rc, note):
    r = {"id": a["id"], "verb": a["verb"], "utc": utc(), "rc": rc, "note": note}
    FLEET["receipts"] = (FLEET["receipts"] + [r])[-10:]
    os.makedirs(ACTIONS_DIR, exist_ok=True)
    with open(ACTED, "a") as f:
        f.write("%s %s %s rc=%s\n" % (a["id"], a.get("nonce", "-"), r["utc"], rc))
    flog("action %s %s: rc %s%s" % (a["id"], a["verb"], rc, (" " + note) if note else ""))


def actions_poll(cfg):
    """The owner's pending actions from the queue (the shell's `actions` verb): each verified against the queue's
    signer, refused for another host, an unknown verb, a passed expiry or an id already acted; acted only when no job
    runs on this host (deferred otherwise, said once); receipted on the next heartbeat."""
    if LANE or not fleet_host(cfg):
        return
    rc, out, err = ssh_cmd(["actions"], timeout=30)
    if rc != 0 or not out.strip() or out.strip() == b"[]":
        return
    try:
        pending = json.loads(out.decode())
    except Exception:
        log("actions: unreadable answer")
        return
    os.makedirs(ACTIONS_DIR, exist_ok=True)
    acted = _acted_ids()
    for env in pending:
        try:
            a_bytes, s_bytes = base64.b64decode(env["action"]), base64.b64decode(env["sig"])
            a = json.loads(a_bytes.decode())
            aid = str(a.get("id") or "")
            if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,80}", aid):
                continue
            if aid in acted:
                continue
            ap = os.path.join(ACTIONS_DIR, aid + ".json")
            with open(ap, "wb") as f:
                f.write(a_bytes)
            with open(ap + ".sig", "wb") as f:
                f.write(s_bytes)
            ok, why = verify_signature(ap, ap + ".sig")
            if not ok:
                _receipt(a, 3, "refused: signature: " + why[:80])
                continue
            if a.get("machine") != cfg["name"]:
                _receipt(a, 3, "refused: for " + str(a.get("machine")))
                continue
            if a.get("verb") not in ACTION_VERBS:
                _receipt(a, 3, "refused: verb")
                continue
            try:
                # a UTC stamp parsed as UTC (calendar.timegm): mktime minus time.timezone read an hour early under BST
                # on rig 1 (23:37 UK, 13 October 2026) and refused a live action as expired
                exp = calendar.timegm(time.strptime(str(a.get("expires_utc")), "%Y-%m-%dT%H:%M:%SZ"))
            except Exception:
                exp = 0
            if time.time() > exp:
                _receipt(a, 3, "refused: expired")
                continue
            if fleet_busy() and a["verb"] not in ("tail", "run"):   # a tail or a menu read may run beside a job
                if aid not in FLEET["deferred"]:
                    FLEET["deferred"].add(aid)
                    flog("action %s %s deferred: a job runs on this host; it acts at the job's end" % (aid, a["verb"]))
                continue
            flog("action %s %s by %s: acting" % (aid, a["verb"], a.get("owner", "?")))
            rc2, note = do_action(a["verb"], a)
            _receipt(a, rc2, note)
            if a["verb"] == "restart-ember":
                FLEET["last_accepted_ts"] = time.time()
        except Exception as ex:
            log("actions: %s" % str(ex)[:120])


def watchdog_rules():
    """The rig's own rules (FEATURES.md tier 1): the profile's `watchdog` block when one was applied, else the fixed
    figures. Keys, each optional: no_shares_min, down_min, hot_margin_c (0 turns the rule off), rate_drop_pct with
    rate_drop_min and expected_hps (the whole rig's validated rate the profile expects), restarts_per_hour,
    reboots_per_hour, reboot (false keeps the watchdog to restarts)."""
    r = {"no_shares_min": WD_NO_SHARES_S // 60, "down_min": WD_DOWN_S // 60, "hot_margin_c": 5, "rate_drop_pct": 0, "rate_drop_min": 10,
         "expected_hps": 0, "restarts_per_hour": WD_RESTARTS_PER_HOUR, "reboots_per_hour": WD_REBOOTS_PER_HOUR, "reboot": True}
    try:
        w = (json.load(open(PROFILE_FILE)) or {}).get("watchdog") or {}
        for k in r:
            if k in w and isinstance(w[k], (int, float, bool)):
                r[k] = w[k]
    except Exception:
        pass
    return r


def watchdog(cfg):
    """The rig's rules (watchdog_rules): no accepted share for no_shares_min with the source connected, the endpoint
    down for down_min, a card within hot_margin_c of its own slow point, or the rig's rate under the profile's
    expected rate by rate_drop_pct for rate_drop_min: restart Ember; the limits per hour; a second restart in the hour
    that brings nothing, then a reboot (when the rules allow one)."""
    if LANE or not fleet_host(cfg) or not ember_unit_kind() or FLEET["paused"]:
        return
    if not ember_enabled():
        return   # the owner's stop stands: a disabled unit is never restarted or rebooted for
    now = time.time()
    rules = watchdog_rules()
    cause = None
    b = ember_block()
    if b.get("up"):
        if b["source"].get("connected") and FLEET["last_accepted_ts"] and now - FLEET["last_accepted_ts"] > rules["no_shares_min"] * 60:
            cause = "no accepted share for %d min" % int((now - FLEET["last_accepted_ts"]) / 60)
        hps = b["totals"].get("validated_hps")
        exp = float(rules["expected_hps"] or 0)
        if not cause and rules["rate_drop_pct"] and exp > 0 and isinstance(hps, (int, float)):
            if hps < exp * (1 - float(rules["rate_drop_pct"]) / 100):
                FLEET.setdefault("low_since", now)
                if now - FLEET["low_since"] > rules["rate_drop_min"] * 60:
                    cause = "rate %.0f H/s under the profile's %.0f by more than %s percent for %d min" % (hps, exp, rules["rate_drop_pct"], int((now - FLEET["low_since"]) / 60))
            else:
                FLEET.pop("low_since", None)
        if not cause and rules["hot_margin_c"]:
            try:
                for bus, f in (card_facts() or {}).items():
                    e, t = f.get("edge_c"), f.get("throttle_c")
                    m = f.get("tlimit_margin_c")
                    if isinstance(e, (int, float)) and isinstance(t, (int, float)) and e >= t - rules["hot_margin_c"]:
                        cause = "%s edge %s C within %s of its slow point %s" % (f.get("model") or bus, e, rules["hot_margin_c"], t)
                        break
                    if isinstance(m, (int, float)) and m <= rules["hot_margin_c"]:
                        cause = "%s %s C from its slow point" % (f.get("model") or bus, m)
                        break
            except Exception:
                pass
    elif ember_active() and FLEET["down_since"] and now - FLEET["down_since"] > rules["down_min"] * 60:
        # the down rule counts only once the endpoint has answered since the unit's last start, or after the unit has
        # run WD_BIND_GRACE_S without binding (rig 1, 23:32 UK, 13 October 2026: Ember tuned seven minutes with no
        # listener on 4068 and the rule restarted a healthy miner; product A names the binding moment)
        started = ember_started_ts()
        if (FLEET.get("seen_up") and started and FLEET["seen_up"] > started) or (started and now - started > WD_BIND_GRACE_S):
            cause = "Ember's endpoint down for %d min" % int((now - FLEET["down_since"]) / 60)
    if not cause or fleet_busy():
        return
    if now - PULL_OK > WD_ROUTE_S:
        if FLEET.get("blind_said") != cause:
            flog("watchdog: %s, but no answer from the queue for %d s (no route): nothing acts blind" % (cause, int(now - PULL_OK)))
            FLEET["blind_said"] = cause
        return
    mem = mem_available_mb()
    if mem is not None and mem < WD_MEM_FLOOR_MB:
        if FLEET.get("mem_said") != cause:
            flog("watchdog: %s, but %d MB available is under the %d MB floor: no restart into a starved host" % (cause, mem, WD_MEM_FLOOR_MB))
            FLEET["mem_said"] = cause
        FLEET["watchdog"] = {"verb": "waits", "cause": "%s; host memory %d MB under the floor" % (cause, mem), "utc": utc()}
        return
    FLEET["restarts"] = [t for t in FLEET["restarts"] if now - t < 3600]
    FLEET["reboots"] = [t for t in FLEET["reboots"] if now - t < 3600]
    if rules["reboot"] and len(FLEET["restarts"]) >= 2 and len(FLEET["reboots"]) < rules["reboots_per_hour"]:
        verb = "reboot"
    elif len(FLEET["restarts"]) < rules["restarts_per_hour"]:
        verb = "restart-ember"
    else:
        if FLEET["watchdog"] is None or FLEET["watchdog"].get("verb") != "waits":
            flog("watchdog: %s; the hour's restarts and reboot used; waiting for a hand from the console" % cause)
            FLEET["watchdog"] = {"verb": "waits", "cause": cause, "utc": utc()}
        return
    rc, note = do_action(verb)
    (FLEET["reboots"] if verb == "reboot" else FLEET["restarts"]).append(now)
    FLEET["watchdog"] = {"verb": verb, "cause": cause, "utc": utc(), "rc": rc}
    FLEET["last_accepted_ts"] = now
    FLEET["down_since"] = None
    flog("watchdog: %s; %s (rc %s%s)" % (cause, verb, rc, (": " + note) if note else ""))


def fleet_tick(cfg):
    try:
        actions_poll(cfg)
        watchdog(cfg)
        fans_guard(cfg)
    except Exception as ex:
        log("fleet: %s" % str(ex)[:120])


def cmd_run(parent=None):
    global PULL_OK
    cfg = load_config()
    os.makedirs(JOBS, exist_ok=True)
    if os.path.exists(STOPFILE) and not LANE:
        os.remove(STOPFILE)
    if not LANE:
        # one default agent per machine (the mini, 11 October 2026 18:44 UK: launchd's KeepAlive and a self-install's
        # guard each started one): a live default agent named by the pid file, or any other `run` process without a
        # lane, means this start ends here
        try:
            old = int(open(PIDFILE).read().split()[0])
        except Exception:
            old = 0
        others = [p for p in _running_pids("igneum-bench-agent.py") if p != old and "--lane" not in (_cmdline(p) or "") and re.search(r"igneum-bench-agent\.py run( |$)", _cmdline(p) or "")]
        if (old and re.search(r"igneum-bench-agent\.py run( |$)", _cmdline(old) or "") and "--lane" not in (_cmdline(old) or "")) or others:
            seen = old if (old and re.search(r"igneum-bench-agent\.py run( |$)", _cmdline(old) or "")) else others[0]
            log("another default agent runs (pid %s: %s); this start ends" % (seen, (_cmdline(seen) or "")[:120]))
            return
    with open(PIDFILE, "w") as f:
        f.write(str(os.getpid()))
    signal.signal(signal.SIGTERM, lambda *a: STOP.set())
    signal.signal(signal.SIGINT, lambda *a: STOP.set())
    global THERMAL_GUARD, RIG_SERIAL
    if str(cfg.get("name", "")).startswith("rig") or cfg.get("thermal_guard"):
        THERMAL_GUARD = os.environ["IGNEUM_THERMAL_GUARD"] = "1"   # the lanes inherit it
    if str(cfg.get("name", "")).startswith("rig") and not LANE:
        # one card at a time by default; config.json "rig_serial": false lifts it to the PCs' per-card lanes for a window
        # (main's ruling of 12 October 2026 17:3x UK: the founder's chassis fans on until 19:20 UK; the flag is set back
        # and the agent restarted between rows when the fans go off). Read at start only; a change needs a restart.
        RIG_SERIAL = cfg.get("rig_serial", True) is not False
        if not RIG_SERIAL:
            log("a rig with rig_serial false: per-card lanes, the one-card rule lifted for this run")
            global POWER_GUARD
            pg = cfg.get("power_guard")
            if isinstance(pg, dict) and pg.get("url"):
                POWER_GUARD = pg
                threading.Thread(target=power_guard_loop, daemon=True).start()
                log("power guard on: the plug at the relay, the cap %s W, one lane per pass under it" % pg.get("max_w", 1500))
            else:
                log("per-card lanes on a rig WITHOUT a power guard in config.json: the lanes start together (main's rule wants the guard)")
    seed_cache_from_jobs()
    log("agent %s up as %s, pid %d%s%s" % (AGENT_VERSION, cfg["name"], os.getpid(), (" (lane %s for %s)" % LANE) if LANE else "",
                                         (" thermal guard on: edge %d C, junction %d C" % (THERMAL_EDGE_C, THERMAL_JUNCTION_C)) if THERMAL_GUARD == "1" else ""))
    if not LANE:
        start_lanes()
        pause_state_load(cfg)   # Ember Fleet: a pause the previous agent left becomes a wanted resume
    threading.Thread(target=sidecar, daemon=True).start()
    last_hb = 0
    try:
        while not STOP.is_set() and not os.path.exists(STOPFILE):
            if LANE and parent:
                try:
                    os.kill(parent, 0)
                except ProcessLookupError:
                    log("lane %s: the default agent (pid %d) is gone; exiting" % (LANE[0], parent))
                    break
            if time.time() - last_hb >= HEARTBEAT_S:
                retry_pending_done()
                added, removed = refresh_identity()
                if LANE and parent is not None and not any(LANE[1].lower() in c.lower() for c in (identity().get("cards") or [])):
                    log("lane %s: its card '%s' is no longer present; ending" % (LANE[0], LANE[1]))
                    heartbeat({"state": "card-removed"})
                    write_state(state="card-removed")
                    break
                if not LANE:
                    if added or removed:
                        log("cards changed: added %s, removed %s" % (added, removed))
                    housekeeping()   # every heartbeat: the lanes, the drop, the relay
                heartbeat({"state": "idle"})
                last_hb = time.time()
                write_state(state="idle", pull_ok=int(PULL_OK))
                if not LANE:
                    fleet_tick(cfg)   # Ember Fleet: the owner's actions and the watchdog, between jobs
            if not fetch_signer():
                STOP.wait(IDLE_POLL_S)
                continue
            if LANE and not lane_may_pull():
                STOP.wait(IDLE_POLL_S)
                continue
            pull = ["pull"]
            if LANE:
                pull.append(base64.urlsafe_b64encode(LANE[1].encode()).decode())
            elif SINGLE_CARD:
                pull.append(base64.urlsafe_b64encode(SINGLE_CARD.encode()).decode() + "+any")   # the one-card machine: card jobs and card-less jobs alike
            rc, out, err = ssh_cmd(pull, timeout=60)
            if rc == 0:
                PULL_OK = time.time()
            if rc != 0 or out.strip() in (b"", b"none"):
                if rc != 0:
                    log("pull: " + err.strip()[:200])
                elif not LANE:
                    ember_resume_if_idle()   # Ember Fleet: the queue answered empty, a wanted resume may start now
                STOP.wait(IDLE_POLL_S)
                continue
            envelope = json.loads(out.decode())
            job_bytes, sig_bytes = base64.b64decode(envelope["job"]), base64.b64decode(envelope["sig"])
            job = json.loads(job_bytes.decode())
            jid = job.get("job_id") or job.get("id") or "unknown"
            # one job directory at a time per lane (the rig hosts have a 60 GB disk): this lane's own finished jobs' work
            # and bin trees go, never another lane's, never a directory without its summary (a live step)
            prune_jobs(jid)
            jobdir = os.path.join(JOBS, jid)
            os.makedirs(jobdir, exist_ok=True)
            with open(os.path.join(jobdir, "lane"), "w") as f:
                f.write(LANE[0] if LANE else "default")
            with open(os.path.join(jobdir, "job.json"), "wb") as f:
                f.write(job_bytes)
            with open(os.path.join(jobdir, "job.sig"), "wb") as f:
                f.write(sig_bytes)
            ok, why = verify_signature(os.path.join(jobdir, "job.json"), os.path.join(jobdir, "job.sig"))
            if not ok or job.get("machine") != cfg["name"]:
                log("REFUSED job %s: %s" % (jid, why if not ok else "job is for " + str(job.get("machine"))))
                ssh_cmd(["done", jid, "refused"], timeout=60)
                continue
            log("job %s cell %s worker %s" % (jid, job["cell"]["name"], job["worker"]["sha256"][:12]))
            heartbeat({"state": "running", "job": jid, "cell": job["cell"]["name"]})
            ember_pause(cfg)   # Ember Fleet: on a rig Ember stops for the job and starts again after it
            try:
                summary = run_job(cfg, job, jobdir)
            finally:
                RUNNING.clear()
                ember_resume(cfg)
            pushed = push_results(cfg, job, jobdir, summary)
            send_done(jid, summary["verdict"], pushed, jobdir)
            log("job %s %s (%ss); pushed %d files" % (jid, summary["verdict"], summary.get("seconds"), len(pushed)))
            last_hb = 0
    finally:
        heartbeat({"state": "stopped"})
        write_state(state="stopped")
        req = os.path.join(DIR, "swap-request")
        if not LANE and os.path.exists(req):
            # the swap: this default agent starts the guard from its own context and logs its pid (0.3.18)
            try:
                gpath = open(req).read().strip()
                os.remove(req)
                gl = None
                try:
                    gl = int(open(os.path.join(DIR, "guard.pid")).read().strip())
                    os.kill(gl, 0)
                except Exception:
                    gl = None
                if gl:
                    log("swap: the step's guard (pid %d) is alive; no second guard" % gl)
                    raise StopIteration
                gpid, how = start_guard(gpath)
                time.sleep(1.5)
                alive = False
                try:
                    os.kill(gpid, 0)
                    alive = True
                except Exception:
                    pass
                with open(os.path.join(DIR, "guard.pid"), "w") as gf:
                    gf.write(str(gpid))
                log("swap: guard %s started as pid %d (%s), alive after 1.5 s: %s" % (gpath, gpid, how, alive))
                with open(os.path.join(DIR, "self-update.log"), "a") as lf:
                    lf.write("%s guard started by the default agent (pid %d, %s, alive %s)\n" % (utc(), gpid, how, alive))
            except StopIteration:
                pass
            except Exception as ex:
                log("swap: guard start failed: %s" % ex)
        try:
            # the pid file goes only when it names this process (an ended duplicate took launchd's agent's file with
            # it on the mini, 11 October 2026 18:47 UK)
            if open(PIDFILE).read().split()[0] == str(os.getpid()):
                os.remove(PIDFILE)
        except (FileNotFoundError, IndexError, ValueError):
            pass
        log("agent down")


def fetch_signer():
    """The queue's public signing key from build-1, into allowed_signers. True when it is in place."""
    if os.path.exists(SIGNERS):
        return True
    rc, out, err = ssh_cmd(["signer"], timeout=30)
    if rc == 0 and out.strip():
        with open(SIGNERS, "w") as f:
            f.write("%s namespaces=\"%s\" %s\n" % (SIGN_IDENTITY, SIGN_NAMESPACE, out.decode().strip()))
        with open(SIGNERS + ".pub", "w") as f:
            f.write(out.decode().strip() + "\n")
        log("queue signing key received: " + subprocess.run(["ssh-keygen", "-lf", SIGNERS + ".pub"], capture_output=True).stdout.decode().strip())
        return True
    log("not authorized yet on build-1 (%s)" % (err.strip().splitlines()[-1] if err.strip() else "rc %d" % rc))
    return False


def cmd_doctor():
    cfg = load_config() if os.path.exists(CONFIG) else {}
    print("agent %s  python %s  platform %s" % (AGENT_VERSION, platform.python_version(), platform.platform()))
    print("dir %s  config %s  name %s  meter %s" % (DIR, "present" if cfg else "none", cfg.get("name"), (cfg.get("meter") or {}).get("kind")))
    for tool in ("ssh", "ssh-keygen", "tar"):
        print("%-11s %s" % (tool, shutil.which(tool) or "MISSING"))
    v = subprocess.run(["ssh", "-V"], capture_output=True).stderr.decode().strip()
    print("ssh version %s" % v)
    y = subprocess.run(["ssh-keygen", "-Y", "verify"], capture_output=True).stderr.decode()
    print("ssh-keygen -Y %s" % ("supported" if "usage" in y.lower() or "missing" in y.lower() or "-f" in y else "NOT supported (OpenSSH 8.1+ needed): " + y.strip()[:80]))
    print("key %s  pin %s  signer %s  pid file %s" % ("present" if os.path.exists(KEY) else "none", "present" if os.path.exists(KNOWN) else "none",
                                                      "present" if os.path.exists(SIGNERS) else "none", PIDFILE))
    if os.path.exists(KEY):
        rc, out, err = ssh_cmd(["signer"], timeout=30)
        print("connection to %s: %s" % (SERVER, "authorized" if rc == 0 else "refused (%s)" % (err.strip().splitlines()[-1] if err.strip() else rc)))


def cmd_register(args):
    name = _arg(args, "--name")
    if not name:
        sys.exit("usage: register --name <machine> [--meter <shelly|tasmota|json> <url>] [--provider <name> --handle <id>]")
    os.makedirs(DIR, mode=0o700, exist_ok=True)
    cfg = load_config() if os.path.exists(CONFIG) else {}
    cfg["name"] = name
    if "--meter" in args:
        i = args.index("--meter")
        cfg["meter"] = {"kind": args[i + 1], "address": args[i + 2], "sample_hz": 1}
    for k in ("provider", "handle", "pcie_link"):     # a rented pod: --provider vast --handle <id>; a rig host: --pcie_link "x1 electrical"
        v = _arg(args, "--" + k)
        if v:
            cfg[k] = v
    with open(CONFIG, "w") as f:
        json.dump(cfg, f, indent=1)
    with open(KNOWN, "w") as f:
        f.write(SERVER_HOSTKEY + "\n")
    if not os.path.exists(KEY) or not os.path.exists(KEY + ".pub"):
        for f in (KEY, KEY + ".pub"):
            if os.path.exists(f):
                os.remove(f)
        subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "igneum-bench-" + name, "-f", KEY],
                       check=True, stdin=subprocess.DEVNULL)
    pub = open(KEY + ".pub").read().strip()
    print("this machine's public key (the private half stays here, %s):\n  %s" % (KEY, pub))
    print("\nthe operator authorizes it on the console host with:\n  ssh %s igneum-bench authorize %s <<< '%s'" % (SERVER, name, pub))
    print("\nthen on this machine:  igneum-bench-agent.py run   (it fetches the queue's signing key once authorized;")
    print("compare the fingerprint it logs with infra/bench/queue-sign.pub on main before trusting a job)")
    if "--wait" in args:
        while not fetch_signer():
            time.sleep(10)
        print("authorized; run: igneum-bench-agent.py run")


def cmd_stop():
    try:
        pid = int(open(PIDFILE).read().strip())
    except (FileNotFoundError, ValueError):
        sys.exit("no agent pid file at " + PIDFILE)
    cmdline = _cmdline(pid)
    if not cmdline:
        sys.exit("pid %d is not running (stale pid file)" % pid)
    if "igneum-bench-agent" not in cmdline:
        sys.exit("pid %d is not the agent (%s); not killed" % (pid, cmdline[:80]))
    print("stopping agent pid %d: %s" % (pid, cmdline.strip()))
    open(STOPFILE, "w").close()
    os.kill(pid, signal.SIGTERM)


def cmd_self_restart():
    """A restart through the queue, no hand on the machine (the update's own path with the installed file): the
    guard waits for this agent to stop, starts the installed file again, and this agent stops after the job."""
    global DIR
    d = os.environ.get("IGNEUM_BENCH_DIR") or DIR
    DIR = d
    _paths()
    target = os.path.join(DIR, "igneum-bench-agent.py")
    guard = os.path.join(DIR, "self-restart-guard.sh")
    with open(guard, "w") as f:
        f.write("""#!/usr/bin/env bash
d=%(dir)s; t="$d/igneum-bench-agent.py"
old=$(cat "$d/agent.pid" 2>/dev/null)
for i in $(seq 1 120); do if [ -z "$old" ] || ! kill -0 "$old" 2>/dev/null; then break; fi; sleep 5; done
rm -f "$d/agent.pid"; sleep 3
if systemctl --user is-enabled igneum-bench-agent >/dev/null 2>&1; then systemctl --user restart igneum-bench-agent; else cd "$d" && env -u IGNEUM_OUT -u IGNEUM_JOB -u IGNEUM_CELL -u IGNEUM_WORKER IGNEUM_BENCH_DIR="$d" HOME="$(dirname "$d")" nohup python3 "$t" run >> "$d/agent.log" 2>&1 & fi
echo "$(date -u +%%FT%%TZ) restarted on request" >> "$d/self-update.log"
""" % {"dir": DIR})
    os.chmod(guard, 0o755)
    subprocess.Popen(["bash", guard], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
    open(STOPFILE, "w").close()
    print("self-restart: guard started; the agent stops after this job and the installed file starts again (%s)" % target)


def cmd_self_install():
    """The self-update path (BENCH-AGENT.md section 8), the Python side: this file is the NEW agent, fetched by sha as a
    job's worker and run as the step "igneum-bench-agent.py self-install". It copies itself over the installed agent
    (the old one kept as .prev), starts a detached guard that waits for the running agent to stop, restarts the agent
    (the user unit if present, else nohup), and rolls back to .prev if no fresh state appears within two minutes."""
    global DIR
    me = os.path.abspath(__file__)
    # the running agent gives a step HOME=<job work dir>, so DIR read from HOME is wrong here (the 0.3.5 update on the
    # mini failed on it); the installed directory is the one this file was fetched under: <DIR>/jobs/<id>/bin/<me>,
    # or IGNEUM_BENCH_DIR when the running agent passes it (0.3.7 and later do)
    d = os.environ.get("IGNEUM_BENCH_DIR")
    if not d or not os.path.isdir(d):
        up = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(me))))
        d = up if os.path.isfile(os.path.join(up, "igneum-bench-agent.py")) or os.path.isdir(os.path.join(up, "jobs")) else DIR
    DIR = d
    _paths()
    target = os.path.join(DIR, "igneum-bench-agent.py")
    new_sha = sha256_file(me)
    if os.path.exists(target) and sha256_file(target) == new_sha:
        print("self-install: already at " + new_sha)
        return
    if os.path.exists(target):
        shutil.copy2(target, target + ".prev")
    shutil.copy2(me, target + ".new")
    if sha256_file(target + ".new") != new_sha:
        raise RuntimeError("self-install: copy sha mismatch")
    os.replace(target + ".new", target)
    os.chmod(target, 0o755)
    print("self-install: installed %s (previous kept as .prev)" % new_sha)
    # meters.py beside the agent, from the harness tree of this very job, so meter-test and the plug, PMD2X and WireView
    # readers work outside a job too (a job's reader prefers the harness tree the job names)
    for cand in (os.path.join(os.path.dirname(me), "meters.py"), os.path.join(os.environ.get("IGNEUM_OUT", ""), "..", "work", "harness", "pow", "harness", "meters.py")):
        if cand and os.path.exists(cand):
            shutil.copy2(cand, os.path.join(DIR, "meters.py"))
            print("self-install: meters.py placed beside the agent (%s)" % sha256_file(cand)[:16])
            break
    # the fleet page beside the agent (FLEET.md: the relay serves it at /fleet on the LAN), from the job's harness tree
    # when the job names site/fleet among its harness paths
    page = os.path.join(os.environ.get("IGNEUM_OUT", ""), "..", "work", "harness", "site", "fleet", "index.html")
    if os.path.exists(page):
        shutil.copy2(page, os.path.join(DIR, "fleet.html"))
        print("self-install: the fleet page placed beside the agent (%s)" % sha256_file(page)[:16])
    guard = os.path.join(DIR, "self-update-guard.sh")
    with open(guard, "w") as f:
        f.write("""#!/usr/bin/env bash
# the self-update guard: wait for the old agent to stop, start the new one, roll back if it does not write state in two minutes
d=%(dir)s; t="$d/igneum-bench-agent.py"; swap=$(date +%%s)
# the old agent is gone when its pid is gone, not when its file is (an agent before 0.3.8 left the file on its stop and
# the guard waited its whole ten minutes on the mini, 10 October 2026); the file is read once and the pid watched
old=$(cat "$d/agent.pid" 2>/dev/null)
for i in $(seq 1 120); do
  if [ -z "$old" ] || ! kill -0 "$old" 2>/dev/null; then break; fi
  [ -f "$d/agent.pid" ] || break
  sleep 5
done
# a launchd job owning the agent (the mini's LaunchAgent, 11 October 2026): launchd restarts the agent itself the
# moment the old one exits, so the guard neither removes the pid file (launchd's fresh agent wrote it; the 18:51 UK
# handover removed it and the guard rolled a good start back) nor starts one of its own: it kickstarts the job
la=""; if command -v launchctl >/dev/null 2>&1 && launchctl print "gui/$(id -u)/network.igneum.bench-agent" >/dev/null 2>&1; then la=1; fi
[ -n "$la" ] || rm -f "$d/agent.pid"
sleep 3
# the new agent starts with the agent directory named and the user's own home, never the step's environment (the
# step runs with HOME set to the job's work directory and IGNEUM_OUT set; the 0.3.7 and 0.3.8 handovers on the mini
# started the new agent with that environment, it read its config under the work directory and died at once)
start() { if [ -n "$la" ]; then launchctl kickstart -k "gui/$(id -u)/network.igneum.bench-agent"; elif systemctl --user is-enabled igneum-bench-agent >/dev/null 2>&1; then systemctl --user restart igneum-bench-agent; else cd "$d" && env -u IGNEUM_OUT -u IGNEUM_JOB -u IGNEUM_CELL -u IGNEUM_WORKER IGNEUM_BENCH_DIR="$d" HOME="$(dirname "$d")" nohup python3 "$t" run >> "$d/agent.log" 2>&1 & fi; }
if [ -n "$la" ]; then np=$(cat "$d/agent.pid" 2>/dev/null); if [ -n "$np" ] && [ "$np" != "$old" ] && kill -0 "$np" 2>/dev/null; then :; else start; fi; else start; fi
# fresh means the NEW agent wrote its pid file and that pid is alive (the old agent's own "stopped" state, written
# after the swap, read as fresh before and hid a dead start)
ok=0
for i in $(seq 1 24); do sleep 5; np=$(cat "$d/agent.pid" 2>/dev/null); if [ -n "$np" ] && [ "$np" != "$old" ] && kill -0 "$np" 2>/dev/null; then ok=1; break; fi; done
if [ "$ok" = 1 ]; then echo "$(date -u +%%FT%%TZ) updated to %(sha)s" >> "$d/self-update.log"
else cp "$t.prev" "$t" 2>/dev/null; start; echo "$(date -u +%%FT%%TZ) rolled back: no fresh state in two minutes" >> "$d/self-update.log"; fi
""" % {"dir": DIR, "sha": new_sha})
    os.chmod(guard, 0o755)
    with open(os.path.join(DIR, "swap-request"), "w") as f:   # the default agent starts a second guard at its exit only when none is alive (0.3.19)
        f.write(guard)
    gpid, how = start_guard(guard)
    with open(os.path.join(DIR, "guard.pid"), "w") as f:
        f.write(str(gpid))
    open(STOPFILE, "w").close()
    print("self-install: guard started (pid %d, %s) and swap requested; the running agent stops after its current work and the new file takes over" % (gpid, how))


def start_guard(guard):
    """Starts the swap guard where it survives this agent's exit and returns (pid, how). Under a systemd user unit (a
    rig) the unit's cgroup dies with the agent (KillMode=mixed took the guard and the lanes on rig 1, 11 October 2026
    20:10 UK, and nothing restarted the unit), so the guard runs as its own transient unit, and its start() restarts
    the agent's unit. The unit is known by /proc/self/cgroup, never by the environment alone: the self-install step
    runs with the job's own environment, where INVOCATION_ID, XDG_RUNTIME_DIR and the session bus are absent (so
    systemd-run is given the user manager's runtime directory and bus by the uid: without them it fails silently and
    the guard fell into the cgroup twice on rig 1, 22:04 and 22:38 UK). Both callers use this: the self-install step
    and the default agent at its exit. Elsewhere (launchd, a plain shell) the guard is a detached child."""
    try:
        in_unit = "igneum-bench-agent.service" in open("/proc/self/cgroup").read()
    except Exception:
        in_unit = False
    if (in_unit or os.environ.get("INVOCATION_ID")) and shutil.which("systemd-run"):
        env = dict(os.environ)
        rd = env.get("XDG_RUNTIME_DIR") or "/run/user/%d" % os.getuid()
        env["XDG_RUNTIME_DIR"] = rd
        env.setdefault("DBUS_SESSION_BUS_ADDRESS", "unix:path=%s/bus" % rd)
        unit = "igneum-bench-guard-%d" % int(time.time())
        try:
            r = subprocess.run(["systemd-run", "--user", "--collect", "--unit=" + unit, "bash", guard], env=env,
                               stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=30)
            out = subprocess.run(["systemctl", "--user", "show", "-p", "MainPID", "--value", unit + ".service"], env=env,
                                 capture_output=True, text=True, timeout=10).stdout.strip()
            pid = int(out.splitlines()[0]) if out else 0
            if pid:
                return pid, "transient unit " + unit
            log("swap: systemd-run gave no pid (rc %s: %s); the guard goes as a detached child" % (r.returncode, (r.stderr or r.stdout).strip()[:120]))
        except Exception as ex:
            log("swap: systemd-run failed (%s); the guard goes as a detached child" % str(ex)[:120])
    gp = subprocess.Popen(["bash", guard], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
    return gp.pid, "detached child"


def cmd_status():
    try:
        print(json.dumps(json.load(open(STATEFILE)), indent=1))
    except FileNotFoundError:
        print("no state yet")


def _arg(args, flag):
    return args[args.index(flag) + 1] if flag in args and args.index(flag) + 1 < len(args) else None


if __name__ == "__main__":
    a = sys.argv[1:]
    if not a:
        sys.exit(__doc__)
    if a[0] == "register":
        cmd_register(a[1:])
    elif a[0] == "run":
        if "--lane" in a:
            LANE = (_arg(a, "--lane"), _arg(a, "--card"))
            _lane_paths()
        cmd_run(int(_arg(a, "--parent")) if "--parent" in a else None)
    elif a[0] == "stop":
        cmd_stop()
    elif a[0] == "status":
        cmd_status()
    elif a[0] == "doctor":
        cmd_doctor()
    elif a[0] == "self-restart":
        cmd_self_restart()
    elif a[0] == "self-install":
        cmd_self_install()
    elif a[0] == "meter-test":
        print(meter_read(load_config()["meter"]))
    elif a[0] == "plug-relay":
        cmd_plug_relay()
    else:
        sys.exit(__doc__)