k8s: exclude passthrough source keys from jbod-mqtt secret

excludeRaw alone left all home_assistant source keys (api_key, vantage_*, etc.)
in the synced secret, so envFrom injected them into the web container. Add
transformation.excludes ['.*'] so only the templated MQTT_USERNAME/MQTT_PASSWORD
remain.

.dockerignore

@@ -6,3 +6,5 @@ __pycache__
 .git
 .env
 *.md
+tmp/
+.venv/

.gitignore

@@ -1,3 +1,5 @@
 __pycache__/
 *.pyc
 .venv/
+tmp/
+jbod-monitor.jsx*

Dockerfile

@@ -1,4 +1,4 @@
-# ---- Stage 1: Build frontend ----
+# ---- Stage: build frontend ----
 FROM node:22-alpine AS frontend-build
 WORKDIR /app/frontend
 COPY frontend/package.json frontend/package-lock.json* ./
@@ -7,8 +7,10 @@ COPY frontend/ ./
 RUN npm run build
 # Output: /app/static/
 
-# ---- Stage 2: Production runtime ----
-FROM python:3.13-slim
+# ---- Target: poller (hardware producer) ----
+# Privileged at runtime; the only image with SAS/SMART tooling. Slim Python
+# deps (just Redis) — no web stack.
+FROM python:3.13-slim AS poller
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
     smartmontools \
@@ -20,15 +22,48 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 
 WORKDIR /app
 
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+COPY requirements-poller.txt .
+RUN pip install --no-cache-dir -r requirements-poller.txt
+
+COPY poller.py .
+COPY services/ services/
+
+CMD ["python", "-u", "poller.py"]
+
+# ---- Target: host-poller (chassis IPMI/iDRAC + CPU + on-metal drives) ----
+# Privileged at runtime (needs /dev/ipmi0 + raw disks). Carries ipmitool +
+# smartmontools; no SAS-shelf tooling.
+FROM python:3.13-slim AS host-poller
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ipmitool \
+    smartmontools \
+    util-linux \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements-poller.txt .
+RUN pip install --no-cache-dir -r requirements-poller.txt
+
+COPY host_poller.py .
+COPY services/ services/
+
+CMD ["python", "-u", "host_poller.py"]
+
+# ---- Target: web (read-only consumer) ----
+# Unprivileged at runtime; no hardware tools. Serves REST/UI + MQTT, reads Redis.
+FROM python:3.13-slim AS web
+
+WORKDIR /app
+
+COPY requirements-web.txt .
+RUN pip install --no-cache-dir -r requirements-web.txt
 
 COPY main.py .
 COPY routers/ routers/
 COPY services/ services/
 COPY models/ models/
-
-# Copy built frontend from stage 1
 COPY --from=frontend-build /app/static/ static/
 
 EXPOSE 8000
@@ -36,4 +71,7 @@ EXPOSE 8000
 HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
     CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/api/health')" || exit 1
 
-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]
+# Single worker: the MQTT publisher is a per-process singleton (multiple
+# workers would connect with the same client_id and flap the broker). The
+# API is a low-traffic internal monitor, so one worker is plenty.
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

README.md

@@ -1,45 +1,127 @@
 # JBOD Monitor
 
-REST API for monitoring drive health in JBOD enclosures on Linux.
+Drive-health monitoring for JBOD enclosures on Linux, with a REST API, web UI,
+and optional Home Assistant MQTT publishing.
 
-Auto-discovers SES enclosures via sysfs, maps drives to physical slots, and exposes SMART health data.
+Auto-discovers SES enclosures via sysfs, maps drives to physical slots, reads
+SMART, and aggregates per-enclosure temperatures (named SES sensors + hotspot).
+
+## Architecture
+
+Two processes, with **Redis as the only interface between them**:
+
+```
+poller (privileged, pid:host)  ──smartctl/sg_ses/zpool/ledctl──>  hardware
+        │  serialized + paced
+        ▼
+      Redis  ◀── reads ──  app (REST API + web UI + MQTT)   [unprivileged]
+        ▲
+        └── locate-LED command queue (app enqueues, poller runs)
+```
+
+- **`poller`** is the *only* process that touches the SAS bus. All hardware
+  commands run behind a single gate (`POLL_CONCURRENCY`, default 1 = fully
+  serialized) with a gap between drives, so it never storms fragile SAS
+  expanders. It sweeps on an interval and writes results to Redis.
+- **`app`** (and the MQTT publisher) are pure Redis readers — unprivileged, no
+  `/dev`. They can restart freely without issuing a single hardware command.
+- **Locate LEDs**: the API enqueues a request in Redis; the poller executes it
+  serialized with everything else.
+
+This split exists for a reason: concurrent/bursty SMART+SES traffic — especially
+repeated on every container restart — can drop SAS expanders and suspend pools.
+One paced owner makes that structurally impossible.
 
 ## Prerequisites
 
 - Linux with SAS/SATA JBODs connected via HBA
-- `smartmontools` — for `smartctl` (SMART data)
-- `sg3-utils` — for `sg_ses` (SES enclosure data)
+- `smartmontools` (`smartctl`), `sg3-utils` (`sg_ses`), `ledmon` (`ledctl`)
+- Redis
 - Python 3.11+
 
 ```bash
-# Debian/Ubuntu
-apt install smartmontools sg3-utils
-
-# RHEL/Fedora
-dnf install smartmontools sg3_utils
+apt install smartmontools sg3-utils ledmon   # Debian/Ubuntu
 ```
 
-## Install
+## Run (docker compose)
+
+Two **independent** stacks so web iterations never recreate the privileged
+poller (which touches the SAS bus). They're built as two images —
+`jbod-poller` (privileged, has the SAS/SMART tooling) and `jbod-web` (slim,
+unprivileged) — and share Redis over host networking.
 
 ```bash
-pip install -r requirements.txt
+# 1. Infra: poller + redis — the stable substrate. Deploy once; touch deliberately.
+docker compose -f compose.infra.yml up -d
+
+# 2. Web: REST/UI/MQTT — iterate freely; this only ever recreates `app`.
+docker compose -f compose.web.yml up -d --build
 ```
 
-## Run
-
-The API needs root access for `smartctl` to query drives:
+Pin both images to a known-good SHA in deploy (`./build.sh` tags `:<sha>` and
+`:latest` for both). To run the two processes by hand instead:
 
 ```bash
-sudo uvicorn main:app --host 0.0.0.0 --port 8000
+REDIS_HOST=localhost sudo -E python poller.py            # producer (needs root)
+REDIS_HOST=localhost uvicorn main:app --port 8000        # consumer
 ```
 
 ## API Endpoints
 
 | Endpoint | Description |
 |---|---|
-| `GET /api/health` | Service health + tool availability |
-| `GET /api/enclosures` | List all discovered SES enclosures |
-| `GET /api/enclosures/{id}/drives` | List drive slots for an enclosure |
+| `GET /api/health` | Service health + `redis`/`mqtt`/`poller_fresh` status |
+| `GET /api/enclosures` | List discovered SES enclosures |
+| `GET /api/enclosures/{id}/drives` | Drive slots for an enclosure |
 | `GET /api/drives/{device}` | SMART detail for a block device |
 | `GET /api/overview` | Aggregate enclosure + drive health |
-| `GET /docs` | Interactive API docs (Swagger UI) |
+| `GET /api/temps` | Per-enclosure temperatures: named SES sensors + hotspot |
+| `POST /api/drives/{device}/led` | Queue a locate-LED change (`state`: `locate`/`off`) |
+| `GET /docs` | Swagger UI |
+
+Read endpoints serve the poller's last sweep; `X-Poll-Age` (seconds) headers and
+the `poller_fresh` health flag surface staleness.
+
+## Poller tuning (env)
+
+| Variable | Default | Description |
+|---|---|---|
+| `POLL_SWEEP_INTERVAL` | `300` | Seconds between full sweeps |
+| `POLL_DRIVE_GAP` | `0.5` | Seconds between each drive's SMART query |
+| `POLL_CONCURRENCY` | `1` | Max concurrent hardware ops (1 = serialized) |
+| `POLL_STARTUP_JITTER` | `10` | Random startup delay to avoid restart storms |
+| `POLL_DATA_TTL` | `900` | Redis TTL for poller data (must exceed sweep interval) |
+| `ZFS_USE_NSENTER` | — | `true` to run `zpool` in the host namespace (containers) |
+
+## Home Assistant (MQTT)
+
+The `app` publishes per-enclosure temperatures via
+[MQTT discovery](https://www.home-assistant.io/integrations/mqtt/#mqtt-discovery)
+— no HA YAML. Each enclosure becomes a device with a **Hotspot** sensor (max
+across SES sensors + housed drive temps; `drive_max_c`/`drive_temp_count` as
+attributes) and one sensor per named SES temperature element. Opt-in via
+`MQTT_HOST`; availability is tracked via an MQTT LWT.
+
+| Variable | Default | Description |
+|---|---|---|
+| `MQTT_HOST` | _(unset)_ | Broker host. Unset → MQTT disabled. |
+| `MQTT_PORT` | `1883` | Broker port |
+| `MQTT_USERNAME` / `MQTT_PASSWORD` | _(unset)_ | Broker credentials |
+| `MQTT_DISCOVERY_PREFIX` | `homeassistant` | HA discovery topic prefix |
+| `MQTT_BASE_TOPIC` | `jbod-monitor` | State/availability topic root |
+| `MQTT_PUBLISH_INTERVAL` | `60` | Seconds between publishes |
+| `MQTT_NODE_ID` | _(hostname)_ | Stable id (disambiguates multiple hosts) |
+
+**Credentials**: simplest is a root-owned `/etc/jbod-monitor/mqtt.env` with
+`MQTT_USERNAME=`/`MQTT_PASSWORD=`, loaded via compose `env_file`. Optionally the
+app can read them from OpenBao at runtime (`MQTT_SECRET_PATH` + `OPENBAO_*`,
+keys via `MQTT_SECRET_USER_KEY`/`MQTT_SECRET_PASS_KEY`) — see
+`services/secrets.py`; it falls back to the env vars if the read fails.
+
+Topics:
+```
+homeassistant/sensor/<node>_enc<id>/hotspot/config   # retained discovery
+homeassistant/sensor/<node>_enc<id>/temp<n>/config    # retained discovery
+jbod-monitor/<node>/status                            # online | offline (LWT)
+jbod-monitor/<node>/enclosure/<id>/state              # {"hotspot_c":42.0, ...}
+```

build.sh

@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Usage: ./build.sh [poller|web|all] [commit message]
+#   web   — build/push ONLY the web image (use this for web iterations so the
+#           poller image is never rebuilt/repushed)
+#   poller— build/push ONLY the poller image (touch deliberately)
+#   all   — both (default)
+#
+# Deploy pins images by SHA (compose.*.yml ship :latest as a convenience only).
+
+REG="docker.adamksmith.xyz"
+
+cd "$(dirname "$0")"
+
+# First arg is the target if it's a known one; otherwise it's treated as the
+# commit message (backward-compatible with `./build.sh "message"`).
+case "${1:-}" in
+    poller)      TARGETS="poller";      MSG="${2:-Build and push images}" ;;
+    host-poller) TARGETS="host-poller"; MSG="${2:-Build and push images}" ;;
+    web)         TARGETS="web";         MSG="${2:-Build and push images}" ;;
+    all|"")      TARGETS="poller host-poller web"; MSG="${2:-Build and push images}" ;;
+    *)           TARGETS="poller host-poller web"; MSG="${1}" ;;
+esac
+TARGET="${TARGETS// /+}"
+
+# Stage and commit all changes
+git add -A
+if git diff --cached --quiet; then
+    echo "No changes to commit, using HEAD"
+else
+    git commit -m "${MSG}"
+fi
+
+SHA=$(git rev-parse --short HEAD)
+
+for target in ${TARGETS}; do
+    img="${REG}/jbod-${target}"
+    echo "Building ${img}:${SHA}"
+    docker build --target "${target}" -t "${img}:${SHA}" -t "${img}:latest" .
+    echo "Pushing ${img}:${SHA}"
+    docker push "${img}:${SHA}"
+    docker push "${img}:latest"
+done
+
+echo "Done (${TARGET}): ${REG}/jbod-{${TARGETS// /,}}:${SHA}"

compose.infra.yml

@@ -0,0 +1,74 @@
+# Infra stack: the hardware poller + Redis. This is the STABLE substrate —
+# deploy once, pin to a known-good image SHA, and touch it deliberately
+# (never as part of a web iteration). Web redeploys use compose.web.yml and
+# can never name these services.
+#
+#   docker compose -f compose.infra.yml up -d      # bring up / update poller+redis
+#
+name: jbod-infra
+
+services:
+  poller:
+    image: docker.adamksmith.xyz/jbod-poller:latest   # pin to a SHA in deploy
+    container_name: jbod-poller
+    restart: unless-stopped
+    privileged: true
+    pid: host
+    network_mode: host
+    volumes:
+      - /dev:/dev
+      - /sys:/sys:ro
+      - /run/udev:/run/udev:ro
+    environment:
+      - TZ=America/Denver
+      - ZFS_USE_NSENTER=true
+      - REDIS_HOST=127.0.0.1
+      - REDIS_PORT=6379
+      - REDIS_DB=0
+      # Pacing — keep the SAS bus quiet. POLL_CONCURRENCY=1 fully serializes.
+      - POLL_SWEEP_INTERVAL=300
+      - POLL_DRIVE_GAP=0.5
+      - POLL_CONCURRENCY=1
+      - POLL_STARTUP_JITTER=10
+      - POLL_DATA_TTL=900
+    depends_on:
+      - redis
+
+  # Chassis producer: IPMI/iDRAC + CPU coretemp + on-metal drives → Redis.
+  # Separate from the SAS poller (different subsystem); writes host_sensors +
+  # host_drives. The web consumer publishes these on the hub device.
+  host-poller:
+    image: docker.adamksmith.xyz/jbod-host-poller:latest   # pin to a SHA in deploy
+    container_name: jbod-host-poller
+    restart: unless-stopped
+    privileged: true
+    network_mode: host
+    volumes:
+      - /dev:/dev
+      - /sys:/sys:ro
+      - /run/udev:/run/udev:ro
+    environment:
+      - TZ=America/Denver
+      - REDIS_HOST=127.0.0.1
+      - REDIS_PORT=6379
+      - REDIS_DB=0
+      - HOST_ENV_INTERVAL=60     # IPMI/iDRAC + CPU — responsive (inlet temp 1/min)
+      - HOST_DRIVE_INTERVAL=300  # on-metal SMART — gentle, isolated from env
+      - POLL_DRIVE_GAP=0.5
+      - POLL_CONCURRENCY=1
+      - POLL_STARTUP_JITTER=10
+      - POLL_DATA_TTL=900
+    depends_on:
+      - redis
+
+  redis:
+    image: redis:7-alpine
+    container_name: jbod-redis
+    restart: unless-stopped
+    network_mode: host
+    volumes:
+      - redis-data:/data
+    command: redis-server --save 60 1 --loglevel warning
+
+volumes:
+  redis-data:

compose.web.yml

