fix(fleet): summarize reserved and observed GPU runs on one line

adapters/claude/tools/fleet/tests/test_gpu_oneline.py

@@ -0,0 +1 @@
+../../../../../tools/fleet/tests/test_gpu_oneline.py

core/OPERATIONS.md

@@ -1518,9 +1518,18 @@ Reservation lifetime follows the exact wrapper PID, start time and PID namespace
 not its launching session. Normal exit releases it; the next launch removes
 proven-dead/reused-PID reservations, and read-only status omits them from its
 view, preserving unknown identity. Existing work is not adopted or interrupted: its probe processes mark
-the GPU occupied. `compute-hosts list` and Fleet's top compute panel show GPU
-reservations, including the session tag, task and start time, before CUDA has
-allocated memory. Host-local storage lets callers on different machines converge
+the GPU occupied. `compute-hosts list` shows GPU reservations before CUDA has
+allocated memory. Fleet's folded compute panel uses one row per GPU, appending
+observed process counts by session tag, largest count first. Only differences
+between reserved and observed counts per session add short `reserved+N` or
+`unreserved+N` labels (`R+N`/`U+N` when space is tight); unavailable observation
+is not an empty process or reservation sample.
+The existing `p` expansion retains reservation owners, tasks, start times and
+process commands. The session summary shortens whole entries to `+N` at narrow
+widths without shifting the utilization and memory columns or adding rows.
+Reservations with unknown device placement remain visible at host level.
+This common read-only renderer applies to all three harnesses and adds no
+flag, setting or reservation mutation. Host-local storage lets callers on different machines converge
 without depending on a shared NAS mount or an installed harness on the target.
 
 `run` starts a command detached under a stable run id and writes its log and

tools/fleet/render.py

@@ -36,6 +36,7 @@
 import shlex
 import sys
 import time
