diff --git a/agent/agent_init.py b/agent/agent_init.py
index 95161f8cdee5..b37a0409d686 100644
--- a/agent/agent_init.py
+++ b/agent/agent_init.py
@@ -33,6 +33,7 @@
 from agent.context_compressor import ContextCompressor
 from agent.iteration_budget import IterationBudget
 from agent.memory_manager import StreamingContextScrubber
+from agent.session_activity import ActivityProvenance
 from agent.model_metadata import (
     MINIMUM_CONTEXT_LENGTH,
     fetch_model_metadata,
@@ -860,6 +861,11 @@ def init_agent(
     # notifications to show progress.
     agent._last_activity_ts: float = time.time()
     agent._last_activity_desc: str = "initializing"
+    # Default / unmigrated paths and _touch_activity stamp unknown; named
+    # provenances are reserved for special writers (e.g. #72424 compression).
+    agent._last_activity_provenance = ActivityProvenance.UNKNOWN
+    # Rate-limit durable SessionDB activity stamps from _touch_activity (#72016).
+    agent._session_activity_last_persist_mono: float = 0.0
     agent._current_tool: str | None = None
     agent._api_call_count: int = 0
     # Opt-out flag for the between-turns MCP tool refresh (build_turn_context).
diff --git a/agent/session_activity.py b/agent/session_activity.py
new file mode 100644
index 000000000000..585d452a53ec
--- /dev/null
+++ b/agent/session_activity.py
@@ -0,0 +1,82 @@
+"""Shared session activity observation contract (#72016 / #72039).
+
+Observation-only: timestamp + bounded description/provenance.
+Notification, timeout, kill, and retry policy stay in their own components.
+Consumers distinguish work (API / tool / compacting / stalled) from the
+description text itself — there is no separate phase enum.
+
+Provenance is a small closed enum of *noun* sources (where the stamp came
+from). The default agent activity clock (``_touch_activity``) stamps
+``unknown`` unless a caller passes an explicit ``provenance=``; named
+values are for special writers.
+"""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Mapping, Optional
+
+ACTIVITY_DESCRIPTION_MAX = 120
+
+
+class ActivityProvenance(str, Enum):
+    """Where a durable/in-memory activity stamp came from."""
+
+    UNKNOWN = "unknown"
+    # Reserved for #72424 writers; not stamped by #72039 call sites yet.
+    AGENT_COMPRESSION = "agent.compression"
+    AGENT_COMPRESSION_TIMEOUT = "agent.compression_timeout"
+    AGENT_COMPRESSION_COOLDOWN = "agent.compression_cooldown"
+
+
+def bound_activity_description(description: Optional[str]) -> str:
+    """Clamp free-form activity text to the shared description budget."""
+    text = (description or "").strip()
+    if len(text) <= ACTIVITY_DESCRIPTION_MAX:
+        return text
+    return text[: ACTIVITY_DESCRIPTION_MAX - 1] + "…"
+
+
+def normalize_activity_provenance(
+    provenance: Optional[ActivityProvenance | str],
+) -> ActivityProvenance:
+    """Return a known provenance, or ``UNKNOWN`` when unset/unrecognized."""
+    if isinstance(provenance, ActivityProvenance):
+        return provenance
+    value = (provenance or "").strip()
+    try:
+        return ActivityProvenance(value)
+    except ValueError:
+        return ActivityProvenance.UNKNOWN
+
+
+def build_activity_snapshot(
+    *,
+    last_activity_at: Optional[float],
+    last_activity_description: Optional[str],
+    last_activity_provenance: Optional[ActivityProvenance | str] = None,
+    now: Optional[float] = None,
+    extra: Optional[Mapping[str, Any]] = None,
+) -> dict[str, Any]:
+    """Build the shared activity snapshot (plus optional caller extras)."""
+    import time as _time
+
+    when = float(last_activity_at) if last_activity_at is not None else None
+    clock = float(now if now is not None else _time.time())
+    desc = bound_activity_description(last_activity_description)
+    prov = normalize_activity_provenance(last_activity_provenance)
+    elapsed = round(clock - when, 1) if when is not None else None
+    snap: dict[str, Any] = {
+        "last_activity_at": when,
+        "last_activity_description": desc,
+        "last_activity_provenance": prov.value,
+        "seconds_since_activity": elapsed,
+        # Short aliases used by existing gateway/delegate readers.
+        "last_activity_ts": when,
+        "last_activity_desc": desc,
+        "description": desc,
+        "provenance": prov.value,
+    }
+    if extra:
+        snap.update(dict(extra))
+    return snap
diff --git a/gateway/run.py b/gateway/run.py
index 37dfd9d92cde..028db1f11fc9 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -19610,21 +19610,25 @@ def _evict_cached_agent(self, session_key: str) -> None:
     def _init_cached_agent_for_turn(agent: Any, interrupt_depth: int) -> None:
         """Reset per-turn state on a cached agent before a new turn starts.
 
-        Both _last_activity_ts and _last_activity_desc are only reset for
-        fresh external turns (depth 0); they are semantically paired —
-        desc describes the activity *at* ts, so updating one without the
-        other would make get_activity_summary() misleading.
-        For interrupt-recursive turns both are preserved so the inactivity
-        watchdog can accumulate stuck-turn idle time and fire the 30-min
-        timeout (#15654).  The depth-0 reset is still needed: a session
-        idle for 29 min would otherwise trip the watchdog before the new
-        turn makes its first API call (#9051).
+        ``_last_activity_ts``, ``_last_activity_desc``, and
+        ``_last_activity_provenance`` are only reset for fresh external
+        turns (depth 0); they are a semantic triple - description and
+        provenance describe the activity *at* ts, so updating one without
+        the others would make get_activity_summary() misleading.
+        For interrupt-recursive turns all three are preserved so the
+        inactivity watchdog can accumulate stuck-turn idle time and fire
+        the 30-min timeout (#15654).  The depth-0 reset is still needed:
+        a session idle for 29 min would otherwise trip the watchdog before
+        the new turn makes its first API call (#9051).
         """
         if interrupt_depth == 0:
+            from agent.session_activity import ActivityProvenance
+
             agent._last_activity_ts = time.time()
             agent._last_activity_desc = "starting new turn (cached)"
+            agent._last_activity_provenance = ActivityProvenance.UNKNOWN
             # Reset the SessionDB flush cursor so the new turn's messages are
-            # fully persisted — a stale value from the previous turn would
+            # fully persisted - a stale value from the previous turn would
             # cause `_flush_messages_to_session_db` to skip new rows (#44327).
             if hasattr(agent, "_last_flushed_db_idx"):
                 agent._last_flushed_db_idx = 0
diff --git a/hermes_cli/status.py b/hermes_cli/status.py
index 7537b85a1d92..fed6e723ebf8 100644
--- a/hermes_cli/status.py
+++ b/hermes_cli/status.py
@@ -60,6 +60,27 @@ def _format_iso_timestamp(value) -> str:
     return parsed.astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
 
 
+def _format_relative_ts(ts: float) -> str:
+    """Format an epoch timestamp as a short relative age for status output."""
+    if not ts:
+        return "?"
+    import time as _time
+    from datetime import datetime
+
+    delta = _time.time() - float(ts)
+    if delta < 60:
+        return "just now"
+    if delta < 3600:
+        return f"{int(delta / 60)}m ago"
+    if delta < 86400:
+        return f"{int(delta / 3600)}h ago"
+    if delta < 172800:
+        return "yesterday"
+    if delta < 604800:
+        return f"{int(delta / 86400)}d ago"
+    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
+
+
 def _configured_model_label(config: dict) -> str:
     """Return the configured default model from config.yaml."""
     model_cfg = config.get("model")
@@ -550,20 +571,29 @@ def _resolve_env(env_ref) -> str:
     # Gateway session count: state.db is the source of truth (#9006);
     # fall back to sessions.json for pre-migration installs.
     _session_count = None
+    _gateway_rows = []
     try:
         from hermes_state import SessionDB
         _db = SessionDB()
         try:
             _lister = getattr(_db, "list_gateway_sessions", None)
             if callable(_lister):
-                _session_count = len(_lister(active_only=True))
+                _gateway_rows = _lister(active_only=True) or []
+                _session_count = len(_gateway_rows)
         finally:
             _db.close()
     except Exception:
         _session_count = None
+        _gateway_rows = []
 
     if _session_count is not None and _session_count > 0:
         print(f"  Active:       {_session_count} session(s)")
+        freshest = max(
+            (float(r.get("last_active") or 0) for r in _gateway_rows),
+            default=0.0,
+        )
+        if freshest > 0:
+            print(f"  Last activity:{_format_relative_ts(freshest):>13}")
     else:
         sessions_file = get_hermes_home() / "sessions" / "sessions.json"
         if sessions_file.exists():
diff --git a/hermes_state.py b/hermes_state.py
index f309acce1277..facf7bb72477 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -28,6 +28,7 @@
 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 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
     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 @@ def list_gateway_sessions(
         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 @@ def get_compression_lock_holder(self, session_id: str) -> Optional[str]:
             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 @@ def get_compression_tip(self, session_id: str) -> Optional[str]:
         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 @@ def get_compression_tip(self, session_id: str) -> Optional[str]:
                         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 @@ def list_sessions_rich(
 
         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 @@ def _like_pattern(needle: str) -> str:
                 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 @@ def _like_pattern(needle: str) -> str:
                          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 @@ def _like_pattern(needle: str) -> str:
                          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 @@ def list_cron_job_runs(
                      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 @@ def _get_session_rich_row(self, session_id: str, compact_rows: bool = False) ->
                      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 @@ def search_sessions(
     ) -> 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 @@ def archive_stale_sessions(
     ) -> 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 @@ def archive_stale_sessions(
                 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 @@ def list_unlinked_telegram_sessions_for_user(
                              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 @@ def list_unlinked_telegram_sessions_for_user(
                              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 = ?
diff --git a/run_agent.py b/run_agent.py
index 703759d8ea62..801b659c2800 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -149,6 +149,7 @@ def _session_source_for_agent(platform: Optional[str]) -> str:
 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,
@@ -3452,7 +3453,12 @@ def _apply_pending_steer_to_tool_results(self, messages: list, num_tool_msgs: in
         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
@@ -3460,9 +3466,23 @@ def _touch_activity(self, desc: str) -> None:
         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
@@ -3473,6 +3493,66 @@ def _touch_activity(self, desc: str) -> None:
                 # 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.
@@ -3673,20 +3753,32 @@ def _check_openrouter_cache_status(self, http_response: Any) -> None:
     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.
@@ -6857,6 +6949,12 @@ def run_conversation(
                     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)
 
diff --git a/tests/agent/test_session_activity.py b/tests/agent/test_session_activity.py
new file mode 100644
index 000000000000..28ede5b64e4c
--- /dev/null
+++ b/tests/agent/test_session_activity.py
@@ -0,0 +1,60 @@
+"""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"
diff --git a/tests/gateway/test_agent_cache.py b/tests/gateway/test_agent_cache.py
index 29e98fc1c483..152da94ac0e6 100644
--- a/tests/gateway/test_agent_cache.py
+++ b/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 @@ def test_fresh_turn_resets_desc(self):
 
         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 @@ def test_interrupt_turn_preserves_desc(self):
             "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
diff --git a/tests/gateway/test_cached_agent_max_iterations.py b/tests/gateway/test_cached_agent_max_iterations.py
index fcd523c70ef6..74e85f16972b 100644
--- a/tests/gateway/test_cached_agent_max_iterations.py
+++ b/tests/gateway/test_cached_agent_max_iterations.py
@@ -21,6 +21,7 @@
 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
 
diff --git a/tests/hermes_cli/test_status.py b/tests/hermes_cli/test_status.py
index 8b09dd8377a3..bbbe01d08b9f 100644
--- a/tests/hermes_cli/test_status.py
+++ b/tests/hermes_cli/test_status.py
@@ -352,3 +352,43 @@ def test_status_function_returns_none_does_not_crash(self, monkeypatch, capsys,
 
         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
diff --git a/tests/run_agent/test_session_activity_persist.py b/tests/run_agent/test_session_activity_persist.py
new file mode 100644
index 000000000000..ce1be14a685d
--- /dev/null
+++ b/tests/run_agent/test_session_activity_persist.py
@@ -0,0 +1,185 @@
+"""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
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index e99785ae5643..77b92fe3c9ba 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -8,6 +8,7 @@
 import pytest
 
 import hermes_state
+from agent.session_activity import ActivityProvenance
 from hermes_state import SCHEMA_SQL, SCHEMA_VERSION, SessionDB
 
 
@@ -4674,6 +4675,110 @@ def test_last_active_fallback_to_started_at(self, db):
         # 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")