feat(status): rich /status snapshots across CLI and gateway

agent/context_compressor.py

@@ -2334,6 +2334,10 @@ def __init__(
         self._tail_token_budget: int | None = None
         self._max_summary_tokens: int | None = None
         self.compression_count = 0
+        # Wall-clock timestamp (epoch seconds) of the most recent successful
+        # compression. Surfaced via /status so users can see *when* the last
+        # compaction happened, not just how many (#7317).
+        self.last_compressed_at: Optional[float] = None
 
         # The "initialized" log reports resolved token budgets, which would
         # force the deferred get_model_context_length() probe to run inside
@@ -6680,6 +6684,7 @@ def _is_nonempty_user_turn(message: Dict[str, Any]) -> bool:
             compressed.append(msg)
 
         self.compression_count += 1
+        self.last_compressed_at = time.time()
 
         compressed = self._sanitize_tool_pairs(compressed)
 

gateway/run.py

@@ -5350,6 +5350,11 @@ def _approval_notify_sync(approval_data: dict) -> None:
             _input_toks = getattr(_agent, "session_prompt_tokens", 0)
             _output_toks = getattr(_agent, "session_completion_tokens", 0)
             _context_length = getattr(_agent.context_compressor, "context_length", 0) or 0
+            _compression_count = getattr(_agent.context_compressor, "compression_count", 0) or 0
+            _last_compressed_at = getattr(_agent.context_compressor, "last_compressed_at", None)
+        else:
+            _compression_count = 0
+            _last_compressed_at = None
         _resolved_model = getattr(_agent, "model", None) if _agent else None
 
         # Sync session_id immediately after run_conversation(). Compression
@@ -5483,6 +5488,8 @@ def _approval_notify_sync(approval_data: dict) -> None:
                 "compacted_in_place": _compacted_in_place,
                 "session_id": effective_session_id,
                 "last_prompt_tokens": _last_prompt_toks,
+                "compression_count": _compression_count,
+                "last_compressed_at": _last_compressed_at,
                 "input_tokens": _input_toks,
                 "output_tokens": _output_toks,
                 "model": _resolved_model,
@@ -5621,6 +5628,8 @@ def _title_failure_cb(task: str, exc: BaseException) -> None:
             "history_offset": _effective_history_offset,
             "compacted_in_place": _compacted_in_place,
             "last_prompt_tokens": _last_prompt_toks,
+            "compression_count": _compression_count,
+            "last_compressed_at": _last_compressed_at,
             "input_tokens": _input_toks,
             "output_tokens": _output_toks,
             "model": _resolved_model,
@@ -17920,11 +17929,14 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                         )
 
             # Token counts and model are now persisted by the agent directly.
-            # Keep only last_prompt_tokens here for context-window tracking and
-            # compression decisions.
+            # Keep last_prompt_tokens, compression_count, and last_compressed_at
+            # here for context-window tracking, compression decisions, and idle
+            # /status display (#7317 — show compaction history + when).
             await self.async_session_store.update_session(
                 session_entry.session_key,
                 last_prompt_tokens=agent_result.get("last_prompt_tokens", 0),
+                compression_count=agent_result.get("compression_count", 0),
+                last_compressed_at=agent_result.get("last_compressed_at"),
             )
 
             # Re-baseline the cached agent's message_count snapshot now that

gateway/session.py

@@ -806,7 +806,17 @@ class SessionEntry:
 
     # Last API-reported prompt tokens (for accurate compression pre-check)
     last_prompt_tokens: int = 0
-
+
+    # How many times context compression has run in this session. Persisted
+    # by the gateway after each turn so idle /status can show compaction
+    # history without a resident agent (#7317).
+    compression_count: int = 0
+
+    # Wall-clock timestamp of the most recent compression. Stored as seconds
+    # since epoch; ``None`` when compression has never run for this session
+    # (#7317 — show *when* as well as *how many*).
+    last_compressed_at: Optional[float] = None
+
     # Set when a session was created because the previous one expired;
     # consumed once by the message handler to inject a notice into context
     was_auto_reset: bool = False
@@ -877,6 +887,8 @@ def to_dict(self) -> Dict[str, Any]:
             "cache_write_tokens": self.cache_write_tokens,
             "total_tokens": self.total_tokens,
             "last_prompt_tokens": self.last_prompt_tokens,
+            "compression_count": self.compression_count,
+            "last_compressed_at": self.last_compressed_at,
             "estimated_cost_usd": self.estimated_cost_usd,
             "cost_status": self.cost_status,
             "expiry_finalized": self.expiry_finalized,