+from collections import Counter
 from .display import label as _user_label, project as _user_project, gpu_owner as _gpu_owner_label
 
 from .model import (fmt_min, dash, project_of, exec_child_is_wait,
@@ -6005,6 +6006,69 @@ def _gpu_reservation_rows(leases, indent, width, unplaced=False):
     return rows
 
 
+def _gpu_activity_summary(gpu, host):
+    """Count observed processes by full session identity, not saved badge text."""
+    def session_key(owner):
+        if (isinstance(owner, dict) and owner.get("kind") == "session"
+                and owner.get("harness") in {"claude", "codex", "opencode"}
+                and isinstance(owner.get("id"), str) and owner["id"]):
+            return owner["harness"], owner["id"]
+        return None
+
+    observed = Counter()
+    for process in gpu.get("processes") or ():
+        if isinstance(process, dict):
+            key = session_key(process.get("session_owner")) or session_key(process.get("owner"))
+            observed[key] += 1
+    reserved = Counter(session_key(lease.get("owner"))
+                       for lease in gpu.get("reservations") or () if isinstance(lease, dict))
+    groups = []
+    for key, count in observed.items():
+        tag = _resolve_session_tag(*key) if key else None
+        groups.append(("[%s]" % tag if tag else "?", count))
+    groups.sort(key=lambda group: (-group[1], group[0]))
+    warnings = []
+    # A failed process query or unmapped fallback cannot prove an empty GPU.
+    if (isinstance(gpu.get("processes"), (list, tuple))
+            and isinstance(gpu.get("reservations"), (list, tuple))
+            and not host.get("process_detail") and not gpu.get("observation_source")):
+        extra_reserved = sum((reserved - observed).values())
+        extra_observed = sum((observed - reserved).values())
+        if extra_reserved:
+            warnings.append("reserved+%d" % extra_reserved)
+        if extra_observed and not host.get("reservation_detail"):
+            warnings.append("unreserved+%d" % extra_observed)
+    return groups, warnings
+
+
+def _gpu_activity_tail(groups, warnings, available):
+    """Fit whole session entries; +N counts omitted sessions, not processes."""
+    warning = " ".join(warnings)
+    if _dw(" · " + warning) > available:
+        warning = warning.replace("unreserved+", "unres+").replace("reserved+", "res+")
+    if _dw(" · " + warning) > available:
+        warning = warning.replace("unres+", "U+").replace("res+", "R+")
+    if _dw(" · " + warning) > available:
+        warning = "/".join("U" if item.startswith("unreserved+") else "R"
+                           for item in warnings)
+    warning_width = _dw(" · " + warning) if warning else 0
+    room = available - warning_width - 3
+    entries = ["%s×%d" % group for group in groups]
+    summary = ""
+    for kept in range(len(entries), -1, -1):
+        parts = entries[:kept]
+        if kept < len(entries):
+            parts = parts + ["+%d" % (len(entries) - kept)]
+        candidate = " ".join(parts)
+        if _dw(candidate) <= room:
+            summary = candidate
+            break
+    tail = [(" · " + summary, "dim")] if summary else []
+    if warning and warning_width <= available:
+        tail.append((" · " + warning, "lvl_y"))
+    return tail
+
+
 def _compute_host_rows(term_width=None, sessions=None, resources=None):
     _compute_host_rows.fold_rows = []
     snapshot = _COMPUTE_HOSTS
@@ -6138,12 +6202,17 @@ def _compute_host_rows(term_width=None, sessions=None, resources=None):
             token = _gpu_token(gpu, max(12, width - _dw(indent) - chip_width), show_name=width >= 100,
                                sessions=sessions, index_width=gpu_index_width,
                                vram_slot=wide_vram_slot)
-            if count:
+            if count or gpu.get("reservations"):
                 _compute_host_rows.fold_rows.append({"line": len(rows),
                     "card_key": _gpu_fold_key(host.get("host"), gpu.get("index")), "folded": folded})
-            rows.append(_clip_segs([(indent, None)] + token + chip, width)[0])
-            rows.extend(_gpu_reservation_rows(gpu.get("reservations") or (), indent, width))
+            row = [(indent, None)] + token + chip
+            if folded:
+                groups, warnings = _gpu_activity_summary(gpu, host)
+                room = width - 1 - sum(_dw(text) for text, _key in row)
+                row += _gpu_activity_tail(groups, warnings, room)
+            rows.append(_clip_segs(row, width)[0])
             if not folded:
+                rows.extend(_gpu_reservation_rows(gpu.get("reservations") or (), indent, width))
                 rows.extend(_gpu_process_rows(gpu, indent, width))
     return rows
 

tools/fleet/tests/fixtures/stage_divider/screen-100.txt

@@ -6,7 +6,7 @@
 ─────
   compute resources  1/1
   ◇ gpu  CPU ────────  — thr  · MEM —/—GB  · SWP —/—GB
-         ⠋ 0: working  —                   UTIL    —  VRAM ────────      —/—GB ▸ 1
+         ⠋ 0: working  —                   UTIL    —  VRAM ────────      —/—GB ▸ 1 · ?×1
 ─────
 
   SESSIONS   narrow · press w to cycle

tools/fleet/tests/fixtures/stage_divider/screen-168.txt

@@ -6,7 +6,7 @@
 ─────
   compute resources  1/1
   ◇ gpu  CPU ────────  — thr  · MEM —/—GB  · SWP —/—GB
-         ⠋ 0: working  —                   UTIL    —  VRAM ────────      —/—GB ▸ 1
+         ⠋ 0: working  —                   UTIL    —  VRAM ────────      —/—GB ▸ 1 · ?×1
 ─────
 
 ● bundle/

tools/fleet/tests/fixtures/stage_divider/screen-60.txt

@@ -6,7 +6,7 @@
 ─────
   compute resources  1/1
   ◇ gpu  CPU ────────  — thr  M —/—G  S —/—G
-         ⠋ 0: UTIL    —  VRAM ────      —/—GB ▸ 1
+         ⠋ 0: UTIL    —  VRAM ────      —/—GB ▸ 1 · ?×1
 ─────
 
   SESSIONS   stack · press w to cycle

tools/fleet/tests/test_f83_compute_hosts.py

@@ -123,11 +123,11 @@ def test_malformed_and_timeout_are_typed_diagnostics(self):
 
 
 class ComputeHostRenderTest(unittest.TestCase):
-    def test_reservation_owner_task_and_start_are_visible_even_with_folded_processes(self):
+    def test_reservation_owner_task_and_start_are_visible_only_when_expanded(self):
         gpu = self.snapshot["hosts"][0]["gpus"][0]
         gpu["reservations"] = [{"owner": {"label": "SR_CorrNet [cd]"},
                                 "task": "css 1ch training", "started_at": 1728000000}]
-        render._ROUTE_FOLD = {render._GPU_FOLD_ALL: True}
+        render._ROUTE_FOLD = {render._GPU_FOLD_ALL: False}
         render.set_compute_hosts(self.snapshot)
         for width in (60, 100, 168):
             rows = render._compute_host_rows(width)

tools/fleet/tests/test_gpu_oneline.py

@@ -0,0 +1,124 @@
+"""Folded GPU rows summarize observations while preserving expanded detail."""
+import copy
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+from fleet import render
+
+
+def owner(harness, tag):
+    return {"kind": "session", "harness": harness, "id": tag}
+
+
+class GPUOneLineTest(unittest.TestCase):
+    def setUp(self):
+        self.addCleanup(setattr, render, "_ROUTE_FOLD", render._ROUTE_FOLD)
+        self.addCleanup(setattr, render, "_COMPUTE_HOSTS", render._COMPUTE_HOSTS)
+        self.addCleanup(setattr, render, "_COMPUTE_HOSTS_SET_AT", render._COMPUTE_HOSTS_SET_AT)
+        self.addCleanup(setattr, render, "_FOLDABLE", render._FOLDABLE)
+        render._ROUTE_FOLD = {render._GPU_FOLD_ALL: True}
+        self.tag = mock.patch.object(render, "_resolve_session_tag", side_effect=lambda h, sid: sid)
+        self.tag.start()
+        self.addCleanup(self.tag.stop)
+        owners = [owner("codex", "bb")] * 3 + [owner("claude", "8e")]
+        self.gpu = {"index": 1, "name": "NVIDIA RTX 6000 Ada Generation",
+                    "utilization_gpu_pct": 100, "memory_used_mib": 23552,
+                    "memory_total_mib": 49152,
+                    "processes": [{"pid": n + 10, "session_owner": o,
+                                   "command": "python train.py --config detailed-name.yaml"}
+                                  for n, o in enumerate(owners)],
+                    "reservations": [{"owner": {**o, "label": "saved [%s]" % o["id"]},
+                                      "started_at": 1728000000, "task": "detailed task"}
+                                     for o in owners]}
+        self.host = {"host": "moving4", "reachable": True, "cpu_count": 64,
+                     "gpus": [self.gpu]}
+        render.set_compute_hosts({"configured": True, "hosts": [self.host]})
+
+    def rows(self, width):
+        return [render._plain(row) for row in render._compute_host_rows(width)]
+
+    def test_matching_runs_fold_to_one_line_and_sorted_session_counts(self):
+        for width in (80, 200):
+            with self.subTest(width=width):
+                lines = self.rows(width)
+                self.assertEqual(len(lines), 3)
+                self.assertIn("▸ 4 · [bb]×3 [8e]×1", lines[2])
+                self.assertNotIn("reserved", lines[2])
+                self.assertNotIn("detailed", "\n".join(lines))
+                self.assertLessEqual(render._dw(lines[2]), width - 1)
+
+    def test_narrow_summary_omits_whole_sessions_without_shifting_metrics(self):
+        before = self.rows(80)[2]
+        for n, harness in enumerate(("codex", "claude", "opencode")):
+            o = owner(harness, "a%d" % n)
+            self.gpu["processes"].append({"session_owner": o})
+            self.gpu["reservations"].append({"owner": o})
+        line = self.rows(80)[2]
+        self.assertIn("[bb]×3 +4", line)
+        for token in ("UTIL", "%", "VRAM", "GB"):
+            self.assertEqual(before.index(token), line.index(token))
+        self.assertLessEqual(render._dw(line), 79)
+
+    def test_same_total_with_different_owners_is_a_mismatch(self):
+        self.gpu["reservations"][-1]["owner"] = owner("opencode", "ff")
+        for width in (80, 200):
+            line = self.rows(width)[2]
+            self.assertIn("res", line)
+            self.assertIn("unres", line)
+            self.assertIn("+1", line)
+            self.assertLessEqual(render._dw(line), width - 1)
+
+    def test_reserved_only_gpu_is_foldable_and_process_only_is_labelled(self):
+        self.gpu["processes"] = []
+        line = self.rows(80)[2]
+        self.assertIn("reserved+4", line)
+        self.assertEqual(len(render._compute_host_rows.fold_rows), 1)
+        self.gpu["processes"] = [{"owner": owner("opencode", "ff")}]
+        self.gpu["reservations"] = []
+        self.assertIn("unreserved+1", self.rows(80)[2])
+
+    def test_two_mismatch_labels_survive_a_long_host_in_80_columns(self):
+        self.host["host"] = "gpu-worker"
+        self.gpu["reservations"][-1]["owner"] = owner("opencode", "ff")
+        line = self.rows(80)[2]
+        self.assertIn("R+1 U+1", line)
+        self.assertIn("UTIL 100%", line)
+        self.assertIn("23/48GB ▸ 4", line)
+        self.assertLessEqual(render._dw(line), 79)
+
+    def test_unknown_queries_do_not_claim_a_reservation_mismatch(self):
+        reservations = self.gpu.pop("reservations")
+        self.assertNotIn("unreserved+", self.rows(200)[2])
+        self.gpu["reservations"] = reservations
+        self.gpu["processes"] = []
+        self.host["process_detail"] = "query failed"
+        self.assertNotIn("reserved+", self.rows(200)[2])
+        self.host.pop("process_detail")
+        self.gpu["observation_source"] = "proc-fallback"
+        self.assertNotIn("reserved+", self.rows(200)[2])
+        self.gpu.pop("observation_source")
+        self.gpu["processes"] = [{"session_owner": owner("opencode", "ff")}]
+        self.gpu["reservations"] = []
+        self.host["reservation_detail"] = "leases unreadable"
+        self.assertNotIn("unreserved+", self.rows(200)[2])
+
+    def test_expanded_detail_and_toggle_keep_original_payload(self):
+        original = copy.deepcopy(self.gpu)
+        render._FOLDABLE = list(render._compute_host_rows.fold_rows)
+        render._toggle_gpu_commands()
+        lines = self.rows(200)
+        self.assertEqual(len(lines), 11)
+        self.assertIn("▾ 4", lines[2])
+        self.assertEqual(sum("reserved saved" in line for line in lines), 4)
+        self.assertEqual(sum("↳" in line for line in lines), 4)
+        self.assertIn("10-04", "\n".join(lines))
+        self.assertIn("detailed task", "\n".join(lines))
+        self.assertIn("detailed-name.yaml", "\n".join(lines))
+        self.assertEqual(original, self.gpu)
+
+
+if __name__ == "__main__":
+    unittest.main()