hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(status): CLI, gateway and TUI /status render the same field set from hermes_cli/status_report.py

Browse Source 
The three /status renderers (hermes_cli/cli_session_mixin.py::_show_session_status,
gateway/slash_commands_status.py::_handle_status_command, tui_gateway/methods_session.py
session.status) each hand-built Session ID / Path / Title / Model (provider) / Created /
Last Activity / Tokens / Agent Running with their own getattr(agent, "model") fallback
chain, their own updated_at/last_updated_at/last_activity_at scan and their own timestamp
format. A fix to one (a new last-activity column, a placeholder change) silently missed the
other two.

hermes_cli/status_report.py::build_status_fields now derives the common facts once and
returns them as structured, display-ready data; status_lines() renders the English
"Label: value" form for the CLI and TUI. The gateway keeps translating through its
existing t("gateway.status.*") catalog keys (no locale change); the CLI keeps reasoning /
approvals / context, the gateway keeps free-tier / context / queue depth / Matrix scope,
the TUI keeps its Project line. tui_gateway/methods_session.py::_status_dt and the
CLI's inline updated_at loop are gone; cli_session_mixin._timestamp_or stays for its
remaining history-timestamp caller.

Behavior change: none intended for populated sessions. Unified edge cases: a
SessionDB row with an unparseable started_at now falls back to now() on the TUI as it
already did on the CLI, and the TUI's fallback on a bad updated_at is the created stamp on
both surfaces.

Test: tests/hermes_cli/test_status_report_contract.py drives the three real renderers with
one session (distinctive model, provider, title, stamps, token count) and asserts each
output carries every common value. Sabotage-verified: builder dropping tokens -> red;
TUI hand-formatting the model line -> red; restored -> green.
This commit is contained in:
teknium1
2026-09-12 19:49:34 -07:00
committed by Teknium
parent 991b23ad8d
commit f537c0b4c3
5 changed files with 262 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
gateway/slash_commands_status.py
View File

