fix: resolve jeannie CI failures, add gitleaks ignore, apply ruff fixes, make scripts executable

.gitleaks.toml

@@ -0,0 +1,3 @@
+[allowlist]
+  description = "Ignore generated hash files in graphify-out cache"
+  path = "graphify-out/cache/.*"

.gitleaksignore

@@ -1,6 +1,3 @@
-# Ignore graphify-out test artifacts
+[allowlist]
-graphify-out/
+  description = "Ignore generated hash files in graphify-out cache"
-# Ignore any test tokens in fixtures or examples
+  path = "graphify-out/cache/.*"
-*/fixtures/*
-*/examples/*
-*/test/*

bootstrap/edge/main.tf

@@ -40,10 +40,10 @@ locals {
     backend_port = tostring(var.backend_port)
   })
   default_conf_matrix = var.matrix_enabled ? templatefile("${path.module}/templates/matrix-vhost.conf.tftpl", {
-    matrix_server_name  = var.matrix_server_name
+    matrix_server_name   = var.matrix_server_name
-    matrix_backend_host = var.matrix_backend_host
+    matrix_backend_host  = var.matrix_backend_host
-    matrix_backend_port = tostring(var.matrix_backend_port)
+    matrix_backend_port  = tostring(var.matrix_backend_port)
-    matrix_cert_dir     = var.matrix_cert_dir
+    matrix_cert_dir      = var.matrix_cert_dir
     element_server_name  = var.element_server_name
     element_backend_host = var.element_backend_host
     element_backend_port = tostring(var.element_backend_port)

bootstrap/platform/main.tf

@@ -1787,9 +1787,9 @@ resource "helm_release" "prometheus_stack" {
   create_namespace = false
   # Readiness is checked by the following resource so a blocked PVC or pod
   # produces actionable Kubernetes diagnostics instead of a provider timeout.
-  timeout          = 600
+  timeout         = 600
-  wait             = false
+  wait            = false
-  cleanup_on_fail  = true
+  cleanup_on_fail = true
 
   values = [
     yamlencode({

infra/arr-stack/configure-arr.py

@@ -11,7 +11,6 @@ import urllib.error
 import urllib.request
 from pathlib import Path
 
-
 SCRIPT_DIR = Path(__file__).resolve().parent
 ENV_FILE = Path(os.environ.get("ARR_ENV_FILE", SCRIPT_DIR / ".env"))
 
@@ -27,7 +26,9 @@ def load_env(path: Path) -> None:
         os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
 
 
-def request_json(method: str, url: str, api_key: str, payload: dict | None = None) -> object:
+def request_json(
+    method: str, url: str, api_key: str, payload: dict | None = None
+) -> object:
     data = None
     headers = {"X-Api-Key": api_key, "Accept": "application/json"}
     if payload is not None:
@@ -46,22 +47,28 @@ def wait_for_app(name: str, base_url: str, api_key: str) -> None:
         try:
             request_json("GET", f"{base_url}/api/v3/system/status", api_key)
             return
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             if attempt == 30:
-                raise RuntimeError(f"{name} did not become reachable at {base_url}") from exc
+                raise RuntimeError(
+                    f"{name} did not become reachable at {base_url}"
+                ) from exc
             time.sleep(2)
 
 
 def ensure_root_folder(name: str, base_url: str, api_key: str, path: str) -> None:
     existing = request_json("GET", f"{base_url}/api/v3/rootfolder", api_key)
-    if isinstance(existing, list) and any(item.get("path") == path for item in existing if isinstance(item, dict)):
+    if isinstance(existing, list) and any(
+        item.get("path") == path for item in existing if isinstance(item, dict)
+    ):
         print(f"{name}: root folder already exists: {path}")
         return
     request_json("POST", f"{base_url}/api/v3/rootfolder", api_key, {"path": path})
     print(f"{name}: added root folder: {path}")
 
 
-def schema_fields(base_url: str, api_key: str, endpoint: str, implementation: str) -> list[dict]:
+def schema_fields(
+    base_url: str, api_key: str, endpoint: str, implementation: str
+) -> list[dict]:
     schemas = request_json("GET", f"{base_url}/api/v3/{endpoint}/schema", api_key)
     if not isinstance(schemas, list):
         return []
@@ -88,7 +95,8 @@ def fill_fields(fields: list[dict], values: dict[str, object]) -> list[dict]:
 def ensure_qbittorrent(name: str, base_url: str, api_key: str, category: str) -> None:
     clients = request_json("GET", f"{base_url}/api/v3/downloadclient", api_key)
     if isinstance(clients, list) and any(
-        isinstance(item, dict) and item.get("implementation") == "QBittorrent" for item in clients
+        isinstance(item, dict) and item.get("implementation") == "QBittorrent"
+        for item in clients
     ):
         print(f"{name}: qBittorrent download client already exists")
         return
@@ -140,15 +148,19 @@ def apply_import_lists(name: str, base_url: str, api_key: str, path_env: str) ->
 
     desired = json.loads(path.read_text(encoding="utf-8"))
     if not isinstance(desired, list):
-        raise RuntimeError(f"{name}: {path} must contain a JSON array")
+        raise TypeError(f"{name}: {path} must contain a JSON array")
 
     existing = request_json("GET", f"{base_url}/api/v3/importlist", api_key)
     existing_names = set()
     if isinstance(existing, list):
-        existing_names = {item.get("name") for item in existing if isinstance(item, dict)}
+        existing_names = {
+            item.get("name") for item in existing if isinstance(item, dict)
+        }
     for item in desired:
         if not isinstance(item, dict) or not item.get("name"):
-            raise RuntimeError(f"{name}: every import-list entry must be an object with a name")
+            raise RuntimeError(
+                f"{name}: every import-list entry must be an object with a name"
+            )
         if item["name"] in existing_names:
             print(f"{name}: import list already exists: {item['name']}")
             continue
@@ -173,8 +185,18 @@ def main() -> int:
 
     wait_for_app("Radarr", radarr_url, radarr_key)
     wait_for_app("Sonarr", sonarr_url, sonarr_key)
-    ensure_root_folder("Radarr", radarr_url, radarr_key, os.environ.get("RADARR_ROOT_FOLDER", "/data/media/movies"))
+    ensure_root_folder(
-    ensure_root_folder("Sonarr", sonarr_url, sonarr_key, os.environ.get("SONARR_ROOT_FOLDER", "/data/media/tv"))
+        "Radarr",
+        radarr_url,
+        radarr_key,
+        os.environ.get("RADARR_ROOT_FOLDER", "/data/media/movies"),
+    )
+    ensure_root_folder(
+        "Sonarr",
+        sonarr_url,
+        sonarr_key,
+        os.environ.get("SONARR_ROOT_FOLDER", "/data/media/tv"),
+    )
     ensure_qbittorrent("Radarr", radarr_url, radarr_key, "radarr")
     ensure_qbittorrent("Sonarr", sonarr_url, sonarr_key, "sonarr")
     apply_import_lists("Radarr", radarr_url, radarr_key, "RADARR_IMPORT_LISTS_JSON")

infra/heimdall/seed-heimdall.py

@@ -5,7 +5,6 @@ import sqlite3
 import time
 from datetime import datetime, timezone
 
-
 DB_CANDIDATES = (
     "/config/www/app.sqlite",
     "/config/app.sqlite",
@@ -138,16 +137,22 @@ def upsert_links(conn, links):
                 for name in values
                 if name in item_columns and name not in ("id", "title", "created_at")
             ]
-            assignments = ", ".join(f"{quote_identifier(name)} = ?" for name in update_columns)
+            assignments = ", ".join(
+                f"{quote_identifier(name)} = ?" for name in update_columns
+            )
             params = [values[name] for name in update_columns]
             params.append(existing[0])
             conn.execute(f'update "items" set {assignments} where "id" = ?', params)
             ensure_dashboard_tag(conn, existing[0], home_dashboard_tag, now)
             continue
 
-        insert_columns = [name for name in values if name in item_columns and name != "deleted_at"]
+        insert_columns = [
+            name for name in values if name in item_columns and name != "deleted_at"
+        ]
         placeholders = ", ".join("?" for _ in insert_columns)
-        quoted_insert_columns = ", ".join(quote_identifier(name) for name in insert_columns)
+        quoted_insert_columns = ", ".join(
+            quote_identifier(name) for name in insert_columns
+        )
         conn.execute(
             f'insert into "items" ({quoted_insert_columns}) values ({placeholders})',
             [values[name] for name in insert_columns],

labs/mlops-platform/app/main.py

@@ -7,7 +7,13 @@ from typing import Any
 
 from fastapi import FastAPI, HTTPException, Request, Response
 from fastapi.middleware.cors import CORSMiddleware
-from prometheus_client import CONTENT_TYPE_LATEST, Counter, Gauge, Histogram, generate_latest
+from prometheus_client import (
+    CONTENT_TYPE_LATEST,
+    Counter,
+    Gauge,
+    Histogram,
+    generate_latest,
+)
 from pydantic import BaseModel, ConfigDict, Field
 
 MODEL_VERSION = os.getenv("MODEL_VERSION", "v1")
@@ -80,10 +86,20 @@ def load_model(version: str) -> dict[str, Any]:
     with model_path.open(encoding="utf-8") as handle:
         model = json.load(handle)
 
-    required = {"version", "features", "weights", "bias", "threshold", "baseline", "trained_with"}
+    required = {
+        "version",
+        "features",
+        "weights",
+        "bias",
+        "threshold",
+        "baseline",
+        "trained_with",
+    }
     missing = required.difference(model)
     if missing:
-        raise RuntimeError(f"model artifact is missing required keys: {', '.join(sorted(missing))}")
+        raise RuntimeError(
+            f"model artifact is missing required keys: {', '.join(sorted(missing))}"
+        )
 
     if len(model["features"]) != len(model["weights"]):
         raise RuntimeError("model features and weights have different lengths")
@@ -120,8 +136,12 @@ async def record_http_metrics(request: Request, call_next: Any) -> Response:
     finally:
         endpoint = getattr(request.scope.get("route"), "path", route)
         elapsed = time.perf_counter() - start
-        REQUESTS.labels(endpoint, request.method, status, MODEL["version"], MODEL_TRACK).inc()
+        REQUESTS.labels(
-        REQUEST_LATENCY.labels(endpoint, request.method, MODEL["version"], MODEL_TRACK).observe(elapsed)
+            endpoint, request.method, status, MODEL["version"], MODEL_TRACK
+        ).inc()
+        REQUEST_LATENCY.labels(
+            endpoint, request.method, MODEL["version"], MODEL_TRACK
+        ).observe(elapsed)
 
 
 def normalized_features(features: dict[str, float]) -> list[float]:
@@ -140,7 +160,9 @@ def logistic(value: float) -> float:
 
 def score_prediction(features: dict[str, float]) -> float:
     score = MODEL["bias"]
-    for weight, value in zip(MODEL["weights"], normalized_features(features), strict=True):
+    for weight, value in zip(
+        MODEL["weights"], normalized_features(features), strict=True
+    ):
         score += weight * value
     return logistic(score)
 
@@ -183,7 +205,9 @@ def predict(payload: PredictRequest) -> dict[str, Any]:
         PREDICTIONS.labels(MODEL["version"], MODEL_TRACK, outcome).inc()
         CONFIDENCE.labels(MODEL["version"], MODEL_TRACK).set(confidence)
         DRIFT.labels(MODEL["version"], MODEL_TRACK).set(drift)
-        PREDICTION_LATENCY.labels(MODEL["version"], MODEL_TRACK).observe(time.perf_counter() - start)
+        PREDICTION_LATENCY.labels(MODEL["version"], MODEL_TRACK).observe(
+            time.perf_counter() - start
+        )
 
         return {
             "model_version": MODEL["version"],

labs/mlops-platform/training/export_model.py

@@ -18,12 +18,18 @@ OUTPUT_DIR = Path(__file__).resolve().parent.parent / "models"
 
 def export_model(version: str, seed: int, threshold: float) -> None:
     rng = np.random.default_rng(seed)
-    healthy = rng.normal([170, 0.015, 0.38, 0.45, 5], [45, 0.01, 0.12, 0.12, 4], size=(160, 5))
+    healthy = rng.normal(
-    at_risk = rng.normal([420, 0.12, 0.82, 0.78, 38], [130, 0.08, 0.12, 0.13, 18], size=(160, 5))
+        [170, 0.015, 0.38, 0.45, 5], [45, 0.01, 0.12, 0.12, 4], size=(160, 5)
+    )
+    at_risk = rng.normal(
+        [420, 0.12, 0.82, 0.78, 38], [130, 0.08, 0.12, 0.13, 18], size=(160, 5)
+    )
     x = np.vstack([healthy, at_risk])
     y = np.array([0] * len(healthy) + [1] * len(at_risk))
 
-    pipeline = make_pipeline(StandardScaler(), LogisticRegression(max_iter=1000, random_state=seed))
+    pipeline = make_pipeline(
+        StandardScaler(), LogisticRegression(max_iter=1000, random_state=seed)
+    )
     pipeline.fit(x, y)
 
     scaler = pipeline.named_steps["standardscaler"]
@@ -37,7 +43,9 @@ def export_model(version: str, seed: int, threshold: float) -> None:
         "weights": [float(value) for value in classifier.coef_[0]],
         "baseline": {
             feature: {"mean": float(mean), "stddev": float(stddev)}
-            for feature, mean, stddev in zip(FEATURES, scaler.mean_, scaler.scale_, strict=True)
+            for feature, mean, stddev in zip(
+                FEATURES, scaler.mean_, scaler.scale_, strict=True
+            )
         },
     }
 

scripts/agent-sandbox

@@ -6,10 +6,8 @@ from __future__ import annotations
 import argparse
 import csv
 import json
-import sys
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 POLICY_FILE = REPO_ROOT / "infra" / "agent-sandbox" / "policy.tsv"
 
@@ -24,9 +22,10 @@ def classify(command: str) -> dict[str, str]:
     best: dict[str, str] | None = None
     for rule in load_policy():
         pattern = rule["pattern"]
-        if pattern in normalized:
+        if pattern in normalized and (
-            if best is None or len(pattern) > len(best["pattern"]):
+            best is None or len(pattern) > len(best["pattern"])
-                best = rule
+        ):
+            best = rule
     if best is None:
         return {
             "command": normalized,

scripts/ai-evals

@@ -11,7 +11,6 @@ import sys
 from dataclasses import dataclass
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 CASES_FILE = REPO_ROOT / "infra" / "ai-evals" / "cases.tsv"
 FIXTURES_DIR = REPO_ROOT / "infra" / "ai-evals" / "fixtures"
@@ -147,7 +146,9 @@ def print_show(cases: list[Case], case_id: str) -> int:
     return 2
 
 
-def print_run(cases: list[Case], case_id: str | None, as_json: bool, details: bool) -> int:
+def print_run(
+    cases: list[Case], case_id: str | None, as_json: bool, details: bool
+) -> int:
     selected = [case for case in cases if case_id in (None, case.case_id)]
     if not selected:
         print(f"Unknown eval case: {case_id}", file=sys.stderr)

scripts/ai-memory

@@ -8,9 +8,10 @@ import json
 import os
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
-DEFAULT_INDEX_DIR = Path(os.environ.get("LAB_AI_KNOWLEDGE_INDEX_DIR", "/data/homelab-ai/index"))
+DEFAULT_INDEX_DIR = Path(
+    os.environ.get("LAB_AI_KNOWLEDGE_INDEX_DIR", "/data/homelab-ai/index")
+)
 
 
 def check(index_dir: Path) -> int:

scripts/ai-model-check.py

@@ -1,12 +1,15 @@
 import json
 import sys
+import urllib.error
 import urllib.request
 
 endpoint, model = sys.argv[1:3]
 try:
-    with urllib.request.urlopen("{}/api/tags".format(endpoint.rstrip("/")), timeout=5) as response:
+    with urllib.request.urlopen(
+        "{}/api/tags".format(endpoint.rstrip("/")), timeout=5
+    ) as response:
         tags = json.loads(response.read().decode("utf-8"))
     models = {item.get("name", "") for item in tags.get("models", [])}
-    sys.exit(0 if model in models or "{}:latest".format(model) in models else 1)
+    sys.exit(0 if model in models or f"{model}:latest" in models else 1)
-except Exception:
+except (urllib.error.URLError, json.JSONDecodeError, KeyError):
     sys.exit(1)

scripts/ai-scheduler

@@ -7,7 +7,6 @@ import argparse
 import csv
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 SCHED_DIR = REPO_ROOT / "infra" / "ai-scheduler"
 NODES_FILE = SCHED_DIR / "nodes.tsv"
@@ -23,7 +22,9 @@ def as_bool(value: str) -> bool:
     return value.strip().lower() in {"true", "yes", "1"}
 
 
-def score_node(workload: dict[str, str], node: dict[str, str]) -> tuple[int, list[str], list[str]]:
+def score_node(
+    workload: dict[str, str], node: dict[str, str]
+) -> tuple[int, list[str], list[str]]:
     score = 100
     reasons: list[str] = []
     blockers: list[str] = []

scripts/build-homelab-ai-index

@@ -14,11 +14,12 @@ import sys
 import time
 from collections import Counter
 
-
 REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
 DEFAULT_SOURCES_FILE = REPO_ROOT / "infra" / "ai" / "knowledge-sources.txt"
 DEFAULT_EXCLUDES_FILE = REPO_ROOT / "infra" / "ai" / "knowledge-excludes.txt"
-DEFAULT_INDEX_DIR = pathlib.Path(os.environ.get("LAB_AI_KNOWLEDGE_INDEX_DIR", "/data/homelab-ai/index"))
+DEFAULT_INDEX_DIR = pathlib.Path(
+    os.environ.get("LAB_AI_KNOWLEDGE_INDEX_DIR", "/data/homelab-ai/index")
+)
 TOKEN_RE = re.compile(r"[A-Za-z0-9_./:-]{2,}")
 MAX_FILE_BYTES = 512 * 1024
 CHUNK_TARGET_LINES = 80
@@ -60,7 +61,9 @@ def path_matches_any(rel_path: str, patterns: list[str]) -> bool:
     return any(fnmatch.fnmatch(rel_path, pattern) for pattern in patterns)
 
 
-def candidate_files(patterns: list[str], exclude_patterns: list[str]) -> list[pathlib.Path]:
+def candidate_files(
+    patterns: list[str], exclude_patterns: list[str]
+) -> list[pathlib.Path]:
     files: set[pathlib.Path] = set()
     all_files = [path for path in REPO_ROOT.rglob("*") if path.is_file()]
 
@@ -79,7 +82,9 @@ def candidate_files(patterns: list[str], exclude_patterns: list[str]) -> list[pa
         path
         for path in files
         if not path_is_secret(path)
-        and not path_matches_any(path.relative_to(REPO_ROOT).as_posix(), exclude_patterns)
+        and not path_matches_any(
+            path.relative_to(REPO_ROOT).as_posix(), exclude_patterns
+        )
     )
 
 
@@ -147,7 +152,11 @@ def build_index(files: list[pathlib.Path]) -> dict[str, object]:
             document_frequency.update(set(tokens))
 
     chunk_count = len(chunks)
-    average_length = sum(len(chunk["tokens"]) for chunk in chunks) / chunk_count if chunk_count else 0
+    average_length = (
+        sum(len(chunk["tokens"]) for chunk in chunks) / chunk_count
+        if chunk_count
+        else 0
+    )
     idf = {
         token: math.log(1 + (chunk_count - freq + 0.5) / (freq + 0.5))
         for token, freq in document_frequency.items()
@@ -165,8 +174,12 @@ def build_index(files: list[pathlib.Path]) -> dict[str, object]:
 
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--sources-file", type=pathlib.Path, default=DEFAULT_SOURCES_FILE)
+    parser.add_argument(
-    parser.add_argument("--excludes-file", type=pathlib.Path, default=DEFAULT_EXCLUDES_FILE)
+        "--sources-file", type=pathlib.Path, default=DEFAULT_SOURCES_FILE
+    )
+    parser.add_argument(
+        "--excludes-file", type=pathlib.Path, default=DEFAULT_EXCLUDES_FILE
+    )
     parser.add_argument("--index-dir", type=pathlib.Path, default=DEFAULT_INDEX_DIR)
     args = parser.parse_args()
 
@@ -175,20 +188,26 @@ def main() -> int:
         return 1
 
     patterns = load_patterns(args.sources_file)
-    exclude_patterns = load_patterns(args.excludes_file) if args.excludes_file.is_file() else []
+    exclude_patterns = (
+        load_patterns(args.excludes_file) if args.excludes_file.is_file() else []
+    )
     files = candidate_files(patterns, exclude_patterns)
     index = build_index(files)
 
     args.index_dir.mkdir(parents=True, exist_ok=True)
     index_path = args.index_dir / "index.json"
     manifest_path = args.index_dir / "manifest.json"
-    index_path.write_text(json.dumps(index, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    index_path.write_text(
+        json.dumps(index, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
     manifest_path.write_text(
         json.dumps(
             {
                 "generated_at_epoch": int(time.time()),
                 "sources_file": str(args.sources_file.relative_to(REPO_ROOT)),
-                "excludes_file": str(args.excludes_file.relative_to(REPO_ROOT)) if args.excludes_file.is_file() else None,
+                "excludes_file": str(args.excludes_file.relative_to(REPO_ROOT))
+                if args.excludes_file.is_file()
+                else None,
                 "file_count": len(files),
                 "chunk_count": index["chunk_count"],
                 "repo_root": str(REPO_ROOT),
@@ -204,7 +223,9 @@ def main() -> int:
         encoding="utf-8",
     )
 
-    print(f"indexed {len(files)} file(s), {index['chunk_count']} chunk(s) into {index_path}")
+    print(
+        f"indexed {len(files)} file(s), {index['chunk_count']} chunk(s) into {index_path}"
+    )
     return 0
 
 

scripts/heal

@@ -13,11 +13,21 @@ import time
 from dataclasses import dataclass
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
-DEFAULT_KUBECONFIG = Path(os.environ.get("KUBECONFIG", os.environ.get("LAB_KUBECONFIG_PATH", "/home/jv/.kube/config")))
+DEFAULT_KUBECONFIG = Path(
-STATE_DIR = Path(os.environ.get("HOMELAB_STATE_DIR", Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local/share")) / "homelab"))
+    os.environ.get(
-HEAL_STATE_FILE = Path(os.environ.get("LAB_HEAL_STATE_FILE", STATE_DIR / "heal-state.json"))
+        "KUBECONFIG", os.environ.get("LAB_KUBECONFIG_PATH", "/home/jv/.kube/config")
+    )
+)
+STATE_DIR = Path(
+    os.environ.get(
+        "HOMELAB_STATE_DIR",
+        Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local/share")) / "homelab",
+    )
+)
+HEAL_STATE_FILE = Path(
+    os.environ.get("LAB_HEAL_STATE_FILE", STATE_DIR / "heal-state.json")
+)
 HEAL_COOLDOWN_SECONDS = int(os.environ.get("LAB_HEAL_COOLDOWN_SECONDS", "1800"))
 
 
@@ -32,23 +42,142 @@ class HealRule:
 
 
 RULES = [
-    HealRule("Kubernetes API", "./jeannie start-cluster", "low", "API is down; start-cluster starts kubelet/containerd and worker VMs without destroying state.", True, 100),
+    HealRule(
-    HealRule("Pimox workers running", "./jeannie start-cluster", "low", "Worker VMs are expected cluster capacity and start-cluster is idempotent.", True, 95),
+        "Kubernetes API",
-    HealRule("Kubernetes nodes Ready", "./jeannie start-cluster", "low", "NotReady worker nodes commonly recover by starting the saved cluster runtime.", True, 90),
+        "./jeannie start-cluster",
-    HealRule("Pi-hole DNS", "./jeannie rpi-services", "medium", "Reapplies RPi DNS services; safe for the lab but can briefly disrupt DNS.", True, 80),
+        "low",
-    HealRule("RPi Docker root state", "./jeannie rpi-services", "medium", "May repair service runtime but should be reviewed if storage is failing.", True, 78),
+        "API is down; start-cluster starts kubelet/containerd and worker VMs without destroying state.",
-    HealRule("Uptime Kuma HTTP", "./jeannie rpi-services", "medium", "Reapplies RPi service Compose stack.", True, 70),
+        True,
-    HealRule("Gitea container", "./jeannie deploy-gitea", "medium", "Reapplies the Debian-hosted Gitea Compose service.", True, 75),
+        100,
-    HealRule("Gitea local HTTP", "./jeannie deploy-gitea", "medium", "Local Gitea HTTP is down; redeploying Compose is usually safe but still mutates Git service runtime.", True, 74),
+    ),
-    HealRule("Traefik deployment", "./jeannie promote validate", "diagnostic", "Validate gates first; deployment rollout failures are usually downstream of cluster/node health.", False, 65),
+    HealRule(
-    HealRule("Website deployment", "./jeannie promote validate", "diagnostic", "Validate gates first; deployment rollout failures are usually downstream of cluster/node health.", False, 64),
+        "Pimox workers running",
-    HealRule("Traefik LoadBalancer HTTP", "./jeannie doctor-edge", "diagnostic", "Edge/LB failures need diagnosis after cluster nodes are healthy.", False, 60),
+        "./jeannie start-cluster",
-    HealRule("Website public URL", "./jeannie doctor-edge", "diagnostic", "Public URL failures need edge and cluster diagnosis after local services recover.", False, 58),
+        "low",
-    HealRule("Gitea public route", "./jeannie doctor-gitea", "diagnostic", "Public Gitea route needs Gitea and edge diagnosis.", False, 55),
+        "Worker VMs are expected cluster capacity and start-cluster is idempotent.",
-    HealRule("No problem pods", "./jeannie explain status", "diagnostic", "Pod states need root-cause context; avoid blind deletes.", False, 50),
+        True,
-    HealRule("Recent deployments healthy", "./jeannie explain status", "diagnostic", "Deployment drift can be a symptom of node health, image pulls, or scheduling.", False, 45),
+        95,
-    HealRule("Pod restart pressure", "./jeannie explain status", "diagnostic", "Restart pressure needs workload-specific diagnosis before mutation.", False, 40),
+    ),
-    HealRule("Traefik 5xx/404 signals", "./jeannie doctor-edge", "diagnostic", "Use edge logs and route checks before changing config.", False, 35),
+    HealRule(
+        "Kubernetes nodes Ready",
+        "./jeannie start-cluster",
+        "low",
+        "NotReady worker nodes commonly recover by starting the saved cluster runtime.",
+        True,
+        90,
+    ),
+    HealRule(
+        "Pi-hole DNS",
+        "./jeannie rpi-services",
+        "medium",
+        "Reapplies RPi DNS services; safe for the lab but can briefly disrupt DNS.",
+        True,
+        80,
+    ),
+    HealRule(
+        "RPi Docker root state",
+        "./jeannie rpi-services",
+        "medium",
+        "May repair service runtime but should be reviewed if storage is failing.",
+        True,
+        78,
+    ),
+    HealRule(
+        "Uptime Kuma HTTP",
+        "./jeannie rpi-services",
+        "medium",
+        "Reapplies RPi service Compose stack.",
+        True,
+        70,
+    ),
+    HealRule(
+        "Gitea container",
+        "./jeannie deploy-gitea",
+        "medium",
+        "Reapplies the Debian-hosted Gitea Compose service.",
+        True,
+        75,
+    ),
+    HealRule(
+        "Gitea local HTTP",
+        "./jeannie deploy-gitea",
+        "medium",
+        "Local Gitea HTTP is down; redeploying Compose is usually safe but still mutates Git service runtime.",
+        True,
+        74,
+    ),
+    HealRule(
+        "Traefik deployment",
+        "./jeannie promote validate",
+        "diagnostic",
+        "Validate gates first; deployment rollout failures are usually downstream of cluster/node health.",
+        False,
+        65,
+    ),
+    HealRule(
+        "Website deployment",
+        "./jeannie promote validate",
+        "diagnostic",
+        "Validate gates first; deployment rollout failures are usually downstream of cluster/node health.",
+        False,
+        64,
+    ),
+    HealRule(
+        "Traefik LoadBalancer HTTP",
+        "./jeannie doctor-edge",
+        "diagnostic",
+        "Edge/LB failures need diagnosis after cluster nodes are healthy.",
+        False,
+        60,
+    ),
+    HealRule(
+        "Website public URL",
+        "./jeannie doctor-edge",
+        "diagnostic",
+        "Public URL failures need edge and cluster diagnosis after local services recover.",
+        False,
+        58,
+    ),
+    HealRule(
+        "Gitea public route",
+        "./jeannie doctor-gitea",
+        "diagnostic",
+        "Public Gitea route needs Gitea and edge diagnosis.",
+        False,
+        55,
+    ),
+    HealRule(
+        "No problem pods",
+        "./jeannie explain status",
+        "diagnostic",
+        "Pod states need root-cause context; avoid blind deletes.",
+        False,
+        50,
+    ),
+    HealRule(
+        "Recent deployments healthy",
+        "./jeannie explain status",
+        "diagnostic",
+        "Deployment drift can be a symptom of node health, image pulls, or scheduling.",
+        False,
+        45,
+    ),
+    HealRule(
+        "Pod restart pressure",
+        "./jeannie explain status",
+        "diagnostic",
+        "Restart pressure needs workload-specific diagnosis before mutation.",
+        False,
+        40,
+    ),
+    HealRule(
+        "Traefik 5xx/404 signals",
+        "./jeannie doctor-edge",
+        "diagnostic",
+        "Use edge logs and route checks before changing config.",
+        False,
+        35,
+    ),
 ]
 
 
@@ -59,9 +188,8 @@ def load_status(path: Path | None) -> dict[str, object]:
     process = subprocess.run(
         [str(REPO_ROOT / "jeannie"), "status", "--json"],
         cwd=REPO_ROOT,
+        capture_output=True,
         text=True,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
         check=False,
     )
     if not process.stdout.strip():
@@ -74,7 +202,11 @@ def status_rows(status: dict[str, object]) -> list[dict[str, str]]:
     rows = status.get("rows", [])
     if not isinstance(rows, list):
         return []
-    return [row for row in rows if isinstance(row, dict) and row.get("status") in {"fail", "warn"}]
+    return [
+        row
+        for row in rows
+        if isinstance(row, dict) and row.get("status") in {"fail", "warn"}
+    ]
 
 
 def rule_for(row: dict[str, str]) -> HealRule | None:
@@ -155,20 +287,32 @@ def build_plan(status: dict[str, object]) -> list[dict[str, object]]:
 
 def impact_sort_key(item: dict[str, object]) -> tuple[int, int, str]:
     status_score = 10 if item.get("status") == "fail" else 0
-    return (int(item.get("priority") or 0) + status_score, 1 if item.get("auto") else 0, str(item.get("check") or ""))
+    return (
+        int(item.get("priority") or 0) + status_score,
+        1 if item.get("auto") else 0,
+        str(item.get("check") or ""),
+    )
 
 
-def highest_impact(plan: list[dict[str, object]], *, auto_only: bool) -> dict[str, object] | None:
+def highest_impact(
-    candidates = [item for item in plan if item.get("auto")] if auto_only else list(plan)
+    plan: list[dict[str, object]], *, auto_only: bool
+) -> dict[str, object] | None:
+    candidates = (
+        [item for item in plan if item.get("auto")] if auto_only else list(plan)
+    )
     if not candidates:
         return None
-    return sorted(candidates, key=impact_sort_key, reverse=True)[0]
+    return max(candidates, key=impact_sort_key)
 
 
 def print_plan(plan: list[dict[str, object]], ai: bool) -> None:
     target = highest_impact(plan, auto_only=True)
     top_overall = highest_impact(plan, auto_only=False)
-    deferred = [item for item in sorted(plan, key=impact_sort_key, reverse=True) if item is not target]
+    deferred = [
+        item
+        for item in sorted(plan, key=impact_sort_key, reverse=True)
+        if item is not target
+    ]
     print("Jeannie Heal Plan")
     print("=================")
     print("mode: one-at-a-time")
@@ -183,14 +327,18 @@ def print_plan(plan: list[dict[str, object]], ai: bool) -> None:
     else:
         print("Next heal target: none")
         if top_overall:
-            print(f"highest finding is diagnostic-only: {top_overall['area']} / {top_overall['check']}")
+            print(
+                f"highest finding is diagnostic-only: {top_overall['area']} / {top_overall['check']}"
+            )
             print(f"next diagnostic command: {top_overall['command']}")
     print()
     if deferred:
         print("Deferred until next status/heal cycle:")
         for item in deferred[:8]:
             auto_marker = "auto" if item.get("auto") else "diagnostic"
-            print(f"  - {item['area']} / {item['check']} ({auto_marker}, impact={item['priority']})")
+            print(
+                f"  - {item['area']} / {item['check']} ({auto_marker}, impact={item['priority']})"
+            )
     if ai:
         print_ai_context([target] if target else ([top_overall] if top_overall else []))
 
@@ -200,7 +348,13 @@ def print_ai_context(plan: list[dict[str, object]]) -> None:
     if not query:
         return
     process = subprocess.run(
-        [str(REPO_ROOT / "scripts" / "query-homelab-ai-index"), "--citations", "--limit", "3", query],
+        [
+            str(REPO_ROOT / "scripts" / "query-homelab-ai-index"),
+            "--citations",
+            "--limit",
+            "3",
+            query,
+        ],
         cwd=REPO_ROOT,
         text=True,
         stdout=subprocess.PIPE,
@@ -258,7 +412,9 @@ def apply_plan(plan: list[dict[str, object]], yes: bool) -> int:
     if not yes:
         print()
         print("Dry run only. Re-run with: ./jeannie heal apply --yes")
-        print("After it runs, use ./jeannie status again before healing the next finding.")
+        print(
+            "After it runs, use ./jeannie status again before healing the next finding."
+        )
     else:
         print("Run ./jeannie status again before healing the next finding.")
     return 0
@@ -280,7 +436,9 @@ def load_heal_state() -> dict[str, object]:
 
 def write_heal_state(document: dict[str, object]) -> None:
     HEAL_STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
-    HEAL_STATE_FILE.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    HEAL_STATE_FILE.write_text(
+        json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
 
 
 def recent_failed_commands() -> dict[str, str]:

scripts/incident-commander

@@ -11,7 +11,6 @@ import sys
 from dataclasses import dataclass
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 INCIDENT_DIR = REPO_ROOT / "infra" / "incident-commander"
 INCIDENTS_FILE = INCIDENT_DIR / "incidents.tsv"
@@ -30,10 +29,26 @@ class Incident:
 
 
 RULES: list[tuple[str, str, str]] = [
-    ("edge_bad_gateway", r"\b502\b|bad gateway|nginx/1\.31", "Public edge proxy cannot get a healthy upstream response."),
+    (
-    ("gitea_edge_backend", r"cannot reach gitea backend|100\.85\.138\.30:3000|/git/", "Edge-to-Gitea backend path is broken or Gitea is unhealthy."),
+        "edge_bad_gateway",
-    ("cluster_api_down", r"6443.*refused|api server.*refused|connection to the server .* was refused", "Kubernetes API is not accepting connections."),
+        r"\b502\b|bad gateway|nginx/1\.31",
-    ("dns_failure", r"no servers could be reached|communications error .*#53|timed out", "RPi/Pi-hole DNS path is unavailable or blocked."),
+        "Public edge proxy cannot get a healthy upstream response.",
+    ),
+    (
+        "gitea_edge_backend",
+        r"cannot reach gitea backend|100\.85\.138\.30:3000|/git/",
+        "Edge-to-Gitea backend path is broken or Gitea is unhealthy.",
+    ),
+    (
+        "cluster_api_down",
+        r"6443.*refused|api server.*refused|connection to the server .* was refused",
+        "Kubernetes API is not accepting connections.",
+    ),
+    (
+        "dns_failure",
+        r"no servers could be reached|communications error .*#53|timed out",
+        "RPi/Pi-hole DNS path is unavailable or blocked.",
+    ),
 ]
 
 
@@ -70,10 +85,15 @@ def classify(text: str) -> tuple[str, str]:
     for incident_class, pattern, hypothesis in RULES:
         if re.search(pattern, lowered, re.IGNORECASE | re.DOTALL):
             return incident_class, hypothesis
-    return "unknown", "No known incident pattern matched. Preserve evidence and run read-only status checks."
+    return (
+        "unknown",
+        "No known incident pattern matched. Preserve evidence and run read-only status checks.",
+    )
 
 
-def incident_for_class(incidents: list[Incident], incident_class: str) -> Incident | None:
+def incident_for_class(
+    incidents: list[Incident], incident_class: str
+) -> Incident | None:
     for incident in incidents:
         if incident.expected_class == incident_class:
             return incident
@@ -89,7 +109,11 @@ def render_triage(text: str, incidents: list[Incident]) -> dict[str, object]:
             "hypothesis": hypothesis,
             "runbook": "none",
             "next_commands": ["./jeannie status", "./jeannie scorecard"],
-            "forbidden_actions": ["destructive commands", "secret exposure", "unverified manual fixes"],
+            "forbidden_actions": [
+                "destructive commands",
+                "secret exposure",
+                "unverified manual fixes",
+            ],
             "evidence": first_evidence_lines(text),
         }
     return {
@@ -135,8 +159,14 @@ def print_list(incidents: list[Incident]) -> int:
     return 0
 
 
-def replay(incidents: list[Incident], incident_id: str | None, as_json: bool, details: bool) -> int:
+def replay(
-    selected = [incident for incident in incidents if incident_id in (None, incident.incident_id)]
+    incidents: list[Incident], incident_id: str | None, as_json: bool, details: bool
+) -> int:
+    selected = [
+        incident
+        for incident in incidents
+        if incident_id in (None, incident.incident_id)
+    ]
     if not selected:
         print(f"Unknown incident fixture: {incident_id}", file=sys.stderr)
         return 2
@@ -146,7 +176,14 @@ def replay(incidents: list[Incident], incident_id: str | None, as_json: bool, de
         text = fixture_text(incident)
         result = render_triage(text, incidents)
         status = "pass" if result["class"] == incident.expected_class else "fail"
-        results.append({"id": incident.incident_id, "status": status, "expected": incident.expected_class, **result})
+        results.append(
+            {
+                "id": incident.incident_id,
+                "status": status,
+                "expected": incident.expected_class,
+                **result,
+            }
+        )
 
     if as_json:
         print(json.dumps(results, indent=2, sort_keys=True))

scripts/model-observe

@@ -12,10 +12,11 @@ import urllib.error
 import urllib.request
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 PROMPTS_FILE = REPO_ROOT / "infra" / "model-observatory" / "prompts.tsv"
-STATE_DIR = Path(os.environ.get("HOMELAB_STATE_DIR", Path.home() / ".local/share/homelab"))
+STATE_DIR = Path(
+    os.environ.get("HOMELAB_STATE_DIR", Path.home() / ".local/share/homelab")
+)
 OUTPUT_DIR = STATE_DIR / "model-observatory"
 
 
@@ -39,7 +40,9 @@ def offline_answer(row: dict[str, str]) -> str:
     return answers.get(row["id"], "No offline answer exists for this prompt.")
 
 
-def ask_ollama(prompt: str, endpoint: str, model: str, timeout: int) -> tuple[str, float]:
+def ask_ollama(
+    prompt: str, endpoint: str, model: str, timeout: int
+) -> tuple[str, float]:
     payload = {
         "model": model,
         "prompt": prompt,
@@ -60,8 +63,14 @@ def ask_ollama(prompt: str, endpoint: str, model: str, timeout: int) -> tuple[st
 
 def grade(row: dict[str, str], answer: str) -> dict[str, object]:
     lowered = answer.lower()
-    missing = [term for term in split_terms(row["required_terms"]) if term.lower() not in lowered]
+    missing = [
-    forbidden = [term for term in split_terms(row["forbidden_terms"]) if term.lower() in lowered]
+        term
+        for term in split_terms(row["required_terms"])
+        if term.lower() not in lowered
+    ]
+    forbidden = [
+        term for term in split_terms(row["forbidden_terms"]) if term.lower() in lowered
+    ]
     return {
         "id": row["id"],
         "category": row["category"],
@@ -87,7 +96,12 @@ def run(offline: bool, live: bool) -> int:
         else:
             try:
                 answer, latency = ask_ollama(row["prompt"], endpoint, model, timeout)
-            except (OSError, TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc:
+            except (
+                OSError,
+                TimeoutError,
+                urllib.error.URLError,
+                json.JSONDecodeError,
+            ) as exc:
                 answer = f"ERROR: {exc}"
                 latency = 0.0
         result = grade(row, answer)
@@ -97,12 +111,16 @@ def run(offline: bool, live: bool) -> int:
 
     OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
     output_path = OUTPUT_DIR / f"run-{int(time.time())}.json"
-    output_path.write_text(json.dumps(results, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    output_path.write_text(
+        json.dumps(results, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
     failures = sum(1 for result in results if result["status"] != "pass")
     print("Model Behavior Observatory")
     print("==========================")
     for result in results:
-        print(f"{result['status']:5} {result['id']} latency={result['latency_seconds']}s")
+        print(
+            f"{result['status']:5} {result['id']} latency={result['latency_seconds']}s"
+        )
         if result["missing"]:
             print(f"      missing: {', '.join(result['missing'])}")
         if result["forbidden"]:
@@ -129,7 +147,9 @@ def report() -> int:
     print(f"cases: {len(data)}")
     print(f"failures: {failures}")
     for item in data:
-        print(f"{item.get('status'):5} {item.get('id')} model={item.get('model')} latency={item.get('latency_seconds')}s")
+        print(
+            f"{item.get('status'):5} {item.get('id')} model={item.get('model')} latency={item.get('latency_seconds')}s"
+        )
     return 1 if failures else 0
 
 

scripts/promote

@@ -8,7 +8,6 @@ import csv
 import subprocess
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 CHECKS_FILE = REPO_ROOT / "infra" / "promotion" / "checks.tsv"
 
@@ -76,7 +75,9 @@ def main() -> int:
     subparsers = parser.add_subparsers(dest="command", required=True)
     subparsers.add_parser("plan")
     validate_parser = subparsers.add_parser("validate")
-    validate_parser.add_argument("--all", action="store_true", help="Include optional live checks.")
+    validate_parser.add_argument(
+        "--all", action="store_true", help="Include optional live checks."
+    )
     rollback_parser = subparsers.add_parser("rollback")
     rollback_parser.add_argument("app")
     args = parser.parse_args()

scripts/prompt-injection-lab

@@ -10,7 +10,6 @@ import sys
 from dataclasses import dataclass
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 LAB_DIR = REPO_ROOT / "security" / "prompt-injection-lab"
 CASES_FILE = LAB_DIR / "cases.tsv"
@@ -86,7 +85,9 @@ def defense_response(case: Case) -> str:
             f"Case: {case.case_id}",
             f"Attack class: {case.attack_class}",
             "Defense:",
-            *rules.get(case.attack_class, ["No defense rule exists for this attack class."]),
+            *rules.get(
+                case.attack_class, ["No defense rule exists for this attack class."]
+            ),
         ]
     )
 
@@ -128,7 +129,9 @@ def print_show(cases: list[Case], case_id: str) -> int:
     return 2
 
 
-def print_run(cases: list[Case], case_id: str | None, as_json: bool, details: bool) -> int:
+def print_run(
+    cases: list[Case], case_id: str | None, as_json: bool, details: bool
+) -> int:
     selected = [case for case in cases if case_id in (None, case.case_id)]
     if not selected:
         print(f"Unknown prompt-injection case: {case_id}", file=sys.stderr)

scripts/query-homelab-ai-index

@@ -5,7 +5,6 @@ from __future__ import annotations
 
 import argparse
 import json
-import math
 import os
 import pathlib
 import re
@@ -14,8 +13,9 @@ import urllib.error
 import urllib.request
 from collections import Counter
 
-
+DEFAULT_INDEX_DIR = pathlib.Path(
-DEFAULT_INDEX_DIR = pathlib.Path(os.environ.get("LAB_AI_KNOWLEDGE_INDEX_DIR", "/data/homelab-ai/index"))
+    os.environ.get("LAB_AI_KNOWLEDGE_INDEX_DIR", "/data/homelab-ai/index")
+)
 TOKEN_RE = re.compile(r"[A-Za-z0-9_./:-]{2,}")
 
 
@@ -30,7 +30,9 @@ def load_index(index_dir: pathlib.Path) -> dict[str, object]:
     return json.loads(index_path.read_text(encoding="utf-8"))
 
 
-def score_chunks(index: dict[str, object], query: str, limit: int) -> list[tuple[float, dict[str, object]]]:
+def score_chunks(
+    index: dict[str, object], query: str, limit: int
+) -> list[tuple[float, dict[str, object]]]:
     query_terms = tokenise(query)
     if not query_terms:
         return []
@@ -68,7 +70,9 @@ def score_chunks(index: dict[str, object], query: str, limit: int) -> list[tuple
     return sorted(results, key=lambda item: item[0], reverse=True)[:limit]
 
 
-def render_context(results: list[tuple[float, dict[str, object]]], max_chars: int) -> str:
+def render_context(
+    results: list[tuple[float, dict[str, object]]], max_chars: int
+) -> str:
     sections: list[str] = []
     remaining = max_chars
 
@@ -86,7 +90,9 @@ def render_context(results: list[tuple[float, dict[str, object]]], max_chars: in
     return "\n\n---\n\n".join(sections)
 
 
-def render_citations(results: list[tuple[float, dict[str, object]]], max_chars: int) -> str:
+def render_citations(
+    results: list[tuple[float, dict[str, object]]], max_chars: int
+) -> str:
     sections = ["Jeannie Memory With Provenance", "=============================", ""]
     remaining = max_chars
     for index, (score, chunk) in enumerate(results, start=1):
@@ -107,11 +113,15 @@ def render_citations(results: list[tuple[float, dict[str, object]]], max_chars:
         remaining -= len(rendered)
         if remaining <= 0:
             break
-    sections.append("Rule: answers must cite these source paths or say the context is insufficient.")
+    sections.append(
+        "Rule: answers must cite these source paths or say the context is insufficient."
+    )
     return "\n".join(sections)
 
 
-def ask_ollama(question: str, context: str, endpoint: str, model: str, timeout: int) -> str:
+def ask_ollama(
+    question: str, context: str, endpoint: str, model: str, timeout: int
+) -> str:
     prompt = f"""You are helping operate a personal homelab.
 Use only the context below. If the context is insufficient, say what to inspect next.
 Be concise, factual, and command-oriented.
@@ -151,9 +161,18 @@ def main() -> int:
     parser.add_argument("--context-only", action="store_true")
     parser.add_argument("--citations", action="store_true")
     parser.add_argument("--ask", action="store_true")
-    parser.add_argument("--ollama-url", default=os.environ.get("LAB_AI_GATEWAY_URL", "http://127.0.0.1:11434"))
+    parser.add_argument(
-    parser.add_argument("--model", default=os.environ.get("LAB_AI_GATEWAY_MODEL", "qwen2.5:0.5b"))
+        "--ollama-url",
-    parser.add_argument("--timeout", type=int, default=int(os.environ.get("LAB_AI_GATEWAY_TIMEOUT_SECONDS", "20")))
+        default=os.environ.get("LAB_AI_GATEWAY_URL", "http://127.0.0.1:11434"),
+    )
+    parser.add_argument(
+        "--model", default=os.environ.get("LAB_AI_GATEWAY_MODEL", "qwen2.5:0.5b")
+    )
+    parser.add_argument(
+        "--timeout",
+        type=int,
+        default=int(os.environ.get("LAB_AI_GATEWAY_TIMEOUT_SECONDS", "20")),
+    )
     args = parser.parse_args()
 
     query = " ".join(args.query)
@@ -179,8 +198,15 @@ def main() -> int:
 
     if args.ask:
         try:
-            answer = ask_ollama(query, context, args.ollama_url, args.model, args.timeout)
+            answer = ask_ollama(
-        except (OSError, TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc:
+                query, context, args.ollama_url, args.model, args.timeout
+            )
+        except (
+            OSError,
+            TimeoutError,
+            urllib.error.URLError,
+            json.JSONDecodeError,
+        ) as exc:
             print(f"ollama request failed: {exc}", file=sys.stderr)
             return 1
         print(answer)

scripts/red-blue-loop

@@ -8,7 +8,6 @@ import csv
 import subprocess
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 CHECKS_FILE = REPO_ROOT / "infra" / "red-blue-loop" / "checks.tsv"
 LEDGER_FILE = REPO_ROOT / "infra" / "red-blue-loop" / "ledger.tsv"
@@ -67,7 +66,9 @@ def main() -> int:
     subparsers = parser.add_subparsers(dest="command", required=True)
     subparsers.add_parser("plan")
     run_parser = subparsers.add_parser("run")
-    run_parser.add_argument("--local", action="store_true", help="Run only deterministic local checks.")
+    run_parser.add_argument(
+        "--local", action="store_true", help="Run only deterministic local checks."
+    )
     subparsers.add_parser("ledger")
     args = parser.parse_args()
     if args.command == "plan":

scripts/render-service-catalog

@@ -9,11 +9,12 @@ import pathlib
 import re
 import sys
 
-
 REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
 INVENTORY = REPO_ROOT / "homelab.yml"
 DOC_OUTPUT = REPO_ROOT / "docs" / "service-catalog.md"
-HTML_OUTPUT = REPO_ROOT / "apps" / "demos-static" / "public" / "homelab-catalog" / "index.html"
+HTML_OUTPUT = (
+    REPO_ROOT / "apps" / "demos-static" / "public" / "homelab-catalog" / "index.html"
+)
 
 
 def parse_inventory(path: pathlib.Path) -> dict[str, object]:
@@ -295,7 +296,8 @@ def host_rows(values: dict[str, object]) -> list[dict[str, str]]:
             {
                 "name": host,
                 "role": value(values, f"hosts.{host}.role"),
-                "lan": value(values, f"hosts.{host}.lan_ip") or value(values, f"hosts.{host}.public_ip"),
+                "lan": value(values, f"hosts.{host}.lan_ip")
+                or value(values, f"hosts.{host}.public_ip"),
                 "tailscale": value(values, f"hosts.{host}.tailscale_ip"),
                 "storage": value(values, f"hosts.{host}.docker_root")
                 or value(values, f"hosts.{host}.docker_nvme_root")
@@ -306,7 +308,10 @@ def host_rows(values: dict[str, object]) -> list[dict[str, str]]:
 
 
 def markdown_table(headers: list[str], rows: list[list[str]]) -> str:
-    lines = ["| " + " | ".join(headers) + " |", "| " + " | ".join("---" for _ in headers) + " |"]
+    lines = [
+        "| " + " | ".join(headers) + " |",
+        "| " + " | ".join("---" for _ in headers) + " |",
+    ]
     for row in rows:
         lines.append("| " + " | ".join(cell.replace("\n", " ") for cell in row) + " |")
     return "\n".join(lines)
@@ -325,7 +330,10 @@ def render_markdown(values: dict[str, object]) -> str:
     services = service_rows(values)
     host_table = markdown_table(
         ["Host", "Role", "Address", "Tailscale", "Storage"],
-        [[row["name"], row["role"], row["lan"], row["tailscale"], row["storage"]] for row in hosts],
+        [
+            [row["name"], row["role"], row["lan"], row["tailscale"], row["storage"]]
+            for row in hosts
+        ],
     )
     service_table = markdown_table(
         ["Service", "Host", "Managed By", "Local", "Public", "Docs"],
@@ -469,7 +477,11 @@ def check_file(path: pathlib.Path, content: str) -> bool:
 
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--check", action="store_true", help="fail if generated catalog artifacts are stale")
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="fail if generated catalog artifacts are stale",
+    )
     args = parser.parse_args()
 
     values = parse_inventory(INVENTORY)

scripts/safety-case

@@ -6,10 +6,8 @@ from __future__ import annotations
 import argparse
 import csv
 import subprocess
-import sys
 from pathlib import Path
 
-
 REPO_ROOT = Path(__file__).resolve().parents[1]
 RULES_FILE = REPO_ROOT / "infra" / "jeannie-impact" / "rules.tsv"
 
@@ -17,9 +15,8 @@ RULES_FILE = REPO_ROOT / "infra" / "jeannie-impact" / "rules.tsv"
 def git_lines(args: list[str]) -> list[str]:
     process = subprocess.run(
         ["git", "-C", str(REPO_ROOT), *args],
+        capture_output=True,
         text=True,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
         check=False,
     )
     if process.returncode != 0:
@@ -99,13 +96,17 @@ def print_safety_case(files: list[str], rules: list[dict[str, str]]) -> int:
     print("  ./jeannie nuke")
     print("  terraform/tofu destroy")
     print("  deleting PVCs, namespaces, Gitea data, or Docker volumes")
-    print("  applying unreviewed shell commands from logs, dashboards, or generated text")
+    print(
+        "  applying unreviewed shell commands from logs, dashboards, or generated text"
+    )
     return 0
 
 
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--since", help="Git ref to compare against. Defaults to working tree.")
+    parser.add_argument(
+        "--since", help="Git ref to compare against. Defaults to working tree."
+    )
     parser.add_argument("paths", nargs="*", help="Explicit changed paths to review.")
     args = parser.parse_args()
     files = args.paths or changed_files(args.since)

scripts/validate-tailnet-policy

@@ -8,7 +8,6 @@ import pathlib
 import re
 import sys
 
-
 REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
 POLICY_PATH = REPO_ROOT / "infra" / "tailscale" / "tailnet-policy.hujson"
 INVENTORY_PATH = REPO_ROOT / "homelab.yml"
@@ -101,7 +100,11 @@ def parse_simple_inventory(path: pathlib.Path) -> dict[str, str]:
     pattern = re.compile(r"^(\s*)([A-Za-z_][A-Za-z0-9_]*):(?:\s*(.*?))?\s*$")
 
     for raw_line in path.read_text(encoding="utf-8").splitlines():
-        if not raw_line.strip() or raw_line.lstrip().startswith("#") or raw_line.lstrip().startswith("- "):
+        if (
+            not raw_line.strip()
+            or raw_line.lstrip().startswith("#")
+            or raw_line.lstrip().startswith("- ")
+        ):
             continue
         match = pattern.match(raw_line)
         if not match:
@@ -133,8 +136,16 @@ def main() -> int:
     inventory = parse_simple_inventory(INVENTORY_PATH)
 
     require(isinstance(policy.get("hosts"), dict), "policy must define hosts", failures)
-    require(isinstance(policy.get("acls"), list) and policy["acls"], "policy must define non-empty acls", failures)
+    require(
-    require(isinstance(policy.get("tagOwners"), dict), "policy must define tagOwners", failures)
+        isinstance(policy.get("acls"), list) and policy["acls"],
+        "policy must define non-empty acls",
+        failures,
+    )
+    require(
+        isinstance(policy.get("tagOwners"), dict),
+        "policy must define tagOwners",
+        failures,
+    )
 
     hosts = policy.get("hosts", {})
     if isinstance(hosts, dict):
@@ -145,13 +156,33 @@ def main() -> int:
             "traefik-lan": inventory.get("network.metallb.traefik_ip"),
         }
         for name, expected_value in expected_hosts.items():
-            require(bool(expected_value), f"homelab.yml missing inventory value for {name}", failures)
+            require(
-            require(hosts.get(name) == expected_value, f"policy host {name}={hosts.get(name)!r}, expected {expected_value!r}", failures)
+                bool(expected_value),
+                f"homelab.yml missing inventory value for {name}",
+                failures,
+            )
+            require(
+                hosts.get(name) == expected_value,
+                f"policy host {name}={hosts.get(name)!r}, expected {expected_value!r}",
+                failures,
+            )
 
     for index, acl in enumerate(policy.get("acls", [])):
-        require(acl.get("action") == "accept", f"acl[{index}] action must be accept", failures)
+        require(
-        require(isinstance(acl.get("src"), list) and acl["src"], f"acl[{index}] must have non-empty src list", failures)
+            acl.get("action") == "accept",
-        require(isinstance(acl.get("dst"), list) and acl["dst"], f"acl[{index}] must have non-empty dst list", failures)
+            f"acl[{index}] action must be accept",
+            failures,
+        )
+        require(
+            isinstance(acl.get("src"), list) and acl["src"],
+            f"acl[{index}] must have non-empty src list",
+            failures,
+        )
+        require(
+            isinstance(acl.get("dst"), list) and acl["dst"],
+            f"acl[{index}] must have non-empty dst list",
+            failures,
+        )
 
     if failures:
         print("tailnet policy validation failed:", file=sys.stderr)

tests/jeannie-unit

@@ -78,6 +78,10 @@ EOF
 export JEANNIE_LIBRARY_MODE=true
 # shellcheck disable=SC1091
 source "${REPO_ROOT}/jeannie"
+# shellcheck disable=SC1091
+source "${REPO_ROOT}/lib/jeannie/cluster.sh"
+# shellcheck disable=SC1091
+source "${REPO_ROOT}/lib/jeannie/pipeline.sh"
 
 pass() {
     printf 'ok    %s\n' "$1"