@@ -0,0 +1,33 @@
+# Web stack: REST API + UI + Home Assistant MQTT. Read-only consumer of the
+# Redis filled by compose.infra.yml. Iterate freely here — this command only
+# ever touches the `app` container, never the poller:
+#
+#   docker compose -f compose.web.yml up -d --build
+#
+# Requires the infra stack (Redis) to be running. Reaches Redis at
+# 127.0.0.1:6379 via host networking (separate compose project, shared host net).
+name: jbod-web
+
+services:
+  app:
+    image: docker.adamksmith.xyz/jbod-web:latest   # pin to a SHA in deploy
+    container_name: jbod-monitor
+    restart: unless-stopped
+    network_mode: host
+    environment:
+      - TZ=America/Denver
+      - UVICORN_LOG_LEVEL=info
+      - REDIS_HOST=127.0.0.1
+      - REDIS_PORT=6379
+      - REDIS_DB=0
+      # Home Assistant MQTT publishing (EMQX, bridged to Mosquitto HA add-on).
+      - MQTT_HOST=10.5.30.3
+      - MQTT_PORT=1883
+      - MQTT_DISCOVERY_PREFIX=homeassistant
+      - MQTT_BASE_TOPIC=jbod-monitor
+      - MQTT_PUBLISH_INTERVAL=60
+    # Broker credentials (MQTT_USERNAME/MQTT_PASSWORD). Optional: container
+    # starts without it (MQTT just won't authenticate).
+    env_file:
+      - path: /etc/jbod-monitor/mqtt.env
+        required: false

docker-compose.yml

@@ -1,17 +0,0 @@
-services:
-  jbod-monitor:
-    build: .
-    image: docker.adamksmith.xyz/jbod-monitor:latest
-    container_name: jbod-monitor
-    restart: unless-stopped
-    privileged: true
-    pid: host
-    network_mode: host
-    volumes:
-      - /dev:/dev
-      - /sys:/sys:ro
-      - /run/udev:/run/udev:ro
-    environment:
-      - TZ=America/Denver
-      - UVICORN_LOG_LEVEL=info
-      - ZFS_USE_NSENTER=true

frontend/src/App.jsx

@@ -698,9 +698,20 @@ function HostDrivesCard({ drives, onSelect, t }) {
                   </span>
                 )}
               </div>
-              {d.physical_drives?.map((pd, i) => (
-                <HostDriveRow key={pd.serial || i} d={pd} onSelect={onSelect} t={t} indent />
+              {d.physical_drives?.length > 0 && (
+                <div style={{ marginLeft: 20, borderLeft: `2px solid ${t.cardBorder}`, paddingLeft: 0 }}>
+                  {d.physical_drives.map((pd, i) => (
+                    <div key={pd.serial || i} style={{ display: "flex", alignItems: "center", marginTop: 4 }}>
+                      <div style={{
+                        width: 16, height: 1, background: t.cardBorder, flexShrink: 0,
+                      }} />
+                      <div style={{ flex: 1 }}>
+                        <HostDriveRow d={pd} onSelect={onSelect} t={t} indent />
+                      </div>
+                    </div>
                   ))}
+                </div>
+              )}
             </React.Fragment>
           ))}
         </div>
@@ -709,7 +720,181 @@ function HostDrivesCard({ drives, onSelect, t }) {
   );
 }
 
+function EnclosureHealthSummary({ health, t }) {
+  if (!health) return null;
+
+  const statusColors = {
+    CRITICAL: t.health.error,
+    WARNING: t.health.warning,
+    OK: t.health.healthy,
+  };
+  const sc = statusColors[health.overall_status] || statusColors.OK;
+
+  const failedPsus = health.psus.filter((p) => p.fail || p.status.toLowerCase() === "critical");
+  const failedFans = health.fans.filter((f) => f.fail);
+  const temps = health.temps.filter((s) => s.temperature_c != null);
+  const tempMin = temps.length > 0 ? Math.min(...temps.map((s) => s.temperature_c)) : null;
+  const tempMax = temps.length > 0 ? Math.max(...temps.map((s) => s.temperature_c)) : null;
+
+  return (
+    <div style={{ display: "flex", alignItems: "center", gap: 8, flexWrap: "wrap", marginTop: 6 }}>
+      {/* Overall badge */}
+      <span style={{
+        display: "inline-flex", alignItems: "center", gap: 5,
+        padding: "2px 10px", borderRadius: 99,
+        background: sc.bg, border: `1px solid ${sc.border}`,
+        fontSize: 11, fontWeight: 700, color: sc.text, letterSpacing: 0.3,
+      }}>
+        <span style={{ width: 6, height: 6, borderRadius: "50%", background: sc.dot }} />
+        {health.overall_status}
+      </span>
+
+      {/* PSU pills */}
+      {health.psus.map((psu) => {
+        const bad = psu.fail || psu.status.toLowerCase() === "critical";
+        const pc = bad ? t.health.error : t.health.healthy;
+        return (
+          <span key={psu.index} style={{
+            display: "inline-flex", alignItems: "center", gap: 4,
+            padding: "2px 8px", borderRadius: 99,
+            background: pc.bg, border: `1px solid ${pc.border}`,
+            fontSize: 10, fontWeight: 600, color: pc.text,
+          }}>
+            <span style={{ width: 5, height: 5, borderRadius: "50%", background: pc.dot }} />
+            PSU {psu.index} {bad ? "FAIL" : "OK"}
+          </span>
+        );
+      })}
+
+      {/* Fans summary */}
+      {health.fans.length > 0 && (
+        <span style={{
+          fontSize: 11, color: failedFans.length > 0 ? t.health.error.text : t.textSecondary,
+          fontWeight: 600,
+        }}>
+          {failedFans.length > 0
+            ? `${failedFans.length}/${health.fans.length} fans failed`
+            : `${health.fans.length} fans OK`}
+        </span>
+      )}
+
+      {/* Temp range */}
+      {tempMin != null && (
+        <span style={{
+          fontSize: 11, color: tempMax >= 45 ? t.health.warning.text : t.textSecondary,
+          fontWeight: 600, fontFamily: "'JetBrains Mono', monospace",
+        }}>
+          {tempMin === tempMax ? `${tempMin}\u00B0C` : `${tempMin}\u2013${tempMax}\u00B0C`}
+        </span>
+      )}
+    </div>
+  );
+}
+
+function EnclosureHealthDetail({ health, t }) {
+  if (!health) return null;
+
+  const sectionStyle = { marginBottom: 12 };
+  const headerStyle = {
+    fontSize: 10, fontWeight: 700, color: t.textMuted,
+    textTransform: "uppercase", letterSpacing: 1, marginBottom: 6,
+  };
+  const rowStyle = {
+    display: "flex", justifyContent: "space-between", alignItems: "center",
+    padding: "4px 0", borderBottom: `1px solid ${t.divider}`, fontSize: 12,
+  };
+
+  return (
+    <div style={{
+      padding: "12px 16px", background: t.surface,
+      borderTop: `1px solid ${t.divider}`, borderBottom: `1px solid ${t.divider}`,
+    }}>
+      <div style={{ display: "grid", gridTemplateColumns: "repeat(auto-fit, minmax(220px, 1fr))", gap: 16 }}>
+        {/* PSUs */}
+        {health.psus.length > 0 && (
+          <div style={sectionStyle}>
+            <div style={headerStyle}>Power Supplies</div>
+            {health.psus.map((psu) => {
+              const bad = psu.fail || psu.status.toLowerCase() === "critical";
+              return (
+                <div key={psu.index} style={rowStyle}>
+                  <span style={{ color: t.textSecondary }}>PSU {psu.index}</span>
+                  <span style={{
+                    fontWeight: 600, color: bad ? t.health.error.text : t.health.healthy.text,
+                    fontFamily: "'JetBrains Mono', monospace",
+                  }}>
+                    {psu.status}{psu.ac_fail ? " (AC fail)" : ""}{psu.dc_fail ? " (DC fail)" : ""}
+                  </span>
+                </div>
+              );
+            })}
+          </div>
+        )}
+
+        {/* Fans */}
+        {health.fans.length > 0 && (
+          <div style={sectionStyle}>
+            <div style={headerStyle}>Fans</div>
+            {health.fans.map((fan) => (
+              <div key={fan.index} style={rowStyle}>
+                <span style={{ color: t.textSecondary }}>Fan {fan.index}</span>
+                <span style={{
+                  fontWeight: 600,
+                  color: fan.fail ? t.health.error.text : t.health.healthy.text,
+                  fontFamily: "'JetBrains Mono', monospace",
+                }}>
+                  {fan.rpm != null ? `${fan.rpm} RPM` : fan.status}
+                  {fan.fail ? " FAIL" : ""}
+                </span>
+              </div>
+            ))}
+          </div>
+        )}
+
+        {/* Temps */}
+        {health.temps.length > 0 && (
+          <div style={sectionStyle}>
+            <div style={headerStyle}>Temperature Sensors</div>
+            {health.temps.map((ts) => (
+              <div key={ts.index} style={rowStyle}>
+                <span style={{ color: t.textSecondary }}>Sensor {ts.index}</span>
+                <span style={{
+                  fontWeight: 600,
+                  color: ts.temperature_c >= 45 ? t.health.warning.text : t.text,
+                  fontFamily: "'JetBrains Mono', monospace",
+                }}>
+                  {ts.temperature_c != null ? `${ts.temperature_c}\u00B0C` : ts.status}
+                </span>
+              </div>
+            ))}
+          </div>
+        )}
+
+        {/* Voltages */}
+        {health.voltages.length > 0 && (
+          <div style={sectionStyle}>
+            <div style={headerStyle}>Voltage Rails</div>
+            {health.voltages.map((vs) => (
+              <div key={vs.index} style={rowStyle}>
+                <span style={{ color: t.textSecondary }}>Rail {vs.index}</span>
+                <span style={{
+                  fontWeight: 600, color: t.text,
+                  fontFamily: "'JetBrains Mono', monospace",
+                }}>
+                  {vs.voltage != null ? `${vs.voltage} V` : vs.status}
+                </span>
+              </div>
+            ))}
+          </div>
+        )}
+      </div>
+    </div>
+  );
+}
+
 function EnclosureCard({ enclosure, view, onSelect, selectedSerial, t }) {
+  const [healthExpanded, setHealthExpanded] = useState(false);
+
   return (
     <div style={{
       background: t.cardBg, borderRadius: 16,
@@ -720,7 +905,7 @@ function EnclosureCard({ enclosure, view, onSelect, selectedSerial, t }) {
       <div style={{
         padding: "16px 20px",
         borderBottom: `1px solid ${t.divider}`,
-        display: "flex", alignItems: "center", justifyContent: "space-between",
+        display: "flex", alignItems: "flex-start", justifyContent: "space-between",
         flexWrap: "wrap", gap: 8,
       }}>
         <div>
@@ -730,11 +915,29 @@ function EnclosureCard({ enclosure, view, onSelect, selectedSerial, t }) {
           <div style={{ fontSize: 12, color: t.textSecondary, marginTop: 2 }}>
             {enclosure.sg_device} &middot; {enclosure.populated_slots}/{enclosure.total_slots} slots populated
           </div>
+          {enclosure.health && (
+            <div style={{ display: "flex", alignItems: "center", gap: 6 }}>
+              <EnclosureHealthSummary health={enclosure.health} t={t} />
+              <button
+                onClick={() => setHealthExpanded(!healthExpanded)}
+                style={{
+                  background: "none", border: "none", cursor: "pointer",
+                  fontSize: 11, color: t.accent, fontWeight: 600,
+                  padding: "2px 6px", marginTop: 6,
+                }}
+              >
+                {healthExpanded ? "Hide details" : "Details"}
+              </button>
+            </div>
+          )}
         </div>
         <div style={{ fontSize: 11, color: t.textMuted, fontFamily: "'JetBrains Mono', monospace" }}>
           ID {enclosure.id}
         </div>
       </div>
+      {healthExpanded && enclosure.health && (
+        <EnclosureHealthDetail health={enclosure.health} t={t} />
+      )}
       <div style={{ padding: 16 }}>
         {view === "grid" ? (
           <GridView enclosure={enclosure} onSelect={onSelect} t={t} />

host_poller.py

@@ -0,0 +1,168 @@
+"""Host-poller — server chassis temps, split into two independent loops.
+
+Separate process/container from the SAS poller. Two concurrent loops in one
+process, sharing the single-instance lock AND the per-process hardware gate
+(so IPMI and SMART never hit hardware at the same instant), but on separate
+schedules:
+
+  - env loop   (fast):  IPMI/iDRAC env + CPU coretemp  -> jbod:host_sensors
+  - drive loop (slow):  on-metal drive SMART           -> jbod:host_drives
+
+Keeping SMART on a gentler cadence than the responsive inlet/CPU reads. Each
+loop has its own heartbeat + restart-storm guard. Host-drive ZFS annotation
+reuses jbod:zfs_map written by the SAS poller.
+"""
+import asyncio
+import logging
+import os
+import random
+import socket
+import time
+
+from services import store
+from services.cache import close_cache, init_cache, redis_available
+from services.host import get_host_drives
+from services.host_sensors import read_coretemp, read_hwmon_env, read_ipmi_temps
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger("host-poller")
+
+ENV_INTERVAL = int(os.environ.get("HOST_ENV_INTERVAL", "60"))
+DRIVE_INTERVAL = int(os.environ.get("HOST_DRIVE_INTERVAL", "300"))
+STARTUP_JITTER = float(os.environ.get("POLL_STARTUP_JITTER", "10"))
+LOCK_TTL = 30
+
+TOKEN = f"{socket.gethostname()}-hostpoll-{os.getpid()}"
+
+
+async def sweep_env() -> None:
+    """IPMI/iDRAC environmental + CPU temps (the responsive loop)."""
+    t0 = time.monotonic()
+    errors: list[str] = []
+    try:
+        ipmi = await read_ipmi_temps()
+    except Exception as e:
+        errors.append(f"ipmi:{e}")
+        ipmi = []
+    env = [s for s in ipmi if s.get("kind") == "env"] + read_hwmon_env()
+    cpu = read_coretemp() or [s for s in ipmi if s.get("kind") == "cpu"]
+
+    await store.set_host_sensors({
+        "node": socket.gethostname(),
+        "env": env,
+        "cpu": cpu,
+        "ts": store.now(),
+    })
+    await store.set_meta({
+        "last_sweep_ts": store.now(),
+        "last_sweep_duration": round(time.monotonic() - t0, 1),
+        "env_count": len(env),
+        "cpu_count": len(cpu),
+        "errors": errors[:20],
+        "interval": ENV_INTERVAL,
+    }, key=store.K_HOST_ENV_META)
+    logger.info("env sweep: %d env, %d cpu, %d error(s)", len(env), len(cpu), len(errors))
+
+
+async def sweep_drives() -> None:
+    """On-metal (non-enclosure) drive SMART (the gentle loop)."""
+    t0 = time.monotonic()
+    errors: list[str] = []
+    drives: list = []
+    try:
+        pool_map = await store.get_zfs_map()
+        drives = await get_host_drives(pool_map)
+        await store.set_host_drives(drives)
+    except Exception as e:
+        errors.append(f"hostdrives:{e}")
+    await store.set_meta({
+        "last_sweep_ts": store.now(),
+        "last_sweep_duration": round(time.monotonic() - t0, 1),
+        "host_drive_count": len(drives),
+        "errors": errors[:20],
+        "interval": DRIVE_INTERVAL,
+    }, key=store.K_HOST_DRIVE_META)
+    logger.info("drive sweep: %d drives, %.1fs, %d error(s)",
+                len(drives), round(time.monotonic() - t0, 1), len(errors))
+
+
+async def run_loop(name: str, fn, interval: int, meta_key: str) -> None:
+    """Run `fn` every `interval`s, with a restart-storm guard from its meta."""
+    meta = await store.get_meta(key=meta_key)
+    if meta and meta.get("last_sweep_ts"):
+        since = store.now() - meta["last_sweep_ts"]
+        if 0 <= since < interval:
+            wait = interval - since
+            logger.info("%s: last sweep %.0fs ago — deferring first %.0fs", name, since, wait)
+            await asyncio.sleep(wait)
+    while True:
+        t0 = time.monotonic()
+        try:
+            await fn()
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            logger.error("%s loop error: %s", name, e)
+        await asyncio.sleep(max(5, interval - (time.monotonic() - t0)))
+
+
+async def lock_refresher() -> None:
+    while True:
+        await asyncio.sleep(LOCK_TTL / 3)
+        try:
+            ok = await store.refresh_lock(TOKEN, LOCK_TTL, key=store.K_HOST_LOCK)
+        except Exception as e:
+            logger.error("lock refresh error (%s) — exiting", e)
+            os._exit(1)
+        if not ok:
+            logger.error("lost host-poller lock — exiting")
+            os._exit(1)
+
+
+def _critical_task_done(task: asyncio.Task) -> None:
+    if task.cancelled():
+        return
+    logger.error("critical task ended unexpectedly — exiting")
+    os._exit(1)
+
+
+async def main() -> None:
+    await init_cache()
+    while not redis_available():
+        logger.warning("Redis unavailable — host-poller needs Redis; retrying in 5s")
+        await asyncio.sleep(5)
+        await init_cache()
+
+    while not await store.acquire_lock(TOKEN, LOCK_TTL, key=store.K_HOST_LOCK):
+        logger.warning("another host-poller holds the lock; waiting %ds", LOCK_TTL // 2)
+        await asyncio.sleep(LOCK_TTL // 2)
+    logger.info("host-poller lock acquired (%s)", TOKEN)
+
+    refresher = asyncio.create_task(lock_refresher())
+    refresher.add_done_callback(_critical_task_done)
+
+    if os.geteuid() != 0:
+        logger.warning("not running as root — ipmitool/smartctl may fail")
+
+    await asyncio.sleep(random.uniform(0, STARTUP_JITTER))
+
+    env_task = asyncio.create_task(
+        run_loop("env", sweep_env, ENV_INTERVAL, store.K_HOST_ENV_META))
+    drive_task = asyncio.create_task(
+        run_loop("drive", sweep_drives, DRIVE_INTERVAL, store.K_HOST_DRIVE_META))
+    try:
+        await asyncio.gather(env_task, drive_task)
+    finally:
+        for t in (refresher, env_task, drive_task):
+            t.cancel()
+        await close_cache()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass

k8s/README.md

@@ -0,0 +1,66 @@
+# jbod-monitor on RKE2 (DaemonSet)
+
+Per-node self-contained deployment for Kubernetes worker nodes that can't run
+the docker-compose stack. Each opted-in node runs one pod (`host-poller` +
+`redis` + `web`) and publishes its own Home Assistant device
+(`JBOD Monitor (<node>)`) over MQTT.
+
+This mirrors the docker-compose stack with **no code changes** — each node is
+independent, exactly like the standalone hosts.
+
+## What it collects
+
+- **host-poller** (privileged, hostPath `/dev`+`/sys`): IPMI/iDRAC env, CPU
+  coretemp, on-metal drive SMART → node-local redis.
+- **web**: reads redis, publishes per-node MQTT discovery to the broker.
+- No SAS poller (no SES enclosures on these nodes). Consequence: host-drive
+  entities have no ZFS pool label (no zfs_map producer).
+
+## Prereqs — verified 2026-06-16
+
+1. **Registry pull**: registry requires auth. `registry-cred-vaultstaticsecret.yaml`
+   syncs `secret/registry/nexus` → `docker-registry-cred` (dockerconfigjson),
+   referenced as `imagePullSecrets` in the DaemonSet. (Mirrors the cluster's
+   existing `*-registry-cred` pattern.)
+2. **VSO**: v1.3.0; `openbao-kubernetes` VaultAuth healthy; `transformation.templates`
+   supported. No existing VSS in the cluster uses transformation, so watch the
+   `jbod-mqtt` VSS status on first apply for template errors.
+3. **MQTT egress**: confirmed — a cluster pod reached `10.5.30.3:1883`.
+4. **Nodes**: pascal/currie/fermi are untainted → no tolerations needed.
+
+## Apply order
+
+```bash
+# 1. Label ONLY the intended nodes (do not blanket-label all workers):
+kubectl label node usco-dc-pascal usco-dc-currie usco-dc-fermi jbod-monitor=enabled
+#    (canary: label just usco-dc-fermi first)
+
+# 2. Namespace + DaemonSet:
+kubectl apply -f k8s/daemonset.yaml
+
+# 3. Secrets (VSO → docker-registry-cred + jbod-mqtt). Apply, then confirm sync:
+kubectl apply -f k8s/registry-cred-vaultstaticsecret.yaml
+kubectl apply -f k8s/mqtt-vaultstaticsecret.yaml
+kubectl -n jbod-monitor get vaultstaticsecret           # SYNCED=True for both
+kubectl -n jbod-monitor get secret jbod-mqtt -o jsonpath='{.data.MQTT_USERNAME}' | base64 -d
+
+# 4. Roll the DaemonSet so pods pick up the secrets if they raced ahead:
+kubectl -n jbod-monitor rollout restart daemonset/jbod-monitor
+```
+
+## Verify
+
+```bash
+kubectl -n jbod-monitor get pods -o wide        # one per labeled node
+kubectl -n jbod-monitor logs <pod> -c host-poller | grep sweep
+kubectl -n jbod-monitor logs <pod> -c web | grep -i mqtt   # "MQTT connected"
+```
+
+Then check Home Assistant for `JBOD Monitor (usco-dc-pascal|currie|fermi)`.
+
+## Notes
+
+- pascal (R620) already has an iDRAC "System Board Inlet Temp" in HA via another
+  integration — its env temp will partially duplicate; drives + CPU are net-new.
+- Pin image tags by SHA (already done: host-poller `514502e`, web `1839e9d`).
+- Removal: `kubectl delete -f k8s/daemonset.yaml` and unlabel the nodes.

k8s/daemonset.yaml

@@ -0,0 +1,103 @@
+# jbod-monitor on RKE2 nodes — per-node self-contained DaemonSet (Model A).
+#
+# Each scheduled node runs one pod = host-poller + redis + web, sharing the
+# pod's localhost (no hostNetwork needed). Only host-poller is privileged and
+# mounts /dev + /sys to read IPMI / SMART / coretemp. web egresses to the MQTT
+# broker and publishes a per-node HA device (MQTT_NODE_ID = spec.nodeName).
+#
+# No SAS poller (these nodes have no SES enclosures). NOTE: without it there is
+# no zfs_map, so host-drive entities won't carry a ZFS pool label.
+#
+# Target nodes are chosen by label — see k8s/README.md (label only
+# pascal/currie/fermi). Apply order: namespace -> VaultStaticSecret -> this.
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: jbod-monitor
+---
+apiVersion: apps/v1
+kind: DaemonSet
+metadata:
+  name: jbod-monitor
+  namespace: jbod-monitor
+  labels:
+    app: jbod-monitor
+spec:
+  selector:
+    matchLabels:
+      app: jbod-monitor
+  template:
+    metadata:
+      labels:
+        app: jbod-monitor
+    spec:
+      # Only nodes explicitly opted in (see README: kubectl label node ...).
+      nodeSelector:
+        jbod-monitor: "enabled"
+      imagePullSecrets:
+        - name: docker-registry-cred   # synced by registry-cred-vaultstaticsecret.yaml
+      containers:
+        - name: redis
+          image: redis:7-alpine
+          args: ["redis-server", "--save", "60", "1", "--loglevel", "warning"]
+          volumeMounts:
+            - name: redis-data
+              mountPath: /data
+          resources:
+            requests: {cpu: 25m, memory: 32Mi}
+            limits: {memory: 128Mi}
+
+        - name: host-poller
+          image: docker.adamksmith.xyz/jbod-host-poller:514502e
+          securityContext:
+            privileged: true   # raw /dev access for ipmitool + smartctl
+          env:
+            - {name: TZ, value: "America/Denver"}
+            - {name: REDIS_HOST, value: "127.0.0.1"}
+            - {name: REDIS_PORT, value: "6379"}
+            - {name: HOST_ENV_INTERVAL, value: "60"}
+            - {name: HOST_DRIVE_INTERVAL, value: "300"}
+            - {name: POLL_CONCURRENCY, value: "1"}
+            - {name: POLL_DRIVE_GAP, value: "0.5"}
+            - {name: POLL_STARTUP_JITTER, value: "10"}
+            - {name: POLL_DATA_TTL, value: "900"}
+          volumeMounts:
+            - {name: dev, mountPath: /dev}
+            - {name: sys, mountPath: /sys, readOnly: true}
+            - {name: udev, mountPath: /run/udev, readOnly: true}
+          resources:
+            requests: {cpu: 25m, memory: 64Mi}
+            limits: {memory: 256Mi}
+
+        - name: web
+          image: docker.adamksmith.xyz/jbod-web:1839e9d
+          env:
+            - {name: TZ, value: "America/Denver"}
+            - {name: REDIS_HOST, value: "127.0.0.1"}
+            - {name: REDIS_PORT, value: "6379"}
+            - {name: MQTT_HOST, value: "10.5.30.3"}
+            - {name: MQTT_PORT, value: "1883"}
+            - {name: MQTT_DISCOVERY_PREFIX, value: "homeassistant"}
+            - {name: MQTT_BASE_TOPIC, value: "jbod-monitor"}
+            - {name: MQTT_PUBLISH_INTERVAL, value: "60"}
+            - name: MQTT_NODE_ID
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
+          envFrom:
+            - secretRef:
+                name: jbod-mqtt   # MQTT_USERNAME / MQTT_PASSWORD (see VaultStaticSecret)
+          resources:
+            requests: {cpu: 25m, memory: 64Mi}
+            limits: {memory: 256Mi}
+
+      volumes:
+        - name: redis-data
+          emptyDir: {}
+        - name: dev
+          hostPath: {path: /dev}
+        - name: sys
+          hostPath: {path: /sys}
+        - name: udev
+          hostPath: {path: /run/udev}

k8s/mqtt-vaultstaticsecret.yaml

@@ -0,0 +1,35 @@
+# MQTT broker creds for the web container, synced from OpenBao by VSO.
+#
+# Reads secret/home_assistant (keys mqtt_user / mqtt_pass) and renders a k8s
+# Secret `jbod-mqtt` with MQTT_USERNAME / MQTT_PASSWORD keys (consumed via
+# envFrom in the DaemonSet). VSO's `vso-read` policy can read secret/data/*,
+# so unlike the claude-read token it CAN read home_assistant — no creds ever
+# touch these manifests.
+#
+# Verify the transformation syntax against the installed VSO version.
+apiVersion: secrets.hashicorp.com/v1beta1
+kind: VaultStaticSecret
+metadata:
+  name: jbod-mqtt
+  namespace: jbod-monitor
+spec:
+  vaultAuthRef: vault-secrets-operator-system/openbao-kubernetes
+  mount: secret
+  type: kv-v2
+  path: home_assistant
+  refreshAfter: 1h
+  destination:
+    name: jbod-mqtt
+    create: true
+    transformation:
+      excludeRaw: true
+      # Drop ALL passthrough source keys (api_key, vantage_*, etc.) — keep only
+      # the templated MQTT_USERNAME/MQTT_PASSWORD below. Without this, envFrom
+      # would inject the entire home_assistant secret into the web container.
+      excludes:
+        - ".*"
+      templates:
+        MQTT_USERNAME:
+          text: '{{ .Secrets.mqtt_user }}'
+        MQTT_PASSWORD:
+          text: '{{ .Secrets.mqtt_pass }}'

k8s/registry-cred-vaultstaticsecret.yaml

@@ -0,0 +1,18 @@
+# Image pull secret for docker.adamksmith.xyz, synced by VSO from OpenBao.
+# Mirrors the cluster's existing *-registry-cred pattern (secret/registry/nexus
+# -> dockerconfigjson). Referenced as imagePullSecrets in the DaemonSet.
+apiVersion: secrets.hashicorp.com/v1beta1
+kind: VaultStaticSecret
+metadata:
+  name: docker-registry-cred
+  namespace: jbod-monitor
+spec:
+  vaultAuthRef: vault-secrets-operator-system/openbao-kubernetes
+  mount: secret
+  path: registry/nexus
+  type: kv-v2
+  refreshAfter: 1h
+  destination:
+    name: docker-registry-cred
+    create: true
+    type: kubernetes.io/dockerconfigjson

main.py

@@ -1,15 +1,18 @@
+import asyncio
 import logging
 import os
+from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 
 from models.schemas import HealthCheck
-from routers import drives, enclosures, leds, overview
-from services.smart import sg_ses_available, smartctl_available
+from routers import drives, enclosures, leds, overview, temps
+from services import store
+from services.cache import close_cache, init_cache, redis_available
+from services.mqtt_publisher import MqttPublisher, _PAHO_AVAILABLE, mqtt_enabled
 
 logging.basicConfig(
     level=logging.INFO,
@@ -17,10 +20,52 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
+_tool_status: dict[str, bool] = {}
+_mqtt_task: asyncio.Task | None = None
+_mqtt_publisher: MqttPublisher | None = None
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """API/web + MQTT consumer. Reads everything from Redis; the dedicated
+    poller process is the only thing that touches hardware."""
+    global _mqtt_task, _mqtt_publisher
+
+    await init_cache()
+    _tool_status["redis"] = redis_available()
+
+    # Optional MQTT publisher for Home Assistant (opt-in via MQTT_HOST).
+    _tool_status["mqtt"] = False
+    if mqtt_enabled():
+        if not _PAHO_AVAILABLE:
+            logger.warning("MQTT_HOST set but paho-mqtt not installed — MQTT disabled")
+        else:
+            try:
+                _mqtt_publisher = MqttPublisher()
+                _mqtt_publisher.start()
+                _mqtt_task = asyncio.create_task(_mqtt_publisher.run())
+                _tool_status["mqtt"] = True
+            except Exception as e:
+                logger.error("Failed to start MQTT publisher: %s", e)
+
+    yield
+
+    if _mqtt_task is not None:
+        _mqtt_task.cancel()
+        try:
+            await _mqtt_task
+        except asyncio.CancelledError:
+            pass
+    if _mqtt_publisher is not None:
+        _mqtt_publisher.stop()
+    await close_cache()
+
+
 app = FastAPI(
     title="JBOD Monitor",
     description="Drive health monitoring for JBOD enclosures",
-    version="0.1.0",
+    version="0.2.0",
+    lifespan=lifespan,
 )
 
 app.add_middleware(
@@ -34,26 +79,23 @@ app.include_router(enclosures.router)
 app.include_router(drives.router)
 app.include_router(leds.router)
 app.include_router(overview.router)
-
-_tool_status: dict[str, bool] = {}
-
-
-@app.on_event("startup")
-async def check_dependencies():
-    _tool_status["smartctl"] = smartctl_available()
-    _tool_status["sg_ses"] = sg_ses_available()
-
-    if not _tool_status["smartctl"]:
-        logger.warning("smartctl not found — install smartmontools for SMART data")
-    if not _tool_status["sg_ses"]:
-        logger.warning("sg_ses not found — install sg3-utils for enclosure SES data")
-    if os.geteuid() != 0:
-        logger.warning("Not running as root — smartctl may fail on some devices")
+app.include_router(temps.router)
 
 
 @app.get("/api/health", response_model=HealthCheck, tags=["health"])
 async def health():
-    return HealthCheck(status="ok", tools=_tool_status)
+    tools = dict(_tool_status)
+    tools["redis"] = redis_available()
+
+    # Poller freshness: is the producer keeping the store warm?
+    meta = await store.get_meta()
+    if meta and meta.get("last_sweep_ts"):
+        age = store.now() - meta["last_sweep_ts"]
+        tools["poller_fresh"] = age < meta.get("sweep_interval", 300) * 3
+    else:
+        tools["poller_fresh"] = False
+
+    return HealthCheck(status="ok", tools=tools)
 
 
 # Serve built frontend static files — mounted last so /api routes take priority.

models/schemas.py

@@ -65,6 +65,45 @@ class SlotWithDrive(BaseModel):
     drive: DriveHealthSummary | None = None
 
 
+class PsuStatus(BaseModel):
+    index: int
+    name: str | None = None
+    status: str
+    fail: bool = False
+    ac_fail: bool = False
+    dc_fail: bool = False
+
+
+class FanStatus(BaseModel):
+    index: int
+    name: str | None = None
+    status: str
+    rpm: int | None = None
+    fail: bool = False
+
+
+class TempSensor(BaseModel):
+    index: int
+    name: str | None = None
+    status: str
+    temperature_c: float | None = None
+
+
+class VoltageSensor(BaseModel):
+    index: int
+    name: str | None = None
+    status: str
+    voltage: float | None = None
+
+
+class EnclosureHealth(BaseModel):
+    overall_status: str = "OK"
+    psus: list[PsuStatus] = []
+    fans: list[FanStatus] = []
+    temps: list[TempSensor] = []
+    voltages: list[VoltageSensor] = []
+
+
 class EnclosureWithDrives(BaseModel):
     id: str
     sg_device: str | None = None
@@ -74,6 +113,7 @@ class EnclosureWithDrives(BaseModel):
     total_slots: int
     populated_slots: int
     slots: list[SlotWithDrive]
+    health: EnclosureHealth | None = None
 
 
 class HostDrive(BaseModel):
@@ -111,3 +151,25 @@ class Overview(BaseModel):
 class HealthCheck(BaseModel):
     status: str
     tools: dict[str, bool]
+
+
+class EnclosureTempSensor(BaseModel):
+    index: int
+    name: str | None = None
+    status: str
+    temperature_c: float | None = None
+
+
+class EnclosureTemps(BaseModel):
+    id: str
+    vendor: str
+    model: str
+    hotspot_c: float | None = None
+    drive_max_c: float | None = None
+    drive_temp_count: int = 0
+    sensors: list[EnclosureTempSensor] = []
+
+
+class TempSnapshot(BaseModel):
+    node: str
+    enclosures: list[EnclosureTemps] = []

poller.py

@@ -0,0 +1,212 @@
+"""Dedicated hardware poller — the sole owner of SAS/SMART/SES I/O.
+
+Runs as its own (privileged, pid:host) container. Everything that touches
+the bus happens here, serialized behind the hardware gate and paced, then
+written to Redis. The API and MQTT services are pure Redis readers and
+never issue a hardware command — so they can crash-loop freely without
+ever storming the backplanes.
+
+Design points that matter for fragile SAS expanders:
+  - one global semaphore (POLL_CONCURRENCY=1) serializes all hardware I/O
+  - a gap (POLL_DRIVE_GAP) between each drive keeps the bus mostly quiet
+  - startup jitter avoids a thundering herd on (re)start
+  - a single-instance Redis lock means two pollers can't both poll
+  - last-good results stay in Redis; a failed sweep never blanks them
+"""
+import asyncio
+import logging
+import os
+import random
+import socket
+import time
+
+from services import store
+from services.cache import close_cache, init_cache, redis_available
+from services.enclosure import discover_enclosures, get_enclosure_status, list_slots
+from services.leds import run_led
+from services.smart import _run_smartctl, sg_ses_available, smartctl_available
+from services.zfs import get_zfs_pool_map
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger("poller")
+
+SWEEP_INTERVAL = int(os.environ.get("POLL_SWEEP_INTERVAL", "300"))
+STARTUP_JITTER = float(os.environ.get("POLL_STARTUP_JITTER", "10"))
+LOCK_TTL = 30
+# Pacing between hardware ops (POLL_DRIVE_GAP) is enforced centrally by the
+# hardware gate, so every op — SMART, SES, host, MegaRAID, ledctl — is spaced.
+
+TOKEN = f"{socket.gethostname()}-{os.getpid()}"
+
+
+async def sweep() -> None:
+    """One full, paced pass over all hardware → Redis."""
+    t0 = time.monotonic()
+    errors: list[str] = []
+
+    # 1. Discover enclosures + slots (sysfs, no bus I/O).
+    enclosures = discover_enclosures()
+    inv_enclosures = []
+    drive_devices: list[str] = []
+    for enc in enclosures:
+        slots = list_slots(enc["id"])
+        inv_enclosures.append({**enc, "slots": slots})
+        for s in slots:
+            if s["device"]:
+                drive_devices.append(s["device"])
+
+    # 2. ZFS topology (gated).
+    pool_map = await store.get_zfs_map()
+    try:
+        pool_map = await get_zfs_pool_map()
+        await store.set_zfs_map(pool_map)
+    except Exception as e:
+        errors.append(f"zfs:{e}")
+
+    # 3. Per-drive SMART (gate serializes + paces each call).
+    for dev in drive_devices:
+        try:
+            data = await _run_smartctl(dev)
+            await store.set_smart(dev, data)
+        except Exception as e:
+            errors.append(f"smart:{dev}:{e}")
+
+    # 4. Per-enclosure SES status (gate serializes + paces each call).
+    for enc in enclosures:
+        if not enc.get("sg_device"):
+            continue
+        try:
+            ses = await get_enclosure_status(enc["sg_device"])
+            if ses is not None:
+                await store.set_ses(enc["id"], ses)
+        except Exception as e:
+            errors.append(f"ses:{enc['id']}:{e}")
+
+    # NOTE: host/on-metal drives are owned by the separate host-poller now.
+
+    # 6. Inventory + heartbeat.
+    await store.set_inventory({"enclosures": inv_enclosures, "ts": store.now()})
+    duration = round(time.monotonic() - t0, 1)
+    await store.set_meta({
+        "last_sweep_ts": store.now(),
+        "last_sweep_duration": duration,
+        "drive_count": len(drive_devices),
+        "enclosure_count": len(enclosures),
+        "errors": errors[:20],
+        "sweep_interval": SWEEP_INTERVAL,
+    })
+    logger.info(
+        "sweep done: %d drives, %d enclosures, %.1fs, %d error(s)",
+        len(drive_devices), len(enclosures), duration, len(errors),
+    )
+
+
+async def led_consumer() -> None:
+    """Drain locate-LED requests from Redis and run them (gated)."""
+    while True:
+        try:
+            cmd = await store.pop_led(timeout=5)
+            if cmd:
+                try:
+                    await run_led(cmd["device"], cmd["state"])
+                    logger.info("LED %s -> %s", cmd.get("device"), cmd.get("state"))
+                except Exception as e:
+                    logger.warning("LED command failed %s: %s", cmd, e)
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            logger.warning("LED consumer error: %s", e)
+            await asyncio.sleep(2)
+
+
+async def lock_refresher() -> None:
+    while True:
+        await asyncio.sleep(LOCK_TTL / 3)
+        # Any failure to confirm we still hold the lock is fatal — keeping the
+        # lock alive is what prevents a second poller from sweeping concurrently.
+        try:
+            ok = await store.refresh_lock(TOKEN, LOCK_TTL)
+        except Exception as e:
+            logger.error("lock refresh error (%s) — exiting to avoid double-polling", e)
+            os._exit(1)
+        if not ok:
+            logger.error("lost poller lock — exiting to avoid double-polling")
+            os._exit(1)
+
+
+def _critical_task_done(task: asyncio.Task) -> None:
+    """Fail-safe: if a critical task ends for any reason other than a clean
+    cancel (shutdown), exit so we never run unguarded."""
+    if task.cancelled():
+        return
+    logger.error("critical task ended unexpectedly — exiting")
+    os._exit(1)
+
+
+async def main() -> None:
+    await init_cache()
+    while not redis_available():
+        logger.warning("Redis unavailable — poller needs Redis; retrying in 5s")
+        await asyncio.sleep(5)
+        await init_cache()
+
+    # Single-instance guard.
+    while not await store.acquire_lock(TOKEN, LOCK_TTL):
+        logger.warning("another poller holds the lock; waiting %ds", LOCK_TTL // 2)
+        await asyncio.sleep(LOCK_TTL // 2)
+    logger.info("poller lock acquired (%s)", TOKEN)
+
+    # Start the refresher IMMEDIATELY — before any pre-sweep sleep. The jitter
+    # and restart-storm deferral below can together exceed LOCK_TTL, so without
+    # an active refresher the lock would expire and a second poller could begin
+    # sweeping concurrently (a hardware-storm risk).
+    refresher = asyncio.create_task(lock_refresher())
+    refresher.add_done_callback(_critical_task_done)
+
+    if not smartctl_available():
+        logger.warning("smartctl not found — install smartmontools")
+    if not sg_ses_available():
+        logger.warning("sg_ses not found — install sg3-utils")
+    if os.geteuid() != 0:
+        logger.warning("not running as root — smartctl/sg_ses may fail")
+
+    # Stagger startup so a restart doesn't immediately storm the bus.
+    jitter = random.uniform(0, STARTUP_JITTER)
+    logger.info("startup jitter %.1fs", jitter)
+    await asyncio.sleep(jitter)
+
+    # Restart-storm guard: if a sweep ran recently (persisted in Redis across
+    # restarts), wait out the remainder of the interval instead of immediately
+    # re-sweeping. This is the key protection against deploy-cycling storms.
+    meta = await store.get_meta()
+    if meta and meta.get("last_sweep_ts"):
+        since = store.now() - meta["last_sweep_ts"]
+        if 0 <= since < SWEEP_INTERVAL:
+            wait = SWEEP_INTERVAL - since
+            logger.info("last sweep %.0fs ago — deferring first sweep %.0fs", since, wait)
+            await asyncio.sleep(wait)
+
+    leds = asyncio.create_task(led_consumer())
+    try:
+        while True:
+            t0 = time.monotonic()
+            try:
+                await sweep()
+            except Exception as e:
+                logger.error("sweep error: %s", e)
+            elapsed = time.monotonic() - t0
+            await asyncio.sleep(max(5, SWEEP_INTERVAL - elapsed))
+    finally:
+        refresher.cancel()
+        leds.cancel()
+        await close_cache()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass

requirements-poller.txt

@@ -0,0 +1,2 @@
+# Poller image — hardware producer. Only needs Redis; no web stack.
+redis>=5.0.0

requirements-web.txt

@@ -0,0 +1,6 @@
+# Web/API image — read-only consumer + MQTT. No hardware tools needed.
+fastapi>=0.115.0
+uvicorn>=0.34.0
+pydantic>=2.10.0
+redis>=5.0.0
+paho-mqtt>=2.1.0

requirements.txt

@@ -1,3 +1,5 @@
 fastapi>=0.115.0
 uvicorn>=0.34.0
 pydantic>=2.10.0
+redis>=5.0.0
+paho-mqtt>=2.1.0

routers/drives.py

@@ -1,28 +1,37 @@
-from fastapi import APIRouter, HTTPException
+import re
+
+from fastapi import APIRouter, HTTPException, Response
 
 from models.schemas import DriveDetail
-from services.smart import get_smart_data
-from services.zfs import get_zfs_pool_map
+from services import store
 
 router = APIRouter(prefix="/api/drives", tags=["drives"])
 
 
 @router.get("/{device}", response_model=DriveDetail)
-async def get_drive_detail(device: str):
-    """Get SMART detail for a specific block device."""
-    try:
-        data = await get_smart_data(device)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
+async def get_drive_detail(device: str, response: Response):
+    """Get SMART detail for a specific block device (from the store)."""
+    if not re.match(r"^[a-zA-Z0-9\-]+$", device):
+        raise HTTPException(status_code=400, detail=f"Invalid device name: {device}")
 
+    data = await store.get_smart(device)
+    if data is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No SMART data for '{device}' yet (poller may not have swept it)",
+        )
     if "error" in data:
         raise HTTPException(status_code=502, detail=data["error"])
 
-    pool_map = await get_zfs_pool_map()
+    pool_map = await store.get_zfs_map()
     zfs_info = pool_map.get(device)
     if zfs_info:
         data["zfs_pool"] = zfs_info["pool"]
         data["zfs_vdev"] = zfs_info["vdev"]
         data["zfs_state"] = zfs_info.get("state")
 
+    meta = await store.get_meta()
+    if meta and meta.get("last_sweep_ts"):
+        response.headers["X-Poll-Age"] = str(int(store.now() - meta["last_sweep_ts"]))
+
     return DriveDetail(**data)

routers/enclosures.py

@@ -1,24 +1,23 @@
 from fastapi import APIRouter, HTTPException
 
 from models.schemas import Enclosure, SlotInfo
-from services.enclosure import discover_enclosures, list_slots
+from services import store
 
 router = APIRouter(prefix="/api/enclosures", tags=["enclosures"])
 
 
 @router.get("", response_model=list[Enclosure])
 async def get_enclosures():
-    """Discover all SES enclosures."""
-    return discover_enclosures()
+    """List all discovered SES enclosures (from the poller inventory)."""
+    inventory = await store.get_inventory()
+    return [Enclosure(**e) for e in inventory.get("enclosures", [])]
 
 
 @router.get("/{enclosure_id}/drives", response_model=list[SlotInfo])
 async def get_enclosure_drives(enclosure_id: str):
-    """List all drive slots for an enclosure."""
-    slots = list_slots(enclosure_id)
-    if not slots:
-        # Check if the enclosure exists at all
-        enclosures = discover_enclosures()
-        if not any(e["id"] == enclosure_id for e in enclosures):
+    """List all drive slots for an enclosure (from the poller inventory)."""
+    inventory = await store.get_inventory()
+    for enc in inventory.get("enclosures", []):
+        if enc["id"] == enclosure_id:
+            return [SlotInfo(**s) for s in enc.get("slots", [])]
     raise HTTPException(status_code=404, detail=f"Enclosure '{enclosure_id}' not found")
-    return slots

routers/leds.py

@@ -1,7 +1,9 @@
+import re
+
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, field_validator
 
-from services.leds import set_led
+from services import store
 
 router = APIRouter(prefix="/api/drives", tags=["leds"])
 
@@ -19,11 +21,12 @@ class LedRequest(BaseModel):
 
 @router.post("/{device}/led")
 async def set_drive_led(device: str, body: LedRequest):
-    """Set the locate LED for a drive."""
-    try:
-        result = await set_led(device, body.state)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-    except RuntimeError as e:
-        raise HTTPException(status_code=502, detail=str(e))
-    return result
+    """Queue a locate-LED change. The poller (sole hardware owner) executes it."""
+    if not re.fullmatch(r"[a-zA-Z0-9]+", device):
+        raise HTTPException(status_code=400, detail=f"Invalid device name: {device}")
+
+    queued = await store.enqueue_led(device, body.state)
+    if not queued:
+        raise HTTPException(status_code=503, detail="LED queue unavailable (Redis down)")
+
+    return {"device": device, "state": body.state, "queued": True}

routers/overview.py

@@ -1,19 +1,17 @@
-import asyncio
 import logging
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Response
 
 from models.schemas import (
     DriveHealthSummary,
+    EnclosureHealth,
     EnclosureWithDrives,
     HostDrive,
     Overview,
     SlotWithDrive,
 )
-from services.enclosure import discover_enclosures, list_slots
-from services.host import get_host_drives
-from services.smart import get_smart_data
-from services.zfs import get_zfs_pool_map
+from services import store
+from services.health import compute_health_status
 
 logger = logging.getLogger(__name__)
 
@@ -21,10 +19,14 @@ router = APIRouter(prefix="/api/overview", tags=["overview"])
 
 
 @router.get("", response_model=Overview)
-async def get_overview():
-    """Aggregate view of all enclosures, slots, and drive health."""
-    enclosures_raw = discover_enclosures()
-    pool_map = await get_zfs_pool_map()
+async def get_overview(response: Response):
+    """Aggregate view of all enclosures, slots, and drive health.
+
+    Pure read from the store — the poller is the only thing that touched
+    hardware to produce this data.
+    """
+    inventory = await store.get_inventory()
+    pool_map = await store.get_zfs_map()
 
     enc_results: list[EnclosureWithDrives] = []
     total_drives = 0
@@ -32,99 +34,78 @@ async def get_overview():
     errors = 0
     all_healthy = True
 
-    for enc in enclosures_raw:
-        slots_raw = list_slots(enc["id"])
-
-        # Gather SMART data for all populated slots concurrently
-        populated = [(s, s["device"]) for s in slots_raw if s["populated"] and s["device"]]
-        smart_tasks = [get_smart_data(dev) for _, dev in populated]
-        smart_results = await asyncio.gather(*smart_tasks, return_exceptions=True)
-
-        smart_map: dict[str, dict] = {}
-        for (slot_info, dev), result in zip(populated, smart_results):
-            if isinstance(result, Exception):
-                logger.warning("SMART query failed for %s: %s", dev, result)
-                smart_map[dev] = {"device": dev, "smart_supported": False}
-            else:
-                smart_map[dev] = result
-
+    for enc in inventory.get("enclosures", []):
         slots_out: list[SlotWithDrive] = []
-        for s in slots_raw:
+        for s in enc.get("slots", []):
+            dev = s.get("device")
             drive_summary = None
-            if s["device"] and s["device"] in smart_map:
-                sd = smart_map[s["device"]]
-                total_drives += 1
 
-                healthy = sd.get("smart_healthy")
-                if healthy is False:
+            if dev:
+                total_drives += 1
+                sd = await store.get_smart(dev)
+                if sd:
+                    status = compute_health_status(sd)
+                    if status == "error":
                         errors += 1
                         all_healthy = False
-                elif healthy is None and sd.get("smart_supported", True):
+                    elif status == "warning":
                         warnings += 1
 
-                # Check for concerning SMART values
-                if sd.get("reallocated_sectors") and sd["reallocated_sectors"] > 0:
-                    warnings += 1
-                if sd.get("pending_sectors") and sd["pending_sectors"] > 0:
-                    warnings += 1
-                if sd.get("uncorrectable_errors") and sd["uncorrectable_errors"] > 0:
-                    warnings += 1
-
-                # Compute health_status for frontend
-                realloc = sd.get("reallocated_sectors") or 0
-                pending = sd.get("pending_sectors") or 0
-                unc = sd.get("uncorrectable_errors") or 0
-                if healthy is False:
-                    health_status = "error"
-                elif realloc > 0 or pending > 0 or unc > 0 or (healthy is None and sd.get("smart_supported", True)):
-                    health_status = "warning"
-                else:
-                    health_status = "healthy"
-
+                    zinfo = pool_map.get(dev, {})
                     drive_summary = DriveHealthSummary(
-                    device=sd["device"],
+                        device=sd.get("device", dev),
                         model=sd.get("model"),
                         serial=sd.get("serial"),
                         wwn=sd.get("wwn"),
                         firmware=sd.get("firmware"),
                         capacity_bytes=sd.get("capacity_bytes"),
-                    smart_healthy=healthy,
+                        smart_healthy=sd.get("smart_healthy"),
                         smart_supported=sd.get("smart_supported", True),
                         temperature_c=sd.get("temperature_c"),
                         power_on_hours=sd.get("power_on_hours"),
                         reallocated_sectors=sd.get("reallocated_sectors"),
                         pending_sectors=sd.get("pending_sectors"),
                         uncorrectable_errors=sd.get("uncorrectable_errors"),
-                    zfs_pool=pool_map.get(sd["device"], {}).get("pool"),
-                    zfs_vdev=pool_map.get(sd["device"], {}).get("vdev"),
-                    zfs_state=pool_map.get(sd["device"], {}).get("state"),
-                    health_status=health_status,
+                        zfs_pool=zinfo.get("pool"),
+                        zfs_vdev=zinfo.get("vdev"),
+                        zfs_state=zinfo.get("state"),
+                        health_status=status,
                     )
-            elif s["populated"]:
+            elif s.get("populated"):
                 total_drives += 1
 
             slots_out.append(SlotWithDrive(
                 slot=s["slot"],
                 populated=s["populated"],
-                device=s["device"],
+                device=dev,
                 drive=drive_summary,
             ))
 
+        ses = await store.get_ses(enc["id"])
+        enc_health = None
+        if ses:
+            enc_health = EnclosureHealth(**ses)
+            if enc_health.overall_status == "CRITICAL":
+                errors += 1
+                all_healthy = False
+            elif enc_health.overall_status == "WARNING":
+                warnings += 1
+
         enc_results.append(EnclosureWithDrives(
             id=enc["id"],
             sg_device=enc.get("sg_device"),
-            vendor=enc["vendor"],
-            model=enc["model"],
-            revision=enc["revision"],
-            total_slots=enc["total_slots"],
-            populated_slots=enc["populated_slots"],
+            vendor=enc.get("vendor", ""),
+            model=enc.get("model", ""),
+            revision=enc.get("revision", ""),
+            total_slots=enc.get("total_slots", 0),
+            populated_slots=enc.get("populated_slots", 0),
             slots=slots_out,
+            health=enc_health,
         ))
 
-    # Host drives (non-enclosure)
-    host_drives_raw = await get_host_drives()
+    # Host (non-enclosure) drives — fully precomputed by the poller.
     host_drives_out: list[HostDrive] = []
-    for hd in host_drives_raw:
+    for hd in await store.get_host_drives():
         total_drives += 1
         hs = hd.get("health_status", "healthy")
         if hs == "error":
@@ -132,7 +113,6 @@ async def get_overview():
             all_healthy = False
         elif hs == "warning":
             warnings += 1
-        # Count physical drives behind RAID controllers
         for pd in hd.get("physical_drives", []):
             total_drives += 1
             pd_hs = pd.get("health_status", "healthy")
@@ -143,6 +123,11 @@ async def get_overview():
                 warnings += 1
         host_drives_out.append(HostDrive(**hd))
 
+    # Surface poller freshness so stale data is visible.
+    meta = await store.get_meta()
+    if meta and meta.get("last_sweep_ts"):
+        response.headers["X-Poll-Age"] = str(int(store.now() - meta["last_sweep_ts"]))
+
     return Overview(
         healthy=all_healthy and errors == 0,
         drive_count=total_drives,

routers/temps.py

@@ -0,0 +1,12 @@
+from fastapi import APIRouter
+
+from models.schemas import TempSnapshot
+from services.temps import build_temp_snapshot
+
+router = APIRouter(prefix="/api/temps", tags=["temps"])
+
+
+@router.get("", response_model=TempSnapshot)
+async def get_temps():
+    """Per-enclosure temperature snapshot: named SES sensors + hotspot."""
+    return await build_temp_snapshot()

services/cache.py

@@ -1,29 +1,70 @@
-import time
+import json
+import logging
+import os
 from typing import Any
 
+import redis.asyncio as redis
 
-class TTLCache:
-    """Simple in-memory TTL cache."""
+logger = logging.getLogger(__name__)
 
-    def __init__(self, ttl_seconds: int = 60):
-        self._ttl = ttl_seconds
-        self._store: dict[str, tuple[float, Any]] = {}
+_redis: redis.Redis | None = None
 
-    def get(self, key: str) -> Any | None:
-        entry = self._store.get(key)
-        if entry is None:
+
+async def init_cache() -> None:
+    """Create Redis connection from environment variables."""
+    global _redis
+    host = os.environ.get("REDIS_HOST", "localhost")
+    port = int(os.environ.get("REDIS_PORT", "6379"))
+    db = int(os.environ.get("REDIS_DB", "0"))
+    try:
+        _redis = redis.Redis(host=host, port=port, db=db, decode_responses=True)
+        await _redis.ping()
+        logger.info("Redis connected at %s:%d/%d", host, port, db)
+    except Exception as e:
+        logger.warning("Redis connection failed: %s — running without cache", e)
+        _redis = None
+
+
+async def close_cache() -> None:
+    """Close Redis connection."""
+    global _redis
+    if _redis is not None:
+        await _redis.aclose()
+        _redis = None
+
+
+def redis_available() -> bool:
+    """Return whether Redis connection is live."""
+    return _redis is not None
+
+
+def get_client() -> "redis.Redis | None":
+    """Return the raw Redis client for primitives not covered by get/set."""
+    return _redis
+
+
+async def cache_get(key: str) -> Any | None:
+    """GET key from Redis, return deserialized value or None on miss/error."""
+    if _redis is None:
         return None
-        ts, value = entry
-        if time.monotonic() - ts > self._ttl:
-            del self._store[key]
+    try:
+        raw = await _redis.get(key)
+        if raw is None:
+            return None
+        return json.loads(raw)
+    except Exception as e:
+        logger.warning("Redis GET %s failed: %s", key, e)
         return None
-        return value
-
-    def set(self, key: str, value: Any) -> None:
-        self._store[key] = (time.monotonic(), value)
-
-    def clear(self) -> None:
-        self._store.clear()
 
 
-smart_cache = TTLCache(ttl_seconds=60)
+async def cache_set(key: str, value: Any, ttl: int = 120) -> None:
+    """SET key in Redis. ttl<=0 stores with no expiry (Redis rejects EX 0)."""
+    if _redis is None:
+        return
+    try:
+        if ttl and ttl > 0:
+            await _redis.set(key, json.dumps(value), ex=ttl)
+        else:
+            await _redis.set(key, json.dumps(value))
+    except Exception as e:
+        logger.warning("Redis SET %s failed: %s", key, e)

services/enclosure.py

@@ -1,7 +1,11 @@
-import os
+import asyncio
 import logging
+import os
+import re
 from pathlib import Path
 
+from services.hwgate import gate
+
 logger = logging.getLogger(__name__)
 
 ENCLOSURE_BASE = Path("/sys/class/enclosure")
@@ -49,6 +53,18 @@ def discover_enclosures() -> list[dict]:
         slots = list_slots(enc_id)
         total = len(slots)
         populated = sum(1 for s in slots if s["populated"])
+        has_devices = any(s["device"] for s in slots)
+
+        # Skip secondary/passive IOM paths: enclosures where SES reports
+        # populated slots but the kernel has no device symlinks for any of
+        # them (the drives are owned by the primary IOM enclosure entry).
+        if populated > 0 and not has_devices:
+            logger.info(
+                "Skipping enclosure %s (%s %s) — %d populated slots but no "
+                "device links (likely a secondary IOM path)",
+                enc_id, vendor, model, populated,
+            )
+            continue
 
         enclosures.append({
             "id": enc_id,
@@ -136,3 +152,196 @@ def _parse_slot_number(entry: Path) -> int | None:
             except ValueError:
                 return None
     return None
+
+
+async def _run_sg_ses(sg_device: str, page: str) -> str | None:
+    """Run sg_ses for a given page, returning decoded stdout or None.
+    Hardware-gated."""
+    try:
+        async with gate():
+            proc = await asyncio.create_subprocess_exec(
+                "sg_ses", f"--page={page}", sg_device,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            logger.warning(
+                "sg_ses page %s failed for %s: %s",
+                page, sg_device, stderr.decode().strip(),
+            )
+            return None
+        return stdout.decode(errors="replace")
+    except FileNotFoundError:
+        logger.warning("sg_ses not found")
+        return None
+    except Exception as e:
+        logger.warning("sg_ses page %s error for %s: %s", page, sg_device, e)
+        return None
+
+
+async def get_enclosure_status(sg_device: str) -> dict | None:
+    """Run sg_ses status page (0x02) and parse enclosure health data.
+
+    Deliberately only the status page: the element descriptor page (0x07)
+    is omitted because it errored on the IOM6/Xyratex expanders here
+    (``couldn't read config page, res=35``) and returned empty names
+    anyway. Elements fall back to generic labels. If descriptor names are
+    ever wanted, fetch 0x07 ONCE at startup and cache — never per poll.
+    """
+    status_text = await _run_sg_ses(sg_device, "0x02")
+    if status_text is None:
+        return None
+    return _parse_ses_page02(status_text)
+
+
+def _norm_type(element_type: str) -> str:
+    """Normalize an SES element type line to a stable key.
+
+    sg_ses appends qualifiers like ", subenclosure id: 0 [ti=2]" to the
+    type line; strip those so the same element type matches across the
+    status (0x02) and element descriptor (0x07) pages.
+    """
+    return element_type.split(",")[0].strip().lower()
+
+
+def _parse_element_descriptors(text: str) -> dict[tuple[str, int], str]:
+    """Parse sg_ses --page=0x07 into {(element_type, index): name}."""
+    names: dict[tuple[str, int], str] = {}
+    sections = re.split(r"(?=\s*Element type:)", text)
+    for section in sections:
+        type_match = re.match(r"\s*Element type:\s*(.+)", section)
+        if not type_match:
+            continue
+        etype = _norm_type(type_match.group(1).strip().rstrip(","))
+        for line in section.splitlines():
+            m = re.match(r"\s*Element (\d+) descriptor:\s*(.*)", line)
+            if not m:
+                continue
+            idx = int(m.group(1))
+            name = m.group(2).strip()
+            # sg_ses prints "<empty>" when an enclosure leaves the descriptor
+            # string blank — treat that as no name so callers fall back to a
+            # generic label.
+            if name and name.lower() != "<empty>":
+                names[(etype, idx)] = name
+    return names
+
+
+def _parse_ses_page02(
+    text: str, names: dict[tuple[str, int], str] | None = None
+) -> dict:
+    """Parse sg_ses --page=0x02 text output into structured health data.
+
+    ``names`` maps (normalized element type, index) -> descriptor name from
+    the element descriptor page (0x07); when present it labels each element.
+    """
+    names = names or {}
+    result = {
+        "overall_status": "OK",
+        "psus": [],
+        "fans": [],
+        "temps": [],
+        "voltages": [],
+    }
+
+    # Split into element type sections.
+    # Each section starts with "Element type: <type>"
+    sections = re.split(r"(?=\s*Element type:)", text)
+
+    for section in sections:
+        type_match = re.match(r"\s*Element type:\s*(.+)", section)
+        if not type_match:
+            continue
+        element_type = type_match.group(1).strip().rstrip(",").lower()
+        etype_key = _norm_type(element_type)
+
+        # Find individual element blocks (skip "Overall descriptor")
+        elements = re.split(r"(?=\s*Element \d+ descriptor:)", section)
+
+        for elem_text in elements:
+            desc_match = re.match(r"\s*Element (\d+) descriptor:", elem_text)
+            if not desc_match:
+                continue
+            idx = int(desc_match.group(1))
+
+            # Extract status line
+            status_match = re.search(r"status:\s*(.+?)(?:,|\n|$)", elem_text, re.IGNORECASE)
+            status = status_match.group(1).strip() if status_match else "Unknown"
+
+            if status.lower() == "not installed":
+                continue
+
+            name = names.get((etype_key, idx))
+
+            if "power supply" in element_type:
+                fail = "Fail=1" in elem_text
+                ac_fail = "AC fail=1" in elem_text
+                dc_fail = "DC fail=1" in elem_text
+                result["psus"].append({
+                    "index": idx,
+                    "name": name,
+                    "status": status,
+                    "fail": fail,
+                    "ac_fail": ac_fail,
+                    "dc_fail": dc_fail,
+                })
+
+            elif "cooling" in element_type or "fan" in element_type:
+                fail = "Fail=1" in elem_text
+                rpm_match = re.search(r"Actual speed[=:]\s*(\d+)\s*rpm", elem_text, re.IGNORECASE)
+                rpm = int(rpm_match.group(1)) if rpm_match else None
+                result["fans"].append({
+                    "index": idx,
+                    "name": name,
+                    "status": status,
+                    "rpm": rpm,
+                    "fail": fail,
+                })
+
+            elif "temperature" in element_type:
+                temp_match = re.search(r"Temperature=\s*([\d.]+)\s*C", elem_text)
+                temp = float(temp_match.group(1)) if temp_match else None
+                # Skip dead/disconnected sensors: 0°C with a non-OK status
+                # is a non-functional sensor slot, not an actual reading.
+                if (temp is None or temp == 0) and status.lower() in (
+                    "unrecoverable", "unknown", "not available",
+                ):
+                    continue
+                result["temps"].append({
+                    "index": idx,
+                    "name": name,
+                    "status": status,
+                    "temperature_c": temp,
+                })
+
+            elif "voltage" in element_type:
+                volt_match = re.search(r"Voltage:\s*([\d.]+)\s*V", elem_text, re.IGNORECASE)
+                if not volt_match:
+                    volt_match = re.search(r"([\d.]+)\s*V", elem_text)
+                voltage = float(volt_match.group(1)) if volt_match else None
+                result["voltages"].append({
+                    "index": idx,
+                    "name": name,
+                    "status": status,
+                    "voltage": voltage,
+                })
+
+    # Derive overall_status from the actual parsed elements rather than
+    # the raw SES header, which counts dead/disconnected sensors as
+    # UNRECOV and inflates severity.
+    all_statuses = (
+        [e["status"] for e in result["psus"]]
+        + [e["status"] for e in result["fans"]]
+        + [e["status"] for e in result["temps"]]
+        + [e["status"] for e in result["voltages"]]
+    )
+    status_lower = [s.lower() for s in all_statuses]
+    if any(s in ("unrecoverable", "critical") for s in status_lower):
+        result["overall_status"] = "CRITICAL"
+    elif any(s in ("noncritical", "non-critical", "warning") for s in status_lower):
+        result["overall_status"] = "WARNING"
+    elif any(s not in ("ok", "unknown") for s in status_lower):
+        result["overall_status"] = "WARNING"
+
+    return result

services/health.py

@@ -0,0 +1,21 @@
+"""Shared drive-health classification.
+
+Single source of truth for turning a parsed SMART dict into a
+healthy/warning/error status, used by both the poller (host drives) and
+the API (enclosure drives) so the rule never drifts between them.
+"""
+
+
+def compute_health_status(smart: dict) -> str:
+    """Return "healthy", "warning", or "error" for a SMART dict."""
+    healthy = smart.get("smart_healthy")
+    realloc = smart.get("reallocated_sectors") or 0
+    pending = smart.get("pending_sectors") or 0
+    unc = smart.get("uncorrectable_errors") or 0
+    supported = smart.get("smart_supported", True)
+
+    if healthy is False:
+        return "error"
+    if realloc > 0 or pending > 0 or unc > 0 or (healthy is None and supported):
+        return "warning"
+    return "healthy"

services/host.py

@@ -3,16 +3,22 @@ import json
 import logging
 
 from services.enclosure import discover_enclosures, list_slots
-from services.smart import get_smart_data, scan_megaraid_drives
-from services.zfs import get_zfs_pool_map
+from services.health import compute_health_status
+from services.hwgate import gate
+from services.smart import _run_smartctl, scan_megaraid_drives
 
 logger = logging.getLogger(__name__)
 
 
-async def get_host_drives() -> list[dict]:
-    """Discover non-enclosure block devices and return SMART data for each."""
+async def get_host_drives(pool_map: dict) -> list[dict]:
+    """Discover non-enclosure block devices and return SMART data for each.
+
+    Poller-side producer: serialized through the hardware gate. ``pool_map``
+    is passed in (computed once per sweep) to avoid a second zpool call.
+    """
     # Get all block devices via lsblk
     try:
+        async with gate():
             proc = await asyncio.create_subprocess_exec(
                 "lsblk", "-d", "-o", "NAME,SIZE,TYPE,MODEL,ROTA,TRAN", "-J",
                 stdout=asyncio.subprocess.PIPE,
@@ -59,32 +65,23 @@ async def get_host_drives() -> list[dict]:
 
         host_devices.append({"name": name, "drive_type": drive_type})
 
-    # Fetch SMART + ZFS data concurrently
-    pool_map = await get_zfs_pool_map()
-    smart_tasks = [get_smart_data(d["name"]) for d in host_devices]
-    smart_results = await asyncio.gather(*smart_tasks, return_exceptions=True)
+    # Fetch SMART for each host disk (serialized by the gate inside _run_smartctl)
+    smart_results = await asyncio.gather(
+        *[_run_smartctl(d["name"]) for d in host_devices],
+        return_exceptions=True,
+    )
 
     results: list[dict] = []
-    for dev_info, smart in zip(host_devices, smart_results):
+    for dev_info, smart_result in zip(host_devices, smart_results):
         name = dev_info["name"]
 
-        if isinstance(smart, Exception):
-            logger.warning("SMART query failed for host drive %s: %s", name, smart)
+        if isinstance(smart_result, Exception):
+            logger.warning("SMART query failed for host drive %s: %s", name, smart_result)
             smart = {"device": name, "smart_supported": False}
-
-        # Compute health_status (same logic as overview.py)
-        healthy = smart.get("smart_healthy")
-        realloc = smart.get("reallocated_sectors") or 0
-        pending = smart.get("pending_sectors") or 0
-        unc = smart.get("uncorrectable_errors") or 0
-
-        if healthy is False:
-            health_status = "error"
-        elif realloc > 0 or pending > 0 or unc > 0 or (healthy is None and smart.get("smart_supported", True)):
-            health_status = "warning"
         else:
-            health_status = "healthy"
+            smart = smart_result
 
+        health_status = compute_health_status(smart)
         zfs_info = pool_map.get(name, {})
 
         results.append({
@@ -95,7 +92,7 @@ async def get_host_drives() -> list[dict]:
             "wwn": smart.get("wwn"),
             "firmware": smart.get("firmware"),
             "capacity_bytes": smart.get("capacity_bytes"),
-            "smart_healthy": healthy,
+            "smart_healthy": smart.get("smart_healthy"),
             "smart_supported": smart.get("smart_supported", True),
             "temperature_c": smart.get("temperature_c"),
             "power_on_hours": smart.get("power_on_hours"),
@@ -114,16 +111,7 @@ async def get_host_drives() -> list[dict]:
     if has_raid:
         megaraid_drives = await scan_megaraid_drives()
         for pd in megaraid_drives:
-            pd_healthy = pd.get("smart_healthy")
-            pd_realloc = pd.get("reallocated_sectors") or 0
-            pd_pending = pd.get("pending_sectors") or 0
-            pd_unc = pd.get("uncorrectable_errors") or 0
-            if pd_healthy is False:
-                pd["health_status"] = "error"
-            elif pd_realloc > 0 or pd_pending > 0 or pd_unc > 0:
-                pd["health_status"] = "warning"
-            else:
-                pd["health_status"] = "healthy"
+            pd["health_status"] = compute_health_status(pd)
             pd["drive_type"] = "physical"
             pd["physical_drives"] = []
 

services/host_sensors.py

@@ -0,0 +1,147 @@
+"""Host/chassis temperature collectors: IPMI (iDRAC) + CPU coretemp.
+
+Used by the host-poller (separate process from the SAS poller). In-band IPMI
+via ipmitool (/dev/ipmi0) gives environmental sensors (Inlet/Exhaust/…); the
+kernel coretemp hwmon gives CPU package temps. IPMI is gated for serialization;
+coretemp is a cheap sysfs read.
+"""
+import asyncio
+import logging
+import os
+import re
+from collections import Counter
+from pathlib import Path
+
+from services.hwgate import gate
+
+logger = logging.getLogger(__name__)
+
+HWMON = Path("/sys/class/hwmon")
+
+
+async def read_ipmi_temps() -> list[dict]:
+    """`ipmitool sdr type temperature` → [{name, temperature_c, status, kind}]."""
+    try:
+        async with gate():
+            proc = await asyncio.create_subprocess_exec(
+                "ipmitool", "sdr", "type", "temperature",
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            out, err = await proc.communicate()
+    except FileNotFoundError:
+        logger.warning("ipmitool not found")
+        return []
+    if proc.returncode != 0:
+        logger.warning("ipmitool failed: %s", err.decode(errors="replace").strip())
+        return []
+    return _parse_ipmi(out.decode(errors="replace"))
+
+
+def _parse_ipmi(text: str) -> list[dict]:
+    # e.g.  "Inlet Temp       | 04h | ok  |  7.1 | 23 degrees C"
+    #       "Temp             | 0Eh | ok  |  3.1 | 40 degrees C"  (CPU1)
+    sensors: list[dict] = []
+    for line in text.splitlines():
+        parts = [p.strip() for p in line.split("|")]
+        if len(parts) < 5:
+            continue
+        name, _sid, status, entity, reading = parts[:5]
+        # Skip relative/derived sensors (e.g. "P1 Therm Margin") — they read in
+        # "degrees C" but are a distance-to-throttle, not an absolute temp.
+        if "margin" in name.lower():
+            continue
+        m = re.search(r"(-?\d+(?:\.\d+)?)\s*degrees?\s*C", reading, re.IGNORECASE)
+        if not m:  # "no reading", "disabled", "ns", etc.
+            continue
+        temp = float(m.group(1))
+        kind = "env"
+        # Dell reports CPU temps as "Temp" under entity 3.x — relabel them.
+        if name.lower() == "temp" and entity.startswith("3."):
+            name = f"CPU {entity.split('.')[-1]}"
+            kind = "cpu"
+        sensors.append({
+            "name": name,
+            "temperature_c": temp,
+            "status": status or "ok",
+            "kind": kind,
+        })
+    return sensors
+
+
+def read_hwmon_env() -> list[dict]:
+    """Read labelled temps from extra hwmon chips as env sensors.
+
+    For boxes without IPMI (e.g. Dell workstations), the BIOS SMM interface
+    (`dell_smm`) exposes CPU/ambient/SODIMM/board temps. Chips are configurable
+    via HOST_HWMON_CHIPS (comma list; default "dell_smm"); set to "" to disable.
+    Names are "<chip> <label>", with the temp index appended when a label
+    repeats within a chip (e.g. dell_smm's multiple "Other" sensors).
+    """
+    chips = [c.strip() for c in os.environ.get("HOST_HWMON_CHIPS", "dell_smm").split(",") if c.strip()]
+    if not chips or not HWMON.is_dir():
+        return []
+    out: list[dict] = []
+    for hw in sorted(HWMON.iterdir()):
+        try:
+            chip = (hw / "name").read_text().strip()
+        except OSError:
+            continue
+        if chip not in chips:
+            continue
+        entries: list[tuple[str, str, int]] = []  # (idx, label, temp)
+        for label_f in sorted(hw.glob("temp*_label")):
+            try:
+                label = label_f.read_text().strip()
+            except OSError:
+                continue
+            input_f = label_f.with_name(label_f.name.replace("_label", "_input"))
+            try:
+                temp = round(int(input_f.read_text().strip()) / 1000)
+            except (OSError, ValueError):
+                continue
+            idx = label_f.name[len("temp"):-len("_label")]
+            entries.append((idx, label, temp))
+        dup = {lbl for lbl, n in Counter(l for _, l, _ in entries).items() if n > 1}
+        for idx, label, temp in entries:
+            name = f"{chip} {label}" + (f" {idx}" if label in dup else "")
+            out.append({"name": name, "temperature_c": temp, "status": "ok", "kind": "env"})
+    return out
+
+
+def read_coretemp() -> list[dict]:
+    """CPU package temps from coretemp hwmon → [{name, temperature_c, kind}]."""
+    cpus: list[dict] = []
+    if not HWMON.is_dir():
+        return cpus
+    for hw in sorted(HWMON.iterdir()):
+        try:
+            if (hw / "name").read_text().strip() != "coretemp":
+                continue
+        except OSError:
+            continue
+        packages: list[dict] = []
+        cores: list[int] = []
+        for label_f in sorted(hw.glob("temp*_label")):
+            try:
+                label = label_f.read_text().strip()
+            except OSError:
+                continue
+            input_f = label_f.with_name(label_f.name.replace("_label", "_input"))
+            try:
+                temp = round(int(input_f.read_text().strip()) / 1000)
+            except (OSError, ValueError):
+                continue
+            low = label.lower()
+            if low.startswith("package id"):
+                pkg = label.split("id")[-1].strip()
+                packages.append({"name": f"CPU {pkg}", "temperature_c": temp, "kind": "cpu"})
+            elif low.startswith("core"):
+                cores.append(temp)
+        # Prefer the per-package sensor; older CPUs (no package sensor) fall back
+        # to the hottest core as the CPU temp.
+        if packages:
+            cpus.extend(packages)
+        elif cores:
+            cpus.append({"name": "CPU", "temperature_c": max(cores), "kind": "cpu"})
+    return cpus

services/hwgate.py

@@ -0,0 +1,61 @@
+"""Global hardware-access gate + pacing.
+
+Every command that touches the SAS bus — smartctl, sg_ses, zpool, lsblk,
+ledctl — must run inside this gate. It does two things:
+
+  1. Serializes hardware I/O (semaphore, default concurrency 1) so the
+     poller can never fire a thundering herd at fragile SAS expanders.
+  2. Paces it: after every hardware op it holds the gate for POLL_DRIVE_GAP
+     seconds, so back-to-back ops (gathered SMART, MegaRAID scans, SES,
+     ledctl) are always spaced — not just the per-drive loop.
+
+This is a per-process gate; only the poller process calls hardware, so
+per-process serialization is sufficient. Created lazily so it binds to the
+running event loop.
+"""
+import asyncio
+import contextlib
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+_sem: asyncio.Semaphore | None = None
+
+
+def _semaphore() -> asyncio.Semaphore:
+    global _sem
+    if _sem is None:
+        n = max(1, int(os.environ.get("POLL_CONCURRENCY", "1")))
+        # Serialized (1) is the safe default. Concurrent hardware I/O can drop
+        # fragile SAS expanders, so >1 is clamped unless explicitly overridden.
+        if n > 1:
+            override = os.environ.get("POLL_ALLOW_UNSAFE_CONCURRENCY", "").lower() in (
+                "1", "true", "yes",
+            )
+            if override:
+                logger.warning(
+                    "POLL_CONCURRENCY=%d with unsafe override — concurrent "
+                    "hardware I/O enabled; backplane must tolerate it", n,
+                )
+            else:
+                logger.warning(
+                    "POLL_CONCURRENCY=%d ignored (clamped to 1 for hardware "
+                    "safety); set POLL_ALLOW_UNSAFE_CONCURRENCY=1 to override", n,
+                )
+                n = 1
+        _sem = asyncio.Semaphore(n)
+    return _sem
+
+
+@contextlib.asynccontextmanager
+async def gate():
+    """Acquire the hardware gate for one op, then hold it for the pacing gap."""
+    sem = _semaphore()
+    async with sem:
+        try:
+            yield
+        finally:
+            gap = float(os.environ.get("POLL_DRIVE_GAP", "0.5"))
+            if gap > 0:
+                await asyncio.sleep(gap)

services/leds.py

@@ -1,9 +1,12 @@
 import asyncio
 import re
 
+from services.hwgate import gate
 
-async def set_led(device: str, state: str) -> dict:
-    """Set the locate LED on a drive via ledctl.
+
+async def run_led(device: str, state: str) -> dict:
+    """Set the locate LED on a drive via ledctl. Hardware-gated; runs in the
+    poller, fed by the Redis LED queue. The API enqueues, never calls this.
 
     Args:
         device: Block device name (e.g. "sda"). Must be alphanumeric.
@@ -14,6 +17,7 @@ async def set_led(device: str, state: str) -> dict:
     if state not in ("locate", "off"):
         raise ValueError(f"Invalid state: {state}")
 
+    async with gate():
         proc = await asyncio.create_subprocess_exec(
             "ledctl",
             f"{state}=/dev/{device}",

services/mqtt_publisher.py

@@ -0,0 +1,321 @@
+"""MQTT publisher with Home Assistant discovery for enclosure temperatures.
+
+Opt-in: only starts when ``MQTT_HOST`` is set. Publishes one retained
+discovery config per HA entity (a device per enclosure, with a hotspot
+sensor plus one sensor per named SES temperature element) and pushes a
+single JSON state message per enclosure on each poll. Availability is
+tracked via an LWT so HA marks entities unavailable if the monitor dies.
+
+Topic layout (defaults)::
+
+    homeassistant/sensor/<node>_enc<id>/hotspot/config   (retained discovery)
+    homeassistant/sensor/<node>_enc<id>/temp<n>/config    (retained discovery)
+    jbod-monitor/<node>/status                             online | offline (LWT)
+    jbod-monitor/<node>/enclosure/<id>/state              {"hotspot_c":.., "sensor_0":..}
+"""
+import asyncio
+import json
+import logging
+import os
+import re
+
+from services import store
+from services.secrets import read_kv2
+from services.temps import build_temp_snapshot, get_node_id
+
+logger = logging.getLogger(__name__)
+
+try:
+    import paho.mqtt.client as mqtt
+    from paho.mqtt.enums import CallbackAPIVersion
+
+    _PAHO_AVAILABLE = True
+except ImportError:  # pragma: no cover - exercised only without the dep
+    mqtt = None
+    CallbackAPIVersion = None
+    _PAHO_AVAILABLE = False
+
+
+def mqtt_enabled() -> bool:
+    """MQTT publishing is opt-in via MQTT_HOST."""
+    return bool(os.environ.get("MQTT_HOST"))
+
+
+def _slug(value: str) -> str:
+    """Sanitize a string for use in MQTT topics / HA unique_ids."""
+    return re.sub(r"[^a-zA-Z0-9_-]", "_", value).strip("_") or "x"
+
+
+def _resolve_credentials() -> tuple[str | None, str | None]:
+    """Resolve broker credentials, preferring OpenBao over plaintext env.
+
+    If ``MQTT_SECRET_PATH`` is set, read username/password from that OpenBao
+    KV v2 path (keys ``username``/``password``). Falls back to the
+    ``MQTT_USERNAME``/``MQTT_PASSWORD`` env vars when unset or on failure.
+    """
+    username = os.environ.get("MQTT_USERNAME") or None
+    password = os.environ.get("MQTT_PASSWORD") or None
+
+    secret_path = os.environ.get("MQTT_SECRET_PATH")
+    if secret_path:
+        user_key = os.environ.get("MQTT_SECRET_USER_KEY", "username")
+        pass_key = os.environ.get("MQTT_SECRET_PASS_KEY", "password")
+        data = read_kv2(secret_path)
+        if data:
+            username = data.get(user_key, username)
+            password = data.get(pass_key, password)
+            logger.info("Loaded MQTT credentials from OpenBao (%s)", secret_path)
+        else:
+            logger.warning(
+                "MQTT_SECRET_PATH set (%s) but OpenBao read failed — "
+                "falling back to env credentials", secret_path,
+            )
+    return username, password
+
+
+class MqttPublisher:
+    def __init__(self) -> None:
+        self.host = os.environ["MQTT_HOST"]
+        self.port = int(os.environ.get("MQTT_PORT", "1883"))
+        self.username, self.password = _resolve_credentials()
+        self.discovery_prefix = os.environ.get("MQTT_DISCOVERY_PREFIX", "homeassistant")
+        self.base_topic = os.environ.get("MQTT_BASE_TOPIC", "jbod-monitor")
+        self.interval = int(os.environ.get("MQTT_PUBLISH_INTERVAL", "60"))
+        self.node_raw = get_node_id()
+        self.node = _slug(self.node_raw)
+        # Parent "hub" device that the per-enclosure devices nest under.
+        self.hub_id = f"{self.node}_jbod_monitor"
+
+        self.availability_topic = f"{self.base_topic}/{self.node}/status"
+        # unique_ids for which we've already published discovery this connection.
+        self._discovered: set[str] = set()
+
+        self._client = mqtt.Client(
+            CallbackAPIVersion.VERSION2,
+            client_id=os.environ.get("MQTT_CLIENT_ID", f"jbod-monitor-{self.node}"),
+        )
+        if self.username:
+            self._client.username_pw_set(self.username, self.password)
+        self._client.will_set(self.availability_topic, "offline", qos=1, retain=True)
+        self._client.reconnect_delay_set(min_delay=1, max_delay=60)
+        self._client.on_connect = self._on_connect
+        self._client.on_disconnect = self._on_disconnect
+
+    def start(self) -> None:
+        logger.info("MQTT connecting to %s:%d", self.host, self.port)
+        self._client.connect_async(self.host, self.port, keepalive=60)
+        self._client.loop_start()
+
+    def stop(self) -> None:
+        try:
+            self._client.publish(self.availability_topic, "offline", qos=1, retain=True)
+        except Exception:
+            pass
+        self._client.loop_stop()
+        try:
+            self._client.disconnect()
+        except Exception:
+            pass
+
+    # paho callbacks (run in the network thread) -----------------------------
+    def _on_connect(self, client, userdata, flags, reason_code, properties=None):
+        if reason_code != 0:
+            logger.warning("MQTT connect failed: %s", reason_code)
+            return
+        logger.info("MQTT connected")
+        # Re-announce discovery + availability on every (re)connect.
+        self._discovered.clear()
+        client.publish(self.availability_topic, "online", qos=1, retain=True)
+
+    def _on_disconnect(self, client, userdata, flags, reason_code, properties=None):
+        logger.warning("MQTT disconnected: %s", reason_code)
+
+    # discovery / state ------------------------------------------------------
+    def _hub_device_block(self) -> dict:
+        """The parent device the enclosures nest under (named, so HA doesn't
+        show an 'Unnamed Device')."""
+        return {
+            "identifiers": [self.hub_id],
+            "name": f"JBOD Monitor ({self.node_raw})",
+            "manufacturer": "jbod-monitor",
+            "model": "SES/SMART poller",
+        }
+
+    def _device_block(self, enc: dict) -> dict:
+        enc_id = enc["id"]
+        vendor = enc.get("vendor", "").strip()
+        model = enc.get("model", "").strip()
+        name = " ".join(p for p in [vendor, model] if p) or "JBOD enclosure"
+        return {
+            "identifiers": [f"{self.node}_enc_{_slug(enc_id)}"],
+            "name": f"JBOD {name} ({enc_id})",
+            "manufacturer": vendor or None,
+            "model": model or None,
+            "via_device": self.hub_id,
+        }
+
+    def _publish_hub(self) -> None:
+        """Announce the hub device via a connectivity binary_sensor, so the
+        parent device is named and shows monitor online/offline status."""
+        uid = f"{self.hub_id}_status"
+        if uid in self._discovered:
+            return
+        payload = {
+            "name": "Status",
+            "unique_id": uid,
+            "device_class": "connectivity",
+            "state_topic": self.availability_topic,
+            "payload_on": "online",
+            "payload_off": "offline",
+            "entity_category": "diagnostic",
+            "device": self._hub_device_block(),
+        }
+        topic = f"{self.discovery_prefix}/binary_sensor/{self.hub_id}/status/config"
+        self._client.publish(topic, json.dumps(payload), qos=1, retain=True)
+        self._discovered.add(uid)
+
+    def _publish_discovery(self, enc: dict) -> None:
+        enc_id = enc["id"]
+        enc_slug = _slug(enc_id)
+        state_topic = f"{self.base_topic}/{self.node}/enclosure/{enc_slug}/state"
+        device = self._device_block(enc)
+
+        def announce(object_id: str, payload: dict) -> None:
+            uid = payload["unique_id"]
+            if uid in self._discovered:
+                return
+            topic = (
+                f"{self.discovery_prefix}/sensor/"
+                f"{self.node}_enc{enc_slug}/{object_id}/config"
+            )
+            self._client.publish(topic, json.dumps(payload), qos=1, retain=True)
+            self._discovered.add(uid)
+
+        common = {
+            "device_class": "temperature",
+            "unit_of_measurement": "°C",
+            "state_class": "measurement",
+            "state_topic": state_topic,
+            "availability_topic": self.availability_topic,
+            "device": device,
+        }
+
+        # Hotspot (max across SES sensors + drive temps).
+        announce("hotspot", {
+            **common,
+            "name": "Hotspot",
+            "unique_id": f"{self.node}_enc{enc_slug}_hotspot",
+            "value_template": "{{ value_json.hotspot_c }}",
+            "icon": "mdi:thermometer-high",
+            "json_attributes_topic": state_topic,
+            "json_attributes_template":
+                '{"drive_max_c": {{ value_json.drive_max_c | default("null") }}, '
+                '"drive_temp_count": {{ value_json.drive_temp_count | default(0) }}}',
+        })
+
+        # One sensor per named SES temperature element.
+        for s in enc.get("sensors", []):
+            idx = s["index"]
+            label = s.get("name") or f"Temp {idx}"
+            announce(f"temp{idx}", {
+                **common,
+                "name": label,
+                "unique_id": f"{self.node}_enc{enc_slug}_temp{idx}",
+                "value_template": f"{{{{ value_json.sensor_{idx} }}}}",
+            })
+
+    def _publish_state(self, enc: dict) -> None:
+        enc_slug = _slug(enc["id"])
+        state_topic = f"{self.base_topic}/{self.node}/enclosure/{enc_slug}/state"
+        payload: dict = {
+            "hotspot_c": enc.get("hotspot_c"),
+            "drive_max_c": enc.get("drive_max_c"),
+            "drive_temp_count": enc.get("drive_temp_count", 0),
+        }
+        for s in enc.get("sensors", []):
+            payload[f"sensor_{s['index']}"] = s.get("temperature_c")
+            payload[f"sensor_{s['index']}_status"] = s.get("status")
+        self._client.publish(state_topic, json.dumps(payload), qos=0, retain=True)
+
+    def publish_snapshot(self, snapshot: dict) -> None:
+        self._publish_hub()  # name the parent device before nesting enclosures
+        for enc in snapshot.get("enclosures", []):
+            self._publish_discovery(enc)
+            self._publish_state(enc)
+
+    def publish_host(self, sensors: dict, drives: list) -> None:
+        """Publish chassis temps (IPMI env, CPU, on-metal drives) on the hub."""
+        self._publish_hub()
+        state_topic = f"{self.base_topic}/{self.node}/host/state"
+        device = self._hub_device_block()
+        payload: dict = {}
+
+        def announce(object_id: str, cfg: dict) -> None:
+            uid = cfg["unique_id"]
+            if uid not in self._discovered:
+                topic = f"{self.discovery_prefix}/sensor/{self.hub_id}/{object_id}/config"
+                self._client.publish(topic, json.dumps(cfg), qos=1, retain=True)
+                self._discovered.add(uid)
+
+        common = {
+            "device_class": "temperature",
+            "unit_of_measurement": "°C",
+            "state_class": "measurement",
+            "state_topic": state_topic,
+            "availability_topic": self.availability_topic,
+            "device": device,
+        }
+
+        def add(kind: str, name: str, temp, icon: str | None = None) -> None:
+            if temp is None or not name:
+                return
+            key = f"{kind}_{_slug(name)}"
+            payload[key] = temp
+            cfg = {
+                **common,
+                "name": name,
+                "unique_id": f"{self.hub_id}_{key}",
+                "value_template": f"{{{{ value_json.{key} }}}}",
+            }
+            if icon:
+                cfg["icon"] = icon
+            announce(key, cfg)
+
+        for s in sensors.get("env", []):
+            add("env", s.get("name"), s.get("temperature_c"))
+        for c in sensors.get("cpu", []):
+            add("cpu", c.get("name"), c.get("temperature_c"), icon="mdi:cpu-64-bit")
+
+        def add_drive(d: dict) -> None:
+            dev = d.get("device")
+            if dev:
+                add("drive", f"Drive {dev}", d.get("temperature_c"), icon="mdi:harddisk")
+
+        for d in drives:
+            add_drive(d)
+            for pd in d.get("physical_drives", []):
+                add_drive(pd)
+
+        self._client.publish(state_topic, json.dumps(payload), qos=0, retain=True)
+
+    async def run(self) -> None:
+        """Background loop: build a snapshot and publish on an interval."""
+        await asyncio.sleep(3)  # let the first SMART poll populate the cache
+        while True:
+            try:
+                snapshot = await build_temp_snapshot()
+                self.publish_snapshot(snapshot)
+                host_sensors = await store.get_host_sensors() or {}
+                host_drives = await store.get_host_drives() or []
+                self.publish_host(host_sensors, host_drives)
+                logger.info(
+                    "MQTT published temps for %d enclosure(s) + host (%d env, %d cpu)",
+                    len(snapshot.get("enclosures", [])),
+                    len(host_sensors.get("env", [])),
+                    len(host_sensors.get("cpu", [])),
+                )
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                logger.error("MQTT publish loop error: %s", e)
+            await asyncio.sleep(self.interval)

services/secrets.py

@@ -0,0 +1,58 @@
+"""Minimal OpenBao (Vault) KV v2 reader for runtime secret injection.
+
+Mirrors the homelab pattern used by other non-k8s services: read
+``<addr>/v1/<mount>/data/<path>`` with an ``X-Vault-Token``. Stdlib only —
+no extra dependency. Token comes from ``OPENBAO_TOKEN`` or, preferably, a
+mounted token file via ``OPENBAO_TOKEN_FILE`` (Docker/host secret).
+"""
+import json
+import logging
+import os
+import urllib.error
+import urllib.request
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_ADDR = "https://vault.adamksmith.xyz"
+DEFAULT_MOUNT = "secret"
+
+
+def _resolve_token() -> str | None:
+    token = os.environ.get("OPENBAO_TOKEN")
+    if token:
+        return token.strip()
+    token_file = os.environ.get("OPENBAO_TOKEN_FILE")
+    if token_file:
+        try:
+            return open(token_file).read().strip()
+        except OSError as e:
+            logger.warning("Could not read OPENBAO_TOKEN_FILE %s: %s", token_file, e)
+    return None
+
+
+def read_kv2(path: str) -> dict | None:
+    """Read a KV v2 secret. ``path`` is relative to the mount (e.g.
+    "jbod-monitor/mqtt"). Returns the inner data dict, or None on failure.
+    """
+    token = _resolve_token()
+    if not token:
+        logger.warning("OpenBao read requested for %s but no token configured", path)
+        return None
+
+    addr = os.environ.get("OPENBAO_ADDR", DEFAULT_ADDR).rstrip("/")
+    mount = os.environ.get("OPENBAO_MOUNT", DEFAULT_MOUNT).strip("/")
+    url = f"{addr}/v1/{mount}/data/{path.strip('/')}"
+
+    req = urllib.request.Request(url, headers={"X-Vault-Token": token})
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            body = json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        logger.warning("OpenBao read %s failed: HTTP %s", path, e.code)
+        return None
+    except Exception as e:
+        logger.warning("OpenBao read %s failed: %s", path, e)
+        return None
+
+    # KV v2 wraps the secret in data.data.
+    return body.get("data", {}).get("data")

services/smart.py

@@ -1,10 +1,11 @@
 import asyncio
 import json
 import logging
+import os
 import re
 import shutil
 
-from services.cache import smart_cache
+from services.hwgate import gate
 
 logger = logging.getLogger(__name__)
 
@@ -26,26 +27,12 @@ def sg_ses_available() -> bool:
     return shutil.which("sg_ses") is not None
 
 
-async def get_smart_data(device: str) -> dict:
-    """Run smartctl -a -j against a device, with caching."""
-    # Sanitize device name: only allow alphanumeric and hyphens
-    if not re.match(r"^[a-zA-Z0-9\-]+$", device):
-        raise ValueError(f"Invalid device name: {device}")
-
-    cached = smart_cache.get(device)
-    if cached is not None:
-        return cached
-
-    result = await _run_smartctl(device)
-    smart_cache.set(device, result)
-    return result
-
-
 async def _run_smartctl(device: str) -> dict:
-    """Execute smartctl and parse JSON output."""
+    """Execute smartctl and parse JSON output. Hardware-gated."""
     dev_path = f"/dev/{device}"
 
     try:
+        async with gate():
             proc = await asyncio.create_subprocess_exec(
                 "smartctl", "-a", "-j", dev_path,
                 stdout=asyncio.subprocess.PIPE,
@@ -148,6 +135,7 @@ def _parse_smart_json(device: str, data: dict) -> dict:
 async def scan_megaraid_drives() -> list[dict]:
     """Discover physical drives behind MegaRAID controllers via smartctl --scan."""
     try:
+        async with gate():
             proc = await asyncio.create_subprocess_exec(
                 "smartctl", "--scan", "-j",
                 stdout=asyncio.subprocess.PIPE,
@@ -173,6 +161,7 @@ async def scan_megaraid_drives() -> list[dict]:
         dev_path = entry["name"]
         dev_type = entry["type"]
         try:
+            async with gate():
                 proc = await asyncio.create_subprocess_exec(
                     "smartctl", "-a", "-j", "-d", dev_type, dev_path,
                     stdout=asyncio.subprocess.PIPE,

services/store.py

@@ -0,0 +1,158 @@
+"""Redis as the sole interface between the poller (producer) and the API/
+MQTT consumers.
+
+The poller writes hardware-derived state here; the API and MQTT publisher
+read it and never touch hardware. Also carries the locate-LED command queue
+(API enqueues, poller drains) and the poller single-instance lock.
+"""
+import json
+import logging
+import os
+import time
+
+from services.cache import cache_get, cache_set, get_client
+
+logger = logging.getLogger(__name__)
+
+# Data TTL: must comfortably exceed the sweep interval so values persist
+# between sweeps (last-good is served stale rather than blanked).
+DATA_TTL = int(os.environ.get("POLL_DATA_TTL", "900"))
+
+K_SMART = "jbod:smart:{}"
+K_SES = "jbod:ses:{}"
+K_ZFS = "jbod:zfs_map"
+K_INVENTORY = "jbod:inventory"
+K_HOST_DRIVES = "jbod:host_drives"
+K_HOST_SENSORS = "jbod:host_sensors"
+K_META = "jbod:poll:meta"
+K_HOST_ENV_META = "jbod:host_poll:env_meta"
+K_HOST_DRIVE_META = "jbod:host_poll:drive_meta"
+K_LED_QUEUE = "jbod:led:queue"
+K_LOCK = "jbod:poll:lock"
+K_HOST_LOCK = "jbod:host_poll:lock"
+
+
+# ── writes (poller) ─────────────────────────────────────────────────────
+async def set_smart(device: str, data: dict) -> None:
+    await cache_set(K_SMART.format(device), data, DATA_TTL)
+
+
+async def set_ses(enc_id: str, data: dict) -> None:
+    await cache_set(K_SES.format(enc_id), data, DATA_TTL)
+
+
+async def set_zfs_map(pool_map: dict) -> None:
+    await cache_set(K_ZFS, pool_map, DATA_TTL)
+
+
+async def set_inventory(inventory: dict) -> None:
+    await cache_set(K_INVENTORY, inventory, DATA_TTL)
+
+
+async def set_host_drives(host_drives: list) -> None:
+    await cache_set(K_HOST_DRIVES, host_drives, DATA_TTL)
+
+
+async def set_host_sensors(sensors: dict) -> None:
+    await cache_set(K_HOST_SENSORS, sensors, DATA_TTL)
+
+
+async def set_meta(meta: dict, key: str = K_META) -> None:
+    # Meta has no TTL — it's the heartbeat; staleness is judged from its ts.
+    await cache_set(key, meta, 0)
+
+
+# ── reads (consumers) ───────────────────────────────────────────────────
+async def get_smart(device: str) -> dict | None:
+    return await cache_get(K_SMART.format(device))
+
+
+async def get_ses(enc_id: str) -> dict | None:
+    return await cache_get(K_SES.format(enc_id))
+
+
+async def get_zfs_map() -> dict:
+    return await cache_get(K_ZFS) or {}
+
+
+async def get_inventory() -> dict:
+    return await cache_get(K_INVENTORY) or {"enclosures": []}
+
+
+async def get_host_drives() -> list:
+    return await cache_get(K_HOST_DRIVES) or []
+
+
+async def get_host_sensors() -> dict | None:
+    return await cache_get(K_HOST_SENSORS)
+
+
+async def get_meta(key: str = K_META) -> dict | None:
+    return await cache_get(key)
+
+
+# ── locate-LED command queue ────────────────────────────────────────────
+async def enqueue_led(device: str, state: str) -> bool:
+    """API side: push a LED request for the poller to execute. Returns False
+    if Redis is unavailable or the push fails (so the API can surface 503)."""
+    client = get_client()
+    if client is None:
+        return False
+    try:
+        await client.rpush(K_LED_QUEUE, json.dumps({"device": device, "state": state}))
+    except Exception as e:
+        logger.warning("LED enqueue failed: %s", e)
+        return False
+    return True
+
+
+async def pop_led(timeout: int = 5) -> dict | None:
+    """Poller side: block up to `timeout`s for the next LED request."""
+    client = get_client()
+    if client is None:
+        return None
+    item = await client.blpop(K_LED_QUEUE, timeout=timeout)
+    if not item:
+        return None
+    _key, raw = item
+    try:
+        return json.loads(raw)
+    except (ValueError, TypeError):
+        return None
+
+
+# ── poller single-instance lock (atomic compare-and-set) ────────────────
+# Acquire if free OR already ours, refreshing the TTL — all in one round trip
+# so there's no GET/SET race that could let two pollers run concurrently.
+_ACQUIRE_LUA = (
+    "local v = redis.call('get', KEYS[1]) "
+    "if v == false or v == ARGV[1] then "
+    "  redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) return 1 "
+    "else return 0 end"
+)
+# Extend only if we still own it (atomic compare-and-expire).
+_REFRESH_LUA = (
+    "if redis.call('get', KEYS[1]) == ARGV[1] then "
+    "  return redis.call('expire', KEYS[1], ARGV[2]) "
+    "else return 0 end"
+)
+
+
+async def acquire_lock(token: str, ttl: int = 30, key: str = K_LOCK) -> bool:
+    """Atomically claim a single-instance lock (or re-own it). True if held."""
+    client = get_client()
+    if client is None:
+        return True  # no Redis → no coordination possible; run anyway
+    return bool(await client.eval(_ACQUIRE_LUA, 1, key, token, ttl))
+
+
+async def refresh_lock(token: str, ttl: int = 30, key: str = K_LOCK) -> bool:
+    """Atomically extend the lock iff we still own it."""
+    client = get_client()
+    if client is None:
+        return True
+    return bool(await client.eval(_REFRESH_LUA, 1, key, token, ttl))
+
+
+def now() -> float:
+    return time.time()

services/temps.py

@@ -0,0 +1,67 @@
+"""Per-enclosure temperature aggregation (read-only consumer).
+
+Reads the poller's data out of Redis — never touches hardware. Each named
+SES temperature sensor plus a derived per-enclosure hotspot (the hottest
+reading across SES sensors *and* the drives housed in that enclosure —
+drive temps are usually the real hotspot signal).
+"""
+import logging
+import os
+import socket
+
+from services import store
+
+logger = logging.getLogger(__name__)
+
+
+def get_node_id() -> str:
+    """Stable identifier for this monitor instance (defaults to hostname)."""
+    return os.environ.get("MQTT_NODE_ID") or socket.gethostname() or "jbod-monitor"
+
+
+async def build_temp_snapshot() -> dict:
+    """Collect per-enclosure temperature metrics from the store."""
+    inventory = await store.get_inventory()
+    out_enclosures: list[dict] = []
+
+    for enc in inventory.get("enclosures", []):
+        ses = await store.get_ses(enc["id"]) or {}
+
+        sensors: list[dict] = []
+        sensor_temps: list[float] = []
+        for t in ses.get("temps", []):
+            temp = t.get("temperature_c")
+            sensors.append({
+                "index": t["index"],
+                "name": t.get("name"),
+                "status": t.get("status", "Unknown"),
+                "temperature_c": temp,
+            })
+            if isinstance(temp, (int, float)) and temp > 0:
+                sensor_temps.append(float(temp))
+
+        # Drive temperatures for drives housed in this enclosure.
+        drive_temps: list[float] = []
+        for slot in enc.get("slots", []):
+            dev = slot.get("device")
+            if not dev:
+                continue
+            sm = await store.get_smart(dev)
+            if not sm:
+                continue
+            temp = sm.get("temperature_c")
+            if isinstance(temp, (int, float)) and temp > 0:
+                drive_temps.append(float(temp))
+
+        all_temps = sensor_temps + drive_temps
+        out_enclosures.append({
+            "id": enc["id"],
+            "vendor": enc.get("vendor", ""),
+            "model": enc.get("model", ""),
+            "hotspot_c": max(all_temps) if all_temps else None,
+            "drive_max_c": max(drive_temps) if drive_temps else None,
+            "drive_temp_count": len(drive_temps),
+            "sensors": sensors,
+        })
+
+    return {"node": get_node_id(), "enclosures": out_enclosures}

services/zfs.py

@@ -4,6 +4,8 @@ import logging
 import re
 from pathlib import Path
 
+from services.hwgate import gate
+
 logger = logging.getLogger(__name__)
 
 # Allow overriding the zpool binary path via env (for bind-mounted host tools)
@@ -15,6 +17,9 @@ async def get_zfs_pool_map() -> dict[str, dict]:
 
     e.g. {"sda": {"pool": "tank", "vdev": "raidz2-0"},
           "sdb": {"pool": "fast", "vdev": "mirror-0"}}
+
+    Hardware-gated producer; the poller persists the result to the store
+    and consumers read it from there.
     """
     pool_map = {}
     try:
@@ -26,6 +31,7 @@ async def get_zfs_pool_map() -> dict[str, dict]:
         else:
             cmd = [ZPOOL_BIN, "status", "-P"]
 
+        async with gate():
             proc = await asyncio.create_subprocess_exec(
                 *cmd,
                 stdout=asyncio.subprocess.PIPE,
@@ -94,6 +100,7 @@ async def get_zfs_pool_map() -> dict[str, dict]:
                     pass
     except FileNotFoundError:
         logger.debug("zpool not available")
+
     return pool_map