feat(gateway): session activity watchdog, stall notify, compress timeout (#72424)

Three mechanisms to detect and notify when gateway sessions stall silently:

1. Mid-turn activity heartbeats stamped to SessionDB so hermes sessions list
   and hermes status show progress during long turns without new message rows.

2. Stall watchdog: when a busy session has pending inbound and the shared
   activity clock is idle past agent.session_stall_timeout (default 300),
   log a WARNING and notify the user once to try /new. Notify-only; does
   not kill the turn.

3. Compaction timeout: fenceless compress_context callers get a progress-aware
   host budget (compression.context_timeout_seconds default 120 idle,
   compression.context_total_ceiling_seconds default 600 ceiling). On timeout,
   cancel via commit fence, skip compaction without dropping messages, and
   continue the turn.

Closes #72016 (slices 1-3; slice 4 cumulative SSE stream-retry deadline
remains a follow-up).

Cherry-picked from PR #72424 by @fangliquanflq.

agent/agent_init.py

@@ -33,6 +33,7 @@ from urllib.parse import parse_qs, urlparse, urlunparse
 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,
@@ -867,6 +868,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 stamped by compression writers (heartbeat / timeout / cooldown).
+    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/conversation_compression.py

@@ -28,6 +28,7 @@ these paths see no behavioural change.
 
 from __future__ import annotations
 
+import concurrent.futures
 import copy
 import inspect
 import json
@@ -40,16 +41,27 @@ import uuid
 import threading
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 from agent.context_engine import (
     automatic_compaction_status_message,
     sanitize_memory_context,
 )
 from agent.model_metadata import estimate_request_tokens_rough
+from agent.session_activity import ActivityProvenance, normalize_activity_provenance
 
 logger = logging.getLogger(__name__)
 
+# Terminal compression outcomes published by host/hygiene timeout or cooldown
+# writers. Detached heartbeat workers must not clobber these back to
+# agent.compression after cancel (otherwise timeout is unobservable).
+_TERMINAL_COMPRESSION_PROVENANCES = frozenset(
+    {
+        ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
+        ActivityProvenance.AGENT_COMPRESSION_COOLDOWN,
+    }
+)
+
 # Stable marker the gateway matches on to re-tag the auto-compaction lifecycle
 # status as ``kind="compacting"`` (tui_gateway/server.py::_status_update), so
 # drivers like the desktop app can show an explicit "Summarizing…" indicator
@@ -284,6 +296,190 @@ class CompressionCommitFence:
         self._lock.release()
 
 
+# Defaults for the in-agent (non-hygiene) progress-aware compress_context wrap.
+# Mirror hermes_cli.config.DEFAULT_CONFIG["compression"] keys of the same name.
+DEFAULT_CONTEXT_TIMEOUT_SECONDS = 120.0
+DEFAULT_CONTEXT_TOTAL_CEILING_SECONDS = 600.0
+
+# Shared daemon pool for sync compress_context timeout wraps — analogous to
+# asyncio's default executor used by gateway session hygiene's
+# ``loop.run_in_executor(None, ...)``, but daemon so a fence-cancelled hung
+# worker cannot block interpreter exit via concurrent.futures' atexit join.
+# Created lazily; never shut down per call (a timed-out worker may still be
+# winding down after fence cancel).
+_compress_timeout_executor = None
+_compress_timeout_executor_lock = threading.Lock()
+
+
+def _get_compress_timeout_executor():
+    """Return the process-wide compress-timeout DaemonThreadPoolExecutor."""
+    global _compress_timeout_executor
+    executor = _compress_timeout_executor
+    if executor is not None:
+        return executor
+    from tools.daemon_pool import DaemonThreadPoolExecutor
+
+    with _compress_timeout_executor_lock:
+        if _compress_timeout_executor is None:
+            # Small pool: compress is rare and heavy. Sized for a few
+            # overlapping calls (live compress + fence-cancelled workers
+            # still winding down), not asyncio's min(32, cpu+4) fan-out.
+            _compress_timeout_executor = DaemonThreadPoolExecutor(
+                max_workers=4,
+                thread_name_prefix="compress-ctx-timeout",
+            )
+        return _compress_timeout_executor
+
+
+def resolve_context_compression_timeouts(
+    compression_cfg: Optional[dict] = None,
+) -> Tuple[float, float]:
+    """Return ``(idle_timeout_seconds, total_ceiling_seconds)``.
+
+    ``idle_timeout_seconds <= 0`` disables the owned progress-aware wrapper.
+    The ceiling is clamped to at least one idle window when the idle budget
+    is positive, matching gateway hygiene semantics.
+    """
+    idle = DEFAULT_CONTEXT_TIMEOUT_SECONDS
+    ceiling = DEFAULT_CONTEXT_TOTAL_CEILING_SECONDS
+    cfg = compression_cfg
+    if cfg is None:
+        try:
+            from hermes_cli.config import load_config
+
+            raw = load_config()
+            maybe = raw.get("compression", {}) if isinstance(raw, dict) else {}
+            cfg = maybe if isinstance(maybe, dict) else {}
+        except Exception:
+            cfg = {}
+    if isinstance(cfg, dict):
+        raw_idle = cfg.get("context_timeout_seconds")
+        if raw_idle is not None:
+            try:
+                parsed = float(raw_idle)
+                # Explicit 0/negative disables; positive values win.
+                idle = parsed
+            except (TypeError, ValueError):
+                pass
+        raw_ceiling = cfg.get("context_total_ceiling_seconds")
+        if raw_ceiling is not None:
+            try:
+                parsed = float(raw_ceiling)
+                if parsed > 0:
+                    ceiling = parsed
+            except (TypeError, ValueError):
+                pass
+    if idle > 0:
+        ceiling = max(ceiling, idle)
+    return idle, ceiling
+
+
+def run_compress_context_with_progress_timeout(
+    *,
+    worker: Callable[[CompressionCommitFence], Tuple[list, str]],
+    messages: list,
+    system_prompt_fallback: Any,
+    idle_timeout_seconds: float,
+    total_ceiling_seconds: float,
+    on_timeout: Optional[Callable[[float, float, float], None]] = None,
+) -> Tuple[list, str]:
+    """Run ``worker(fence)`` under a sync progress-aware timeout.
+
+    The idle budget is inactivity-based (same idea as gateway session hygiene):
+    streamed summary progress via :meth:`CompressionCommitFence.touch_progress`
+    extends the wait. A hard ceiling still bounds a degenerate trickle stream.
+
+    When cancellation wins before the commit boundary, returns
+    ``(messages, system_prompt_fallback)`` immediately and leaves the worker
+    thread detached — the fence prevents a late commit from mutating session
+    state. When the worker already entered the commit boundary, waits for that
+    commit to finish and returns its result.
+
+    ``system_prompt_fallback`` may be a string or a zero-arg callable resolved
+    only on the timeout path, so successful compression never pays for (or
+    fails on) an eager prompt rebuild.
+    """
+    if idle_timeout_seconds <= 0:
+        raise ValueError(
+            "run_compress_context_with_progress_timeout requires "
+            "idle_timeout_seconds > 0; call compress_context directly to disable"
+        )
+
+    def _resolve_fallback_prompt() -> str:
+        if callable(system_prompt_fallback):
+            return system_prompt_fallback()
+        return system_prompt_fallback
+
+    fence = CompressionCommitFence()
+    ceiling = max(float(total_ceiling_seconds), float(idle_timeout_seconds))
+    idle = float(idle_timeout_seconds)
+    # Sync mirror of gateway session-hygiene's run_in_executor(None, ...) +
+    # wait_for loop (gateway/run.py): offload compress_context onto the shared
+    # daemon pool, poll with an inactivity budget + total ceiling, then
+    # fence-cancel on timeout so a late commit cannot land. Daemon workers
+    # match tool_executor: a cancelled hung summary must not block process exit.
+    from tools.thread_context import propagate_context_to_thread
+
+    executor = _get_compress_timeout_executor()
+    # Bare pool workers start with an empty ContextVar map; propagate the
+    # parent conversation/approval context into the worker.
+    future = executor.submit(propagate_context_to_thread(worker), fence)
+    wait_started = time.monotonic()
+    while True:
+        waited = time.monotonic() - wait_started
+        remaining_ceiling = ceiling - waited
+        if remaining_ceiling <= 0:
+            break
+        wait_slice = min(idle, remaining_ceiling)
+        try:
+            return future.result(timeout=wait_slice)
+        except concurrent.futures.TimeoutError:
+            waited = time.monotonic() - wait_started
+            since_progress = fence.seconds_since_progress()
+            if since_progress < idle and waited < ceiling:
+                logger.info(
+                    "Context compression still streaming after %.0fs "
+                    "(last progress %.1fs ago) — extending wait "
+                    "(ceiling %.0fs)",
+                    waited,
+                    since_progress,
+                    ceiling,
+                )
+                continue
+            break
+
+    cancelled: Optional[bool] = None
+    while cancelled is None:
+        cancelled = fence.try_cancel_before_commit()
+        if cancelled is None:
+            time.sleep(0.001)
+    if not cancelled:
+        return future.result()
+
+    waited = time.monotonic() - wait_started
+    since_progress = fence.seconds_since_progress()
+    if on_timeout is not None:
+        try:
+            on_timeout(idle, waited, since_progress)
+        except Exception:
+            logger.debug(
+                "compress_context timeout callback failed",
+                exc_info=True,
+            )
+    else:
+        logger.warning(
+            "Context compression made no progress for %.1fs "
+            "(total wait %.1fs, ceiling %.1fs); continuing without "
+            "compression",
+            since_progress,
+            waited,
+            ceiling,
+        )
+    # Leave the future on the shared pool: fence cancel won, so a late
+    # commit cannot land (same detachment model as gateway hygiene).
+    return messages, _resolve_fallback_prompt()
+
+
 def _lock_api_is_absent_on_session_db(lock_db: Any) -> bool:
     """Whether the live in-memory SessionDB class structurally predates locks.
 
@@ -592,7 +788,9 @@ class _CompressionActivityHeartbeat:
         )
 
     def start(self) -> "_CompressionActivityHeartbeat":
-        self._touch("context compression started")
+        # A new compression episode always republishes agent.compression even
+        # if a prior timeout/cooldown stamp is still on the agent.
+        self._touch("context compression started", allow_terminal_overwrite=True)
         self._thread.start()
         return self
 
@@ -602,11 +800,17 @@ class _CompressionActivityHeartbeat:
             self._thread.join(timeout=1.0)
         self._touch(desc)
 
-    def _touch(self, desc: str) -> None:
+    def _touch(self, desc: str, *, allow_terminal_overwrite: bool = False) -> None:
         try:
+            if not allow_terminal_overwrite:
+                current = normalize_activity_provenance(
+                    getattr(self._agent, "_last_activity_provenance", None)
+                )
+                if current in _TERMINAL_COMPRESSION_PROVENANCES:
+                    return
             touch = getattr(self._agent, "_touch_activity", None)
             if callable(touch):
-                touch(desc)
+                touch(desc, provenance=ActivityProvenance.AGENT_COMPRESSION)
         except Exception:
             logger.debug("compression activity heartbeat touch failed", exc_info=True)
 
@@ -1791,12 +1995,15 @@ def compress_context(
         # thread-local and the compress call is synchronous on this thread,
         # so it cannot leak into unrelated auxiliary calls.
         #
-        # Fenceless callers (CLI /compress, in-loop auto-compress) install a
-        # no-op hook: nobody polls their progress, but an ACTIVE hook is what
-        # switches the summary call onto the streamed path — giving every
-        # compression path the same two guarantees: the configured timeout
-        # acts on inactivity (slow models finish), and a byte-trickling
-        # provider that keeps the connection alive forever is cut off at the
+        # Callers that pass no commit_fence install a no-op progress hook
+        # here.  AIAgent._compress_context injects an owned fence for
+        # fenceless callers so the host-level progress-aware wait can
+        # extend on streamed tokens; gateway hygiene already passes its
+        # own fence.  An ACTIVE hook (even a no-op) is what switches the
+        # summary call onto the streamed path — giving every compression
+        # path the same two guarantees: the configured timeout acts on
+        # inactivity (slow models finish), and a byte-trickling provider
+        # that keeps the connection alive forever is cut off at the
         # streamed total ceiling (see _aux_stream_total_ceiling) instead of
         # outliving the SDK's inactivity timeout indefinitely.
         from agent.auxiliary_client import aux_progress_hook

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"
+    # Compression writers (#72424 / activity contract): heartbeat, host timeout, cooldown.
+    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

cli-config.yaml.example

@@ -793,6 +793,12 @@ agent:
   # Set to 0 to disable the warning.
   # gateway_timeout_warning: 900
 
+  # Session stall watchdog (seconds). When a busy session has a pending
+  # inbound follow-up and the agent activity clock is idle this long, the
+  # gateway logs a WARNING and notifies the user to try /new. Does not kill
+  # the turn (see gateway_timeout). 0 = disable. Default 300.
+  # session_stall_timeout: 300
+
   # Graceful drain timeout for gateway stop/restart (seconds).
   # Default 0 = no drain: a restart interrupts in-flight agents immediately,
   # cleans up, and exits. Set a positive value only if you want a grace

gateway/run.py

@@ -1986,6 +1986,10 @@ if _config_path.exists():
                 os.environ["HERMES_AGENT_TIMEOUT_WARNING"] = str(_agent_cfg["gateway_timeout_warning"])
             if "gateway_notify_interval" in _agent_cfg:
                 os.environ["HERMES_AGENT_NOTIFY_INTERVAL"] = str(_agent_cfg["gateway_notify_interval"])
+            if "session_stall_timeout" in _agent_cfg:
+                os.environ["HERMES_SESSION_STALL_TIMEOUT"] = str(
+                    _agent_cfg["session_stall_timeout"]
+                )
             if "restart_drain_timeout" in _agent_cfg:
                 os.environ["HERMES_RESTART_DRAIN_TIMEOUT"] = str(_agent_cfg["restart_drain_timeout"])
             if "gateway_auto_continue_freshness" in _agent_cfg:
@@ -2287,6 +2291,9 @@ _CONVERSATION_SCOPED_STATE: tuple = (
     "_pending_model_notes",
     "_last_resolved_model",
     "_queued_events",
+    # Stall-watchdog "already notified" latch (#72016). Cleared on /new so a
+    # fresh conversation can warn again if it later stalls with pending inbound.
+    "_session_stall_notified",
     # Staged-but-never-consumed sidecar notes (turn aborted between staging
     # and run_sync) must not leak into a future conversation's first user
     # message — session keys are source-derived and REUSED.
@@ -3506,6 +3513,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # /new and /reset.  /model and other mid-session operations
         # preserve the queue.
         self._queued_events: Dict[str, List[MessageEvent]] = {}
+        # Session keys that already received a stall notification for the
+        # current stall episode (cleared when pending clears / activity resumes
+        # / conversation boundary). See gateway.session_stall.
+        self._session_stall_notified: Dict[str, bool] = {}
         self._pending_native_image_paths_by_session: Dict[str, List[str]] = {}
         self._busy_ack_ts: Dict[str, float] = {}  # last busy-ack timestamp per session (debounce)
         self._session_run_generation: Dict[str, int] = {}
@@ -8691,6 +8702,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # Start background session expiry watcher to finalize expired sessions
         self._spawn_supervised(self._session_expiry_watcher, "session_expiry_watcher")
 
+        # Stall watchdog: pending inbound + stale agent activity → warn user
+        # to /new (does not kill the turn; see agent.session_stall_timeout).
+        self._spawn_supervised(self._session_stall_watcher, "session_stall_watcher")
+
         # Start background kanban notifier — delivers `completed`, `blocked`,
         # `spawn_auto_blocked`, and `crashed` events to gateway subscribers
         # so human-in-the-loop workflows hear back without polling.
@@ -9297,6 +9312,204 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     break
                 await asyncio.sleep(1)
 
+    def _session_stall_timeout_seconds(self) -> float:
+        """Return configured stall timeout (seconds); 0 disables the watchdog."""
+        return _float_env("HERMES_SESSION_STALL_TIMEOUT", 300)
+
+    def _iter_gateway_adapters(self):
+        """Yield every live platform adapter (default + multiplex profiles)."""
+        seen: set[int] = set()
+        for adapter in list(getattr(self, "adapters", {}).values()):
+            if adapter is None:
+                continue
+            aid = id(adapter)
+            if aid in seen:
+                continue
+            seen.add(aid)
+            yield adapter
+        for amap in list(getattr(self, "_profile_adapters", {}).values()):
+            for adapter in list(amap.values()):
+                if adapter is None:
+                    continue
+                aid = id(adapter)
+                if aid in seen:
+                    continue
+                seen.add(aid)
+                yield adapter
+
+    def _session_activity_for_stall(self, session_key: str) -> Optional[dict]:
+        """Return the shared activity snapshot for stall progress (#72039).
+
+        Single progress source: ``AIAgent.get_activity_summary()`` /
+        ``agent.session_activity``. No turn-start or pending-inbound clocks.
+        """
+        agent = (getattr(self, "_running_agents", None) or {}).get(session_key)
+        if agent is None or agent is _AGENT_PENDING_SENTINEL:
+            return None
+        if not hasattr(agent, "get_activity_summary"):
+            return None
+        try:
+            summary = agent.get_activity_summary()
+        except Exception:
+            return None
+        return summary if isinstance(summary, dict) else None
+
+    async def _check_session_stalls(self, timeout_seconds: float) -> int:
+        """Scan pending inbound sessions and notify once per stall episode.
+
+        Returns the number of notifications sent this pass (for tests).
+        """
+        from gateway.session_stall import (
+            format_session_stall_notification,
+            resolve_session_idle_seconds_from_activity,
+            should_clear_session_stall_notification,
+            should_emit_session_stall_notification,
+        )
+
+        notified_map = getattr(self, "_session_stall_notified", None)
+        if notified_map is None:
+            notified_map = {}
+            self._session_stall_notified = notified_map
+
+        sent = 0
+        now = time.time()
+        candidates: Dict[str, tuple[Any, Any]] = {}
+
+        for adapter in self._iter_gateway_adapters():
+            pending_slot = getattr(adapter, "_pending_messages", None) or {}
+            for session_key, event in list(pending_slot.items()):
+                if session_key and session_key not in candidates and event is not None:
+                    candidates[session_key] = (adapter, event)
+
+        for session_key, overflow in list(
+            (getattr(self, "_queued_events", None) or {}).items()
+        ):
+            if not session_key or session_key in candidates or not overflow:
+                continue
+            event = overflow[0]
+            source = getattr(event, "source", None)
+            adapter = (
+                self._adapter_for_source(source) if source is not None else None
+            )
+            if adapter is None:
+                continue
+            candidates[session_key] = (adapter, event)
+
+        for session_key, (adapter, pending_event) in list(candidates.items()):
+            has_pending = pending_event is not None
+            activity = (
+                self._session_activity_for_stall(session_key) if has_pending else None
+            )
+            idle_seconds = (
+                resolve_session_idle_seconds_from_activity(activity, now=now)
+                if has_pending
+                else None
+            )
+            already = bool(notified_map.get(session_key))
+            if should_clear_session_stall_notification(
+                timeout_seconds=timeout_seconds,
+                idle_seconds=idle_seconds,
+                has_pending_inbound=has_pending,
+            ):
+                notified_map.pop(session_key, None)
+                already = False
+            if not should_emit_session_stall_notification(
+                timeout_seconds=timeout_seconds,
+                idle_seconds=idle_seconds,
+                has_pending_inbound=has_pending,
+                already_notified=already,
+            ):
+                continue
+
+            if idle_seconds is None:
+                continue
+            mins = max(1, int(idle_seconds // 60))
+            activity = activity or {}
+            logger.warning(
+                "Session stall detected: session=%s idle=%.0fs "
+                "(timeout=%.0fs, ~%d min); pending inbound present "
+                "| last_activity=%s | provenance=%s",
+                session_key,
+                idle_seconds,
+                timeout_seconds,
+                mins,
+                activity.get("last_activity_desc")
+                or activity.get("last_activity_description")
+                or "unknown",
+                activity.get("provenance")
+                or activity.get("last_activity_provenance")
+                or "unknown",
+            )
+            source = getattr(pending_event, "source", None)
+            chat_id = getattr(source, "chat_id", None) if source is not None else None
+            if not chat_id:
+                logger.warning(
+                    "Session stall notify skipped (no chat_id): session=%s",
+                    session_key,
+                )
+                # Cannot deliver; latch to avoid log spam every tick.
+                notified_map[session_key] = True
+                continue
+            try:
+                metadata = (
+                    self._thread_metadata_for_source(source)
+                    if source is not None and hasattr(self, "_thread_metadata_for_source")
+                    else None
+                )
+                result = await adapter.send(
+                    str(chat_id),
+                    format_session_stall_notification(idle_seconds),
+                    metadata=metadata,
+                )
+                # Adapters often return SendResult(success=False) instead of raising.
+                if result is not None and getattr(result, "success", True) is False:
+                    logger.warning(
+                        "Session stall notify failed for %s: %s",
+                        session_key,
+                        getattr(result, "error", "send returned success=False"),
+                    )
+                    continue  # do not latch; retry next tick
+                sent += 1
+                notified_map[session_key] = True
+            except Exception as exc:
+                logger.warning(
+                    "Session stall notify failed for %s: %s",
+                    session_key,
+                    exc,
+                )
+                # Do not latch — retry next watcher tick until delivery or episode clear.
+
+        # Drop latches for sessions that no longer appear in any pending map.
+        for key in list(notified_map.keys()):
+            if key not in candidates:
+                notified_map.pop(key, None)
+
+        return sent
+
+    async def _session_stall_watcher(self, interval: float = 30.0):
+        """Periodic pending-inbound + stale-activity stall watchdog (#72016).
+
+        Progress comes only from ``get_activity_summary()`` (#72039).
+        Pending inbound is a notify policy gate, not a progress clock.
+        Notify-only: does not kill the turn (contrast ``gateway_timeout`` /
+        ``shutdown_watchdog``).
+        """
+        # Short initial delay so startup reconnect noise does not false-fire.
+        await asyncio.sleep(min(30.0, max(1.0, float(interval))))
+        while self._running:
+            try:
+                timeout = self._session_stall_timeout_seconds()
+                if timeout > 0:
+                    await self._check_session_stalls(timeout)
+            except Exception as exc:
+                logger.debug("Session stall watcher error: %s", exc)
+            # Interruptible sleep
+            steps = max(1, int(float(interval)))
+            for _ in range(steps):
+                if not self._running:
+                    break
+                await asyncio.sleep(1)
+
     def _active_profile_name(self) -> str:
         """Return the profile name this gateway represents."""
         try:
@@ -13745,6 +13958,23 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                                                 self._hygiene_compression_failure_cooldowns[
                                                     session_entry.session_id
                                                 ] = time.time() + _hyg_failure_cooldown_seconds
+                                            try:
+                                                from agent.session_activity import (
+                                                    ActivityProvenance,
+                                                )
+
+                                                _hyg_agent._touch_activity(
+                                                    "session hygiene compression timed out",
+                                                    provenance=(
+                                                        ActivityProvenance.AGENT_COMPRESSION_TIMEOUT
+                                                    ),
+                                                )
+                                            except Exception:
+                                                logger.debug(
+                                                    "hygiene compression timeout "
+                                                    "activity stamp failed",
+                                                    exc_info=True,
+                                                )
                                             logger.warning(
                                                 "Session hygiene compression for session %s "
                                                 "made no progress for %.1fs "
@@ -13890,6 +14120,23 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                                             self._hygiene_compression_failure_cooldowns[
                                                 session_entry.session_id
                                             ] = time.time() + _hyg_failure_cooldown_seconds
+                                        try:
+                                            from agent.session_activity import (
+                                                ActivityProvenance,
+                                            )
+
+                                            _hyg_agent._touch_activity(
+                                                "session hygiene compression aborted",
+                                                provenance=(
+                                                    ActivityProvenance.AGENT_COMPRESSION_COOLDOWN
+                                                ),
+                                            )
+                                        except Exception:
+                                            logger.debug(
+                                                "hygiene compression abort "
+                                                "activity stamp failed",
+                                                exc_info=True,
+                                            )
                                         _err = getattr(_comp, "_last_summary_error", None) or "unknown error"
                                         # Force-redact: provider exception text
                                         # may contain credentials; this message
@@ -19610,21 +19857,25 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     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

gateway/session_stall.py

@@ -0,0 +1,121 @@
+"""Gateway session stall notification policy (#72016 item 2).
+
+Consumes the shared activity observation contract from
+``agent.session_activity`` / ``AIAgent.get_activity_summary()``
+(#72039) as the **single progress source**. This module owns only the
+notify-once policy for "pending inbound + stale progress"; it does not
+invent a parallel progress clock from turn-start or inbound event
+timestamps.
+
+Boundaries (keep separate):
+- ``gateway/shutdown_watchdog.py`` — process / event-loop liveness
+- ``gateway/delivery_ledger.py`` — outbound delivery obligations
+- Pending inbound here is a stall *policy gate* (queued follow-up exists),
+  not an outbound obligation and not a progress timestamp.
+
+Notification / timeout / kill / retry policy stay in their own components;
+the shared contract remains observation-only (timestamp + bounded
+description + provenance).
+"""
+
+from __future__ import annotations
+
+import math
+from typing import Any, Mapping, Optional
+
+
+def should_emit_session_stall_notification(
+    *,
+    timeout_seconds: float,
+    idle_seconds: Optional[float],
+    has_pending_inbound: bool,
+    already_notified: bool,
+) -> bool:
+    """Return True when a stall warning should be sent for this session."""
+    if timeout_seconds <= 0:
+        return False
+    if not has_pending_inbound:
+        return False
+    if already_notified:
+        return False
+    if idle_seconds is None:
+        return False
+    return idle_seconds >= timeout_seconds
+
+
+def should_clear_session_stall_notification(
+    *,
+    timeout_seconds: float,
+    idle_seconds: Optional[float],
+    has_pending_inbound: bool,
+) -> bool:
+    """Return True when a prior stall notice may be cleared (episode ended)."""
+    if not has_pending_inbound:
+        return True
+    if timeout_seconds <= 0:
+        return True
+    # Unknown progress: hold the latch. Do not treat observation gaps as recovery.
+    if idle_seconds is None:
+        return False
+    return idle_seconds < timeout_seconds
+
+
+def format_session_stall_notification(idle_seconds: float) -> str:
+    """User-facing stall warning (ASCII minutes; matches issue #72016 copy)."""
+    mins = max(1, int(idle_seconds // 60))
+    return (
+        f"⚠️ Agent session appears stalled (last activity {mins} min ago). "
+        f"Try /new to reset."
+    )
+
+
+def resolve_session_idle_seconds_from_activity(
+    activity: Optional[Mapping[str, Any]],
+    *,
+    now: Optional[float] = None,
+) -> Optional[float]:
+    """Idle seconds from a shared activity snapshot only (#72039 contract).
+
+    Prefers ``seconds_since_activity`` when present and finite; otherwise
+    derives from ``last_activity_at`` / ``last_activity_ts``. Returns
+    ``None`` when there is no usable progress timestamp — callers must
+    not fall back to turn-start or pending-inbound clocks.
+    """
+    if not activity:
+        return None
+
+    elapsed = activity.get("seconds_since_activity")
+    if elapsed is not None:
+        try:
+            idle = float(elapsed)
+        except (TypeError, ValueError):
+            idle = None
+        else:
+            if math.isfinite(idle):
+                if idle < 0:
+                    return 0.0
+                return idle
+            # Non-finite: fall through to last_activity_at / last_activity_ts
+
+    ts = activity.get("last_activity_at")
+    if ts is None:
+        ts = activity.get("last_activity_ts")
+    if ts is None:
+        return None
+    try:
+        when = float(ts)
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(when):
+        return None
+
+    if now is None:
+        import time as _time
+
+        clock = float(_time.time())
+    else:
+        clock = float(now)
+    idle = clock - when
+    if idle < 0:
+        return 0.0
+    return idle

hermes_cli/config.py

@@ -951,6 +951,12 @@ DEFAULT_CONFIG = {
         # tools or receiving API responses.  Only fires when the agent has
         # been completely idle for this duration.  0 = unlimited.
         "gateway_timeout": 1800,
+        # Session stall watchdog (seconds): when a gateway session has a
+        # pending inbound message AND the running agent has not updated its
+        # activity clock for this long, log a WARNING and notify the user to
+        # try /new. Distinct from gateway_timeout (which kills the turn) and
+        # gateway_notify_interval ("still working" heartbeats). 0 = disable.
+        "session_stall_timeout": 300,
         # Graceful drain timeout for gateway stop/restart (seconds).
         # The gateway stops accepting new work, waits for running agents
         # to finish, then interrupts any remaining runs after the timeout.
@@ -1501,6 +1507,16 @@ DEFAULT_CONFIG = {
                                       # while tokens are still moving — bounds a degenerate
                                       # trickle stream. Clamped to >= hygiene_timeout_seconds.
         "hygiene_failure_cooldown_seconds": 300,  # skip repeated failed hygiene attempts for this session
+        "context_timeout_seconds": 120,  # inactivity budget for in-agent compress_context
+                                      # (conversation loop, /compress, preflight, etc.).
+                                      # Same progress-aware semantics as hygiene_timeout_seconds:
+                                      # streamed summary tokens extend the wait; only a silent
+                                      # worker is cut off. 0 = disable the owned wrapper
+                                      # (callers that already pass commit_fence, e.g. gateway
+                                      # hygiene, never use this path).
+        "context_total_ceiling_seconds": 600,  # absolute cap on in-agent compress_context wait
+                                      # even while tokens are still moving. Clamped to
+                                      # >= context_timeout_seconds when the idle budget is > 0.
         "protect_first_n": 3,         # non-system head messages always preserved
                                       # verbatim, in ADDITION to the system prompt
                                       # (which is always implicitly protected). Set to

hermes_cli/dump.py

@@ -236,6 +236,7 @@ def _config_overrides(config: dict) -> dict[str, str]:
     interesting_paths = [
         ("agent", "max_turns"),
         ("agent", "gateway_timeout"),
+        ("agent", "session_stall_timeout"),
         ("agent", "tool_use_enforcement"),
         ("terminal", "backend"),
         ("terminal", "docker_image"),

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 show_status(args):
     # 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():

hermes_state.py

@@ -28,6 +28,7 @@ import time
 from pathlib import Path
 
 from agent.memory_manager import sanitize_context
+from agent.session_activity import ActivityProvenance
 from agent.message_sanitization import _sanitize_surrogates
 from agent.skill_commands import (
     SKILL_EXCERPT_JOINT,
@@ -230,6 +231,56 @@ def _ephemeral_child_sql(alias: str = "s") -> str:
     )
 
 
+def _sql_session_last_active(alias: str = "s") -> str:
+    """SQL expression for session recency used by list/status surfaces.
+
+    Freshest of ``last_activity_at`` (mid-turn agent activity heartbeat) and
+    the latest message timestamp, then fall back to ``started_at``.
+
+    Must not prefer a stale heartbeat over a newer message: durable
+    heartbeats are rate-limited (~60s), so after a turn writes messages
+    ``last_activity_at`` can lag ``MAX(messages.timestamp)``.
+    """
+    msg_max = (
+        f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m "
+        f"WHERE _act_m.session_id = {alias}.id)"
+    )
+    return (
+        f"COALESCE("
+        f"(SELECT MAX(_act_v.v) FROM ("
+        f"SELECT {alias}.last_activity_at AS v "
+        f"UNION ALL "
+        f"SELECT {msg_max}"
+        f") _act_v), "
+        f"{alias}.started_at)"
+    )
+
+
+def _sql_session_last_active_by_id(session_id_expr: str) -> str:
+    """Same freshest-of expression keyed by a session-id SQL expression."""
+    msg_max = (
+        f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m "
+        f"WHERE _act_m.session_id = {session_id_expr})"
+    )
+    activity = (
+        f"(SELECT last_activity_at FROM sessions _act_s "
+        f"WHERE _act_s.id = {session_id_expr})"
+    )
+    started = (
+        f"(SELECT started_at FROM sessions _act_s "
+        f"WHERE _act_s.id = {session_id_expr})"
+    )
+    return (
+        f"COALESCE("
+        f"(SELECT MAX(_act_v.v) FROM ("
+        f"SELECT {activity} AS v "
+        f"UNION ALL "
+        f"SELECT {msg_max}"
+        f") _act_v), "
+        f"{started})"
+    )
+
+
 def _collect_delegate_child_ids(conn, parent_ids: List[str]) -> List[str]:
     """Delegate-subagent ids to cascade-delete with *parent_ids*.
 
@@ -1182,6 +1233,9 @@ CREATE TABLE IF NOT EXISTS sessions (
     cost_source TEXT,
     pricing_version TEXT,
     title TEXT,
+    last_activity_at REAL,
+    last_activity_description TEXT,
+    last_activity_provenance TEXT,
     api_call_count INTEGER DEFAULT 0,
     handoff_state TEXT,
     handoff_platform TEXT,
@@ -4002,13 +4056,9 @@ class SessionDB:
         filters on ``source``; ``active_only`` restricts to sessions that
         have not ended.
         """
-        query = """
+        query = f"""
             SELECT sessions.*,
-                   COALESCE(
-                       (SELECT MAX(m.timestamp) FROM messages m
-                        WHERE m.session_id = sessions.id),
-                       sessions.started_at
-                   ) AS last_active
+                   {_sql_session_last_active("sessions")} AS last_active
             FROM sessions
             WHERE session_key IS NOT NULL
               AND started_at = (
@@ -4829,6 +4879,87 @@ class SessionDB:
             return None
         return row["holder"] if isinstance(row, sqlite3.Row) else row[0]
 
+    def touch_session_activity(
+        self,
+        session_id: str,
+        ts: Optional[float] = None,
+        *,
+        description: Optional[str] = None,
+        provenance: Optional[ActivityProvenance] = None,
+    ) -> None:
+        """Stamp durable mid-turn session activity (observation-only).
+
+        Called (rate-limited) from ``AIAgent._touch_activity`` so gateway/CLI
+        surfaces and stall consumers observe API/tool/compaction activity
+        even when no new message row has been written yet (#72016 / #72039).
+
+        Never moves ``last_activity_at`` backwards. When the timestamp
+        advances, bounded ``last_activity_description`` /
+        ``last_activity_provenance`` are written with it. No-ops when
+        ``session_id`` is empty or the row does not exist.
+        """
+        if not session_id:
+            return
+        from agent.session_activity import (
+            bound_activity_description,
+            normalize_activity_provenance,
+        )
+
+        when = float(ts if ts is not None else time.time())
+        desc = bound_activity_description(description)
+        prov = normalize_activity_provenance(provenance).value
+
+        def _do(conn):
+            conn.execute(
+                "UPDATE sessions SET "
+                "last_activity_at = ?, "
+                "last_activity_description = ?, "
+                "last_activity_provenance = ? "
+                "WHERE id = ? AND (last_activity_at IS NULL OR last_activity_at < ?)",
+                (when, desc, prov, session_id, when),
+            )
+
+        self._execute_write(_do)
+
+    def clear_session_activity_labels(self, session_id: str) -> None:
+        """Clear mid-turn activity labels after a turn ends.
+
+        Keeps ``last_activity_at`` intact so idle / watchdog clocks stay
+        continuous. Description and provenance are observation labels for
+        *what was happening at* that timestamp during an active turn; once
+        the turn is idle they must not keep advertising "compressing" /
+        "executing tool" (#72039).
+        """
+        if not session_id:
+            return
+        from agent.session_activity import ActivityProvenance
+
+        def _do(conn):
+            conn.execute(
+                "UPDATE sessions SET "
+                "last_activity_description = ?, "
+                "last_activity_provenance = ? "
+                "WHERE id = ?",
+                ("", ActivityProvenance.UNKNOWN.value, session_id),
+            )
+
+        self._execute_write(_do)
+
+    def get_session_activity(self, session_id: str) -> Optional[Dict[str, Any]]:
+        """Return the durable activity snapshot for *session_id*, or None."""
+        if not session_id:
+            return None
+        row = self.get_session(session_id)
+        if not row:
+            return None
+        from agent.session_activity import build_activity_snapshot
+
+        return build_activity_snapshot(
+            last_activity_at=row.get("last_activity_at"),
+            last_activity_description=row.get("last_activity_description"),
+            last_activity_provenance=row.get("last_activity_provenance"),
+        )
+
     def update_session_meta(
         self,
         session_id: str,
@@ -5792,14 +5923,14 @@ class SessionDB:
         for _ in range(100):
             with self._lock:
                 cursor = self._conn.execute(
-                    """
+                    f"""
                     SELECT child.id
                     FROM sessions parent
                     JOIN sessions child ON child.parent_session_id = parent.id
                     WHERE parent.id = ?
                       AND parent.end_reason = 'compression'
-                      AND json_extract(COALESCE(child.model_config, '{}'), '$._branched_from') IS NULL
-                      AND json_extract(COALESCE(child.model_config, '{}'), '$._delegate_from') IS NULL
+                      AND json_extract(COALESCE(child.model_config, '{{}}'), '$._branched_from') IS NULL
+                      AND json_extract(COALESCE(child.model_config, '{{}}'), '$._delegate_from') IS NULL
                       AND COALESCE(child.source, '') != 'tool'
                     ORDER BY
                       CASE
@@ -5807,10 +5938,7 @@ class SessionDB:
                         WHEN child.ended_at IS NULL THEN 1
                         ELSE 2
                       END,
-                      COALESCE(
-                        (SELECT MAX(m.timestamp) FROM messages m WHERE m.session_id = child.id),
-                        child.started_at
-                      ) DESC,
+                      {_sql_session_last_active("child")} DESC,
                       child.started_at DESC,
                       child.id DESC
                     LIMIT 1
@@ -5896,7 +6024,8 @@ class SessionDB:
 
         Returns dicts with keys: id, source, model, title, started_at, ended_at,
         message_count, preview (first 60 chars of first user message),
-        last_active (timestamp of last message).
+        last_active (freshest of last_activity_at heartbeat and latest
+        message timestamp, else started_at).
 
         Uses a single query with correlated subqueries instead of N+2 queries.
 
@@ -6064,10 +6193,7 @@ class SessionDB:
                 chain_max AS (
                     SELECT
                         root_id,
-                        MAX(COALESCE(
-                            (SELECT MAX(m.timestamp) FROM messages m WHERE m.session_id = cur_id),
-                            (SELECT started_at FROM sessions ss WHERE ss.id = cur_id)
-                        )) AS effective_last_active
+                        MAX({_sql_session_last_active_by_id("cur_id")}) AS effective_last_active
                     FROM chain
                     GROUP BY root_id
                 )
@@ -6079,10 +6205,7 @@ class SessionDB:
                          ORDER BY m.timestamp, m.id LIMIT 1),
                         ''
                     ) AS _preview_raw,
-                    COALESCE(
-                        (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
-                        s.started_at
-                    ) AS last_active,
+                    {_sql_session_last_active("s")} AS last_active,
                     COALESCE(cm.effective_last_active, s.started_at) AS _effective_last_active
                 FROM sessions s
                 LEFT JOIN chain_max cm ON cm.root_id = s.id
@@ -6104,10 +6227,7 @@ class SessionDB:
                          ORDER BY m.timestamp, m.id LIMIT 1),
                         ''
                     ) AS _preview_raw,
-                    COALESCE(
-                        (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
-                        s.started_at
-                    ) AS last_active
+                    {_sql_session_last_active("s")} AS last_active
                 FROM sessions s
                 {where_sql}
                 ORDER BY s.started_at DESC
@@ -6202,10 +6322,7 @@ class SessionDB:
                      ORDER BY m.timestamp, m.id LIMIT 1),
                     ''
                 ) AS _preview_raw,
-                COALESCE(
-                    (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
-                    s.started_at
-                ) AS last_active
+                {_sql_session_last_active("s")} AS last_active
             FROM sessions s
             WHERE s.source = 'cron' AND s.id >= ? AND s.id < ?
             ORDER BY s.started_at DESC, s.id DESC
@@ -6240,10 +6357,7 @@ class SessionDB:
                      ORDER BY m.timestamp, m.id LIMIT 1),
                     ''
                 ) AS _preview_raw,
-                COALESCE(
-                    (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
-                    s.started_at
-                ) AS last_active
+                {_sql_session_last_active("s")} AS last_active
             FROM sessions s
             WHERE s.id = ?
         """
@@ -8678,22 +8792,18 @@ class SessionDB:
     ) -> List[Dict[str, Any]]:
         """List sessions, optionally filtered by source.
 
-        Returns rows enriched with a computed ``last_active`` column (latest
-        message timestamp for the session, falling back to ``started_at``),
-        ordered by most-recently-used first.
+        Returns rows enriched with a computed ``last_active`` column
+        (freshest of ``last_activity_at`` and latest message timestamp,
+        else ``started_at``), ordered by most-recently-used first.
 
-        Pass ``workspace_key`` to scope rows to one workspace — matching
+        Pass ``workspace_key`` to scope rows to one workspace - matching
         :func:`workspace_key` semantics (git repo root, else cwd). Used by
         ``hermes -c``/``--resume`` so the "last" session is the last one in
         the *current* workspace, not the global MRU.
         """
         select_with_last_active = (
-            "SELECT s.*, COALESCE(m.last_active, s.started_at) AS last_active "
+            f"SELECT s.*, {_sql_session_last_active('s')} AS last_active "
             "FROM sessions s "
-            "LEFT JOIN ("
-            "SELECT session_id, MAX(timestamp) AS last_active "
-            "FROM messages GROUP BY session_id"
-            ") m ON m.session_id = s.id "
         )
         where_clauses = []
         params: list = []
@@ -9852,8 +9962,9 @@ class SessionDB:
     ) -> int:
         """Archive every session untouched for at least ``idle_days`` days.
 
-        "Touched" is the latest message timestamp (falling back to
-        ``started_at``) — i.e. real recency, not creation time — so a session
+        "Touched" is the freshest of ``last_activity_at`` and the latest
+        message timestamp (else ``started_at``) — i.e. real recency, not
+        creation time — so a session
         created long ago but active yesterday is spared, while an old
         abandoned one (even a still-open one) is swept. Unlike
         :meth:`archive_sessions`, this method can also archive unended
@@ -9882,11 +9993,7 @@ class SessionDB:
                 WHERE s.archived = 0
                   AND COALESCE(s.end_reason, '') <> 'compression'
                   {pin_clause}
-                  AND COALESCE(
-                        (SELECT MAX(m.timestamp) FROM messages m
-                         WHERE m.session_id = s.id),
-                        s.started_at
-                      ) < ?
+                  AND {_sql_session_last_active("s")} < ?
                 ORDER BY s.started_at ASC
                 """,
                 (cutoff,),
@@ -10472,10 +10579,7 @@ class SessionDB:
                              ORDER BY m.timestamp, m.id LIMIT 1),
                             ''
                         ) AS _preview_raw,
-                        COALESCE(
-                            (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
-                            s.started_at
-                        ) AS last_active
+                        {_sql_session_last_active("s")} AS last_active
                     FROM sessions s
                     WHERE s.source = 'telegram'
                       AND s.user_id = ?
@@ -10501,10 +10605,7 @@ class SessionDB:
                              ORDER BY m.timestamp, m.id LIMIT 1),
                             ''
                         ) AS _preview_raw,
-                        COALESCE(
-                            (SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
-                            s.started_at
-                        ) AS last_active
+                        {_sql_session_last_active("s")} AS last_active
                     FROM sessions s
                     WHERE s.source = 'telegram'
                       AND s.user_id = ?

run_agent.py

@@ -149,6 +149,7 @@ from agent.memory_manager import sanitize_context
 from agent.error_classifier import FailoverReason
 from agent.redact import redact_sensitive_text
 from agent.message_content import flatten_message_text
+from agent.session_activity import ActivityProvenance
 from agent.model_metadata import (
     estimate_request_tokens_rough,  # noqa: F401  # re-exported for tests that mock.patch("run_agent.estimate_request_tokens_rough")
     is_local_endpoint,
@@ -963,6 +964,12 @@ class AIAgent:
             from agent.conversation_compression import (
                 CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE,
             )
+            # cooldown + anti-thrash (ineffective) are both "compression blocked".
+            if _warn_kind in ("cooldown", "ineffective"):
+                self._touch_activity(
+                    f"compression blocked ({reason})",
+                    provenance=ActivityProvenance.AGENT_COMPRESSION_COOLDOWN,
+                )
             self._emit_warning(
                 CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE.format(
                     tokens=preflight_tokens,
@@ -3462,7 +3469,12 @@ class AIAgent:
         from agent.agent_runtime_helpers import apply_pending_steer_to_tool_results
         return apply_pending_steer_to_tool_results(self, messages, num_tool_msgs)
 
-    def _touch_activity(self, desc: str) -> None:
+    def _touch_activity(
+        self,
+        desc: str,
+        *,
+        provenance: Optional[ActivityProvenance] = None,
+    ) -> None:
         """Update the last-activity timestamp and description (thread-safe).
 
         Also bridges to the kanban board's heartbeat fields when this
@@ -3470,9 +3482,23 @@ class AIAgent:
         so the dispatcher watchdog doesn't reclaim an actively-running
         worker as stale (#31752). Bridge is rate-limited (60s) and
         best-effort — it never raises into the agent loop.
+
+        Separately, rate-limits a durable SessionDB activity projection
+        (``last_activity_at`` + bounded description/provenance) so
+        CLI/Gateway consumers share one observation source (#72016 / #72039).
+
+        ``provenance`` defaults to ``unknown`` (the ordinary agent activity
+        clock). Named values are for special writers (e.g. compression);
+        ordinary call sites should leave the default.
         """
+        from agent.session_activity import (
+            bound_activity_description,
+            normalize_activity_provenance,
+        )
+
         self._last_activity_ts = time.time()
-        self._last_activity_desc = desc
+        self._last_activity_desc = bound_activity_description(desc)
+        self._last_activity_provenance = normalize_activity_provenance(provenance)
         if os.environ.get("HERMES_KANBAN_TASK"):
             try:
                 from tools.kanban_tools import heartbeat_current_worker_from_env
@@ -3483,6 +3509,66 @@ class AIAgent:
                 # covers import-time failures (kanban_tools unavailable,
                 # etc.) on niche deployment surfaces.
                 pass
+        self._persist_session_activity_if_due()
+
+    def _persist_session_activity_if_due(self) -> None:
+        """Best-effort durable activity heartbeat for SessionDB consumers.
+
+        Rate-limited to one write per 60s per agent (same cadence as the
+        kanban auto-heartbeat). Fail-open: never raises into the agent loop.
+        """
+        session_id = getattr(self, "session_id", None)
+        session_db = getattr(self, "_session_db", None)
+        if not session_id or session_db is None:
+            return
+        touch = getattr(session_db, "touch_session_activity", None)
+        if not callable(touch):
+            return
+        now_mono = time.monotonic()
+        last_mono = getattr(self, "_session_activity_last_persist_mono", 0.0)
+        if (now_mono - last_mono) < 60.0:
+            return
+        self._session_activity_last_persist_mono = now_mono
+        try:
+            from agent.session_activity import normalize_activity_provenance
+
+            touch(
+                session_id,
+                getattr(self, "_last_activity_ts", None),
+                description=getattr(self, "_last_activity_desc", None),
+                provenance=normalize_activity_provenance(
+                    getattr(self, "_last_activity_provenance", None)
+                ),
+            )
+        except Exception:
+            # Never let durable heartbeat I/O break the agent loop.
+            pass
+
+    def _reset_activity_labels_after_turn(self) -> None:
+        """Drop mid-turn activity labels once the turn is no longer running.
+
+        Keeps ``_last_activity_ts`` so idle/watchdog clocks stay continuous
+        across interrupt-recursive turns (#15654) and between turns. Clears
+        description + provenance so idle cached agents / SessionDB listings
+        do not keep advertising the last mid-turn stamp (e.g. compression
+        or tool execution) after the turn ended (#72039).
+        """
+        from agent.session_activity import ActivityProvenance
+
+        self._last_activity_desc = ""
+        self._last_activity_provenance = ActivityProvenance.UNKNOWN
+        session_id = getattr(self, "session_id", None)
+        session_db = getattr(self, "_session_db", None)
+        if not session_id or session_db is None:
+            return
+        clear = getattr(session_db, "clear_session_activity_labels", None)
+        if not callable(clear):
+            return
+        try:
+            clear(session_id)
+        except Exception:
+            # Never let durable cleanup I/O break turn teardown.
+            pass
 
     def _capture_rate_limits(self, http_response: Any) -> None:
         """Parse x-ratelimit-* headers from an HTTP response and cache the state.
@@ -3694,20 +3780,32 @@ class AIAgent:
     def get_activity_summary(self) -> dict:
         """Return a snapshot of the agent's current activity for diagnostics.
 
-        Called by the gateway timeout handler to report what the agent was doing
-        when it was killed, and by the periodic "still working" notifications.
+        Exposes the shared activity observation contract
+        (``last_activity_at`` / ``last_activity_description`` /
+        ``last_activity_provenance``) plus short aliases
+        (``last_activity_ts`` / ``last_activity_desc`` / …) for existing
+        gateway and delegate readers.
         """
-        elapsed = time.time() - self._last_activity_ts
-        return {
-            "last_activity_ts": self._last_activity_ts,
-            "last_activity_desc": self._last_activity_desc,
-            "seconds_since_activity": round(elapsed, 1),
-            "current_tool": self._current_tool,
-            "api_call_count": self._api_call_count,
-            "max_iterations": self.max_iterations,
-            "budget_used": self.iteration_budget.used,
-            "budget_max": self.iteration_budget.max_total,
-        }
+        from agent.session_activity import (
+            ActivityProvenance,
+            build_activity_snapshot,
+        )
+
+        provenance = getattr(self, "_last_activity_provenance", None)
+        if provenance is None:
+            provenance = ActivityProvenance.UNKNOWN
+        return build_activity_snapshot(
+            last_activity_at=getattr(self, "_last_activity_ts", None),
+            last_activity_description=getattr(self, "_last_activity_desc", None) or "",
+            last_activity_provenance=provenance,
+            extra={
+                "current_tool": self._current_tool,
+                "api_call_count": self._api_call_count,
+                "max_iterations": self.max_iterations,
+                "budget_used": self.iteration_budget.used,
+                "budget_max": self.iteration_budget.max_total,
+            },
+        )
 
     def shutdown_memory_provider(self, messages: list = None) -> None:
         """Shut down the memory provider and context engine — call at actual session boundaries.
@@ -6597,7 +6695,11 @@ class AIAgent:
         auto-compress abort.  Auto-compress callers use the default
         ``force=False``.
         """
-        from agent.conversation_compression import compress_context
+        from agent.conversation_compression import (
+            compress_context,
+            resolve_context_compression_timeouts,
+            run_compress_context_with_progress_timeout,
+        )
         from agent.portal_tags import (
             get_conversation_context,
             reset_conversation_context,
@@ -6623,12 +6725,109 @@ class AIAgent:
             if root:
                 token = set_conversation_context(root)
         try:
-            return compress_context(
-                self, messages, system_message,
-                approx_tokens=approx_tokens, task_id=task_id, focus_topic=focus_topic,
-                force=force,
-                defer_context_engine_notification=defer_context_engine_notification,
-                commit_fence=commit_fence,
+            def _run(fence=None):
+                return compress_context(
+                    self, messages, system_message,
+                    approx_tokens=approx_tokens, task_id=task_id,
+                    focus_topic=focus_topic,
+                    force=force,
+                    defer_context_engine_notification=(
+                        defer_context_engine_notification
+                    ),
+                    commit_fence=fence,
+                )
+
+            # Callers that already own a progress-aware wait (gateway session
+            # hygiene) pass commit_fence and must not be double-wrapped.
+            if commit_fence is not None:
+                return _run(commit_fence)
+
+            idle_timeout, total_ceiling = resolve_context_compression_timeouts()
+            if idle_timeout <= 0:
+                return _run(None)
+
+            # Resolve the fallback prompt lazily on timeout only. Eager
+            # rebuild here would raise before compress_context runs whenever
+            # _cached_system_prompt is unset and _build_system_prompt fails
+            # (lock-refresher / noop-exception tests rely on that path).
+            def _fallback_prompt():
+                cached = getattr(self, "_cached_system_prompt", None)
+                if cached:
+                    return cached
+                try:
+                    return self._build_system_prompt(system_message)
+                except Exception:
+                    logger.debug(
+                        "compress_context timeout fallback prompt rebuild "
+                        "failed; using raw system_message",
+                        exc_info=True,
+                    )
+                    return system_message or ""
+
+            def _on_timeout(idle, waited, since_progress):
+                logger.warning(
+                    "Context compression made no progress for %.1fs "
+                    "(total wait %.1fs, ceiling %.1fs); continuing without "
+                    "compression",
+                    since_progress,
+                    waited,
+                    total_ceiling,
+                )
+                touch = getattr(self, "_touch_activity", None)
+                if callable(touch):
+                    try:
+                        touch(
+                            "context compression timed out",
+                            provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
+                        )
+                    except Exception:
+                        logger.debug(
+                            "compress_context timeout activity touch failed",
+                            exc_info=True,
+                        )
+                # Same timeout cooldown ladder as summary-LLM timeouts
+                # (#62452): avoid re-burning the full idle budget every turn.
+                compressor = getattr(self, "context_compressor", None)
+                if compressor is not None:
+                    streak = (
+                        getattr(compressor, "_consecutive_timeout_failures", 0) + 1
+                    )
+                    compressor._consecutive_timeout_failures = streak
+                    ladder = (60, 300, 900)
+                    cooldown = ladder[min(streak, len(ladder)) - 1]
+                    record = getattr(
+                        compressor, "_record_compression_failure_cooldown", None
+                    )
+                    if callable(record):
+                        try:
+                            record(
+                                float(cooldown),
+                                "host compress_context timeout "
+                                "(no summary progress)",
+                            )
+                        except Exception:
+                            logger.debug(
+                                "failed to record compress_context timeout "
+                                "cooldown",
+                                exc_info=True,
+                            )
+                emit = getattr(self, "_emit_warning", None)
+                if callable(emit):
+                    emit(
+                        "⚠ Context compression timed out "
+                        f"after {idle:.1f}s with no output from the summary "
+                        "model. No messages were dropped — continuing without "
+                        "compression. Run /compress to retry, /new for a clean "
+                        "session, or check auxiliary.compression."
+                    )
+
+            return run_compress_context_with_progress_timeout(
+                worker=_run,
+                messages=messages,
+                system_prompt_fallback=_fallback_prompt,
+                idle_timeout_seconds=idle_timeout,
+                total_ceiling_seconds=total_ceiling,
+                on_timeout=_on_timeout,
             )
         finally:
             # Restore whatever the caller had, so a compaction never leaks its
@@ -6894,6 +7093,12 @@ class AIAgent:
                     moa_config=moa_config,
                 )
             finally:
+                # Always clear mid-turn labels when the turn exits — including
+                # interrupted early returns that skip finalize_turn. Keep ts.
+                try:
+                    self._reset_activity_labels_after_turn()
+                except Exception:
+                    pass
                 reset_accounting_context(acct_token)
                 reset_conversation_context(token)
 

tests/agent/test_compress_context_progress_timeout.py

@@ -0,0 +1,394 @@
+"""Progress-aware timeout around in-agent compress_context (#72016).
+
+In-loop / preflight / manual ``/compress`` paths historically waited on
+``compress_context`` with no host-level inactivity budget. Gateway session
+hygiene already had a progress-aware wait; these tests pin the same contract
+for the owned wrapper used when callers do not pass a ``commit_fence``.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+from agent.conversation_compression import (
+    CompressionCommitFence,
+    resolve_context_compression_timeouts,
+    run_compress_context_with_progress_timeout,
+)
+
+
+class TestResolveContextCompressionTimeouts:
+    def test_defaults_when_empty_cfg(self):
+        idle, ceiling = resolve_context_compression_timeouts({})
+        assert idle == 120.0
+        assert ceiling == 600.0
+
+    def test_zero_idle_disables_wrapper(self):
+        idle, ceiling = resolve_context_compression_timeouts(
+            {"context_timeout_seconds": 0}
+        )
+        assert idle == 0.0
+        assert ceiling == 600.0
+
+    def test_ceiling_clamped_to_idle(self):
+        idle, ceiling = resolve_context_compression_timeouts(
+            {
+                "context_timeout_seconds": 90,
+                "context_total_ceiling_seconds": 30,
+            }
+        )
+        assert idle == 90.0
+        assert ceiling == 90.0
+
+
+class TestRunCompressContextWithProgressTimeout:
+    def test_silent_worker_times_out_and_preserves_messages(self):
+        original = [{"role": "user", "content": "keep-me"}]
+        started = threading.Event()
+        release = threading.Event()
+        commit_attempted = threading.Event()
+
+        def worker(fence: CompressionCommitFence):
+            started.set()
+            assert release.wait(timeout=2)
+            if not fence.begin_commit():
+                return ([{"role": "assistant", "content": "should-not-land"}], "x")
+            try:
+                commit_attempted.set()
+                return ([{"role": "assistant", "content": "too-late"}], "x")
+            finally:
+                fence.finish_commit()
+
+        warnings = []
+
+        result_msgs, result_prompt = run_compress_context_with_progress_timeout(
+            worker=worker,
+            messages=original,
+            system_prompt_fallback="fallback-prompt",
+            idle_timeout_seconds=0.05,
+            total_ceiling_seconds=0.2,
+            on_timeout=lambda idle, waited, since: warnings.append(
+                (idle, waited, since)
+            ),
+        )
+
+        assert started.wait(timeout=1)
+        # Give the waiter time to cancel before releasing the worker.
+        time.sleep(0.15)
+        release.set()
+        # Worker may still be winding down; fence cancel must have won.
+        deadline = time.time() + 1.0
+        while time.time() < deadline and not commit_attempted.is_set():
+            time.sleep(0.01)
+
+        assert result_msgs is original
+        assert result_prompt == "fallback-prompt"
+        assert warnings, "timeout callback should fire"
+        assert not commit_attempted.is_set(), (
+            "cancelled fence must block late session mutation"
+        )
+
+    def test_progress_extends_idle_budget_until_success(self):
+        original = [{"role": "user", "content": "a"}]
+        compressed = [{"role": "user", "content": "summarized"}]
+        fence_holder: dict = {}
+
+        def worker(fence: CompressionCommitFence):
+            fence_holder["fence"] = fence
+            # Keep ticking within each idle window so the waiter extends.
+            for _ in range(6):
+                time.sleep(0.04)
+                fence.touch_progress()
+            if not fence.begin_commit():
+                return (original, "aborted")
+            try:
+                return (compressed, "ok-prompt")
+            finally:
+                fence.finish_commit()
+
+        result_msgs, result_prompt = run_compress_context_with_progress_timeout(
+            worker=worker,
+            messages=original,
+            system_prompt_fallback="fallback",
+            idle_timeout_seconds=0.1,
+            total_ceiling_seconds=1.0,
+        )
+
+        assert result_msgs == compressed
+        assert result_prompt == "ok-prompt"
+        assert "fence" in fence_holder
+
+    def test_commit_started_before_timeout_returns_worker_result(self):
+        original = [{"role": "user", "content": "a"}]
+        compressed = [{"role": "assistant", "content": "done"}]
+        entered = threading.Event()
+
+        def worker(fence: CompressionCommitFence):
+            assert fence.begin_commit()
+            entered.set()
+            try:
+                time.sleep(0.2)
+                return (compressed, "committed")
+            finally:
+                fence.finish_commit()
+
+        result_msgs, result_prompt = run_compress_context_with_progress_timeout(
+            worker=worker,
+            messages=original,
+            system_prompt_fallback="fallback",
+            idle_timeout_seconds=0.05,
+            total_ceiling_seconds=0.05,
+        )
+
+        assert entered.wait(timeout=1)
+        assert result_msgs == compressed
+        assert result_prompt == "committed"
+
+    def test_rejects_non_positive_idle(self):
+        with pytest.raises(ValueError):
+            run_compress_context_with_progress_timeout(
+                worker=lambda fence: ([], ""),
+                messages=[],
+                system_prompt_fallback="",
+                idle_timeout_seconds=0,
+                total_ceiling_seconds=1,
+            )
+
+
+    def test_propagates_conversation_context_into_worker(self):
+        from agent.portal_tags import (
+            get_conversation_context,
+            reset_conversation_context,
+            set_conversation_context,
+        )
+
+        seen = {}
+        token = set_conversation_context("conv-timeout-ctx")
+        try:
+            def worker(fence: CompressionCommitFence):
+                seen["ctx"] = get_conversation_context()
+                if not fence.begin_commit():
+                    return ([], "")
+                try:
+                    return ([{"role": "user", "content": "ok"}], "p")
+                finally:
+                    fence.finish_commit()
+
+            msgs, prompt = run_compress_context_with_progress_timeout(
+                worker=worker,
+                messages=[{"role": "user", "content": "x"}],
+                system_prompt_fallback="fallback",
+                idle_timeout_seconds=1.0,
+                total_ceiling_seconds=2.0,
+            )
+        finally:
+            reset_conversation_context(token)
+
+        assert seen.get("ctx") == "conv-timeout-ctx"
+        assert prompt == "p"
+        assert msgs[0]["content"] == "ok"
+
+    def test_runs_worker_off_caller_thread(self):
+        """Mirror gateway run_in_executor: compress work must leave the caller thread."""
+        caller = threading.current_thread().ident
+        seen = {}
+
+        def worker(fence: CompressionCommitFence):
+            seen["worker"] = threading.current_thread().ident
+            if not fence.begin_commit():
+                return ([], "")
+            try:
+                return ([{"role": "user", "content": "ok"}], "p")
+            finally:
+                fence.finish_commit()
+
+        msgs, prompt = run_compress_context_with_progress_timeout(
+            worker=worker,
+            messages=[],
+            system_prompt_fallback="",
+            idle_timeout_seconds=1.0,
+            total_ceiling_seconds=1.0,
+        )
+        assert seen.get("worker") is not None
+        assert seen["worker"] != caller
+        assert prompt == "p"
+        assert msgs[0]["content"] == "ok"
+
+    def test_reuses_module_shared_executor(self):
+        from tools.daemon_pool import DaemonThreadPoolExecutor
+        from agent import conversation_compression as mod
+
+        first = mod._get_compress_timeout_executor()
+        second = mod._get_compress_timeout_executor()
+        assert first is second
+        assert isinstance(first, DaemonThreadPoolExecutor)
+
+
+class TestCompressContextForwarderOwnsTimeout:
+    """AIAgent._compress_context wraps when no caller fence is supplied."""
+
+    def test_owned_timeout_skips_hung_compress(self, monkeypatch):
+        from run_agent import AIAgent
+
+        agent = object.__new__(AIAgent)
+        agent.session_id = "s1"
+        agent._cached_system_prompt = "sys"
+        agent._emit_warning = MagicMock()
+        agent._touch_activity = MagicMock()
+        agent._build_system_prompt = MagicMock(return_value="sys")
+        agent._conversation_root_id = MagicMock(return_value=None)
+        agent.context_compressor = MagicMock()
+        agent.context_compressor._consecutive_timeout_failures = 0
+        agent.context_compressor._record_compression_failure_cooldown = MagicMock()
+
+        hang = threading.Event()
+        calls = {"n": 0}
+
+        def fake_compress(agent_obj, messages, system_message, **kwargs):
+            calls["n"] += 1
+            fence = kwargs.get("commit_fence")
+            assert fence is not None
+            hang.wait(timeout=2)
+            if not fence.begin_commit():
+                return messages, "sys"
+            try:
+                return ([{"role": "assistant", "content": "nope"}], "sys")
+            finally:
+                fence.finish_commit()
+
+        monkeypatch.setattr(
+            "agent.conversation_compression.compress_context",
+            fake_compress,
+        )
+        monkeypatch.setattr(
+            "agent.conversation_compression.resolve_context_compression_timeouts",
+            lambda compression_cfg=None: (0.05, 0.2),
+        )
+        monkeypatch.setattr(
+            "agent.portal_tags.get_conversation_context",
+            lambda: object(),
+        )
+
+        original = [{"role": "user", "content": "stay"}]
+        out_msgs, out_prompt = AIAgent._compress_context(
+            agent, original, "sys"
+        )
+        hang.set()
+
+        assert out_msgs is original
+        assert out_prompt == "sys"
+        assert calls["n"] == 1
+        agent._emit_warning.assert_called_once()
+        assert agent.context_compressor._consecutive_timeout_failures == 1
+        agent.context_compressor._record_compression_failure_cooldown.assert_called_once()
+        cooldown_args = (
+            agent.context_compressor._record_compression_failure_cooldown.call_args[0]
+        )
+        assert cooldown_args[0] == 60.0
+        assert "host compress_context timeout" in cooldown_args[1]
+        from agent.session_activity import ActivityProvenance
+
+        agent._touch_activity.assert_called_with(
+            "context compression timed out",
+            provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
+        )
+
+    def test_fallback_prompt_resolved_lazily_on_timeout(self, monkeypatch):
+        """Eager prompt rebuild must not run before compression starts."""
+        from run_agent import AIAgent
+
+        agent = object.__new__(AIAgent)
+        agent.session_id = "s1"
+        agent._cached_system_prompt = None
+        agent._emit_warning = MagicMock()
+        agent._touch_activity = MagicMock()
+        agent._conversation_root_id = MagicMock(return_value=None)
+        agent.context_compressor = MagicMock()
+        agent.context_compressor._consecutive_timeout_failures = 0
+        agent.context_compressor._record_compression_failure_cooldown = MagicMock()
+        builds = {"n": 0}
+
+        def boom_build(*_a, **_kw):
+            builds["n"] += 1
+            raise RuntimeError("prompt rebuild boom")
+
+        agent._build_system_prompt = boom_build
+
+        hang = threading.Event()
+
+        def fake_compress(agent_obj, messages, system_message, **kwargs):
+            hang.wait(timeout=2)
+            fence = kwargs.get("commit_fence")
+            if fence is not None and not fence.begin_commit():
+                return messages, "sys"
+            return messages, "sys"
+
+        monkeypatch.setattr(
+            "agent.conversation_compression.compress_context",
+            fake_compress,
+        )
+        monkeypatch.setattr(
+            "agent.conversation_compression.resolve_context_compression_timeouts",
+            lambda compression_cfg=None: (0.05, 0.2),
+        )
+        monkeypatch.setattr(
+            "agent.portal_tags.get_conversation_context",
+            lambda: object(),
+        )
+
+        original = [{"role": "user", "content": "stay"}]
+        out_msgs, out_prompt = AIAgent._compress_context(
+            agent, original, "sys"
+        )
+        hang.set()
+
+        assert out_msgs is original
+        assert out_prompt == "sys"
+        # Fallback rebuild runs only on the timeout return path.
+        assert builds["n"] == 1
+        agent._emit_warning.assert_called_once()
+
+    def test_caller_fence_bypasses_owned_wrapper(self, monkeypatch):
+        from run_agent import AIAgent
+
+        agent = object.__new__(AIAgent)
+        agent.session_id = "s1"
+        agent._cached_system_prompt = "sys"
+        agent._conversation_root_id = MagicMock(return_value=None)
+
+        seen = {}
+
+        def fake_compress(agent_obj, messages, system_message, **kwargs):
+            seen["fence"] = kwargs.get("commit_fence")
+            return ([{"role": "assistant", "content": "ok"}], "sys")
+
+        monkeypatch.setattr(
+            "agent.conversation_compression.compress_context",
+            fake_compress,
+        )
+        # If the owned wrapper ran, this would raise — prove we never call it.
+        monkeypatch.setattr(
+            "agent.conversation_compression.run_compress_context_with_progress_timeout",
+            lambda **kwargs: (_ for _ in ()).throw(
+                AssertionError("owned wrapper must not run")
+            ),
+        )
+        monkeypatch.setattr(
+            "agent.portal_tags.get_conversation_context",
+            lambda: object(),
+        )
+
+        fence = CompressionCommitFence()
+        msgs, prompt = AIAgent._compress_context(
+            agent,
+            [{"role": "user", "content": "x"}],
+            "sys",
+            commit_fence=fence,
+        )
+        assert seen["fence"] is fence
+        assert prompt == "sys"
+        assert msgs[0]["content"] == "ok"

tests/agent/test_compression_concurrent_fork.py

@@ -127,7 +127,7 @@ def test_compression_activity_heartbeat_touches_agent_during_long_compress(tmp_p
     agent = _build_agent_with_db(db, session_id)
     agent._compression_activity_heartbeat_interval = 0.1
     touch_calls: list[str] = []
-    agent._touch_activity = lambda desc: touch_calls.append(desc)
+    agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
 
     def _slow_compress(*_a, **_kw):
         _wait_for_touch(touch_calls, "context compression in progress")
@@ -156,7 +156,7 @@ def test_compression_activity_heartbeat_stops_on_compress_exception(tmp_path: Pa
     agent = _build_agent_with_db(db, session_id)
     agent._compression_activity_heartbeat_interval = 0.1
     touch_calls: list[str] = []
-    agent._touch_activity = lambda desc: touch_calls.append(desc)
+    agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
 
     def _failing_compress(*_a, **_kw):
         _wait_for_touch(touch_calls, "context compression in progress")
@@ -182,7 +182,7 @@ def test_compression_activity_heartbeat_ignores_touch_errors(tmp_path: Path) ->
 
     agent = _build_agent_with_db(db, session_id)
     agent._compression_activity_heartbeat_interval = 0.1
-    agent._touch_activity = lambda _desc: (_ for _ in ()).throw(RuntimeError("touch boom"))
+    agent._touch_activity = lambda _desc, **_kw: (_ for _ in ()).throw(RuntimeError("touch boom"))
     messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
 
     compressed, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
@@ -207,7 +207,7 @@ def test_compression_activity_heartbeat_strict_signature_fallback_releases_lock(
     agent = _build_agent_with_db(db, session_id)
     agent._compression_activity_heartbeat_interval = "not-a-number"
     touch_calls: list[str] = []
-    agent._touch_activity = lambda desc: touch_calls.append(desc)
+    agent._touch_activity = lambda desc, **_kw: touch_calls.append(desc)
     messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
 
     strict_calls: list[int | None] = []
@@ -240,7 +240,13 @@ def test_compression_activity_heartbeat_nonfinite_interval_falls_back(tmp_path:
 
     agent = _build_agent_with_db(db, session_id)
     touch_calls: list[str] = []
-    agent._touch_activity = lambda desc: touch_calls.append(desc)
+    touch_provenances: list = []
+
+    def _capture(desc, *, provenance=None):
+        touch_calls.append(desc)
+        touch_provenances.append(provenance)
+
+    agent._touch_activity = _capture
 
     heartbeat = _CompressionActivityHeartbeat(agent, interval_seconds=float("inf"))
 
@@ -248,7 +254,104 @@ def test_compression_activity_heartbeat_nonfinite_interval_falls_back(tmp_path:
     heartbeat.start()
     heartbeat.stop()
     assert touch_calls == ["context compression started", "context compression completed"]
+    from agent.session_activity import ActivityProvenance
 
+    assert touch_provenances == [
+        ActivityProvenance.AGENT_COMPRESSION,
+        ActivityProvenance.AGENT_COMPRESSION,
+    ]
+
+
+def test_compression_heartbeat_does_not_clobber_timeout_provenance() -> None:
+    """Detached heartbeat/stop must not overwrite a host timeout stamp."""
+    from types import SimpleNamespace
+
+    from agent.conversation_compression import _CompressionActivityHeartbeat
+    from agent.session_activity import ActivityProvenance
+
+    agent = SimpleNamespace(
+        _last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
+        _last_activity_desc="context compression timed out",
+        touches=[],
+    )
+
+    def _touch(desc, *, provenance=None):
+        agent.touches.append((desc, provenance))
+        agent._last_activity_provenance = provenance
+        agent._last_activity_desc = desc
+
+    agent._touch_activity = _touch
+
+    hb = _CompressionActivityHeartbeat(agent, interval_seconds=60.0)
+    hb._touch("context compression in progress")
+    hb.stop("context compression completed")
+
+    assert agent.touches == []
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION_TIMEOUT
+    assert agent._last_activity_desc == "context compression timed out"
+
+
+def test_compression_heartbeat_does_not_clobber_cooldown_provenance() -> None:
+    """Cooldown/abort stamps must also survive a late heartbeat stop."""
+    from types import SimpleNamespace
+
+    from agent.conversation_compression import _CompressionActivityHeartbeat
+    from agent.session_activity import ActivityProvenance
+
+    agent = SimpleNamespace(
+        _last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION_COOLDOWN,
+        _last_activity_desc="compression blocked (cooldown: 30s remaining)",
+        touches=[],
+    )
+
+    def _touch(desc, *, provenance=None):
+        agent.touches.append((desc, provenance))
+        agent._last_activity_provenance = provenance
+        agent._last_activity_desc = desc
+
+    agent._touch_activity = _touch
+
+    hb = _CompressionActivityHeartbeat(agent, interval_seconds=60.0)
+    hb._touch("context compression in progress")
+    hb.stop("context compression failed")
+
+    assert agent.touches == []
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION_COOLDOWN
+
+
+def test_compression_heartbeat_start_republishes_after_terminal_provenance() -> None:
+    """A new compression episode may overwrite a prior timeout/cooldown stamp."""
+    from types import SimpleNamespace
+
+    from agent.conversation_compression import _CompressionActivityHeartbeat
+    from agent.session_activity import ActivityProvenance
+
+    agent = SimpleNamespace(
+        _last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
+        _last_activity_desc="context compression timed out",
+        touches=[],
+    )
+
+    def _touch(desc, *, provenance=None):
+        agent.touches.append((desc, provenance))
+        agent._last_activity_provenance = provenance
+        agent._last_activity_desc = desc
+
+    agent._touch_activity = _touch
+
+    hb = _CompressionActivityHeartbeat(agent, interval_seconds=60.0)
+    hb.start()
+    hb.stop()
+
+    assert agent.touches[0] == (
+        "context compression started",
+        ActivityProvenance.AGENT_COMPRESSION,
+    )
+    assert agent.touches[-1] == (
+        "context compression completed",
+        ActivityProvenance.AGENT_COMPRESSION,
+    )
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
 
 
 def test_concurrent_compression_does_not_fork_session(tmp_path: Path) -> None:

tests/agent/test_session_activity.py

@@ -0,0 +1,82 @@
+"""Unit tests for the shared session activity observation contract."""
+
+from agent.session_activity import (
+    ActivityProvenance,
+    bound_activity_description,
+    build_activity_snapshot,
+    normalize_activity_provenance,
+)
+
+
+def test_bound_activity_description_truncates():
+    long = "x" * 200
+    out = bound_activity_description(long)
+    assert len(out) == 120
+    assert out.endswith("…")
+
+
+def test_normalize_activity_provenance_defaults_to_unknown():
+    assert normalize_activity_provenance(None) is ActivityProvenance.UNKNOWN
+    assert normalize_activity_provenance("") is ActivityProvenance.UNKNOWN
+    assert normalize_activity_provenance("not-a-real-source") is ActivityProvenance.UNKNOWN
+    assert normalize_activity_provenance("agent.activity") is ActivityProvenance.UNKNOWN
+    assert (
+        normalize_activity_provenance(ActivityProvenance.AGENT_COMPRESSION)
+        is ActivityProvenance.AGENT_COMPRESSION
+    )
+    assert (
+        normalize_activity_provenance("agent.compression_timeout")
+        is ActivityProvenance.AGENT_COMPRESSION_TIMEOUT
+    )
+
+
+def test_build_activity_snapshot_includes_compat_aliases():
+    snap = build_activity_snapshot(
+        last_activity_at=100.0,
+        last_activity_description="starting API call #1",
+        last_activity_provenance=ActivityProvenance.UNKNOWN,
+        now=110.0,
+        extra={"api_call_count": 1},
+    )
+    assert snap["last_activity_at"] == 100.0
+    assert snap["last_activity_description"] == "starting API call #1"
+    assert snap["last_activity_provenance"] == "unknown"
+    assert snap["seconds_since_activity"] == 10.0
+    assert snap["last_activity_ts"] == 100.0
+    assert snap["last_activity_desc"] == "starting API call #1"
+    assert snap["description"] == "starting API call #1"
+    assert snap["api_call_count"] == 1
+    assert "phase" not in snap
+    assert "last_progress_at" not in snap
+
+
+def test_build_activity_snapshot_maps_missing_provenance_to_unknown():
+    snap = build_activity_snapshot(
+        last_activity_at=1.0,
+        last_activity_description="starting new turn (cached)",
+        last_activity_provenance=None,
+        now=2.0,
+    )
+    assert snap["last_activity_provenance"] == "unknown"
+
+
+def test_build_activity_snapshot_preserves_compression_transition_provenances():
+    """Compaction / timeout / cooldown share the observation source (#72424)."""
+    for provenance, desc in (
+        (ActivityProvenance.AGENT_COMPRESSION, "context compression in progress"),
+        (ActivityProvenance.AGENT_COMPRESSION_TIMEOUT, "context compression timed out"),
+        (
+            ActivityProvenance.AGENT_COMPRESSION_COOLDOWN,
+            "compression blocked (cooldown: 30s remaining)",
+        ),
+    ):
+        snap = build_activity_snapshot(
+            last_activity_at=50.0,
+            last_activity_description=desc,
+            last_activity_provenance=provenance,
+            now=55.0,
+        )
+        assert snap["last_activity_provenance"] == provenance.value
+        assert snap["provenance"] == provenance.value
+        assert snap["last_activity_description"] == desc
+        assert snap["seconds_since_activity"] == 5.0

tests/gateway/test_agent_cache.py

@@ -1598,10 +1598,13 @@ class TestCachedAgentInactivityReset:
     """
 
     def _fake_agent(self, stale_seconds: float = 1800.0):
+        from agent.session_activity import ActivityProvenance
+
         m = MagicMock()
         m._last_activity_ts = _FAKE_NOW - stale_seconds
         m._api_call_count = 10
         m._last_activity_desc = "previous turn activity"
+        m._last_activity_provenance = ActivityProvenance.AGENT_COMPRESSION
         return m
 
     def test_fresh_turn_resets_idle_clock(self):
@@ -1635,6 +1638,20 @@ class TestCachedAgentInactivityReset:
 
         assert agent._last_activity_desc == "starting new turn (cached)"
 
+    def test_fresh_turn_resets_provenance(self):
+        """interrupt_depth=0: provenance resets with ts/desc (#72039)."""
+        from agent.session_activity import ActivityProvenance
+        from gateway.run import GatewayRunner
+
+        agent = self._fake_agent()
+        assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
+
+        with patch("gateway.run.time") as mock_time:
+            mock_time.time.return_value = _FAKE_NOW
+            GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0)
+
+        assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
+
     def test_interrupt_turn_preserves_idle_clock(self):
         """interrupt_depth=1: clock preserved so accumulated stuck-turn
         idle time is not discarded by an interrupt-recursive re-entry (#15654)."""
@@ -1663,6 +1680,17 @@ class TestCachedAgentInactivityReset:
             "it describes the activity *at* _last_activity_ts"
         )
 
+    def test_interrupt_turn_preserves_provenance(self):
+        """interrupt_depth=1: provenance preserved with ts/desc."""
+        from agent.session_activity import ActivityProvenance
+        from gateway.run import GatewayRunner
+
+        agent = self._fake_agent(stale_seconds=1200.0)
+
+        GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=1)
+
+        assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
+
     def test_deep_interrupt_recursion_preserves_idle_clock(self):
         """interrupt_depth=MAX-1: clock still preserved at any non-zero depth."""
         from gateway.run import GatewayRunner

tests/gateway/test_cached_agent_max_iterations.py

@@ -21,6 +21,7 @@ import time
 from types import SimpleNamespace
 
 from agent.iteration_budget import IterationBudget
+from agent.session_activity import ActivityProvenance
 
 
 def _make_cached_agent(max_iterations: int) -> SimpleNamespace:
@@ -32,6 +33,7 @@ def _make_cached_agent(max_iterations: int) -> SimpleNamespace:
     return SimpleNamespace(
         _last_activity_ts=time.time() - 1000,
         _last_activity_desc="previous turn",
+        _last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION,
         _api_call_count=42,
         _last_flushed_db_idx=5,
         max_iterations=max_iterations,
@@ -53,6 +55,7 @@ def test_init_cached_agent_for_turn_does_not_touch_max_iterations():
     # Per-turn state was reset...
     assert agent._api_call_count == 0
     assert agent._last_activity_desc == "starting new turn (cached)"
+    assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
     assert agent._last_flushed_db_idx == 0
     # ...but the iteration budget was NOT changed by the helper itself.
     assert agent.max_iterations == 90
@@ -67,6 +70,7 @@ def test_init_cached_agent_preserves_max_iterations_on_interrupt_depth():
 
     # Activity timestamps preserved for the inactivity watchdog (#15654)...
     assert agent._last_activity_desc == "previous turn"
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
     # ...and max_iterations untouched.
     assert agent.max_iterations == 200
 

tests/gateway/test_config_env_bridge_authority.py

@@ -44,6 +44,7 @@ def _run_gateway_import(hermes_home: Path, initial_env: dict[str, str]) -> dict[
             "HERMES_MAX_ITERATIONS",
             "HERMES_AGENT_TIMEOUT",
             "HERMES_AGENT_TIMEOUT_WARNING",
+            "HERMES_SESSION_STALL_TIMEOUT",
             "HERMES_GATEWAY_BUSY_INPUT_MODE",
             "HERMES_GATEWAY_BUSY_TEXT_MODE",
             "HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT",
@@ -126,16 +127,19 @@ def test_config_gateway_timeout_wins_over_stale_env(hermes_home: Path) -> None:
     _write_config(hermes_home, agent_cfg={
         "gateway_timeout": 1800,
         "gateway_timeout_warning": 900,
+        "session_stall_timeout": 300,
     })
     _write_env(hermes_home, {
         "HERMES_AGENT_TIMEOUT": "60",
         "HERMES_AGENT_TIMEOUT_WARNING": "30",
+        "HERMES_SESSION_STALL_TIMEOUT": "15",
     })
 
     env = _run_gateway_import(hermes_home, initial_env={})
 
     assert env.get("HERMES_AGENT_TIMEOUT") == "1800"
     assert env.get("HERMES_AGENT_TIMEOUT_WARNING") == "900"
+    assert env.get("HERMES_SESSION_STALL_TIMEOUT") == "300"
 
 
 def test_config_display_busy_input_mode_wins_over_stale_env(hermes_home: Path) -> None:

tests/gateway/test_session_stall_watchdog.py

@@ -0,0 +1,462 @@
+"""Tests for gateway session stall watchdog (#72016 item 2)."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from agent.session_activity import ActivityProvenance, build_activity_snapshot
+from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL
+from gateway.session_stall import (
+    format_session_stall_notification,
+    resolve_session_idle_seconds_from_activity,
+    should_clear_session_stall_notification,
+    should_emit_session_stall_notification,
+)
+
+
+class _FakeAdapter:
+    def __init__(self):
+        self._pending_messages = {}
+        self.sent = []
+
+    async def send(self, chat_id, content, metadata=None):
+        self.sent.append(
+            {"chat_id": chat_id, "content": content, "metadata": metadata}
+        )
+
+
+class _FakeAgent:
+    """Exposes the shared #72039 activity snapshot as the sole progress source."""
+
+    def __init__(
+        self,
+        last_activity_ts: float,
+        *,
+        description: str = "api call",
+        provenance: ActivityProvenance = ActivityProvenance.UNKNOWN,
+    ):
+        self._last_activity_ts = last_activity_ts
+        self._last_activity_desc = description
+        self._last_activity_provenance = provenance
+
+    def get_activity_summary(self):
+        return build_activity_snapshot(
+            last_activity_at=self._last_activity_ts,
+            last_activity_description=self._last_activity_desc,
+            last_activity_provenance=self._last_activity_provenance,
+        )
+
+
+class _AgentWithoutSummary:
+    """Agent with a raw clock but no shared summary consumer API."""
+
+    def __init__(self, last_activity_ts: float):
+        self._last_activity_ts = last_activity_ts
+
+
+def test_should_emit_requires_pending_and_idle():
+    assert should_emit_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=400,
+        has_pending_inbound=True,
+        already_notified=False,
+    )
+    assert not should_emit_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=400,
+        has_pending_inbound=False,
+        already_notified=False,
+    )
+    assert not should_emit_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=100,
+        has_pending_inbound=True,
+        already_notified=False,
+    )
+    assert not should_emit_session_stall_notification(
+        timeout_seconds=0,
+        idle_seconds=9999,
+        has_pending_inbound=True,
+        already_notified=False,
+    )
+    assert not should_emit_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=400,
+        has_pending_inbound=True,
+        already_notified=True,
+    )
+
+
+def test_should_clear_when_pending_gone_or_activity_resumes():
+    assert should_clear_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=400,
+        has_pending_inbound=False,
+    )
+    assert should_clear_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=10,
+        has_pending_inbound=True,
+    )
+    assert not should_clear_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=400,
+        has_pending_inbound=True,
+    )
+
+
+def test_should_clear_holds_latch_when_idle_unknown():
+    assert not should_clear_session_stall_notification(
+        timeout_seconds=300,
+        idle_seconds=None,
+        has_pending_inbound=True,
+    )
+
+
+def test_format_session_stall_notification_minutes():
+    msg = format_session_stall_notification(125)
+    assert "2 min ago" in msg
+    assert "/new" in msg
+    assert format_session_stall_notification(30).count("1 min ago") == 1
+
+
+def test_resolve_idle_uses_shared_activity_snapshot_only():
+    now = 1_000_000.0
+    snap = build_activity_snapshot(
+        last_activity_at=now - 120,
+        last_activity_description="tool: terminal",
+        last_activity_provenance=ActivityProvenance.UNKNOWN,
+        now=now,
+    )
+    assert resolve_session_idle_seconds_from_activity(snap, now=now) == 120.0
+    assert resolve_session_idle_seconds_from_activity(None, now=now) is None
+    assert resolve_session_idle_seconds_from_activity({}, now=now) is None
+
+
+def test_resolve_idle_prefers_seconds_since_activity_field():
+    idle = resolve_session_idle_seconds_from_activity(
+        {
+            "seconds_since_activity": 42.5,
+            "last_activity_at": 1.0,  # must be ignored when seconds present
+        },
+        now=999.0,
+    )
+    assert idle == 42.5
+
+
+def _runner_for_stall(adapter: _FakeAdapter) -> GatewayRunner:
+    r = GatewayRunner.__new__(GatewayRunner)
+    r._running = True
+    r.adapters = {"fake": adapter}
+    r._profile_adapters = {}
+    r._running_agents = {}
+    r._running_agents_ts = {}
+    r._queued_events = {}
+    r._session_stall_notified = {}
+    r._thread_metadata_for_source = lambda source, *a, **k: {
+        "thread_id": getattr(source, "thread_id", None)
+    }
+    return r
+
+
+def _pending_event(chat_id: str = "chat-1", thread_id: str | None = None):
+    source = SimpleNamespace(chat_id=chat_id, thread_id=thread_id, platform=None)
+    return SimpleNamespace(text="follow-up", source=source, timestamp=time.time())
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_notifies_once(monkeypatch):
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    monkeypatch.setenv("HERMES_SESSION_STALL_TIMEOUT", "60")
+    session_key = "agent:main:telegram:dm:1"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+
+    sent = await runner._check_session_stalls(60)
+    assert sent == 1
+    assert len(adapter.sent) == 1
+    assert "/new" in adapter.sent[0]["content"]
+    assert runner._session_stall_notified.get(session_key) is True
+
+    # Second pass must not spam.
+    sent2 = await runner._check_session_stalls(60)
+    assert sent2 == 0
+    assert len(adapter.sent) == 1
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_skips_fresh_activity():
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:2"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 5)
+
+    sent = await runner._check_session_stalls(60)
+    assert sent == 0
+    assert adapter.sent == []
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_skips_without_pending():
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:3"
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 999)
+
+    sent = await runner._check_session_stalls(60)
+    assert sent == 0
+    assert adapter.sent == []
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_clears_latch_when_pending_drains():
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:4"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 200)
+
+    assert await runner._check_session_stalls(60) == 1
+    assert session_key in runner._session_stall_notified
+
+    adapter._pending_messages.clear()
+    assert await runner._check_session_stalls(60) == 0
+    assert session_key not in runner._session_stall_notified
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_skips_pending_sentinel_without_activity():
+    """Pending construction has no shared activity snapshot — no parallel clocks."""
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:5"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _AGENT_PENDING_SENTINEL
+    runner._running_agents_ts[session_key] = time.time() - 90
+
+    sent = await runner._check_session_stalls(60)
+    assert sent == 0
+    assert adapter.sent == []
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_ignores_raw_clock_without_summary():
+    """Do not fall back to agent._last_activity_ts outside get_activity_summary()."""
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:raw"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _AgentWithoutSummary(time.time() - 999)
+    runner._running_agents_ts[session_key] = time.time() - 999
+
+    sent = await runner._check_session_stalls(60)
+    assert sent == 0
+    assert adapter.sent == []
+
+
+@pytest.mark.asyncio
+async def test_session_stall_watcher_disabled_when_timeout_zero(monkeypatch):
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    monkeypatch.setenv("HERMES_SESSION_STALL_TIMEOUT", "0")
+    session_key = "agent:main:telegram:dm:6"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 999)
+
+    task = asyncio.create_task(runner._session_stall_watcher(interval=1))
+    await asyncio.sleep(0.05)
+    # Force one check path via timeout read; watcher should no-op when 0.
+    assert runner._session_stall_timeout_seconds() == 0.0
+    runner._running = False
+    await asyncio.wait_for(task, timeout=2)
+    assert adapter.sent == []
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_queued_events_overflow_notifies():
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:overflow"
+    event = _pending_event()
+    runner._queued_events[session_key] = [event]
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+    runner._adapter_for_source = lambda source: adapter
+
+    assert await runner._check_session_stalls(60) == 1
+    assert adapter.sent and "/new" in adapter.sent[0]["content"]
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_scans_profile_adapters():
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(_FakeAdapter())  # empty primary path unused
+    runner.adapters = {}
+    runner._profile_adapters = {"coder": {"fake": adapter}}
+    session_key = "agent:coder:telegram:dm:1"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+
+    assert await runner._check_session_stalls(60) == 1
+    assert len(adapter.sent) == 1
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_logs_compression_provenance(caplog):
+    """Stale compression stamps still stall, but provenance stays visible."""
+    import logging
+
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:compress"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(
+        time.time() - 120,
+        description="compressing context",
+        provenance=ActivityProvenance.AGENT_COMPRESSION,
+    )
+
+    with caplog.at_level(logging.WARNING):
+        assert await runner._check_session_stalls(60) == 1
+    assert any("agent.compression" in r.message for r in caplog.records)
+    assert any("compressing context" in r.message for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_skips_active_compression_heartbeat():
+    """Fresh agent.compression heartbeats are progress, not a silent stall."""
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:compacting"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(
+        time.time() - 5,
+        description="context compression in progress",
+        provenance=ActivityProvenance.AGENT_COMPRESSION,
+    )
+
+    assert await runner._check_session_stalls(60) == 0
+    assert adapter.sent == []
+    assert session_key not in runner._session_stall_notified
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_does_not_renotify_after_summary_gap():
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:gap"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+
+    assert await runner._check_session_stalls(60) == 1
+
+    # Transient observation gap (no summary API).
+    runner._running_agents[session_key] = _AgentWithoutSummary(time.time() - 999)
+    assert await runner._check_session_stalls(60) == 0
+    assert runner._session_stall_notified.get(session_key) is True
+
+    # Stale progress returns — must not spam again in the same episode.
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+    assert await runner._check_session_stalls(60) == 0
+    assert len(adapter.sent) == 1
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_retries_after_send_failure():
+    class _FailThenOk(_FakeAdapter):
+        def __init__(self):
+            super().__init__()
+            self.calls = 0
+
+        async def send(self, chat_id, content, metadata=None):
+            self.calls += 1
+            if self.calls == 1:
+                raise RuntimeError("boom")
+            await super().send(chat_id, content, metadata=metadata)
+
+    adapter = _FailThenOk()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:retry"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+
+    assert await runner._check_session_stalls(60) == 0
+    assert session_key not in runner._session_stall_notified
+    assert await runner._check_session_stalls(60) == 1
+    assert runner._session_stall_notified.get(session_key) is True
+    assert len(adapter.sent) == 1
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_retries_after_soft_send_failure():
+    class _SoftFailThenOk(_FakeAdapter):
+        def __init__(self):
+            super().__init__()
+            self.calls = 0
+
+        async def send(self, chat_id, content, metadata=None):
+            from gateway.platforms.base import SendResult
+
+            self.calls += 1
+            if self.calls == 1:
+                return SendResult(success=False, error="chat not found")
+            await super().send(chat_id, content, metadata=metadata)
+            return SendResult(success=True)
+
+    adapter = _SoftFailThenOk()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:soft"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+
+    assert await runner._check_session_stalls(60) == 0
+    assert session_key not in runner._session_stall_notified
+    assert await runner._check_session_stalls(60) == 1
+    assert runner._session_stall_notified.get(session_key) is True
+    assert len(adapter.sent) == 1
+
+
+@pytest.mark.asyncio
+async def test_check_session_stalls_renotifies_after_resume_then_restall():
+    adapter = _FakeAdapter()
+    runner = _runner_for_stall(adapter)
+    session_key = "agent:main:telegram:dm:episode"
+    adapter._pending_messages[session_key] = _pending_event()
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+
+    assert await runner._check_session_stalls(60) == 1
+
+    # Activity resumes (still pending) — clears latch for a new episode.
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 5)
+    assert await runner._check_session_stalls(60) == 0
+    assert session_key not in runner._session_stall_notified
+
+    # Stall again — second episode may notify once more.
+    runner._running_agents[session_key] = _FakeAgent(time.time() - 120)
+    assert await runner._check_session_stalls(60) == 1
+    assert len(adapter.sent) == 2
+
+
+def test_resolve_idle_rejects_nonfinite_seconds_since_activity():
+    now = 1_000_000.0
+    idle = resolve_session_idle_seconds_from_activity(
+        {
+            "seconds_since_activity": float("nan"),
+            "last_activity_at": now - 15,
+        },
+        now=now,
+    )
+    assert idle == 15.0
+
+
+def test_session_stall_timeout_in_default_config():
+    from hermes_cli.config import DEFAULT_CONFIG
+
+    timeout = DEFAULT_CONFIG["agent"]["session_stall_timeout"]
+    assert isinstance(timeout, (int, float))
+    assert timeout > 0  # enabled by default; 0 would disable the watchdog

tests/hermes_cli/test_status.py

@@ -352,3 +352,43 @@ class TestShowStatusXaiOAuth:
 
         assert "xAI OAuth" in out
         assert "not logged in (run: hermes auth add xai-oauth)" in out
+
+
+def test_show_status_reports_gateway_session_last_activity(monkeypatch, capsys, tmp_path):
+    """hermes status should surface freshest gateway last_active (#72016)."""
+    from hermes_cli import status as status_mod
+    import hermes_cli.auth as auth_mod
+    import hermes_cli.gateway as gateway_mod
+    import hermes_state
+    import time
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(status_mod, "get_env_path", lambda: tmp_path / ".env", raising=False)
+    monkeypatch.setattr(status_mod, "get_hermes_home", lambda: tmp_path, raising=False)
+    monkeypatch.setattr(status_mod, "load_config", lambda: {"model": "gpt-5.4"}, raising=False)
+    monkeypatch.setattr(status_mod, "resolve_requested_provider", lambda requested=None: "openai-codex", raising=False)
+    monkeypatch.setattr(status_mod, "resolve_provider", lambda requested=None, **kwargs: "openai-codex", raising=False)
+    monkeypatch.setattr(status_mod, "provider_label", lambda provider: "OpenAI Codex", raising=False)
+    monkeypatch.setattr(auth_mod, "get_nous_auth_status", lambda: {}, raising=False)
+    monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
+    monkeypatch.setattr(auth_mod, "get_qwen_auth_status", lambda: {}, raising=False)
+    monkeypatch.setattr(auth_mod, "get_xai_oauth_auth_status", lambda: {}, raising=False)
+    monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
+
+    class _FakeDB:
+        def list_gateway_sessions(self, active_only=True):
+            return [
+                {"id": "gw-old", "last_active": time.time() - 7200},
+                {"id": "gw-new", "last_active": time.time() - 90},
+            ]
+
+        def close(self):
+            return None
+
+    monkeypatch.setattr(hermes_state, "SessionDB", _FakeDB)
+
+    status_mod.show_status(SimpleNamespace(all=False, deep=False))
+    output = capsys.readouterr().out
+    assert "Active:       2 session(s)" in output
+    assert "Last activity:" in output
+    assert "1m ago" in output

tests/run_agent/test_codex_app_server_compaction.py

@@ -69,10 +69,12 @@ class DummyAgent:
         self.events = []
         self.built_prompts = []
         self.touch_calls = []
+        self.touch_provenances = []
         self._compression_activity_heartbeat_interval = 0.1
 
-    def _touch_activity(self, desc):
+    def _touch_activity(self, desc, *, provenance=None):
         self.touch_calls.append(desc)
+        self.touch_provenances.append(provenance)
 
     def _emit_status(self, message):
         self.statuses.append(message)
@@ -136,6 +138,12 @@ def test_codex_app_server_compaction_heartbeat_refreshes_activity_while_waiting(
     assert "context compression started" in agent.touch_calls
     assert "context compression in progress" in agent.touch_calls
     assert agent.touch_calls[-1] == "context compression completed"
+    from agent.session_activity import ActivityProvenance
+
+    assert agent.touch_provenances
+    assert all(
+        p is ActivityProvenance.AGENT_COMPRESSION for p in agent.touch_provenances
+    )
 
 
 def test_codex_app_server_manual_compression_routes_to_codex_thread():

tests/run_agent/test_session_activity_persist.py

@@ -0,0 +1,263 @@
+"""Durable session activity projection from AIAgent._touch_activity (#72016)."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import run_agent
+from agent.session_activity import ActivityProvenance
+
+
+def _agent_with_db(session_id: str = "sess-1"):
+    agent = SimpleNamespace(
+        session_id=session_id,
+        _session_db=MagicMock(),
+        _last_activity_ts=0.0,
+        _last_activity_desc="",
+        _last_activity_provenance=ActivityProvenance.UNKNOWN,
+        _session_activity_last_persist_mono=0.0,
+        _current_tool=None,
+        _api_call_count=0,
+        max_iterations=10,
+        iteration_budget=SimpleNamespace(used=0, max_total=10),
+    )
+    agent._touch_activity = run_agent.AIAgent._touch_activity.__get__(agent, SimpleNamespace)
+    agent._persist_session_activity_if_due = (
+        run_agent.AIAgent._persist_session_activity_if_due.__get__(agent, SimpleNamespace)
+    )
+    agent._reset_activity_labels_after_turn = (
+        run_agent.AIAgent._reset_activity_labels_after_turn.__get__(agent, SimpleNamespace)
+    )
+    agent.get_activity_summary = run_agent.AIAgent.get_activity_summary.__get__(
+        agent, SimpleNamespace
+    )
+    return agent
+
+
+def test_touch_activity_persists_session_activity_once_per_minute(monkeypatch):
+    agent = _agent_with_db()
+    mono = {"t": 1000.0}
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1_700_000_000.0)
+    monkeypatch.setattr(run_agent.time, "monotonic", lambda: mono["t"])
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+    agent._touch_activity("starting API call #1")
+    agent._session_db.touch_session_activity.assert_called_once_with(
+        "sess-1",
+        1_700_000_000.0,
+        description="starting API call #1",
+        provenance=ActivityProvenance.UNKNOWN,
+    )
+
+    agent._session_db.touch_session_activity.reset_mock()
+    mono["t"] = 1030.0  # within 60s window
+    agent._touch_activity("receiving stream response")
+    agent._session_db.touch_session_activity.assert_not_called()
+
+    mono["t"] = 1061.0
+    agent._touch_activity("API call #1 completed")
+    agent._session_db.touch_session_activity.assert_called_once_with(
+        "sess-1",
+        1_700_000_000.0,
+        description="API call #1 completed",
+        provenance=ActivityProvenance.UNKNOWN,
+    )
+
+
+def test_touch_activity_skips_persist_without_session_db(monkeypatch):
+    agent = _agent_with_db()
+    agent._session_db = None
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1.0)
+    monkeypatch.setattr(run_agent.time, "monotonic", lambda: 1.0)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+    agent._touch_activity("starting API call #1")
+    assert agent._last_activity_desc == "starting API call #1"
+    assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
+
+
+def test_touch_activity_accepts_named_provenance(monkeypatch):
+    agent = _agent_with_db()
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1_700_000_000.0)
+    monkeypatch.setattr(run_agent.time, "monotonic", lambda: 1000.0)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+    agent._touch_activity(
+        "compressing context",
+        provenance=ActivityProvenance.AGENT_COMPRESSION,
+    )
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
+    agent._session_db.touch_session_activity.assert_called_once_with(
+        "sess-1",
+        1_700_000_000.0,
+        description="compressing context",
+        provenance=ActivityProvenance.AGENT_COMPRESSION,
+    )
+
+    agent._session_db.touch_session_activity.reset_mock()
+    agent._session_activity_last_persist_mono = 0.0
+    agent._touch_activity("starting API call #1")
+    assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
+    agent._session_db.touch_session_activity.assert_called_once_with(
+        "sess-1",
+        1_700_000_000.0,
+        description="starting API call #1",
+        provenance=ActivityProvenance.UNKNOWN,
+    )
+
+
+def test_touch_activity_persist_errors_are_swallowed(monkeypatch):
+    agent = _agent_with_db()
+    agent._session_db.touch_session_activity.side_effect = RuntimeError("db locked")
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1.0)
+    monkeypatch.setattr(run_agent.time, "monotonic", lambda: 1.0)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+    agent._touch_activity("tool completed: terminal (1.0s)")
+    assert agent._last_activity_desc == "tool completed: terminal (1.0s)"
+
+
+def test_get_activity_summary_exposes_shared_activity_contract(monkeypatch):
+    agent = _agent_with_db()
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1_700_000_010.0)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    agent._last_activity_ts = 1_700_000_000.0
+    agent._last_activity_desc = "executing tool: terminal"
+    agent._last_activity_provenance = ActivityProvenance.UNKNOWN
+
+    summary = agent.get_activity_summary()
+    assert summary["last_activity_at"] == 1_700_000_000.0
+    assert summary["last_activity_description"] == "executing tool: terminal"
+    assert summary["last_activity_provenance"] == "unknown"
+    assert summary["seconds_since_activity"] == 10.0
+    assert summary["last_activity_ts"] == 1_700_000_000.0
+    assert summary["last_activity_desc"] == "executing tool: terminal"
+    assert "phase" not in summary
+    assert "last_progress_at" not in summary
+
+
+def test_reset_activity_labels_after_turn_keeps_ts_and_clears_labels():
+    """Turn-end cleanup must not bump ts (watchdog continuity) but must
+    clear mid-turn description/provenance and force a durable label clear.
+    """
+    agent = _agent_with_db()
+    agent._last_activity_ts = 1_700_000_000.0
+    agent._last_activity_desc = "compressing context"
+    agent._last_activity_provenance = ActivityProvenance.AGENT_COMPRESSION
+    # Still inside the 60s persist window from a prior heartbeat — label
+    # clear must bypass that rate limit via clear_session_activity_labels.
+    agent._session_activity_last_persist_mono = 1_000.0
+
+    agent._reset_activity_labels_after_turn()
+
+    assert agent._last_activity_ts == 1_700_000_000.0
+    assert agent._last_activity_desc == ""
+    assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
+    agent._session_db.clear_session_activity_labels.assert_called_once_with("sess-1")
+    agent._session_db.touch_session_activity.assert_not_called()
+
+
+def test_reset_activity_labels_after_turn_skips_db_without_session():
+    agent = _agent_with_db()
+    agent.session_id = None
+    agent._last_activity_ts = 42.0
+    agent._last_activity_desc = "executing tool: terminal"
+    agent._last_activity_provenance = ActivityProvenance.AGENT_COMPRESSION
+
+    agent._reset_activity_labels_after_turn()
+
+    assert agent._last_activity_ts == 42.0
+    assert agent._last_activity_desc == ""
+    assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
+    agent._session_db.clear_session_activity_labels.assert_not_called()
+
+
+def test_reset_activity_labels_after_turn_swallows_db_errors():
+    agent = _agent_with_db()
+    agent._last_activity_ts = 99.0
+    agent._last_activity_desc = "starting API call #1"
+    agent._last_activity_provenance = ActivityProvenance.UNKNOWN
+    agent._session_db.clear_session_activity_labels.side_effect = RuntimeError("db locked")
+
+    agent._reset_activity_labels_after_turn()
+
+    assert agent._last_activity_ts == 99.0
+    assert agent._last_activity_desc == ""
+    assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
+
+
+def test_warn_context_overflow_blocked_stamps_compression_cooldown(monkeypatch):
+    agent = _agent_with_db()
+    agent._last_ctx_overflow_warn = None
+    agent._emit_warning = MagicMock()
+    agent._touch_activity = run_agent.AIAgent._touch_activity.__get__(
+        agent, SimpleNamespace
+    )
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1_700_000_100.0)
+    monkeypatch.setattr(run_agent.time, "monotonic", lambda: 2000.0)
+
+    run_agent.AIAgent._warn_context_overflow_blocked(
+        agent, "cooldown: 30s remaining", 80_000, 40_000
+    )
+
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION_COOLDOWN
+    assert "compression blocked" in agent._last_activity_desc
+    agent._emit_warning.assert_called_once()
+
+    # Deduped re-entry must not re-touch or re-emit.
+    agent._session_db.touch_session_activity.reset_mock()
+    prev_desc = agent._last_activity_desc
+    run_agent.AIAgent._warn_context_overflow_blocked(
+        agent, "cooldown: 29s remaining", 80_000, 40_000
+    )
+    assert agent._last_activity_desc == prev_desc
+    agent._emit_warning.assert_called_once()
+
+
+def test_warn_context_overflow_blocked_stamps_cooldown_for_ineffective(monkeypatch):
+    agent = _agent_with_db()
+    agent._last_ctx_overflow_warn = None
+    agent._emit_warning = MagicMock()
+    agent._touch_activity = run_agent.AIAgent._touch_activity.__get__(
+        agent, SimpleNamespace
+    )
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1_700_000_100.0)
+    monkeypatch.setattr(run_agent.time, "monotonic", lambda: 2000.0)
+
+    run_agent.AIAgent._warn_context_overflow_blocked(
+        agent, "ineffective: last pass saved 0 tokens", 80_000, 40_000
+    )
+
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION_COOLDOWN
+    agent._emit_warning.assert_called_once()
+
+
+def test_compression_transition_provenances_surface_in_activity_summary(monkeypatch):
+    """Compaction / timeout / cooldown publish through get_activity_summary."""
+    agent = _agent_with_db()
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(run_agent.time, "time", lambda: 1_700_000_200.0)
+    monkeypatch.setattr(run_agent.time, "monotonic", lambda: 3000.0)
+
+    transitions = (
+        (
+            ActivityProvenance.AGENT_COMPRESSION,
+            "context compression in progress",
+        ),
+        (
+            ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
+            "context compression timed out",
+        ),
+        (
+            ActivityProvenance.AGENT_COMPRESSION_COOLDOWN,
+            "compression blocked (cooldown: 30s remaining)",
+        ),
+    )
+    for provenance, desc in transitions:
+        agent._touch_activity(desc, provenance=provenance)
+        summary = agent.get_activity_summary()
+        assert summary["last_activity_provenance"] == provenance.value
+        assert summary["provenance"] == provenance.value
+        assert summary["last_activity_description"] == desc
+        assert summary["last_activity_desc"] == desc

tests/test_hermes_state.py

@@ -8,6 +8,7 @@ from unittest import mock
 import pytest
 
 import hermes_state
+from agent.session_activity import ActivityProvenance
 from hermes_state import SCHEMA_SQL, SCHEMA_VERSION, SessionDB
 
 
@@ -4674,6 +4675,110 @@ class TestListSessionsRich:
         # No messages, so last_active falls back to started_at
         assert sessions[0]["last_active"] == sessions[0]["started_at"]
 
+    def test_last_active_prefers_session_activity_heartbeat(self, db):
+        """Mid-turn agent heartbeats must advance last_active without new messages (#72016)."""
+        db.create_session("s1", "cli")
+        db.append_message("s1", "user", "hello")
+        with db._lock:
+            db._conn.execute(
+                "UPDATE messages SET timestamp=? WHERE session_id=? AND role=?",
+                (1_700_000_000.0, "s1", "user"),
+            )
+            db._conn.commit()
+
+        before = db.list_sessions_rich()[0]["last_active"]
+        heartbeat = 1_700_000_500.0
+        db.touch_session_activity(
+            "s1",
+            heartbeat,
+            description="starting API call #1",
+            provenance=ActivityProvenance.UNKNOWN,
+        )
+        after = db.list_sessions_rich()[0]["last_active"]
+        assert after == heartbeat
+        assert after > before
+
+        row = db.get_session("s1")
+        assert row["last_activity_at"] == heartbeat
+        assert row["last_activity_description"] == "starting API call #1"
+        assert row["last_activity_provenance"] == "unknown"
+
+        activity = db.get_session_activity("s1")
+        assert activity["last_activity_at"] == heartbeat
+        assert activity["last_activity_description"] == "starting API call #1"
+        assert "phase" not in activity
+
+        # Never move last_activity_at backwards.
+        db.touch_session_activity("s1", heartbeat - 100, description="ignored")
+        assert db.get_session("s1")["last_activity_at"] == heartbeat
+        assert db.get_session("s1")["last_activity_description"] == "starting API call #1"
+
+    def test_clear_session_activity_labels_keeps_timestamp(self, db):
+        """Turn-end label clear must wipe desc/provenance without moving ts."""
+        db.create_session("s1", "cli")
+        heartbeat = 1_700_000_500.0
+        db.touch_session_activity(
+            "s1",
+            heartbeat,
+            description="compressing context",
+            provenance=ActivityProvenance.AGENT_COMPRESSION,
+        )
+        row = db.get_session("s1")
+        assert row["last_activity_at"] == heartbeat
+        assert row["last_activity_description"] == "compressing context"
+        assert row["last_activity_provenance"] == "agent.compression"
+
+        db.clear_session_activity_labels("s1")
+        row = db.get_session("s1")
+        assert row["last_activity_at"] == heartbeat
+        assert row["last_activity_description"] == ""
+        assert row["last_activity_provenance"] == "unknown"
+        activity = db.get_session_activity("s1")
+        assert activity["last_activity_at"] == heartbeat
+        assert activity["last_activity_description"] == ""
+        assert activity["last_activity_provenance"] == "unknown"
+
+    def test_last_active_uses_newer_message_over_stale_heartbeat(self, db):
+        """Rate-limited heartbeats can lag message writes; last_active must take max."""
+        db.create_session("s1", "cli")
+        db.append_message("s1", "user", "hello")
+        with db._lock:
+            db._conn.execute(
+                "UPDATE messages SET timestamp=? WHERE session_id=?",
+                (1_700_000_800.0, "s1"),
+            )
+            db._conn.commit()
+        db.touch_session_activity("s1", 1_700_000_500.0, description="api")  # older than message
+        assert db.list_sessions_rich()[0]["last_active"] == 1_700_000_800.0
+
+    def test_list_gateway_sessions_last_active_uses_activity_heartbeat(self, db):
+        db.create_session(
+            "gw-1",
+            "telegram",
+            session_key="agent:main:telegram:dm:c1",
+            chat_id="c1",
+            chat_type="dm",
+        )
+        db.append_message("gw-1", "user", "ping")
+        with db._lock:
+            db._conn.execute(
+                "UPDATE messages SET timestamp=? WHERE session_id=?",
+                (1_700_000_000.0, "gw-1"),
+            )
+            db._conn.commit()
+
+        heartbeat = 1_700_000_900.0
+        db.touch_session_activity(
+            "gw-1",
+            heartbeat,
+            description="compressing context",
+        )
+        rows = db.list_gateway_sessions(active_only=True)
+        assert len(rows) == 1
+        assert rows[0]["last_active"] == heartbeat
+        activity = db.get_session_activity("gw-1")
+        assert activity["last_activity_description"] == "compressing context"
+
     def test_order_by_last_active_surfaces_recently_touched_older_session_first(self, db):
         t0 = 1709500000.0
         db.create_session("old", "cli")

website/docs/user-guide/configuration.md

@@ -754,6 +754,8 @@ compression:
   hygiene_timeout_seconds: 30                       # Max seconds of NO summary-model output before hygiene compression is cut off
   hygiene_total_ceiling_seconds: 600                # Absolute cap on the hygiene wait even while tokens are still streaming
   hygiene_failure_cooldown_seconds: 300             # Skip repeated failed hygiene attempts for this session
+  context_timeout_seconds: 120                      # Inactivity budget for in-agent compress_context (loop /compress / preflight) — see below
+  context_total_ceiling_seconds: 600                # Absolute cap on in-agent compress_context wait even while tokens are still streaming
   proactive_prune_tokens: 0                         # Opt-in tokens trigger for the no-LLM tool-result prune (0 = off; see below)
   proactive_prune_min_result_chars: 8000            # Prune's summarize pass only touches tool results larger than this (clamped >= 200)
   proactive_prune_min_reclaim_tokens: 4096          # Prune only commits when it reclaims at least this many tokens (0 = commit any)
@@ -780,6 +782,10 @@ Older configs with `compression.summary_model`, `compression.summary_provider`,
 
 `hygiene_failure_cooldown_seconds` controls that per-session cooldown after a hygiene compression timeout or abort. During the cooldown, the gateway skips repeated hygiene attempts for the same oversized session so every incoming message does not block on the same broken auxiliary backend. `/compress`, `/reset`, or a healthy later turn can still recover the session.
 
+`context_timeout_seconds` (default `120`) is the same **inactivity budget** for in-agent `compress_context` — the conversation loop, preflight compaction, and manual `/compress` — so a hung summary model cannot stall a session indefinitely. Streamed summary tokens extend the wait; only a silent worker is cut off. On timeout Hermes skips compaction, keeps the existing messages, and warns the user. Set to `0` to disable. Gateway session hygiene keeps its own `hygiene_timeout_seconds` path and is not double-wrapped.
+
+`context_total_ceiling_seconds` (default `600`) bounds the in-agent wait even while tokens are still moving. It is clamped to at least `context_timeout_seconds`.
+
 `protect_first_n` controls how many **non-system** head messages are pinned across every compaction. Default `3` — the opening user/assistant exchange survives every summarizer pass so the original goal stays visible. On long-running rolling-compaction sessions where the opening turn is no longer relevant, set `protect_first_n: 0` to pin nothing but the system prompt + summary + tail. The system prompt itself is always preserved regardless of this setting.
 
 `threshold_tokens` sets an optional **absolute token cap** for the compression trigger. When set, compression fires at the lower of the ratio-based `threshold` and this absolute count — so compression never fires later than the user's preferred token number regardless of which model is active. This solves the problem where switching between models with different context windows (e.g. 1M → 400K) shifts the absolute trigger point. The cap is clamped to the model's context length, so setting it higher than the model supports is safe — the ratio-based threshold is used instead. Default `null` (disabled — ratio-based threshold only). The cap survives model switches and fallback activations.

website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/configuration.md

@@ -556,6 +556,8 @@ compression:
   target_ratio: 0.20                                # 保留为最近尾部的阈值分数
   protect_last_n: 20                                # 保持未压缩的最少最近消息数
   hygiene_hard_message_limit: 5000                  # Gateway 安全阀 —— 见下文
+  context_timeout_seconds: 120                      # Agent 侧 compress_context 无进展超时（秒）—— 见下文
+  context_total_ceiling_seconds: 600                # Agent 侧 compress_context 总等待上限（秒）
 
 # 摘要模型/provider 在 auxiliary: 下配置：
 auxiliary:
@@ -571,6 +573,10 @@ auxiliary:
 
 `hygiene_hard_message_limit` 是仅限 gateway 的**预压缩安全阀**。它的存在是为了打破一个死循环：当超大会话的 API 调用持续断开时，gateway 永远收不到 token 使用数据，基于 token 的阈值因此无法触发，于是 transcript 持续增长、断开愈发严重。这个基于消息数的下限仅凭消息数量触发（无论 API 是否失败，消息数始终已知），强制压缩以恢复会话。默认 `5000` —— 远高于任何正常会话，包括做数千次短轮次的大上下文（1M+）模型，它们早就在 token 阈值处压缩了。对于异常平台可调得更高；要强制更积极的压缩则调低。在运行中的 gateway 上编辑此值将在下一条消息时生效（见下文）。
 
+`context_timeout_seconds`（默认 `120`）是 agent 侧 `compress_context`（对话循环、预检压缩、手动 `/compress`）的**无进展超时**，语义与 gateway 会话预压缩（session hygiene）的 inactivity 预算相同：摘要模型仍在流式出 token 时会延长等待；仅当完全无输出时才跳过压缩并保留原消息。设为 `0` 可关闭。Gateway 会话预压缩仍使用自己的 `hygiene_timeout_seconds`，不会被双重包装。
+
+`context_total_ceiling_seconds`（默认 `600`）限制即使仍有 token 推进时的 agent 侧总等待时间，并会被钳制为至少等于 `context_timeout_seconds`。
+
 :::tip Gateway 热重载压缩和上下文长度
 从最近的版本开始，在运行中的 gateway 上编辑 `config.yaml` 中的 `model.context_length` 或任何 `compression.*` 键将在下一条消息时生效 —— 无需 gateway 重启、`/reset` 或会话轮换。缓存的 agent 签名包含这些键，因此 gateway 在检测到更改时会透明地重建 agent。API 密钥和工具/技能配置仍需要通常的重载路径。
 :::