@@ -16,6 +16,7 @@ from agent.i18n import t
from gateway.config import Platform
from gateway.platforms.event import MessageEvent
from gateway.session_transcript import TranscriptReadError
from hermes_cli.status_report import build_status_fields
# Log-record parity with gateway/run.py and the origin module.
logger = logging.getLogger("gateway.run")
@@ -236,17 +237,21 @@ class GatewayStatusCommandsMixin:
status_agent, persisted_route, session_row, session_entry
)
stamp = "%Y-%m-%d %H:%M"
fields = build_status_fields(
session_entry.session_id, None, session_row, title=title, model=model_name, provider=provider_name,
created=session_entry.created_at, last_activity=session_entry.updated_at,
tokens=db_total_tokens, agent_running=is_running,
)
lines = [t("gateway.status.header"), "",
t("gateway.status.session_id", session_id=session_entry.session_id)]
if title:
lines.append(t("gateway.status.title", title=title))
lines += [t("gateway.status.created", timestamp=session_entry.created_at.strftime(stamp)),
t("gateway.status.last_activity", timestamp=session_entry.updated_at.strftime(stamp))]
if model_name and provider_name:
lines.append(t("gateway.status.model_provider", model=model_name, provider=provider_name))
elif model_name:
lines.append(t("gateway.status.model", model=model_name))
t("gateway.status.session_id", session_id=fields["session_id"])]
if fields["title"]:
lines.append(t("gateway.status.title", title=fields["title"]))
lines += [t("gateway.status.created", timestamp=fields["created"]),
t("gateway.status.last_activity", timestamp=fields["last_activity"])]
if fields["model"] and fields["provider"]:
lines.append(t("gateway.status.model_provider", model=fields["model"], provider=fields["provider"]))
elif fields["model"]:
lines.append(t("gateway.status.model", model=fields["model"]))
try:
from hermes_cli.auth import resolve_provider
from hermes_cli.anon_auth import guest_carries_inference
@@ -266,8 +271,8 @@ class GatewayStatusCommandsMixin:
pct=f"{mark}{pct}"))
elif context_used:
lines.append(t("gateway.status.context_used", used=mark + _fmt(context_used)))
state = t("gateway.status.state_yes") if is_running else t("gateway.status.state_no")
lines += [t("gateway.status.tokens", tokens=_fmt(db_total_tokens)),
state = t("gateway.status.state_yes") if fields["agent_running"] else t("gateway.status.state_no")
lines += [t("gateway.status.tokens", tokens=fields["tokens"]),
t("gateway.status.agent_running", state=state)]
if queue_depth:
lines.append(t("gateway.status.queued", count=queue_depth))

33
hermes_cli/cli_session_mixin.py
View File

@@ -245,26 +245,18 @@ class CLISessionMixin:
def _show_session_status(self):
"""Show gateway-style status for the current CLI session."""
from hermes_constants import display_hermes_home
from hermes_cli.status_report import build_status_fields, status_lines
session_meta = {}
if self._session_db:
with contextlib.suppress(Exception):
session_meta = self._session_db.get_session(self.session_id) or {}
title = (session_meta.get("title") or "").strip()
created_at = _timestamp_or(session_meta.get("started_at"), self.session_start)
updated_at = created_at
for field in ("updated_at", "last_updated_at", "last_activity_at"):
candidate = _timestamp_or(session_meta.get(field), None)
if candidate is not None:
updated_at = candidate
break
agent = getattr(self, "agent", None)
total_tokens = getattr(agent, "session_total_tokens", 0) or 0
provider = getattr(self, "provider", None) or "unknown"
model = getattr(self, "model", None) or "(unknown)"
is_running = bool(getattr(self, "_agent_running", False))
fields = build_status_fields(
self.session_id, agent, session_meta,
model=getattr(self, "model", None), provider=getattr(self, "provider", None),
created_fallback=self.session_start, agent_running=bool(getattr(self, "_agent_running", False)),
)
reasoning_label = None
rc = getattr(agent, "reasoning_config", None) or getattr(self, "reasoning_config", None)
@@ -301,12 +293,7 @@ class CLISessionMixin:
except Exception:
ctx_label = None
lines = [
"Hermes CLI Status", "", f"Session ID: {self.session_id}", f"Path: {display_hermes_home()}",
]
if title:
lines.append(f"Title: {title}")
lines.append(f"Model: {model} ({provider})")
lines = ["Hermes CLI Status", "", *status_lines(fields, "session_id", "path", "title", "model")]
try:
from agent.i18n import t
from hermes_cli.auth import resolve_provider
@@ -320,11 +307,7 @@ class CLISessionMixin:
for label, value in optional:
if value:
lines.append(f"{label}: {value}")
lines.extend([
f"Created: {created_at.strftime('%Y-%m-%d %H:%M')}",
f"Last Activity: {updated_at.strftime('%Y-%m-%d %H:%M')}",
f"Tokens: {total_tokens:,}",
f"Agent Running: {'Yes' if is_running else 'No'}"])
lines.extend(status_lines(fields, "created", "last_activity", "tokens", "agent_running"))
self._console_print("\n".join(lines), highlight=False, markup=False)
def _list_recent_sessions(self, limit: int = 10) -> list[dict[str, Any]]:

116
hermes_cli/status_report.py Normal file
View File

@@ -0,0 +1,116 @@
"""The one field set behind ``/status`` on every surface.
The CLI (``cli_session_mixin._show_session_status``), the messaging gateway
(``gateway/slash_commands_status._handle_status_command``) and the TUI backend
(``tui_gateway/methods_session`` ``session.status``) all report the same session facts —
id, home path, title, model route, created / last-activity stamps, lifetime tokens, running
flag. Each used to derive them independently (three ``getattr(agent, "model")`` fallback
chains, three ``updated_at`` candidate scans, three timestamp formats to keep aligned).
They now all call :func:`build_status_fields`; a surface only adds its own header, labels
(the gateway translates through ``t("gateway.status.*")``) and surface-specific extras.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
STATUS_STAMP = "%Y-%m-%d %H:%M"
UNKNOWN_MODEL = "(unknown)"
UNKNOWN_PROVIDER = "unknown"
# Newest-first: a row carries at most one of these, depending on which writer last touched it.
_LAST_ACTIVITY_FIELDS = ("updated_at", "last_updated_at", "last_activity_at")
# English labels shared by the CLI and TUI renderers (the gateway has its own i18n catalog).
STATUS_LABELS = {
"session_id": "Session ID",
"path": "Path",
"title": "Title",
"model": "Model",
"created": "Created",
"last_activity": "Last Activity",
"tokens": "Tokens",
"agent_running": "Agent Running",
}
def timestamp_or(value: Any, default: datetime | None) -> datetime | None:
"""``datetime.fromtimestamp(value)`` or *default* when the value is missing/unparseable."""
if not value:
return default
try:
return datetime.fromtimestamp(float(value))
except (TypeError, ValueError, OverflowError, OSError):
return default
def session_timestamps(meta: dict, created_fallback: datetime | None = None) -> tuple[datetime, datetime]:
"""``(created, last_activity)`` from a SessionDB row; last activity falls back to created."""
created = timestamp_or(meta.get("started_at"), created_fallback) or datetime.now()
updated = next(
(stamp for stamp in (timestamp_or(meta.get(f), None) for f in _LAST_ACTIVITY_FIELDS) if stamp),
created,
)
return created, updated
def build_status_fields(
session_id: str,
agent: Any,
meta: dict | None,
*,
title: str | None = None,
model: str | None = None,
provider: str | None = None,
created: datetime | None = None,
last_activity: datetime | None = None,
created_fallback: datetime | None = None,
tokens: int | None = None,
agent_running: bool = False,
) -> dict[str, Any]:
"""Common ``/status`` facts, pre-formatted for display.
``agent`` wins for model / provider / tokens; the keyword values are the surface's fallback
(CLI ``self.model``, TUI metadata mirror, gateway's resolved route) and are also what a
surface passes when it has no live agent. ``created`` / ``last_activity`` override the
``meta`` row scan for surfaces whose session store is authoritative (gateway SessionEntry).
"""
from hermes_constants import display_hermes_home
meta = meta or {}
if created is None or last_activity is None:
row_created, row_updated = session_timestamps(meta, created_fallback)
created = created or row_created
last_activity = last_activity or row_updated
if tokens is None:
tokens = getattr(agent, "session_total_tokens", 0) or 0
row_title = meta.get("title") if title is None else title
return {
"session_id": str(session_id or ""),
"path": display_hermes_home(),
"title": (row_title or "").strip(),
"model": getattr(agent, "model", None) or model or "",
"provider": getattr(agent, "provider", None) or provider or "",
"created": created.strftime(STATUS_STAMP),
"last_activity": last_activity.strftime(STATUS_STAMP),
"tokens": f"{int(tokens or 0):,}",
"agent_running": bool(agent_running),
}
def status_lines(fields: dict[str, Any], *keys: str) -> list[str]:
"""``Label: value`` lines for *keys* in order (English surfaces). An empty title is skipped;
an unresolved model/provider shows a placeholder rather than an empty parenthesis."""
lines: list[str] = []
for key in keys:
value = fields[key]
if key == "title" and not value:
continue
if key == "model":
value = f"{fields['model'] or UNKNOWN_MODEL} ({fields['provider'] or UNKNOWN_PROVIDER})"
elif key == "agent_running":
value = "Yes" if value else "No"
lines.append(f"{STATUS_LABELS[key]}: {value}")
return lines

112
tests/hermes_cli/test_status_report_contract.py Normal file
View File

@@ -0,0 +1,112 @@
"""Contract: the CLI, gateway and TUI /status renderers report the SAME common facts.
Every surface renders through ``hermes_cli.status_report.build_status_fields``; each keeps
its own header, labels (the gateway is i18n) and extras. This test feeds the three real
renderers the same session facts and asserts every common value appears in each output —
a contract between one builder and three renderers, not a snapshot of any layout.
"""
from __future__ import annotations
import asyncio
import threading
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
from gateway.config import Platform
from gateway.platforms.event import MessageEvent
from gateway.session import SessionEntry, SessionSource
from hermes_cli.cli_session_mixin import CLISessionMixin
SESSION_ID = "status-contract-7f3a"
MODEL = "vendor/distinctive-model-9000"
PROVIDER = "distinctive-provider"
TITLE = "Distinctive status title"
CREATED = datetime(2031, 3, 14, 15, 9)
UPDATED = datetime(2031, 3, 15, 1, 5)
TOKENS = 1_234_567
HOME = "~/.hermes-status-contract"
COMMON_VALUES = (
SESSION_ID, MODEL, PROVIDER, TITLE,
CREATED.strftime("%Y-%m-%d %H:%M"), UPDATED.strftime("%Y-%m-%d %H:%M"), f"{TOKENS:,}",
)
def _meta() -> dict:
return {"title": TITLE, "started_at": CREATED.timestamp(), "updated_at": UPDATED.timestamp()}
def _agent():
return SimpleNamespace(model=MODEL, provider=PROVIDER, session_total_tokens=TOKENS, reasoning_config=None)
def _render_cli() -> str:
db = MagicMock()
db.get_session.return_value = _meta()
rendered: list[str] = []
cli = SimpleNamespace(
_session_db=db, session_id=SESSION_ID, session_start=datetime(2000, 1, 1), agent=_agent(),
provider=PROVIDER, model=MODEL, _agent_running=True, reasoning_config=None, show_reasoning=None,
session_key="", _get_status_bar_snapshot=lambda: {},
_console_print=lambda text, **_kw: rendered.append(text),
)
CLISessionMixin._show_session_status(cli)
return "\n".join(rendered)
def _render_gateway() -> str:
from gateway.run import GatewayRunner
source = SessionSource(platform=Platform.TELEGRAM, user_id="u1", chat_id="c1", user_name="t", chat_type="dm")
entry = SessionEntry(session_key="telegram:c1", session_id=SESSION_ID, created_at=CREATED, updated_at=UPDATED,
platform=Platform.TELEGRAM, chat_type="dm", total_tokens=0)
runner = object.__new__(GatewayRunner)
runner.adapters = {}
runner._running_agents = {"telegram:c1": _agent()}
runner._queue_depth = lambda *_a, **_k: 0
runner._run_in_executor_with_context = AsyncMock(return_value=False)
runner.session_store = MagicMock()
runner.session_store.get_or_create_session.return_value = entry
runner._session_db = None
# Token totals come from the SQLite row (never from SessionEntry); model route from the live agent.
runner._status_session_db_facts = AsyncMock(return_value=(TITLE, {}, TOKENS, {}))
event = MessageEvent(text="/status", source=source, message_id="m1")
return asyncio.run(runner._handle_status_command(event))
def _render_tui() -> str:
from tui_gateway import server
class _DB:
def get_session(self, key):
return _meta() if key == SESSION_ID else {}
session = {
"agent": _agent(), "session_key": SESSION_ID, "history": [], "history_lock": threading.Lock(),
"history_version": 0, "running": True, "attached_images": [], "image_counter": 0, "cols": 80,
"slash_worker": None, "show_reasoning": False, "tool_progress_mode": "all",
}
server._sessions["status-contract-sid"] = session
try:
with patch.object(server, "_get_db", lambda: _DB()):
resp = server.handle_request(
{"id": "1", "method": "session.status", "params": {"session_id": "status-contract-sid"}})
finally:
server._sessions.pop("status-contract-sid", None)
return resp["result"]["output"]
def test_three_status_surfaces_report_the_same_common_fields():
with patch("hermes_constants.display_hermes_home", return_value=HOME), \
patch("tools.approval_context._get_approval_mode", side_effect=RuntimeError("n/a")):
outputs = {"cli": _render_cli(), "gateway": _render_gateway(), "tui": _render_tui()}
for surface, text in outputs.items():
for value in COMMON_VALUES:
assert value in text, f"{surface} /status lost {value!r}:\n{text}"
# Path is English-only (the gateway catalog has no path line); running flag is per-surface wording.
assert HOME in outputs["cli"] and HOME in outputs["tui"]
assert "Agent Running: Yes" in outputs["cli"] and "Agent Running: Yes" in outputs["tui"]
assert "Yes" in outputs["gateway"]

31
tui_gateway/methods_session.py
View File

@@ -1663,34 +1663,21 @@ def _try_get_session(db, key: str) -> dict:
return {}
def _status_dt(value, fallback=None):
if value:
with contextlib.suppress(Exception):
return datetime.fromtimestamp(float(value))
return fallback or datetime.now()
@_session_method("session.status")
def _(rid, params: dict, session: dict) -> dict:
from hermes_constants import display_hermes_home
from hermes_cli.status_report import build_status_fields, status_lines
key = session.get("session_key") or params.get("session_id") or ""
agent = session.get("agent")
meta = _status_row(session, params, key)
created = _status_dt(meta.get("started_at"))
updated = next((_status_dt(meta[f], created) for f in ("updated_at", "last_updated_at", "last_activity_at")
if meta.get(f)), created)
mirror = _metadata_mirror(session)
provider = getattr(agent, "provider", None) or mirror.get("provider") or "unknown"
model = getattr(agent, "model", None) or mirror.get("model") or "(unknown)"
fields = build_status_fields(
key, session.get("agent"), _status_row(session, params, key),
model=mirror.get("model"), provider=mirror.get("provider"),
tokens=_session_usage_snapshot(session).get("total"), agent_running=bool(session.get("running")),
)
project = _project_info_for_cwd(_display_session_cwd(session))
title = (meta.get("title") or "").strip()
lines = [
"Hermes TUI Status", "", f"Session ID: {key}", f"Path: {display_hermes_home()}",
*([f"Project: {project['name']}"] if project else []), *([f"Title: {title}"] if title else []),
f"Model: {model} ({provider})", f"Created: {created.strftime('%Y-%m-%d %H:%M')}",
f"Last Activity: {updated.strftime('%Y-%m-%d %H:%M')}",
f"Tokens: {int(_session_usage_snapshot(session).get('total') or 0):,}",
f"Agent Running: {'Yes' if session.get('running') else 'No'}"]
"Hermes TUI Status", "", *status_lines(fields, "session_id", "path"),
*([f"Project: {project['name']}"] if project else []),
*status_lines(fields, "title", "model", "created", "last_activity", "tokens", "agent_running")]
return _ok(rid, {"output": "\n".join(lines)})