feat(hermes_state): Gateway session activity tracking so silent stalls are visible

agent/agent_init.py

@@ -33,6 +33,7 @@
 from agent.context_compressor import ContextCompressor
 from agent.iteration_budget import IterationBudget
 from agent.memory_manager import StreamingContextScrubber
+from agent.session_activity import ActivityProvenance
 from agent.model_metadata import (
     MINIMUM_CONTEXT_LENGTH,
     fetch_model_metadata,
@@ -860,6 +861,11 @@ def init_agent(
     # notifications to show progress.
     agent._last_activity_ts: float = time.time()
     agent._last_activity_desc: str = "initializing"
+    # Default / unmigrated paths and _touch_activity stamp unknown; named
+    # provenances are reserved for special writers (e.g. #72424 compression).
+    agent._last_activity_provenance = ActivityProvenance.UNKNOWN
+    # Rate-limit durable SessionDB activity stamps from _touch_activity (#72016).
+    agent._session_activity_last_persist_mono: float = 0.0
     agent._current_tool: str | None = None
     agent._api_call_count: int = 0
     # Opt-out flag for the between-turns MCP tool refresh (build_turn_context).

agent/session_activity.py

@@ -0,0 +1,82 @@
+"""Shared session activity observation contract (#72016 / #72039).
+
+Observation-only: timestamp + bounded description/provenance.
+Notification, timeout, kill, and retry policy stay in their own components.
+Consumers distinguish work (API / tool / compacting / stalled) from the
+description text itself — there is no separate phase enum.
+
+Provenance is a small closed enum of *noun* sources (where the stamp came
+from). The default agent activity clock (``_touch_activity``) stamps
+``unknown`` unless a caller passes an explicit ``provenance=``; named
+values are for special writers.
+"""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Mapping, Optional
+
+ACTIVITY_DESCRIPTION_MAX = 120
+
+
+class ActivityProvenance(str, Enum):
+    """Where a durable/in-memory activity stamp came from."""
+
+    UNKNOWN = "unknown"
+    # Reserved for #72424 writers; not stamped by #72039 call sites yet.
+    AGENT_COMPRESSION = "agent.compression"
+    AGENT_COMPRESSION_TIMEOUT = "agent.compression_timeout"
+    AGENT_COMPRESSION_COOLDOWN = "agent.compression_cooldown"
+
+
+def bound_activity_description(description: Optional[str]) -> str:
+    """Clamp free-form activity text to the shared description budget."""
+    text = (description or "").strip()
+    if len(text) <= ACTIVITY_DESCRIPTION_MAX:
+        return text
+    return text[: ACTIVITY_DESCRIPTION_MAX - 1] + "…"
+
+
+def normalize_activity_provenance(
+    provenance: Optional[ActivityProvenance | str],
+) -> ActivityProvenance:
+    """Return a known provenance, or ``UNKNOWN`` when unset/unrecognized."""
+    if isinstance(provenance, ActivityProvenance):
+        return provenance
+    value = (provenance or "").strip()
+    try:
+        return ActivityProvenance(value)
+    except ValueError:
+        return ActivityProvenance.UNKNOWN
+
+
+def build_activity_snapshot(
+    *,
+    last_activity_at: Optional[float],
+    last_activity_description: Optional[str],
+    last_activity_provenance: Optional[ActivityProvenance | str] = None,
+    now: Optional[float] = None,
+    extra: Optional[Mapping[str, Any]] = None,
+) -> dict[str, Any]:
+    """Build the shared activity snapshot (plus optional caller extras)."""
+    import time as _time
+
+    when = float(last_activity_at) if last_activity_at is not None else None
+    clock = float(now if now is not None else _time.time())
+    desc = bound_activity_description(last_activity_description)
+    prov = normalize_activity_provenance(last_activity_provenance)
+    elapsed = round(clock - when, 1) if when is not None else None
+    snap: dict[str, Any] = {
+        "last_activity_at": when,
+        "last_activity_description": desc,
+        "last_activity_provenance": prov.value,
+        "seconds_since_activity": elapsed,
+        # Short aliases used by existing gateway/delegate readers.
+        "last_activity_ts": when,
+        "last_activity_desc": desc,
+        "description": desc,
+        "provenance": prov.value,
+    }
+    if extra:
+        snap.update(dict(extra))
+    return snap

gateway/run.py

@@ -19610,21 +19610,25 @@ def _evict_cached_agent(self, session_key: str) -> None:
     def _init_cached_agent_for_turn(agent: Any, interrupt_depth: int) -> None:
         """Reset per-turn state on a cached agent before a new turn starts.
 
-        Both _last_activity_ts and _last_activity_desc are only reset for
-        fresh external turns (depth 0); they are semantically paired —
-        desc describes the activity *at* ts, so updating one without the
-        other would make get_activity_summary() misleading.
-        For interrupt-recursive turns both are preserved so the inactivity
-        watchdog can accumulate stuck-turn idle time and fire the 30-min
-        timeout (#15654).  The depth-0 reset is still needed: a session
-        idle for 29 min would otherwise trip the watchdog before the new
-        turn makes its first API call (#9051).
+        ``_last_activity_ts``, ``_last_activity_desc``, and
+        ``_last_activity_provenance`` are only reset for fresh external
+        turns (depth 0); they are a semantic triple - description and
+        provenance describe the activity *at* ts, so updating one without
+        the others would make get_activity_summary() misleading.
+        For interrupt-recursive turns all three are preserved so the
+        inactivity watchdog can accumulate stuck-turn idle time and fire
+        the 30-min timeout (#15654).  The depth-0 reset is still needed:
+        a session idle for 29 min would otherwise trip the watchdog before
+        the new turn makes its first API call (#9051).
         """
         if interrupt_depth == 0:
+            from agent.session_activity import ActivityProvenance
+
             agent._last_activity_ts = time.time()
             agent._last_activity_desc = "starting new turn (cached)"
+            agent._last_activity_provenance = ActivityProvenance.UNKNOWN
             # Reset the SessionDB flush cursor so the new turn's messages are
-            # fully persisted — a stale value from the previous turn would
+            # fully persisted - a stale value from the previous turn would
             # cause `_flush_messages_to_session_db` to skip new rows (#44327).
             if hasattr(agent, "_last_flushed_db_idx"):
                 agent._last_flushed_db_idx = 0

hermes_cli/status.py

@@ -60,6 +60,27 @@ def _format_iso_timestamp(value) -> str:
     return parsed.astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
 
 
+def _format_relative_ts(ts: float) -> str:
+    """Format an epoch timestamp as a short relative age for status output."""
+    if not ts:
+        return "?"
+    import time as _time
+    from datetime import datetime
+
+    delta = _time.time() - float(ts)
+    if delta < 60:
+        return "just now"
+    if delta < 3600:
+        return f"{int(delta / 60)}m ago"
+    if delta < 86400:
+        return f"{int(delta / 3600)}h ago"
+    if delta < 172800:
+        return "yesterday"
+    if delta < 604800:
+        return f"{int(delta / 86400)}d ago"
+    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
+
+
 def _configured_model_label(config: dict) -> str:
     """Return the configured default model from config.yaml."""
     model_cfg = config.get("model")
@@ -550,20 +571,29 @@ def _resolve_env(env_ref) -> str:
     # Gateway session count: state.db is the source of truth (#9006);
     # fall back to sessions.json for pre-migration installs.
     _session_count = None
+    _gateway_rows = []
     try:
         from hermes_state import SessionDB
         _db = SessionDB()
         try:
             _lister = getattr(_db, "list_gateway_sessions", None)
             if callable(_lister):
-                _session_count = len(_lister(active_only=True))
+                _gateway_rows = _lister(active_only=True) or []
+                _session_count = len(_gateway_rows)
         finally:
             _db.close()
     except Exception:
         _session_count = None
+        _gateway_rows = []
 
     if _session_count is not None and _session_count > 0:
         print(f"  Active:       {_session_count} session(s)")
+        freshest = max(
+            (float(r.get("last_active") or 0) for r in _gateway_rows),
+            default=0.0,
+        )
+        if freshest > 0:
+            print(f"  Last activity:{_format_relative_ts(freshest):>13}")
     else:
         sessions_file = get_hermes_home() / "sessions" / "sessions.json"
         if sessions_file.exists():