@@ -958,6 +970,8 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionEntry":
             cache_write_tokens=data.get("cache_write_tokens", 0),
             total_tokens=data.get("total_tokens", 0),
             last_prompt_tokens=data.get("last_prompt_tokens", 0),
+            compression_count=data.get("compression_count", 0),
+            last_compressed_at=data.get("last_compressed_at"),
             estimated_cost_usd=data.get("estimated_cost_usd", 0.0),
             cost_status=data.get("cost_status", "unknown"),
             expiry_finalized=data.get("expiry_finalized", data.get("memory_flushed", False)),
@@ -2639,6 +2653,8 @@ def update_session(
         self,
         session_key: str,
         last_prompt_tokens: int = None,
+        compression_count: int = None,
+        last_compressed_at: Optional[float] = None,
     ) -> None:
         """Update lightweight session metadata after an interaction."""
         with self._lock:
@@ -2649,6 +2665,10 @@ def update_session(
             entry.updated_at = _now()
             if last_prompt_tokens is not None:
                 entry.last_prompt_tokens = last_prompt_tokens
+            if compression_count is not None:
+                entry.compression_count = compression_count
+            if last_compressed_at is not None:
+                entry.last_compressed_at = last_compressed_at
             # Snapshot peer fields while still holding _lock: a concurrent
             # reset/heal may rewrite the entry, and mixing old and new
             # fields would record a torn peer row.

hermes_cli/status_format.py

@@ -0,0 +1,81 @@
+"""Shared formatting helpers for /status snapshot displays.
+
+Used by both the CLI (cli.py) and gateway (gateway/slash_commands.py) to render
+consistent status output without duplicating logic.
+"""
+
+from datetime import datetime
+from typing import Any, Optional
+
+
+def format_status_relative_time(ts: Optional[datetime]) -> str:
+    """Return a compact relative timestamp for status displays."""
+    if not ts:
+        return "unknown"
+    delta = datetime.now() - ts
+    seconds = max(int(delta.total_seconds()), 0)
+    if seconds < 60:
+        return "just now"
+    if seconds < 3600:
+        return f"{seconds // 60}m ago"
+    if seconds < 86400:
+        return f"{seconds // 3600}h ago"
+    return f"{seconds // 86400}d ago"
+
+
+def format_status_cost(amount: Optional[float], status: str) -> Optional[str]:
+    """Return a human-readable cost label for status output."""
+    normalized = (status or "").strip().lower()
+    if normalized == "included":
+        return "included"
+    if amount is None and normalized in ("", "unknown", "none"):
+        return None
+    if amount is None:
+        return normalized
+    label = f"${amount:,.4f}"
+    if normalized in ("", "actual"):
+        return label
+    if normalized == "estimated":
+        return f"{label} est."
+    return f"{label} {normalized}"
+
+
+def format_reasoning_effort_label(config: Optional[dict]) -> str:
+    """Return the effective reasoning effort label."""
+    if config is None:
+        return "medium"
+    if config.get("enabled") is False:
+        return "none"
+    return str(config.get("effort") or "medium")
+
+
+def format_api_mode_label(api_mode: Optional[str]) -> Optional[str]:
+    """Convert internal API mode names into compact user-facing labels."""
+    if not api_mode:
+        return None
+    labels = {
+        "chat_completions": "Chat Completions",
+        "codex_responses": "Responses",
+        "anthropic_messages": "Anthropic Messages",
+    }
+    return labels.get(api_mode, str(api_mode).replace("_", " ").title())
+
+
+def safe_status_int(value: Any, default: int = 0) -> int:
+    """Best-effort integer coercion for status values."""
+    try:
+        if value is None:
+            return default
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def safe_status_float(value: Any) -> Optional[float]:
+    """Best-effort float coercion for status values."""
+    try:
+        if value is None:
+            return None
+        return float(value)
+    except (TypeError, ValueError):
+        return None

hermes_state.py

@@ -5091,6 +5091,58 @@ def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
             row = cursor.fetchone()
         return self._session_row_dict(row) if row else None
 
+    def get_session_token_totals(self, session_id: str) -> Optional[Dict[str, int]]:
+        """Get token totals for a session from SessionDB.
+
+        Returns a dict with input_tokens, output_tokens, cache_read_tokens,
+        cache_write_tokens, reasoning_tokens, and total_tokens (sum of all).
+        Returns None if the session is not found.
+        """
+        self.flush_token_counts()
+        with self._read_ctx() as conn:
+            cursor = conn.execute(
+                """SELECT input_tokens, output_tokens, cache_read_tokens,
+                          cache_write_tokens, reasoning_tokens
+                   FROM sessions WHERE id = ?""",
+                (session_id,),
+            )
+            row = cursor.fetchone()
+        if row:
+            totals = {
+                "input_tokens": row["input_tokens"] or 0,
+                "output_tokens": row["output_tokens"] or 0,
+                "cache_read_tokens": row["cache_read_tokens"] or 0,
+                "cache_write_tokens": row["cache_write_tokens"] or 0,
+                "reasoning_tokens": row["reasoning_tokens"] or 0,
+            }
+            totals["total_tokens"] = sum(totals.values())
+            return totals
+        return None
+
+    def get_session_last_active(self, session_id: str) -> Optional[float]:
+        """Return the latest activity timestamp for a session.
+
+        Uses the most recent message timestamp when available, otherwise falls
+        back to the session's ``started_at`` value. Returns ``None`` when the
+        session does not exist.
+        """
+        with self._read_ctx() as conn:
+            cursor = conn.execute(
+                """
+                SELECT COALESCE(
+                    (SELECT MAX(m.timestamp) FROM messages m WHERE m.session_id = s.id),
+                    s.started_at
+                ) AS last_active
+                FROM sessions s
+                WHERE s.id = ?
+                """,
+                (session_id,),
+            )
+            row = cursor.fetchone()
+        if not row or row["last_active"] is None:
+            return None
+        return float(row["last_active"])
+
     def resolve_session_id(self, session_id_or_prefix: str) -> Optional[str]:
         """Resolve an exact or uniquely prefixed session ID to the full ID.
 

locales/af.yaml

@@ -328,6 +328,28 @@ Future messages in this room will use that transcript until `/reset` or another
     state_no:              "Nee"
     queued:                "**Opgehoopte opvolge:** {count}"
     platforms:             "**Verbinde Platforms:** {platforms}"
+    # Rich snapshot additions (PR #8355 retarget). Composed by the retargeted
+    # _handle_status_command in gateway/slash_commands.py; each line is a
+    # separately localized phrase so translators control markup.
+    usage:                 "**Usage:** {input} in · {output} out · {total} total"
+    usage_cost:            "**Usage:** {input} in · {output} out · {total} total · **Cost:** {cost}"
+    cache:                 "**Cache:** {read} read · {write} write"
+    cache_hit:             "**Cache:** {read} read · {write} write · {pct}% hit"
+    cache_reasoning:       "**Cache:** {read} read · {write} write · {pct}% hit · {reasoning} reasoning"
+    cache_reasoning_only:  "**Cache:** {read} read · {write} write · {reasoning} reasoning"
+    cost:                  "**Cost:** {cost}"
+    context_compactions:   "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count}"
+    context_compactions_at: "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count} (last {when})"
+    compactions:           "**Compactions:** {count}"
+    compactions_at:        "**Compactions:** {count} (last {when})"
+    runtime:               "**Runtime:** {parts}"
+    queue_state:           "**Queue:** depth {depth} · **State:** {state}"
+    state_running:         "running"
+    state_idle:            "idle"
+    chats:                 "**Chats:** {platforms}"
+    services:              "**Services:** {platforms}"
+    transport_webhook:     "Transport webhook"
+    transport_polling:     "Transport polling"
 
   stop:
     stopped_pending:       "⚡ Gestop. Die agent het nog nie begin nie — jy kan met hierdie sessie voortgaan."

locales/ar.yaml

@@ -330,6 +330,28 @@ gateway:
     state_no:              "لا"
     queued:                "**المتابعات في الطابور:** {count}"
     platforms:             "**المنصّات المتصلة:** {platforms}"
+    # Rich snapshot additions (PR #8355 retarget). Composed by the retargeted
+    # _handle_status_command in gateway/slash_commands.py; each line is a
+    # separately localized phrase so translators control markup.
+    usage:                 "**Usage:** {input} in · {output} out · {total} total"
+    usage_cost:            "**Usage:** {input} in · {output} out · {total} total · **Cost:** {cost}"
+    cache:                 "**Cache:** {read} read · {write} write"
+    cache_hit:             "**Cache:** {read} read · {write} write · {pct}% hit"
+    cache_reasoning:       "**Cache:** {read} read · {write} write · {pct}% hit · {reasoning} reasoning"
+    cache_reasoning_only:  "**Cache:** {read} read · {write} write · {reasoning} reasoning"
+    cost:                  "**Cost:** {cost}"
+    context_compactions:   "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count}"
+    context_compactions_at: "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count} (last {when})"
+    compactions:           "**Compactions:** {count}"
+    compactions_at:        "**Compactions:** {count} (last {when})"
+    runtime:               "**Runtime:** {parts}"
+    queue_state:           "**Queue:** depth {depth} · **State:** {state}"
+    state_running:         "running"
+    state_idle:            "idle"
+    chats:                 "**Chats:** {platforms}"
+    services:              "**Services:** {platforms}"
+    transport_webhook:     "Transport webhook"
+    transport_polling:     "Transport polling"
 
   stop:
     stopped_pending:       "⚡ تم الإيقاف. لم يكن الوكيل قد بدأ بعد — يمكنك متابعة هذه الجلسة."

locales/de.yaml

@@ -328,6 +328,28 @@ Future messages in this room will use that transcript until `/reset` or another
     state_no:              "Nein"
     queued:                "**Wartende Folgenachrichten:** {count}"
     platforms:             "**Verbundene Plattformen:** {platforms}"
+    # Rich snapshot additions (PR #8355 retarget). Composed by the retargeted
+    # _handle_status_command in gateway/slash_commands.py; each line is a
+    # separately localized phrase so translators control markup.
+    usage:                 "**Usage:** {input} in · {output} out · {total} total"
+    usage_cost:            "**Usage:** {input} in · {output} out · {total} total · **Cost:** {cost}"
+    cache:                 "**Cache:** {read} read · {write} write"
+    cache_hit:             "**Cache:** {read} read · {write} write · {pct}% hit"
+    cache_reasoning:       "**Cache:** {read} read · {write} write · {pct}% hit · {reasoning} reasoning"
+    cache_reasoning_only:  "**Cache:** {read} read · {write} write · {reasoning} reasoning"
+    cost:                  "**Cost:** {cost}"
+    context_compactions:   "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count}"
+    context_compactions_at: "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count} (last {when})"
+    compactions:           "**Compactions:** {count}"
+    compactions_at:        "**Compactions:** {count} (last {when})"
+    runtime:               "**Runtime:** {parts}"
+    queue_state:           "**Queue:** depth {depth} · **State:** {state}"
+    state_running:         "running"
+    state_idle:            "idle"
+    chats:                 "**Chats:** {platforms}"
+    services:              "**Services:** {platforms}"
+    transport_webhook:     "Transport webhook"
+    transport_polling:     "Transport polling"
 
   stop:
     stopped_pending:       "⚡ Gestoppt. Der Agent hatte noch nicht begonnen — Sie können diese Sitzung fortsetzen."

locales/en.yaml

@@ -341,6 +341,28 @@ gateway:
     state_no:              "No"
     queued:                "**Queued follow-ups:** {count}"
     platforms:             "**Connected Platforms:** {platforms}"
+    # Rich snapshot additions (PR #8355 retarget). Composed by the retargeted
+    # _handle_status_command in gateway/slash_commands.py; each line is a
+    # separately localized phrase so translators control markup.
+    usage:                 "**Usage:** {input} in · {output} out · {total} total"
+    usage_cost:            "**Usage:** {input} in · {output} out · {total} total · **Cost:** {cost}"
+    cache:                 "**Cache:** {read} read · {write} write"
+    cache_hit:             "**Cache:** {read} read · {write} write · {pct}% hit"
+    cache_reasoning:       "**Cache:** {read} read · {write} write · {pct}% hit · {reasoning} reasoning"
+    cache_reasoning_only:  "**Cache:** {read} read · {write} write · {reasoning} reasoning"
+    cost:                  "**Cost:** {cost}"
+    context_compactions:   "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count}"
+    context_compactions_at: "**Context:** {used} / {total} ({pct}%) · **Compactions:** {count} (last {when})"
+    compactions:           "**Compactions:** {count}"
+    compactions_at:        "**Compactions:** {count} (last {when})"
+    runtime:               "**Runtime:** {parts}"
+    queue_state:           "**Queue:** depth {depth} · **State:** {state}"
+    state_running:         "running"
+    state_idle:            "idle"
+    chats:                 "**Chats:** {platforms}"
+    services:              "**Services:** {platforms}"
+    transport_webhook:     "Transport webhook"
+    transport_polling:     "Transport polling"
 
   stop:
     stopped_pending:       "⚡ Stopped. The agent hadn't started yet — you can continue this session."