From 1ef8ae4d3dbda7a8c466d000b8a803ea2eb5f184 Mon Sep 17 00:00:00 2001
From: Brad Stein <Brad.Stein@gmail.com>
Date: Sun, 1 Feb 2026 17:55:07 -0300
Subject: [PATCH] atlasbot: add hardware usage top summary

---
 atlasbot/snapshot/builder.py | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/atlasbot/snapshot/builder.py b/atlasbot/snapshot/builder.py
index 8e93e18..8e13ae4 100644
--- a/atlasbot/snapshot/builder.py
+++ b/atlasbot/snapshot/builder.py
@@ -1580,6 +1580,7 @@ def _append_hardware_usage(lines: list[str], summary: dict[str, Any]) -> None:
     if not isinstance(usage, list) or not usage:
         return
     parts = []
+    tops: dict[str, tuple[str, float]] = {}
     for entry in usage[:5]:
         if not isinstance(entry, dict):
             continue
@@ -1595,8 +1596,27 @@ def _append_hardware_usage(lines: list[str], summary: dict[str, Any]) -> None:
         label += f" cpu={_format_float(cpu)} ram={_format_float(ram)}"
         label += f" io={_format_rate_bytes(io)} net={_format_rate_bytes(net)}"
         parts.append(label)
+        for metric, value in (("cpu", cpu), ("ram", ram), ("io", io), ("net", net), ("load", load)):
+            if isinstance(value, (int, float)):
+                current = tops.get(metric)
+                if current is None or float(value) > current[1]:
+                    tops[metric] = (hardware, float(value))
     if parts:
         lines.append("hardware_usage_avg: " + "; ".join(parts))
+    if tops:
+        top_parts = []
+        for metric in ("cpu", "ram", "io", "net", "load"):
+            entry = tops.get(metric)
+            if not entry:
+                continue
+            hardware, value = entry
+            if metric in {"io", "net"}:
+                rendered = _format_rate_bytes(value)
+            else:
+                rendered = _format_float(value)
+            top_parts.append(f"{metric}={hardware} ({rendered})")
+        if top_parts:
+            lines.append("hardware_usage_top: " + "; ".join(top_parts))
 
 
 def _append_cluster_watchlist(lines: list[str], summary: dict[str, Any]) -> None: