require effective config for slots ttl

agent/nix_builder_autoscaler/cli.py

@@ -96,10 +96,7 @@ def _slot_age_seconds(slot: dict[str, Any]) -> float | None:
     return (datetime.now(UTC) - dt).total_seconds()
 
 
-def _slot_ttl(slot: dict[str, Any], policy: dict[str, Any] | None, active_slots: int) -> str:
-    if policy is None:
-        return "-"
-
+def _slot_ttl(slot: dict[str, Any], policy: dict[str, Any], active_slots: int) -> str:
     capacity = policy.get("capacity")
     scheduler = policy.get("scheduler")
     if not isinstance(capacity, dict) or not isinstance(scheduler, dict):
@@ -138,7 +135,7 @@ def _slot_ttl(slot: dict[str, Any], policy: dict[str, Any] | None, active_slots:
     return "-"
 
 
-def _print_slots(data: list[dict[str, Any]], policy: dict[str, Any] | None = None) -> None:
+def _print_slots(data: list[dict[str, Any]], policy: dict[str, Any]) -> None:
     active_slots = sum(1 for slot in data if str(slot.get("state", "")) not in {"empty", "error"})
     rows: list[list[str]] = []
     for slot in data:
@@ -193,16 +190,15 @@ def _print_status_summary(data: dict[str, Any]) -> None:
     _print_table(["metric", "value"], rows)
 
 
-def _get_effective_config(socket_path: str) -> dict[str, Any] | None:
-    try:
-        status, data = _uds_request(socket_path, "GET", "/v1/config/effective")
-    except OSError:
-        return None
+def _get_effective_config(socket_path: str) -> dict[str, Any]:
+    status, data = _uds_request(socket_path, "GET", "/v1/config/effective")
     if status < 200 or status >= 300:
-        return None
+        msg = "failed to fetch effective config"
+        raise RuntimeError(msg)
     if isinstance(data, dict):
         return data
-    return None
+    msg = "invalid effective config payload"
+    raise RuntimeError(msg)
 
 
 def _bulk_slot_action(socket_path: str, action: str) -> dict[str, Any]:
@@ -395,7 +391,11 @@ def main() -> None:
         print(json.dumps(data, indent=2))
     elif args.command == "slots":
         if isinstance(data, list):
-            policy = _get_effective_config(args.socket)
+            try:
+                policy = _get_effective_config(args.socket)
+            except RuntimeError as err:
+                print(str(err))
+                raise SystemExit(1) from err
             _print_slots(data, policy)
         else:
             _print_error(data)