feat(cli): per-turn summary line and live token flow in the spinner

Ports Claude Code's post-turn accounting (Edited N files +X -Y · Worked for Ns). Display-only, quiet-mode aware, config-gated.
This commit is contained in:
teknium1 2026-07-26 14:29:02 -07:00 committed by Teknium
parent e769560c76
commit ce997f9e62
5 changed files with 809 additions and 0 deletions

304
agent/turn_summary.py Normal file
View file

@ -0,0 +1,304 @@
"""Per-turn accounting for the interactive CLI.
Two display-only pieces live here:
* :class:`TurnSummaryCollector` a tiny observer that rides the existing
``tool_progress_callback`` feed (``tool.completed`` events already carry
the tool name and its raw result) and tallies what a turn actually did.
It holds **no** agent-loop state: the display layer already sees every
tool call, so nothing new is threaded through the conversation loop.
* :func:`format_turn_summary` a pure formatter that turns a tally plus a
wall-clock duration into one dim line, e.g.::
12.4s · edited 2 files +18 -3 · read 4 files · ran 3 commands
Ported from Claude Code's post-turn accounting line
("Edited 1 file +6 -2, read 1 file … Worked for 10s").
:func:`format_token_flow` is the spinner-side counterpart: a cumulative
token readout appended to the live elapsed timer (`` 1.2k tok``).
Everything in this module is pure/side-effect free apart from the
collector's own counters, which makes it directly unit-testable without a
terminal, an agent, or a network call.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
__all__ = [
"TurnSummaryCollector",
"TurnTally",
"format_turn_summary",
"format_token_flow",
"format_elapsed",
]
# Leading glyph for the summary line. Deliberately not an emoji — the line is
# meant to read as terminal chrome, not as agent speech.
SUMMARY_PREFIX = ""
# A turn that called no tools and finished this fast has nothing worth
# reporting (plain chat reply). Below the threshold the formatter returns "".
_MIN_TOOLLESS_SECONDS = 2.0
# Max number of "verb + count" segments rendered before collapsing the rest
# into a "+N more" tail, so a 12-tool turn cannot blow past one line.
_MAX_SEGMENTS = 4
# Tool name -> (verb, singular noun, plural noun).
#
# Verbs are past tense because the line is printed *after* the turn. Tools not
# listed here fall into a generic "called N tools" bucket rather than inventing
# phrasing for plugin/MCP tools whose semantics we don't know.
_VERB_GROUPS: dict[str, tuple[str, str, str]] = {
"write_file": ("edited", "file", "files"),
"patch": ("edited", "file", "files"),
"read_file": ("read", "file", "files"),
"web_extract": ("read", "page", "pages"),
"terminal": ("ran", "command", "commands"),
"execute_code": ("ran", "script", "scripts"),
"search_files": ("searched", "path", "paths"),
"web_search": ("searched the web", "time", "times"),
"session_search": ("searched sessions", "time", "times"),
"browser_navigate": ("browsed", "page", "pages"),
"skill_view": ("read", "skill", "skills"),
"skill_manage": ("updated", "skill", "skills"),
"skills_list": ("listed skills", "time", "times"),
"todo": ("updated", "task list", "task lists"),
"delegate_task": ("delegated", "task", "tasks"),
"memory": ("updated", "memory", "memories"),
}
# Verb groups that carry file-edit line deltas (+X -Y) when known.
_EDIT_VERB = "edited"
# Render order: edits first (the thing users most want confirmed), then reads,
# then commands. Anything else follows in first-seen order.
_VERB_PRIORITY: tuple[str, ...] = ("edited", "read", "ran")
# Tools whose results may report a unified diff we can count lines from.
_DIFF_RESULT_TOOLS = frozenset({"patch"})
@dataclass
class TurnTally:
"""What a single turn did, as observed from the tool-progress feed."""
# verb -> {noun_plural: count}; keeps insertion order for stable rendering.
verbs: dict[str, dict[str, int]] = field(default_factory=dict)
# Tools with no curated verb, counted together.
other_tools: int = 0
# Aggregated unified-diff line deltas across edit tools, when reported.
lines_added: int = 0
lines_removed: int = 0
# True once at least one edit tool reported a countable diff, so the
# formatter knows the difference between "+0 -0" and "unknown".
has_line_deltas: bool = False
@property
def total_tools(self) -> int:
counted = sum(sum(nouns.values()) for nouns in self.verbs.values())
return counted + self.other_tools
def _count_diff_lines(diff: str) -> tuple[int, int]:
"""Count added/removed lines in unified-diff text.
File headers (``+++``/``---``) are excluded so a one-line edit does not
read as three additions.
"""
added = removed = 0
for line in diff.splitlines():
if line.startswith("+++") or line.startswith("---"):
continue
if line.startswith("+"):
added += 1
elif line.startswith("-"):
removed += 1
return added, removed
def _extract_line_deltas(tool_name: str, result: Any) -> tuple[int, int] | None:
"""Pull (added, removed) from a tool result, or None when unavailable.
Only tools that already report a diff in their result payload are
inspected we never shell out to git and never re-read files to
synthesise a delta.
"""
if tool_name not in _DIFF_RESULT_TOOLS:
return None
payload: Any = result
if isinstance(payload, str):
text = payload.strip()
if not text.startswith("{"):
return None
try:
import json
# strict=False tolerates literal control characters inside strings
# (raw newlines in an embedded diff), which some tool serialisers
# emit. A tally line is never worth failing over formatting.
payload = json.loads(text, strict=False)
except Exception:
return None
if not isinstance(payload, dict):
return None
diff = payload.get("diff")
if not isinstance(diff, str) or not diff.strip():
return None
return _count_diff_lines(diff)
class TurnSummaryCollector:
"""Accumulate per-turn tool tallies from the tool-progress feed.
Wired into the CLI's existing ``_on_tool_progress`` handler: the display
layer already receives every ``tool.completed`` event with the tool name
and raw result, so no agent-loop bookkeeping is added.
"""
def __init__(self) -> None:
self._tally = TurnTally()
def begin(self) -> None:
"""Start a fresh turn (drops any prior tally)."""
self._tally = TurnTally()
def record_tool(
self,
tool_name: str | None,
*,
result: Any = None,
is_error: bool = False,
) -> None:
"""Record one completed tool call.
Failed calls are skipped: a summary claiming "edited 2 files" when one
write was denied would be exactly the over-claim the file-mutation
verifier exists to catch.
"""
if not tool_name or is_error:
return
# Internal/pseudo tools (``_thinking``) are not user-visible work.
if tool_name.startswith("_"):
return
group = _VERB_GROUPS.get(tool_name)
if group is None:
self._tally.other_tools += 1
return
verb, _singular, plural = group
nouns = self._tally.verbs.setdefault(verb, {})
nouns[plural] = nouns.get(plural, 0) + 1
if verb == _EDIT_VERB:
deltas = _extract_line_deltas(tool_name, result)
if deltas is not None:
added, removed = deltas
self._tally.lines_added += added
self._tally.lines_removed += removed
self._tally.has_line_deltas = True
@property
def tally(self) -> TurnTally:
return self._tally
def render(self, elapsed_seconds: float) -> str:
"""Render this turn's summary line (see :func:`format_turn_summary`)."""
return format_turn_summary(elapsed_seconds, self._tally)
def format_elapsed(seconds: float) -> str:
"""Format a wall-clock duration compactly (``12.4s`` / ``2m05s``)."""
if seconds < 0:
seconds = 0.0
if seconds < 60:
return f"{seconds:.1f}s"
minutes, rest = divmod(int(round(seconds)), 60)
return f"{minutes}m{rest:02d}s"
def _pluralize(count: int, plural_noun: str) -> str:
"""Return ``"1 file"`` / ``"3 files"`` from a plural noun form."""
if count == 1:
singular = plural_noun
if plural_noun.endswith("ies"):
singular = plural_noun[:-3] + "y"
elif plural_noun.endswith("ses"):
singular = plural_noun[:-2]
elif plural_noun.endswith("s"):
singular = plural_noun[:-1]
return f"1 {singular}"
return f"{count} {plural_noun}"
def _ordered_verbs(tally: TurnTally) -> list[str]:
"""Verbs in render order: priority verbs first, then first-seen order."""
seen = list(tally.verbs.keys())
ranked = [v for v in _VERB_PRIORITY if v in tally.verbs]
ranked += [v for v in seen if v not in _VERB_PRIORITY]
return ranked
def format_turn_summary(
elapsed_seconds: float,
tally: TurnTally | None,
*,
max_segments: int = _MAX_SEGMENTS,
) -> str:
"""Render the per-turn accounting line, or ``""`` when there's nothing to say.
Pure function no config lookups, no terminal access, no I/O. Gating
(``display.turn_summary``, quiet mode, CLI-only) is the caller's job.
"""
if tally is None:
tally = TurnTally()
segments: list[str] = []
for verb in _ordered_verbs(tally):
nouns = tally.verbs[verb]
parts = [_pluralize(count, plural) for plural, count in nouns.items() if count]
if not parts:
continue
segment = f"{verb} {', '.join(parts)}"
if verb == _EDIT_VERB and tally.has_line_deltas:
segment += f" +{tally.lines_added} -{tally.lines_removed}"
segments.append(segment)
if tally.other_tools:
segments.append(f"called {_pluralize(tally.other_tools, 'tools')}")
if not segments and tally.total_tools == 0 and elapsed_seconds < _MIN_TOOLLESS_SECONDS:
return ""
if max_segments > 0 and len(segments) > max_segments:
hidden = len(segments) - max_segments
segments = segments[:max_segments] + [f"+{hidden} more"]
pieces = [format_elapsed(elapsed_seconds)] + segments
return f"{SUMMARY_PREFIX} " + " · ".join(pieces)
def format_token_flow(output_tokens: Any, *, arrow: str = "") -> str:
"""Render cumulative turn tokens for the live spinner (``↓ 1.2k tok``).
Returns ``""`` for a non-positive count so the spinner shows nothing
rather than a misleading `` 0 tok`` before the first API response lands.
"""
try:
count = int(output_tokens)
except (TypeError, ValueError):
return ""
if count <= 0:
return ""
if count < 1000:
return f"{arrow} {count} tok"
if count < 1_000_000:
return f"{arrow} {count / 1000:.1f}k tok"
return f"{arrow} {count / 1_000_000:.1f}M tok"

120
cli.py
View file

@ -4158,6 +4158,22 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
# Inline diff previews for write actions (display.inline_diffs in config.yaml)
self._inline_diffs_enabled = CLI_CONFIG["display"].get("inline_diffs", True)
# Per-turn accounting (display.turn_summary / display.spinner_token_flow).
# Both are CLI-only, display-only chrome. The collector rides the
# tool-progress feed this class already receives, so no agent-loop
# bookkeeping is involved.
self._turn_summary_enabled = bool(CLI_CONFIG["display"].get("turn_summary", True))
self._spinner_token_flow_enabled = bool(
CLI_CONFIG["display"].get("spinner_token_flow", True)
)
self._turn_summary_collector = None
self._turn_summary_start = 0.0
self._turn_token_baseline = 0
# True only while an interactive (run()-loop) turn is in flight. Single
# query, -Q, and gateway paths never set it, which is what keeps the
# summary line out of non-interactive surfaces.
self._interactive_turn = False
# Submitted multiline user-message preview (display.user_message_preview in config.yaml)
_ump = CLI_CONFIG["display"].get("user_message_preview", {})
if not isinstance(_ump, dict):
@ -5295,6 +5311,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
txt = getattr(self, "_spinner_text", "")
if not txt:
return ""
flow = self._spinner_token_flow()
t0 = getattr(self, "_tool_start_time", 0) or 0
if t0 > 0:
elapsed = time.monotonic() - t0
@ -5306,9 +5323,100 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
else:
# Keep width stable before the 60s rollover as well.
elapsed_str = f"{elapsed:5.1f}s"
if flow:
return f" {txt} ({elapsed_str} · {flow})"
return f" {txt} ({elapsed_str})"
if flow:
return f" {txt} ({flow})"
return f" {txt}"
# ── Per-turn accounting (display.turn_summary / spinner_token_flow) ──
#
# Both features are CLI-only chrome. The tally is observed from the
# tool-progress callback this class already receives on every tool call,
# so nothing is threaded through the agent loop. Token flow reads the
# agent's cumulative session counters (bumped per API call in
# agent/conversation_loop.py) and subtracts a per-turn baseline.
def _spinner_token_flow(self) -> str:
"""Cumulative output tokens for the running turn, for the spinner."""
if not getattr(self, "_spinner_token_flow_enabled", False):
return ""
if not getattr(self, "_agent_running", False):
return ""
agent = getattr(self, "agent", None)
if agent is None:
return ""
try:
from agent.turn_summary import format_token_flow
produced = (getattr(agent, "session_output_tokens", 0) or 0) - (
getattr(self, "_turn_token_baseline", 0) or 0
)
return format_token_flow(produced)
except Exception:
return ""
def _turn_summary_is_active(self) -> bool:
"""Whether the per-turn summary line should render for this surface.
Gated off for: the config key, quiet/tool-progress-off mode, and any
non-interactive path (single query, ``-Q``, gateway/messaging) those
surfaces either want machine-readable output or carry their own footer.
"""
if not getattr(self, "_turn_summary_enabled", False):
return False
if getattr(self, "tool_progress_mode", "all") == "off":
return False
agent = getattr(self, "agent", None)
if agent is not None and getattr(agent, "quiet_mode", False):
return False
if not getattr(self, "_interactive_turn", False):
return False
return True
def _turn_summary_begin(self) -> None:
"""Start per-turn accounting for the turn that is about to run."""
try:
from agent.turn_summary import TurnSummaryCollector
collector = getattr(self, "_turn_summary_collector", None)
if collector is None:
collector = TurnSummaryCollector()
self._turn_summary_collector = collector
collector.begin()
self._turn_summary_start = time.monotonic()
agent = getattr(self, "agent", None)
self._turn_token_baseline = (
getattr(agent, "session_output_tokens", 0) or 0
) if agent is not None else 0
except Exception:
self._turn_summary_collector = None
def _turn_summary_record(self, function_name, result, is_error: bool) -> None:
"""Feed one completed tool call into the active tally."""
collector = getattr(self, "_turn_summary_collector", None)
if collector is None:
return
try:
collector.record_tool(function_name, result=result, is_error=bool(is_error))
except Exception:
pass
def _turn_summary_emit(self) -> None:
"""Print the post-turn accounting line, when enabled for this surface."""
collector = getattr(self, "_turn_summary_collector", None)
if collector is None or not self._turn_summary_is_active():
return
try:
started = getattr(self, "_turn_summary_start", 0.0) or 0.0
elapsed = max(0.0, time.monotonic() - started) if started else 0.0
line = collector.render(elapsed)
if line:
_cprint(f" {_DIM}{line}{_RST}")
except Exception:
logger.debug("Turn summary render failed", exc_info=True)
# ── Petdex mascot (base-CLI pet pane) ───────────────────────────────
#
# Parity with the TUI: a half-block sprite rendered as a prompt_toolkit
@ -11301,6 +11409,11 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
if event_type == "tool.completed":
self._tool_start_time = 0.0
# Per-turn accounting: this feed already sees every tool call with
# its result, so the summary line needs no agent-loop state.
self._turn_summary_record(
function_name, kwargs.get("result"), kwargs.get("is_error", False)
)
# Print stacked scrollback line for "new" / "all" / "verbose" modes.
# "verbose" was previously omitted here, so non-streaming model
# calls (MoA aggregator, copilot-acp) rendered each tool only into
@ -15957,8 +16070,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
# Regular chat - run agent
self._agent_running = True
self._interactive_turn = True
self._pet_turn_error = False
self._pet_reasoning = False
self._turn_summary_begin()
app.invalidate() # Refresh status line
try:
@ -15971,6 +16086,11 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
self._last_scrollback_tool = ""
self._pet_reasoning = False
self._pet_react_turn_end()
# Post-turn accounting line (display.turn_summary).
# Emitted after the response box, before the prompt
# returns, so it reads as a footer for the turn.
self._turn_summary_emit()
self._interactive_turn = False
app.invalidate() # Refresh status line

View file

@ -1990,6 +1990,16 @@ DEFAULT_CONFIG = {
# tool name. Applies to CLI spinner + gateway/desktop tool-progress.
# Custom/plugin/MCP tools always fall back to the raw preview.
"friendly_tool_labels": True,
# CLI-only post-turn accounting line printed after each interactive turn:
# "⋯ 12.4s · edited 2 files +18 -3 · read 4 files · ran 3 commands".
# Observed from the tool-progress feed the CLI already receives; never
# printed in quiet/non-interactive paths or in gateway/messaging
# surfaces (those have their own runtime footer).
"turn_summary": True,
# CLI-only: append cumulative turn output tokens to the live spinner
# timer ("⚡ Reading file ( 2.3s · ↓ 1.2k tok)"). Updates as each API
# call in the turn reports usage.
"spinner_token_flow": True,
# How gateway tool-progress is grouped on platforms that support message
# editing: "accumulate" (default) edits one bubble in place; "separate"
# sends one message per tool (the pre-v0.9 behavior, noisier). Only

View file

@ -0,0 +1,346 @@
"""Tests for per-turn accounting: summary formatter, collector, spinner flow, gating.
The formatter and collector are pure (no terminal, no agent, no network), so
they're exercised directly. The gating test drives the real CLI methods on a
stub object with only the attributes the gate reads, so quiet-mode /
config-false behaviour is verified against the shipped code path rather than
against a re-implementation.
"""
import pytest
from agent.turn_summary import (
TurnSummaryCollector,
TurnTally,
format_elapsed,
format_token_flow,
format_turn_summary,
)
# ── format_elapsed ──────────────────────────────────────────────────────────
@pytest.mark.parametrize(
"seconds,expected",
[
(0.0, "0.0s"),
(12.44, "12.4s"),
(59.9, "59.9s"),
(60.0, "1m00s"),
(125.0, "2m05s"),
(-3.0, "0.0s"),
],
)
def test_format_elapsed(seconds, expected):
assert format_elapsed(seconds) == expected
# ── format_turn_summary: pure formatter ─────────────────────────────────────
def test_zero_tools_fast_turn_renders_nothing():
"""A quick chat reply with no tool calls has nothing to summarise."""
assert format_turn_summary(0.8, TurnTally()) == ""
def test_zero_tools_slow_turn_still_reports_wall_time():
"""A long toolless turn (big model, no tools) is worth timing."""
assert format_turn_summary(31.2, TurnTally()) == "⋯ 31.2s"
def test_single_edit_with_line_deltas():
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool(
"patch",
result='{"success": true, "diff": "--- a/x.py\\n+++ b/x.py\\n@@\\n+new\\n-old\\n"}',
)
assert collector.render(6.0) == "⋯ 6.0s · edited 1 file +1 -1"
def test_mixed_verbs_render_in_priority_order():
collector = TurnSummaryCollector()
collector.begin()
for _ in range(4):
collector.record_tool("read_file", result="contents")
for _ in range(3):
collector.record_tool("terminal", result="ok")
collector.record_tool("write_file", result='{"bytes_written": 10}')
collector.record_tool("write_file", result='{"bytes_written": 12}')
line = collector.render(12.4)
# Edits first, then reads, then commands — regardless of call order.
assert line == "⋯ 12.4s · edited 2 files · read 4 files · ran 3 commands"
def test_pluralization_singular_and_plural():
one = TurnTally(verbs={"read": {"files": 1}})
many = TurnTally(verbs={"read": {"files": 3}})
assert format_turn_summary(1.0, one) == "⋯ 1.0s · read 1 file"
assert format_turn_summary(1.0, many) == "⋯ 1.0s · read 3 files"
def test_pluralization_irregular_nouns():
"""The singulariser handles -ies / -ses without producing 'memorie'."""
mem = TurnTally(verbs={"updated": {"memories": 1}})
assert format_turn_summary(1.0, mem) == "⋯ 1.0s · updated 1 memory"
times = TurnTally(verbs={"searched the web": {"times": 1}})
assert format_turn_summary(1.0, times) == "⋯ 1.0s · searched the web 1 time"
def test_missing_line_deltas_omits_plus_minus():
"""write_file reports no diff, so we count the edit and skip +/-."""
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("write_file", result='{"bytes_written": 42}')
line = collector.render(3.0)
assert line == "⋯ 3.0s · edited 1 file"
assert "+" not in line and " -" not in line
def test_patch_result_without_diff_field_omits_deltas():
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("patch", result='{"success": true}')
assert collector.render(2.0) == "⋯ 2.0s · edited 1 file"
def test_diff_headers_not_counted_as_line_changes():
collector = TurnSummaryCollector()
collector.begin()
diff = "--- a/f.py\n+++ b/f.py\n@@ -1,2 +1,3 @@\n ctx\n+a\n+b\n-c\n"
collector.record_tool("patch", result={"success": True, "diff": diff})
assert collector.render(1.0) == "⋯ 1.0s · edited 1 file +2 -1"
def test_json_string_result_with_raw_newlines_still_parses():
"""Some serialisers emit literal newlines inside the diff string."""
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("patch", result='{"success": true, "diff": "@@\n+a\n+b\n-c\n"}')
assert collector.render(1.0) == "⋯ 1.0s · edited 1 file +2 -1"
def test_long_tallies_truncate_to_more_tail():
tally = TurnTally(
verbs={
"edited": {"files": 1},
"read": {"files": 2},
"ran": {"commands": 3},
"searched": {"paths": 4},
"browsed": {"pages": 5},
"delegated": {"tasks": 6},
}
)
line = format_turn_summary(9.0, tally)
assert line == "⋯ 9.0s · edited 1 file · read 2 files · ran 3 commands · searched 4 paths · +2 more"
assert line.count("·") == 5
def test_max_segments_configurable():
tally = TurnTally(verbs={"edited": {"files": 1}, "read": {"files": 2}, "ran": {"commands": 1}})
assert format_turn_summary(5.0, tally, max_segments=1) == "⋯ 5.0s · edited 1 file · +2 more"
def test_none_tally_is_safe():
assert format_turn_summary(0.1, None) == ""
# ── collector semantics ────────────────────────────────────────────────────
def test_failed_tools_are_not_counted():
"""A denied write must not be summarised as a successful edit."""
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("write_file", result='{"error": "denied"}', is_error=True)
assert collector.tally.total_tools == 0
assert collector.render(4.0) == "⋯ 4.0s"
def test_internal_and_empty_tool_names_ignored():
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("_thinking")
collector.record_tool(None)
collector.record_tool("")
assert collector.tally.total_tools == 0
def test_unknown_tools_bucket_into_generic_count():
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("some_mcp__weird_tool", result="{}")
collector.record_tool("another_plugin_tool", result="{}")
assert collector.render(2.0) == "⋯ 2.0s · called 2 tools"
def test_begin_resets_previous_turn():
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("read_file", result="x")
collector.begin()
assert collector.tally.total_tools == 0
assert collector.render(0.5) == ""
def test_line_deltas_aggregate_across_edits():
collector = TurnSummaryCollector()
collector.begin()
collector.record_tool("patch", result={"success": True, "diff": "@@\n+a\n+b\n-c\n"})
collector.record_tool("patch", result={"success": True, "diff": "@@\n+d\n-e\n-f\n"})
assert collector.render(7.5) == "⋯ 7.5s · edited 2 files +3 -3"
def test_malformed_result_payloads_do_not_raise():
collector = TurnSummaryCollector()
collector.begin()
for bad in ("not json at all", "", None, 42, [], {"diff": None}):
collector.record_tool("patch", result=bad)
assert collector.tally.verbs["edited"]["files"] == 6
assert collector.tally.has_line_deltas is False
# ── spinner token flow (PART B) ────────────────────────────────────────────
@pytest.mark.parametrize(
"tokens,expected",
[
(0, ""),
(-5, ""),
(32, "↓ 32 tok"),
(999, "↓ 999 tok"),
(1200, "↓ 1.2k tok"),
(25_400, "↓ 25.4k tok"),
(2_500_000, "↓ 2.5M tok"),
],
)
def test_format_token_flow(tokens, expected):
assert format_token_flow(tokens) == expected
def test_format_token_flow_bad_input_is_empty():
assert format_token_flow(None) == ""
assert format_token_flow("lots") == ""
# ── gating: quiet mode / config false / non-interactive ────────────────────
class _StubAgent:
def __init__(self, quiet_mode=False, session_output_tokens=0):
self.quiet_mode = quiet_mode
self.session_output_tokens = session_output_tokens
def _make_cli(**overrides):
"""Bind the real CLI accounting methods onto a minimal stub object.
Avoids constructing HermesCLI (which loads config, sessions, and a
prompt_toolkit app) while still exercising the shipped gate + emit code.
"""
import cli as cli_module
class _Stub:
_turn_summary_enabled = True
_spinner_token_flow_enabled = True
tool_progress_mode = "all"
_interactive_turn = True
_agent_running = True
agent = None
_turn_summary_collector = None
_turn_summary_start = 0.0
_turn_token_baseline = 0
_spinner_text = "⚡ reading file"
_tool_start_time = 0
_turn_summary_is_active = cli_module.HermesCLI._turn_summary_is_active
_turn_summary_begin = cli_module.HermesCLI._turn_summary_begin
_turn_summary_record = cli_module.HermesCLI._turn_summary_record
_turn_summary_emit = cli_module.HermesCLI._turn_summary_emit
_spinner_token_flow = cli_module.HermesCLI._spinner_token_flow
_render_spinner_text = cli_module.HermesCLI._render_spinner_text
stub = _Stub()
for key, value in overrides.items():
setattr(stub, key, value)
return stub
def _emit_and_capture(stub, monkeypatch):
printed = []
import cli as cli_module
monkeypatch.setattr(cli_module, "_cprint", lambda text: printed.append(text))
stub._turn_summary_begin()
stub._turn_summary_record("read_file", "contents", False)
stub._turn_summary_emit()
return printed
def test_gating_enabled_prints_summary(monkeypatch):
stub = _make_cli()
printed = _emit_and_capture(stub, monkeypatch)
assert len(printed) == 1
assert "read 1 file" in printed[0]
def test_gating_quiet_mode_prints_nothing(monkeypatch):
stub = _make_cli(agent=_StubAgent(quiet_mode=True))
assert _emit_and_capture(stub, monkeypatch) == []
def test_gating_config_false_prints_nothing(monkeypatch):
stub = _make_cli(_turn_summary_enabled=False)
assert _emit_and_capture(stub, monkeypatch) == []
def test_gating_tool_progress_off_prints_nothing(monkeypatch):
stub = _make_cli(tool_progress_mode="off")
assert _emit_and_capture(stub, monkeypatch) == []
def test_gating_non_interactive_prints_nothing(monkeypatch):
"""Single-query / -Q / gateway paths never set _interactive_turn."""
stub = _make_cli(_interactive_turn=False)
assert _emit_and_capture(stub, monkeypatch) == []
def test_spinner_token_flow_appears_when_enabled():
stub = _make_cli(agent=_StubAgent(session_output_tokens=1200))
assert stub._spinner_token_flow() == "↓ 1.2k tok"
assert "↓ 1.2k tok" in stub._render_spinner_text()
def test_spinner_token_flow_uses_per_turn_baseline():
stub = _make_cli(
agent=_StubAgent(session_output_tokens=5200), _turn_token_baseline=5000
)
assert stub._spinner_token_flow() == "↓ 200 tok"
def test_spinner_token_flow_config_false_is_silent():
stub = _make_cli(
_spinner_token_flow_enabled=False, agent=_StubAgent(session_output_tokens=9000)
)
assert stub._spinner_token_flow() == ""
assert "tok" not in stub._render_spinner_text()
def test_spinner_token_flow_silent_without_agent_or_idle():
assert _make_cli(agent=None)._spinner_token_flow() == ""
assert (
_make_cli(agent=_StubAgent(session_output_tokens=500), _agent_running=False)
._spinner_token_flow()
== ""
)
def test_turn_summary_config_defaults_present():
from hermes_cli.config import DEFAULT_CONFIG
display = DEFAULT_CONFIG["display"]
assert display["turn_summary"] is True
assert display["spinner_token_flow"] is True

View file

@ -1524,6 +1524,8 @@ display:
show_cost: false # Show estimated $ cost in the CLI status bar
timestamps: false # When true, prefixes user and assistant labels with [HH:MM] timestamps in the CLI / TUI transcript
tool_preview_length: 0 # Max chars for tool call previews (0 = no limit, show full paths/commands)
turn_summary: true # CLI only: print a one-line post-turn accounting footer after each interactive turn
spinner_token_flow: true # CLI only: append live cumulative turn tokens to the spinner timer
runtime_footer: # Gateway: append a runtime-context footer to final replies
enabled: false
fields: ["model", "context_pct", "cwd"]
@ -1532,6 +1534,33 @@ display:
language: en # UI language for static messages (approval prompts, some gateway replies). en | zh | zh-hant | ja | de | es | fr | tr | uk | af | ko | it | ga | pt | ru | hu
```
### Per-turn summary and spinner token flow
`display.turn_summary` (default `true`) prints one dim accounting line after each **interactive CLI** turn, summarising what that turn actually did:
```
⋯ 12.4s · edited 2 files +18 -3 · read 4 files · ran 3 commands
```
The tally is observed from the tool-progress feed the CLI already receives, so it costs nothing extra. Details:
- Wall time is the turn's real duration (`2m05s` past the one-minute mark).
- Tool calls are grouped by verb (`edited`, `read`, `ran`, `searched`, …) with correct pluralisation; plugin/MCP tools without a curated verb collapse into `called N tools`.
- `+X -Y` line deltas appear only when the tool result already reports a diff (currently `patch`). Hermes never shells out to git to compute them, so a `write_file` edit is counted without a delta.
- **Failed tool calls are not counted** — a denied write never renders as a successful edit (see the [file-mutation verifier](#file-mutation-verifier) for the complementary warning).
- Long turns cap at four verb segments plus a `+N more` tail so the line never wraps.
- A fast turn with no tool calls prints nothing at all.
`display.spinner_token_flow` (default `true`) appends the running turn's cumulative output tokens to the CLI spinner's live timer:
```
⚡ Reading cli.py ( 2.3s · ↓ 1.2k tok)
```
The count is per-turn (session totals are baselined at turn start) and updates as each API call in the turn reports usage. Nothing renders before the first usage report lands, so you never see a misleading `↓ 0 tok`.
Both keys are display-only and CLI-only: they are suppressed in quiet mode, when `display.tool_progress` is `off`, in single-query/`-Q` batch runs, and in gateway/messaging surfaces (those use `display.runtime_footer` instead). Set either key to `false` to turn it off.
### File-mutation verifier
When `display.file_mutation_verifier` is `true` (default), Hermes appends a one-line advisory to the assistant's final response whenever a `write_file` or `patch` call failed during the turn and was never superseded by a successful write to the same path. This catches the "batch of parallel patches, half silently fail, model summarises success" class of over-claim without requiring you to manually run `git status` after every edit.