diff --git a/adapters/claude/tools/fleet/tests/test_gpu_oneline.py b/adapters/claude/tools/fleet/tests/test_gpu_oneline.py new file mode 120000 index 000000000..22ac064c9 --- /dev/null +++ b/adapters/claude/tools/fleet/tests/test_gpu_oneline.py @@ -0,0 +1 @@ +../../../../../tools/fleet/tests/test_gpu_oneline.py \ No newline at end of file diff --git a/core/OPERATIONS.md b/core/OPERATIONS.md index 1bfffc5b7..03b242520 100644 --- a/core/OPERATIONS.md +++ b/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 diff --git a/tools/fleet/render.py b/tools/fleet/render.py index be1d9fd29..707cecffe 100644 --- a/tools/fleet/render.py +++ b/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 diff --git a/tools/fleet/tests/fixtures/stage_divider/screen-100.txt b/tools/fleet/tests/fixtures/stage_divider/screen-100.txt index 9abe5c19a..59ac4607c 100644 --- a/tools/fleet/tests/fixtures/stage_divider/screen-100.txt +++ b/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 diff --git a/tools/fleet/tests/fixtures/stage_divider/screen-168.txt b/tools/fleet/tests/fixtures/stage_divider/screen-168.txt index c723dfb23..0ebb43e70 100644 --- a/tools/fleet/tests/fixtures/stage_divider/screen-168.txt +++ b/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/ diff --git a/tools/fleet/tests/fixtures/stage_divider/screen-60.txt b/tools/fleet/tests/fixtures/stage_divider/screen-60.txt index d34bfd7cd..432461d8a 100644 --- a/tools/fleet/tests/fixtures/stage_divider/screen-60.txt +++ b/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 diff --git a/tools/fleet/tests/test_f83_compute_hosts.py b/tools/fleet/tests/test_f83_compute_hosts.py index b8d997c97..2ca743b6b 100644 --- a/tools/fleet/tests/test_f83_compute_hosts.py +++ b/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) diff --git a/tools/fleet/tests/test_gpu_oneline.py b/tools/fleet/tests/test_gpu_oneline.py new file mode 100755 index 000000000..8a42c4f61 --- /dev/null +++ b/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()