feat(cli): show ▶ N indicator in status bar when /background tasks are running (#27175)

Surface live background-task count in the prompt_toolkit status bar so users
can see at a glance that a /background task exists and is running — no need
to ask the agent about it (the agent has no visibility into bg sessions by
design).

- _get_status_bar_snapshot now reports active_background_tasks from len()
  of the live _background_tasks dict (entries are removed in the task
  thread's finally block, so this reflects truly-running tasks)
- Indicator shown only on medium (<76) and wide (>=76) tiers; narrow (<52)
  stays minimal since it's already cramped
- No invalidate plumbing needed: status bar fragments are pulled via lambda
  on every redraw, and the bg thread already calls _app.invalidate() on exit

Refs #8568
This commit is contained in:
Teknium 2026-05-16 16:51:29 -07:00 committed by GitHub
parent 6f817e1447
commit 226cee43d9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 129 additions and 0 deletions

25
cli.py
View file

@ -3113,8 +3113,19 @@ class HermesCLI:
"session_total_tokens": 0,
"session_api_calls": 0,
"compressions": 0,
"active_background_tasks": 0,
}
# Count live /background tasks. The dict entry is removed in the
# task thread's finally block, so len() reflects truly-running tasks.
# len() on a CPython dict is atomic; safe to read without a lock.
try:
bg_tasks = getattr(self, "_background_tasks", None)
if bg_tasks:
snapshot["active_background_tasks"] = len(bg_tasks)
except Exception:
pass
if not agent:
return snapshot
@ -3350,6 +3361,9 @@ class HermesCLI:
compressions = snapshot.get("compressions", 0)
if compressions:
parts.append(f"🗜️ {compressions}")
bg_count = snapshot.get("active_background_tasks", 0)
if bg_count:
parts.append(f"{bg_count}")
parts.append(duration_label)
if yolo_active:
parts.append("⚠ YOLO")
@ -3366,6 +3380,9 @@ class HermesCLI:
parts = [f"{snapshot['model_short']}", context_label, percent_label]
if compressions:
parts.append(f"🗜️ {compressions}")
bg_count = snapshot.get("active_background_tasks", 0)
if bg_count:
parts.append(f"{bg_count}")
parts.append(duration_label)
prompt_elapsed = snapshot.get("prompt_elapsed")
if prompt_elapsed:
@ -3406,6 +3423,7 @@ class HermesCLI:
percent_label = f"{percent}%" if percent is not None else "--"
if width < 76:
compressions = snapshot.get("compressions", 0)
bg_count = snapshot.get("active_background_tasks", 0)
frags = [
("class:status-bar", ""),
("class:status-bar-strong", snapshot["model_short"]),
@ -3415,6 +3433,9 @@ class HermesCLI:
if compressions:
frags.append(("class:status-bar-dim", " · "))
frags.append((self._compression_count_style(compressions), f"🗜️ {compressions}"))
if bg_count:
frags.append(("class:status-bar-dim", " · "))
frags.append(("class:status-bar-strong", f"{bg_count}"))
frags.extend([
("class:status-bar-dim", " · "),
("class:status-bar-dim", duration_label),
@ -3433,6 +3454,7 @@ class HermesCLI:
bar_style = self._status_bar_context_style(percent)
compressions = snapshot.get("compressions", 0)
bg_count = snapshot.get("active_background_tasks", 0)
frags = [
("class:status-bar", ""),
("class:status-bar-strong", snapshot["model_short"]),
@ -3446,6 +3468,9 @@ class HermesCLI:
if compressions:
frags.append(("class:status-bar-dim", ""))
frags.append((self._compression_count_style(compressions), f"🗜️ {compressions}"))
if bg_count:
frags.append(("class:status-bar-dim", ""))
frags.append(("class:status-bar-strong", f"{bg_count}"))
frags.extend([
("class:status-bar-dim", ""),
("class:status-bar-dim", duration_label),

View file

@ -0,0 +1,104 @@
"""Tests for the /background indicator in the CLI status bar.
The classic prompt_toolkit status bar shows ` N` when N tasks launched via
`/background` are still running. Source of truth is `self._background_tasks`
(a Dict[str, threading.Thread]); entries are removed in the task thread's
finally block, so len() reflects truly-running tasks.
"""
import threading
from datetime import datetime
from cli import HermesCLI
def _stub_thread() -> threading.Thread:
"""Return a Thread instance that's never started — pure dict-value stand-in."""
return threading.Thread(target=lambda: None)
def _make_cli():
"""Bare-metal HermesCLI for snapshot/build tests (no __init__ side effects)."""
cli_obj = HermesCLI.__new__(HermesCLI)
cli_obj.model = "anthropic/claude-opus-4.6"
cli_obj.agent = None
cli_obj._background_tasks = {}
# The snapshot reads session_start to compute duration; supply a stub.
cli_obj.session_start = datetime.now()
return cli_obj
def test_snapshot_reports_zero_when_no_background_tasks():
cli_obj = _make_cli()
snap = cli_obj._get_status_bar_snapshot()
assert snap["active_background_tasks"] == 0
def test_snapshot_counts_live_background_tasks():
cli_obj = _make_cli()
cli_obj._background_tasks = {"bg_a": _stub_thread(), "bg_b": _stub_thread()}
snap = cli_obj._get_status_bar_snapshot()
assert snap["active_background_tasks"] == 2
def test_snapshot_safe_when_background_tasks_attr_missing():
"""Older HermesCLI instances (tests with __new__, etc.) may lack the attr."""
cli_obj = HermesCLI.__new__(HermesCLI)
cli_obj.model = "x"
cli_obj.agent = None
cli_obj.session_start = datetime.now()
# No _background_tasks at all — must not raise.
snap = cli_obj._get_status_bar_snapshot()
assert snap["active_background_tasks"] == 0
def test_plain_text_status_omits_indicator_when_idle():
cli_obj = _make_cli()
text = cli_obj._build_status_bar_text(width=80)
assert "" not in text
def test_plain_text_status_shows_indicator_when_active():
cli_obj = _make_cli()
cli_obj._background_tasks = {"bg_a": _stub_thread()}
text = cli_obj._build_status_bar_text(width=80)
assert "▶ 1" in text
def test_plain_text_status_shows_higher_count():
cli_obj = _make_cli()
cli_obj._background_tasks = {
"a": _stub_thread(),
"b": _stub_thread(),
"c": _stub_thread(),
}
text = cli_obj._build_status_bar_text(width=80)
assert "▶ 3" in text
def test_narrow_width_omits_bg_indicator():
"""The narrow tier (<52) is already cramped — bg is secondary, drop it."""
cli_obj = _make_cli()
cli_obj._background_tasks = {"bg_a": _stub_thread()}
text = cli_obj._build_status_bar_text(width=40)
assert "" not in text
def test_fragments_include_bg_segment_when_active():
cli_obj = _make_cli()
cli_obj._background_tasks = {"a": _stub_thread(), "b": _stub_thread()}
cli_obj._status_bar_visible = True
# _get_status_bar_fragments asks _get_tui_terminal_width(); stub it wide.
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 "▶ 2" in rendered
def test_fragments_omit_bg_segment_when_idle():
cli_obj = _make_cli()
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 "" not in rendered