feat(cli): show live background terminal-process count in status bar (#32061)

The CLI status bar tracked /background agent tasks (▶ N) but not shell
processes spawned via terminal(background=true). Both kinds of work can
run concurrently and a user has no in-bar signal for shell processes.

Add an independent indicator (⚙ N) sourced from
tools.process_registry.process_registry._running. The two indicators
render side-by-side when both are active (▶ 1 │ ⚙ 2), hidden when their
count is zero. Renders at all four status-bar tiers (text fallback +
prompt_toolkit fragments, narrow + wide widths). The narrow <52 tier
still drops both for space — unchanged.

New ProcessRegistry.count_running() returns len(_running) without
acquiring _lock; CPython dict len is atomic and we're polling on every
status-bar tick, so lock-free is the right tradeoff.
This commit is contained in:
Teknium 2026-05-25 05:35:02 -07:00 committed by GitHub
parent 2b16de0ec3
commit 1c3c364287
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 123 additions and 0 deletions

23
cli.py
View file

@ -3503,6 +3503,7 @@ class HermesCLI:
"session_api_calls": 0,
"compressions": 0,
"active_background_tasks": 0,
"active_background_processes": 0,
}
# Count live /background tasks. The dict entry is removed in the
@ -3515,6 +3516,14 @@ class HermesCLI:
except Exception:
pass
# Count live background terminal processes (terminal tool background
# sessions tracked by tools.process_registry). Cheap O(1) read.
try:
from tools.process_registry import process_registry
snapshot["active_background_processes"] = process_registry.count_running()
except Exception:
pass
if not agent:
return snapshot
@ -3753,6 +3762,9 @@ class HermesCLI:
bg_count = snapshot.get("active_background_tasks", 0)
if bg_count:
parts.append(f"{bg_count}")
bg_proc_count = snapshot.get("active_background_processes", 0)
if bg_proc_count:
parts.append(f"{bg_proc_count}")
parts.append(duration_label)
if yolo_active:
parts.append("⚠ YOLO")
@ -3772,6 +3784,9 @@ class HermesCLI:
bg_count = snapshot.get("active_background_tasks", 0)
if bg_count:
parts.append(f"{bg_count}")
bg_proc_count = snapshot.get("active_background_processes", 0)
if bg_proc_count:
parts.append(f"{bg_proc_count}")
parts.append(duration_label)
prompt_elapsed = snapshot.get("prompt_elapsed")
if prompt_elapsed:
@ -3813,6 +3828,7 @@ class HermesCLI:
if width < 76:
compressions = snapshot.get("compressions", 0)
bg_count = snapshot.get("active_background_tasks", 0)
bg_proc_count = snapshot.get("active_background_processes", 0)
frags = [
("class:status-bar", ""),
("class:status-bar-strong", snapshot["model_short"]),
@ -3825,6 +3841,9 @@ class HermesCLI:
if bg_count:
frags.append(("class:status-bar-dim", " · "))
frags.append(("class:status-bar-strong", f"{bg_count}"))
if bg_proc_count:
frags.append(("class:status-bar-dim", " · "))
frags.append(("class:status-bar-strong", f"{bg_proc_count}"))
frags.extend([
("class:status-bar-dim", " · "),
("class:status-bar-dim", duration_label),
@ -3844,6 +3863,7 @@ class HermesCLI:
bar_style = self._status_bar_context_style(percent)
compressions = snapshot.get("compressions", 0)
bg_count = snapshot.get("active_background_tasks", 0)
bg_proc_count = snapshot.get("active_background_processes", 0)
frags = [
("class:status-bar", ""),
("class:status-bar-strong", snapshot["model_short"]),
@ -3860,6 +3880,9 @@ class HermesCLI:
if bg_count:
frags.append(("class:status-bar-dim", ""))
frags.append(("class:status-bar-strong", f"{bg_count}"))
if bg_proc_count:
frags.append(("class:status-bar-dim", ""))
frags.append(("class:status-bar-strong", f"{bg_proc_count}"))
frags.extend([
("class:status-bar-dim", ""),
("class:status-bar-dim", duration_label),

View file

@ -102,3 +102,90 @@ def test_fragments_omit_bg_segment_when_idle():
frags = cli_obj._get_status_bar_fragments()
rendered = "".join(text for _style, text in frags)
assert "" not in rendered
# ── Background terminal-process indicator (⚙ N) ───────────────────────────
# Source of truth is tools.process_registry.process_registry._running (a dict
# of currently-running shell processes spawned by terminal(background=true)).
# Distinct from /background tasks above: ▶ counts agent threads, ⚙ counts
# shell processes. Both can be active simultaneously.
class _FakeRunningRegistry:
"""Minimal stand-in for process_registry; exposes count_running()."""
def __init__(self, count: int) -> None:
self._count = count
def count_running(self) -> int:
return self._count
def _patch_process_registry(monkeypatch, count: int) -> None:
import tools.process_registry as pr_mod
monkeypatch.setattr(pr_mod, "process_registry", _FakeRunningRegistry(count))
def test_snapshot_reports_zero_when_no_background_processes(monkeypatch):
cli_obj = _make_cli()
_patch_process_registry(monkeypatch, 0)
snap = cli_obj._get_status_bar_snapshot()
assert snap["active_background_processes"] == 0
def test_snapshot_counts_live_background_processes(monkeypatch):
cli_obj = _make_cli()
_patch_process_registry(monkeypatch, 3)
snap = cli_obj._get_status_bar_snapshot()
assert snap["active_background_processes"] == 3
def test_snapshot_safe_when_process_registry_raises(monkeypatch):
"""If count_running() raises the snapshot stays at 0; no propagate."""
cli_obj = _make_cli()
import tools.process_registry as pr_mod
class _BoomRegistry:
def count_running(self):
raise RuntimeError("boom")
monkeypatch.setattr(pr_mod, "process_registry", _BoomRegistry())
snap = cli_obj._get_status_bar_snapshot()
assert snap["active_background_processes"] == 0
def test_plain_text_status_shows_proc_indicator_when_active(monkeypatch):
cli_obj = _make_cli()
_patch_process_registry(monkeypatch, 2)
text = cli_obj._build_status_bar_text(width=80)
assert "⚙ 2" in text
def test_plain_text_status_omits_proc_indicator_when_idle(monkeypatch):
cli_obj = _make_cli()
_patch_process_registry(monkeypatch, 0)
text = cli_obj._build_status_bar_text(width=80)
assert "" not in text
def test_fragments_include_proc_segment_when_active(monkeypatch):
cli_obj = _make_cli()
_patch_process_registry(monkeypatch, 1)
cli_obj._status_bar_visible = True
cli_obj._get_tui_terminal_width = lambda: 120 # type: ignore[method-assign]
frags = cli_obj._get_status_bar_fragments()
rendered = "".join(text for _style, text in frags)
assert "⚙ 1" in rendered
def test_indicators_independent_agents_and_processes(monkeypatch):
"""▶ (agent tasks) and ⚙ (shell processes) render side-by-side."""
cli_obj = _make_cli()
cli_obj._background_tasks = {"bg_a": _stub_thread()}
_patch_process_registry(monkeypatch, 2)
cli_obj._status_bar_visible = True
cli_obj._get_tui_terminal_width = lambda: 120 # type: ignore[method-assign]
frags = cli_obj._get_status_bar_fragments()
rendered = "".join(text for _style, text in frags)
assert "▶ 1" in rendered
assert "⚙ 2" in rendered

View file

@ -1235,6 +1235,19 @@ class ProcessRegistry:
except Exception as e:
return {"status": "error", "error": str(e)}
def count_running(self) -> int:
"""Return the count of currently-running background processes.
Cheap O(1) read of the running dict, suitable for status-bar polling
on every render tick. CPython dict ``len()`` is atomic; callers do not
need to hold ``self._lock``. Reflects ``_running`` only: sessions are
moved to ``_finished`` when their subprocess exits.
"""
try:
return len(self._running)
except Exception:
return 0
def list_sessions(self, task_id: str = None) -> list:
"""List all running and recently-finished processes."""
with self._lock: