From c40b1ca1d2c2875a145e888c43147ed4306f1c17 Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Sun, 26 Jul 2026 22:56:52 +0800 Subject: [PATCH 1/7] fix(hermes_state): persist mid-turn session activity for CLI listings Stamp sessions.last_activity_at from AIAgent._touch_activity (rate-limited) so hermes sessions list and hermes status observe API/tool/compaction progress while a gateway turn is still writing no message rows (#72016). --- agent/agent_init.py | 2 + hermes_cli/status.py | 32 ++++- hermes_state.py | 110 ++++++++++-------- run_agent.py | 29 +++++ tests/hermes_cli/test_status.py | 39 +++++++ .../test_session_activity_persist.py | 67 +++++++++++ tests/test_hermes_state.py | 47 ++++++++ 7 files changed, 277 insertions(+), 49 deletions(-) create mode 100644 tests/run_agent/test_session_activity_persist.py diff --git a/agent/agent_init.py b/agent/agent_init.py index 955ca11cff8d..6ec42ab38869 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -860,6 +860,8 @@ def init_agent( # notifications to show progress. agent._last_activity_ts: float = time.time() agent._last_activity_desc: str = "initializing" + # 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/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 32fcffac585f..17cfb461dd90 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -212,6 +212,23 @@ 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. + + Preference order: + 1. ``last_activity_at`` — mid-turn agent heartbeat (API/tool/compaction) + 2. latest message timestamp + 3. ``started_at`` + """ + return ( + f"COALESCE(" + f"{alias}.last_activity_at, " + f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m " + f"WHERE _act_m.session_id = {alias}.id), " + f"{alias}.started_at)" + ) + + def _collect_delegate_child_ids(conn, parent_ids: List[str]) -> List[str]: """Delegate-subagent ids to cascade-delete with *parent_ids*. @@ -1164,6 +1181,7 @@ 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, api_call_count INTEGER DEFAULT 0, handoff_state TEXT, handoff_platform TEXT, @@ -3984,13 +4002,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 = ( @@ -4811,6 +4825,33 @@ 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, + ) -> None: + """Stamp ``sessions.last_activity_at`` for mid-turn agent liveness. + + Called (rate-limited) from ``AIAgent._touch_activity`` so gateway/CLI + session listings and ``hermes status`` observe API/tool/compaction + progress even when no new message row has been written yet (#72016). + + Never moves the timestamp backwards. No-ops when ``session_id`` is + empty or the row does not exist. + """ + if not session_id: + return + when = float(ts if ts is not None else time.time()) + + def _do(conn): + conn.execute( + "UPDATE sessions SET last_activity_at = ? " + "WHERE id = ? AND (last_activity_at IS NULL OR last_activity_at < ?)", + (when, session_id, when), + ) + + self._execute_write(_do) + def update_session_meta( self, session_id: str, @@ -5774,14 +5815,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 @@ -5789,10 +5830,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 @@ -5878,7 +5916,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 (timestamp of last agent activity heartbeat, else last + message, else started_at). Uses a single query with correlated subqueries instead of N+2 queries. @@ -6047,6 +6086,7 @@ def _like_pattern(needle: str) -> str: SELECT root_id, MAX(COALESCE( + (SELECT last_activity_at FROM sessions ss WHERE ss.id = cur_id), (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 @@ -6061,10 +6101,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 @@ -6086,10 +6123,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 @@ -6184,10 +6218,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 @@ -6222,10 +6253,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 = ? """ @@ -8664,12 +8692,8 @@ def search_sessions( ordered by most-recently-used first. """ 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 " ) with self._lock: if source: @@ -9821,11 +9845,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,), @@ -10401,10 +10421,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 = ? @@ -10430,10 +10447,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 1adf170e5e01..0e7055bf5343 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3438,6 +3438,11 @@ 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 ``sessions.last_activity_at`` + stamp via SessionDB so ``hermes sessions list`` / ``hermes status`` + observe mid-turn API/tool/compaction progress across process + boundaries (#72016). """ self._last_activity_ts = time.time() self._last_activity_desc = desc @@ -3451,7 +3456,31 @@ 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 listings. + 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: + touch(session_id, getattr(self, "_last_activity_ts", None)) + except Exception: + # Never let durable heartbeat I/O break the agent loop. + pass def _capture_rate_limits(self, http_response: Any) -> None: """Parse x-ratelimit-* headers from an HTTP response and cache the state. diff --git a/tests/hermes_cli/test_status.py b/tests/hermes_cli/test_status.py index 8b09dd8377a3..a7d3b87ea455 100644 --- a/tests/hermes_cli/test_status.py +++ b/tests/hermes_cli/test_status.py @@ -352,3 +352,42 @@ 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..82bbcf0860ac --- /dev/null +++ b/tests/run_agent/test_session_activity_persist.py @@ -0,0 +1,67 @@ +"""Durable session activity heartbeats from AIAgent._touch_activity (#72016).""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import run_agent + + +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="", + _session_activity_last_persist_mono=0.0, + ) + 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) + ) + return agent + + +def test_touch_activity_persists_session_heartbeat_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 + ) + + 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 + ) + + +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" + + +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)" diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 6f3704fa3e23..e6ee1f40f3c9 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -4649,6 +4649,53 @@ 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) + 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 + + # Never move last_activity_at backwards. + db.touch_session_activity("s1", heartbeat - 100) + assert db.get_session("s1")["last_activity_at"] == heartbeat + + 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) + rows = db.list_gateway_sessions(active_only=True) + assert len(rows) == 1 + assert rows[0]["last_active"] == heartbeat + def test_order_by_last_active_surfaces_recently_touched_older_session_first(self, db): t0 = 1709500000.0 db.create_session("old", "cli") From 45b165f449ff48abe6ac38069427c71aaa1915c0 Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Sun, 26 Jul 2026 23:02:47 +0800 Subject: [PATCH 2/7] fix(hermes_state): take freshest of activity heartbeat and message time Prefer max(last_activity_at, latest message) for session last_active so a rate-limited mid-turn heartbeat cannot rank a session behind newer message timestamps in listings, status, and idle archive. --- hermes_state.py | 59 ++++++++++++++++++++++++++++---------- run_agent.py | 1 + tests/test_hermes_state.py | 13 +++++++++ 3 files changed, 58 insertions(+), 15 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index 17cfb461dd90..b925b07c2f66 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -215,20 +215,53 @@ 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. - Preference order: - 1. ``last_activity_at`` — mid-turn agent heartbeat (API/tool/compaction) - 2. latest message timestamp - 3. ``started_at`` + Freshest of ``last_activity_at`` (mid-turn agent 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"{alias}.last_activity_at, " - f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m " - f"WHERE _act_m.session_id = {alias}.id), " + 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*. @@ -6085,11 +6118,7 @@ def _like_pattern(needle: str) -> str: chain_max AS ( SELECT root_id, - MAX(COALESCE( - (SELECT last_activity_at FROM sessions ss WHERE ss.id = cur_id), - (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 ) @@ -8687,9 +8716,9 @@ 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. """ select_with_last_active = ( f"SELECT s.*, {_sql_session_last_active('s')} AS last_active " diff --git a/run_agent.py b/run_agent.py index 0e7055bf5343..e0a16fd6f45b 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3481,6 +3481,7 @@ def _persist_session_activity_if_due(self) -> None: except Exception: # Never let durable heartbeat I/O break the agent loop. pass + def _capture_rate_limits(self, http_response: Any) -> None: """Parse x-ratelimit-* headers from an HTTP response and cache the state. diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index e6ee1f40f3c9..e3826b13e5fe 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -4674,6 +4674,19 @@ def test_last_active_prefers_session_activity_heartbeat(self, db): db.touch_session_activity("s1", heartbeat - 100) assert db.get_session("s1")["last_activity_at"] == heartbeat + 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) # 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", From 9a8f9ca3feec316c9e05a491687b0fb6305bdda3 Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Sun, 26 Jul 2026 23:06:07 +0800 Subject: [PATCH 3/7] fix(hermes_state): align list_sessions_rich docs with freshest-of last_active Document that last_active is max(heartbeat, message time), and separate the status last-activity test from the preceding class with PEP8 spacing. --- hermes_state.py | 4 ++-- tests/hermes_cli/test_status.py | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index b925b07c2f66..344368958963 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -5949,8 +5949,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 agent activity heartbeat, else last - message, else started_at). + 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. diff --git a/tests/hermes_cli/test_status.py b/tests/hermes_cli/test_status.py index a7d3b87ea455..bbbe01d08b9f 100644 --- a/tests/hermes_cli/test_status.py +++ b/tests/hermes_cli/test_status.py @@ -353,6 +353,7 @@ 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 @@ -390,4 +391,4 @@ def close(self): output = capsys.readouterr().out assert "Active: 2 session(s)" in output assert "Last activity:" in output - assert "1m ago" in output + assert "1m ago" in output From bf6329ddbcf555765654b5258fdda2b7b924d0cc Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Sun, 26 Jul 2026 23:09:30 +0800 Subject: [PATCH 4/7] fix(hermes_state): document freshest-of touch for idle session archive Clarify that archive_stale_sessions ages on max(last_activity_at, latest message), matching the shared last_active SQL helper. --- hermes_state.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index 344368958963..1abc42f258a5 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -9844,8 +9844,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. This differs from :meth:`archive_sessions`, which ages on ``started_at`` and only ended From f83a16d316460ca0ed1456335b2455e15ba15162 Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Mon, 27 Jul 2026 19:00:51 +0800 Subject: [PATCH 5/7] feat(agent): persist bounded activity description and provenance Extend the shared session activity observation contract so durable SessionDB heartbeats carry description/provenance alongside the timestamp, matching in-memory get_activity_summary for #72039 consumers. --- agent/agent_init.py | 4 + agent/session_activity.py | 82 ++++++++++++++++++ hermes_state.py | 51 +++++++++-- run_agent.py | 85 ++++++++++++++----- tests/agent/test_session_activity.py | 60 +++++++++++++ .../test_session_activity_persist.py | 73 +++++++++++++++- tests/test_hermes_state.py | 28 +++++- 7 files changed, 344 insertions(+), 39 deletions(-) create mode 100644 agent/session_activity.py create mode 100644 tests/agent/test_session_activity.py diff --git a/agent/agent_init.py b/agent/agent_init.py index 6ec42ab38869..6dd3a47e5e2e 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,9 @@ 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 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/hermes_state.py b/hermes_state.py index 1abc42f258a5..243a0d0521fd 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, @@ -215,8 +216,8 @@ 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 heartbeat) and the - latest message timestamp, then fall back to ``started_at``. + 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 @@ -1215,6 +1216,8 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A 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, @@ -4862,29 +4865,59 @@ def touch_session_activity( self, session_id: str, ts: Optional[float] = None, + *, + description: Optional[str] = None, + provenance: Optional[ActivityProvenance] = None, ) -> None: - """Stamp ``sessions.last_activity_at`` for mid-turn agent liveness. + """Stamp durable mid-turn session activity (observation-only). Called (rate-limited) from ``AIAgent._touch_activity`` so gateway/CLI - session listings and ``hermes status`` observe API/tool/compaction - progress even when no new message row has been written yet (#72016). + surfaces and stall consumers observe API/tool/compaction activity + even when no new message row has been written yet (#72016 / #72039). - Never moves the timestamp backwards. No-ops when ``session_id`` is - empty or the row does not exist. + 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 = ? " + "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, session_id, when), + (when, desc, prov, session_id, when), ) 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, diff --git a/run_agent.py b/run_agent.py index e0a16fd6f45b..a96f05d4aa91 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, @@ -3430,7 +3431,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 @@ -3439,13 +3445,22 @@ def _touch_activity(self, desc: str) -> None: worker as stale (#31752). Bridge is rate-limited (60s) and best-effort — it never raises into the agent loop. - Separately, rate-limits a durable ``sessions.last_activity_at`` - stamp via SessionDB so ``hermes sessions list`` / ``hermes status`` - observe mid-turn API/tool/compaction progress across process - boundaries (#72016). + 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 @@ -3459,7 +3474,7 @@ def _touch_activity(self, desc: str) -> None: self._persist_session_activity_if_due() def _persist_session_activity_if_due(self) -> None: - """Best-effort durable activity heartbeat for SessionDB listings. + """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. @@ -3477,11 +3492,25 @@ def _persist_session_activity_if_due(self) -> None: return self._session_activity_last_persist_mono = now_mono try: - touch(session_id, getattr(self, "_last_activity_ts", None)) + 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 TypeError: + # Older SessionDB stubs may only accept (session_id, ts). + try: + touch(session_id, getattr(self, "_last_activity_ts", None)) + except Exception: + pass except Exception: # Never let durable heartbeat I/O break the agent loop. pass - def _capture_rate_limits(self, http_response: Any) -> None: """Parse x-ratelimit-* headers from an HTTP response and cache the state. @@ -3681,20 +3710,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. 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/run_agent/test_session_activity_persist.py b/tests/run_agent/test_session_activity_persist.py index 82bbcf0860ac..05e70c5973f6 100644 --- a/tests/run_agent/test_session_activity_persist.py +++ b/tests/run_agent/test_session_activity_persist.py @@ -1,9 +1,10 @@ -"""Durable session activity heartbeats from AIAgent._touch_activity (#72016).""" +"""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"): @@ -12,16 +13,24 @@ def _agent_with_db(session_id: str = "sess-1"): _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.get_activity_summary = run_agent.AIAgent.get_activity_summary.__get__( + agent, SimpleNamespace + ) return agent -def test_touch_activity_persists_session_heartbeat_once_per_minute(monkeypatch): +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) @@ -30,7 +39,10 @@ def test_touch_activity_persists_session_heartbeat_once_per_minute(monkeypatch): agent._touch_activity("starting API call #1") agent._session_db.touch_session_activity.assert_called_once_with( - "sess-1", 1_700_000_000.0 + "sess-1", + 1_700_000_000.0, + description="starting API call #1", + provenance=ActivityProvenance.UNKNOWN, ) agent._session_db.touch_session_activity.reset_mock() @@ -41,7 +53,10 @@ def test_touch_activity_persists_session_heartbeat_once_per_minute(monkeypatch): 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 + "sess-1", + 1_700_000_000.0, + description="API call #1 completed", + provenance=ActivityProvenance.UNKNOWN, ) @@ -54,6 +69,37 @@ def test_touch_activity_skips_persist_without_session_db(monkeypatch): 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): @@ -65,3 +111,22 @@ def test_touch_activity_persist_errors_are_swallowed(monkeypatch): 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 diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index e3826b13e5fe..77b40c0f5946 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 @@ -4662,17 +4663,30 @@ def test_last_active_prefers_session_activity_heartbeat(self, db): before = db.list_sessions_rich()[0]["last_active"] heartbeat = 1_700_000_500.0 - db.touch_session_activity("s1", heartbeat) + 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) + 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_last_active_uses_newer_message_over_stale_heartbeat(self, db): """Rate-limited heartbeats can lag message writes; last_active must take max.""" @@ -4684,7 +4698,7 @@ def test_last_active_uses_newer_message_over_stale_heartbeat(self, db): (1_700_000_800.0, "s1"), ) db._conn.commit() - db.touch_session_activity("s1", 1_700_000_500.0) # older than message + 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): @@ -4704,10 +4718,16 @@ def test_list_gateway_sessions_last_active_uses_activity_heartbeat(self, db): db._conn.commit() heartbeat = 1_700_000_900.0 - db.touch_session_activity("gw-1", heartbeat) + 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 From f2c28e7e966ce35ea83274acfae60fc51e5fb78e Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Mon, 27 Jul 2026 19:05:44 +0800 Subject: [PATCH 6/7] feat(gateway): reset activity provenance on cached turns Keep gateway cache turn resets on the ts/desc/provenance triple, and drop the TypeError persist fallback that could silently omit description/provenance. --- gateway/run.py | 24 +++++++++------- run_agent.py | 7 +---- tests/gateway/test_agent_cache.py | 28 +++++++++++++++++++ .../test_cached_agent_max_iterations.py | 4 +++ 4 files changed, 47 insertions(+), 16 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index aac6a192555f..25e4f3cc7d79 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -19393,21 +19393,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/run_agent.py b/run_agent.py index a96f05d4aa91..03352011512d 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3502,15 +3502,10 @@ def _persist_session_activity_if_due(self) -> None: getattr(self, "_last_activity_provenance", None) ), ) - except TypeError: - # Older SessionDB stubs may only accept (session_id, ts). - try: - touch(session_id, getattr(self, "_last_activity_ts", None)) - except Exception: - pass except Exception: # Never let durable heartbeat I/O break the agent loop. pass + def _capture_rate_limits(self, http_response: Any) -> None: """Parse x-ratelimit-* headers from an HTTP response and cache the state. 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 From b69da0e27f3fdb27a66fc67dc8338504b8e5384d Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Mon, 27 Jul 2026 19:29:49 +0800 Subject: [PATCH 7/7] feat(agent): clear mid-turn activity labels when a turn ends Keep last_activity_at for idle/watchdog continuity, but wipe description/provenance in memory and SessionDB so idle sessions do not keep advertising the last mid-turn stamp. --- hermes_state.py | 24 +++++++++ run_agent.py | 32 +++++++++++ .../test_session_activity_persist.py | 53 +++++++++++++++++++ tests/test_hermes_state.py | 25 +++++++++ 4 files changed, 134 insertions(+) diff --git a/hermes_state.py b/hermes_state.py index 243a0d0521fd..1397b45051e8 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -4903,6 +4903,30 @@ def _do(conn): 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: diff --git a/run_agent.py b/run_agent.py index 03352011512d..8b97ab501e0f 100644 --- a/run_agent.py +++ b/run_agent.py @@ -3506,6 +3506,32 @@ def _persist_session_activity_if_due(self) -> None: # 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. @@ -6822,6 +6848,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/run_agent/test_session_activity_persist.py b/tests/run_agent/test_session_activity_persist.py index 05e70c5973f6..ce1be14a685d 100644 --- a/tests/run_agent/test_session_activity_persist.py +++ b/tests/run_agent/test_session_activity_persist.py @@ -24,6 +24,9 @@ def _agent_with_db(session_id: str = "sess-1"): 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 ) @@ -130,3 +133,53 @@ def test_get_activity_summary_exposes_shared_activity_contract(monkeypatch): 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 77b40c0f5946..144881dcc813 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -4688,6 +4688,31 @@ def test_last_active_prefers_session_activity_heartbeat(self, db): 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")