mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-21 16:18:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7
hermes-agent/tests/gateway/test_session_reset_notify.py
Teknium 9fc0074bac fix(gateway): unify reset boundaries vs recovery — promote accidental ends, honor mode=none, adapter-aware resume guidance 
Unifies the two gateway subsystems that were fighting each other: the
'never lose a session' recovery machinery (#54878 stale-route self-heal,
find_latest_gateway_session_for_peer reopening agent_close/ws_orphan_reap
rows) and the session reset/expiry machinery (expiry watcher, /new,
/resume, resume_pending freshness gate).

The unified contract:
- INTENTIONAL boundaries (expiry finalization, auto-reset, /new,
  /resume switch) are recorded durably via promote_to_session_reset(),
  which upgrades accidental recoverable end_reasons (agent_close,
  ws_orphan_reap) to the explicit boundary while preserving other
  explicit reasons (compression, etc.). Recovery then correctly refuses
  to resurrect them.
- ACCIDENTAL ends (crash, cleanup bug, mistaken reaper) stay
  recoverable — genuine crash recovery is untouched.

On top of the cherry-picked contributor commits:
- promote_to_session_reset widened to ws_orphan_reap + parameterized
  reason so auto-reset paths stay auditable (idle/daily/suspended/
  resume_pending_expired) (#61220, #61993, #63539)
- get_or_create_session auto-reset, reset_session (/new), and
  switch_session (/resume) all write through the promote path — the
  first-reason-wins end_session no-op could previously leave a reset
  session resurrectable behind a stale agent_close row (#61993)
- resume_pending freshness gate now honors session_reset.mode=none:
  explicit opt-out of automatic resets also opts out of the zombie
  gate (#61052)
- resume recovery note extracted to build_resume_recovery_note() and
  made adapter-aware via a new interactive_resume capability flag:
  webhook/api_server auto-resume turns now CONTINUE the interrupted
  task instead of emitting an unanswerable 'session restored'
  acknowledgement that abandoned the work (#57056)
- tests updated to call the real note builder instead of mirroring it

E2E-validated against a real SessionDB + SessionStore in a temp
HERMES_HOME: expiry->agent_close->no-resurrection, /new promote,
crash recovery preserved, mode=none opt-out, routing-table flag sync.
2026-07-17 04:51:39 -07:00

482 lines
18 KiB
Python
Raw Blame History

"""Tests for session auto-reset notifications.
Verifies that:
- _should_reset() returns a reason string ("idle" or "daily") instead of bool
- SessionEntry captures auto_reset_reason
- SessionResetPolicy.notify controls whether notifications are sent
- notify_exclude_platforms skips notifications for excluded platforms
- resume_pending_expired auto-reset sets the correct reason and DB end_reason
"""
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
from gateway.config import (
GatewayConfig,
Platform,
SessionResetPolicy,
)
from gateway.session import SessionEntry, SessionSource, SessionStore
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_source(platform=Platform.TELEGRAM, chat_id="123", user_id="u1"):
return SessionSource(
platform=platform,
chat_id=chat_id,
user_id=user_id,
)
def _make_store(policy=None, tmp_path=None, has_active_processes_fn=None):
config = GatewayConfig()
if policy:
config.default_reset_policy = policy
store = SessionStore(
sessions_dir=tmp_path or "/tmp/test-sessions",
config=config,
has_active_processes_fn=has_active_processes_fn,
)
return store
# ---------------------------------------------------------------------------
# _should_reset returns reason string
# ---------------------------------------------------------------------------
class TestShouldResetReason:
def test_returns_none_when_not_expired(self, tmp_path):
store = _make_store(
SessionResetPolicy(mode="both", idle_minutes=60, at_hour=4),
tmp_path,
)
entry = SessionEntry(
session_key="test",
session_id="s1",
created_at=datetime.now(),
updated_at=datetime.now(), # just updated
)
source = _make_source()
assert store._should_reset(entry, source) is None
def test_returns_idle_when_idle_expired(self, tmp_path):
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=30),
tmp_path,
)
entry = SessionEntry(
session_key="test",
session_id="s1",
created_at=datetime.now() - timedelta(hours=2),
updated_at=datetime.now() - timedelta(hours=1), # 60min ago > 30min threshold
)
source = _make_source()
assert store._should_reset(entry, source) == "idle"
def test_returns_daily_when_daily_boundary_crossed(self, tmp_path):
now = datetime.now()
store = _make_store(
SessionResetPolicy(mode="daily", at_hour=now.hour),
tmp_path,
)
entry = SessionEntry(
session_key="test",
session_id="s1",
created_at=now - timedelta(days=2),
updated_at=now - timedelta(days=1), # last active yesterday
)
source = _make_source()
assert store._should_reset(entry, source) == "daily"
def test_returns_none_when_mode_is_none(self, tmp_path):
store = _make_store(
SessionResetPolicy(mode="none"),
tmp_path,
)
entry = SessionEntry(
session_key="test",
session_id="s1",
created_at=datetime.now() - timedelta(days=30),
updated_at=datetime.now() - timedelta(days=30),
)
source = _make_source()
assert store._should_reset(entry, source) is None
def test_returns_none_when_active_process_check_raises(self, tmp_path):
def _raise(_session_key):
raise RuntimeError("process registry unavailable")
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=30),
tmp_path,
has_active_processes_fn=_raise,
)
entry = SessionEntry(
session_key="test",
session_id="s1",
created_at=datetime.now() - timedelta(hours=2),
updated_at=datetime.now() - timedelta(hours=1),
)
source = _make_source()
assert store._should_reset(entry, source) is None
def test_is_session_expired_fails_closed_when_active_process_check_raises(self, tmp_path):
def _raise(_session_key):
raise RuntimeError("process registry unavailable")
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=30),
tmp_path,
has_active_processes_fn=_raise,
)
entry = SessionEntry(
session_key="test",
session_id="s1",
platform=Platform.TELEGRAM,
chat_type="dm",
created_at=datetime.now() - timedelta(hours=2),
updated_at=datetime.now() - timedelta(hours=1),
)
assert store._is_session_expired(entry) is False
# ---------------------------------------------------------------------------
# SessionEntry captures reason
# ---------------------------------------------------------------------------
class TestSessionEntryReason:
def test_auto_reset_reason_stored(self, tmp_path):
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=1),
tmp_path,
)
source = _make_source()
# Create initial session
entry1 = store.get_or_create_session(source)
assert not entry1.was_auto_reset
# Age it past the idle threshold
entry1.updated_at = datetime.now() - timedelta(minutes=5)
store._save()
# Next call should create a new session with reason
entry2 = store.get_or_create_session(source)
assert entry2.was_auto_reset is True
assert entry2.auto_reset_reason == "idle"
assert entry2.session_id != entry1.session_id
def test_reset_had_activity_false_when_no_tokens(self, tmp_path):
"""Expired session with no tokens → reset_had_activity=False."""
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=1),
tmp_path,
)
source = _make_source()
entry1 = store.get_or_create_session(source)
# No tokens used — session was idle with no conversation
entry1.updated_at = datetime.now() - timedelta(minutes=5)
store._save()
entry2 = store.get_or_create_session(source)
assert entry2.was_auto_reset is True
assert entry2.reset_had_activity is False
def test_reset_had_activity_true_when_tokens_used(self, tmp_path):
"""Expired session with tokens → reset_had_activity=True."""
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=1),
tmp_path,
)
source = _make_source()
entry1 = store.get_or_create_session(source)
# Simulate some conversation happened (last_prompt_tokens is the field
# written on every turn; total_tokens is never persisted).
entry1.last_prompt_tokens = 5000
entry1.updated_at = datetime.now() - timedelta(minutes=5)
store._save()
entry2 = store.get_or_create_session(source)
assert entry2.was_auto_reset is True
assert entry2.reset_had_activity is True
# ---------------------------------------------------------------------------
# SessionResetPolicy notify config
# ---------------------------------------------------------------------------
class TestResetPolicyNotify:
def test_notify_defaults_true(self):
policy = SessionResetPolicy()
assert policy.notify is True
def test_notify_exclude_defaults(self):
policy = SessionResetPolicy()
assert "api_server" in policy.notify_exclude_platforms
assert "webhook" in policy.notify_exclude_platforms
def test_from_dict_with_notify_false(self):
policy = SessionResetPolicy.from_dict({"notify": False})
assert policy.notify is False
def test_from_dict_with_custom_excludes(self):
policy = SessionResetPolicy.from_dict({
"notify_exclude_platforms": ["api_server", "webhook", "homeassistant"],
})
assert "homeassistant" in policy.notify_exclude_platforms
def test_from_dict_preserves_defaults_on_missing_keys(self):
policy = SessionResetPolicy.from_dict({})
assert policy.notify is True
assert "api_server" in policy.notify_exclude_platforms
def test_to_dict_roundtrip(self):
original = SessionResetPolicy(
mode="idle",
notify=False,
notify_exclude_platforms=("api_server",),
)
restored = SessionResetPolicy.from_dict(original.to_dict())
assert restored.notify == original.notify
assert restored.notify_exclude_platforms == original.notify_exclude_platforms
assert restored.mode == original.mode
# ---------------------------------------------------------------------------
# SessionEntry to_dict / from_dict roundtrip for auto-reset fields
# ---------------------------------------------------------------------------
class TestSessionEntryAutoResetRoundtrip:
def test_was_auto_reset_persists_across_roundtrip(self, tmp_path):
"""was_auto_reset=True survives to_dict() → from_dict() (gateway restart)."""
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=1),
tmp_path,
)
source = _make_source()
entry = store.get_or_create_session(source)
entry.updated_at = datetime.now() - timedelta(minutes=5)
store._save()
entry2 = store.get_or_create_session(source)
assert entry2.was_auto_reset is True
assert entry2.auto_reset_reason == "idle"
assert entry2.session_id != entry.session_id
# Simulate gateway restart: reload from disk
store._loaded = False
store._entries.clear()
store._ensure_loaded()
reloaded = store._entries.get(entry2.session_key)
assert reloaded is not None
assert reloaded.was_auto_reset is True
assert reloaded.auto_reset_reason == "idle"
def test_reset_had_activity_persists_across_roundtrip(self, tmp_path):
"""reset_had_activity survives to_dict() → from_dict() (gateway restart)."""
store = _make_store(
SessionResetPolicy(mode="idle", idle_minutes=1),
tmp_path,
)
source = _make_source()
entry = store.get_or_create_session(source)
entry.last_prompt_tokens = 1000
entry.updated_at = datetime.now() - timedelta(minutes=5)
store._save()
entry2 = store.get_or_create_session(source)
assert entry2.reset_had_activity is True
store._loaded = False
store._entries.clear()
store._ensure_loaded()
reloaded = store._entries.get(entry2.session_key)
assert reloaded is not None
assert reloaded.reset_had_activity is True
def test_auto_reset_reason_none_roundtrip(self, tmp_path):
"""auto_reset_reason=None (no reset) survives roundtrip cleanly."""
store = _make_store(tmp_path=tmp_path)
source = _make_source()
entry = store.get_or_create_session(source)
assert entry.was_auto_reset is False
store._loaded = False
store._entries.clear()
store._ensure_loaded()
reloaded = store._entries.get(entry.session_key)
assert reloaded is not None
assert reloaded.was_auto_reset is False
assert reloaded.auto_reset_reason is None
assert reloaded.reset_had_activity is False
# ---------------------------------------------------------------------------
# resume_pending_expired: auto_reset_reason and DB end_reason (#58933)
# ---------------------------------------------------------------------------
def _make_db_mock() -> MagicMock:
"""Return a SessionDB mock with safe defaults for all lookup methods."""
db = MagicMock()
db.get_session.return_value = None
db.get_compression_tip.return_value = None # avoids MagicMock leaking into session_id
db.find_latest_gateway_session_for_peer.return_value = None
db.reopen_session.return_value = None
db.create_session.return_value = None
return db
def _make_store_with_db(tmp_path, db_mock=None, policy=None) -> SessionStore:
"""Build a SessionStore with a mock SessionDB, bypassing disk load."""
cfg_policy = policy or SessionResetPolicy(mode="none")
config = GatewayConfig(default_reset_policy=cfg_policy)
with patch("gateway.session.SessionStore._ensure_loaded"):
store = SessionStore(sessions_dir=tmp_path, config=config)
store._db = db_mock if db_mock is not None else _make_db_mock()
store._loaded = True
return store
class TestResumePendingExpiredAutoReset:
"""resume_pending sessions past the freshness window should fire
was_auto_reset=True with auto_reset_reason='resume_pending_expired' and
persist that reason to state.db (#58933)."""
def _seed_stale_resume_pending(self, store, source, freshness_seconds=3600):
"""Create a session, mark it resume_pending, then backdate the mark
past the freshness window so get_or_create_session treats it as a
zombie."""
entry = store.get_or_create_session(source)
store.mark_resume_pending(entry.session_key)
with store._lock:
entry = store._entries[entry.session_key]
entry.last_resume_marked_at = (
datetime.now() - timedelta(seconds=freshness_seconds + 60)
)
entry.updated_at = datetime.now() # keep updated_at fresh
store._save()
return entry
def test_stale_resume_pending_sets_auto_reset_reason(
self, tmp_path, monkeypatch
):
"""Stale resume_pending triggers was_auto_reset=True with reason
'resume_pending_expired', NOT 'idle'."""
monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
# The freshness gate requires an opted-in reset policy — mode "none"
# disables it entirely (#61052). Use a huge idle window so only the
# freshness gate (not the idle policy) can fire.
store = _make_store_with_db(
tmp_path,
policy=SessionResetPolicy(mode="idle", idle_minutes=999999),
)
source = _make_source()
old = self._seed_stale_resume_pending(store, source)
new = store.get_or_create_session(source)
assert new.session_id != old.session_id, "should have created a new session"
assert new.was_auto_reset is True
assert new.auto_reset_reason == "resume_pending_expired"
def test_stale_resume_pending_had_activity_flag(
self, tmp_path, monkeypatch
):
"""reset_had_activity reflects whether the old session was used."""
monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
store = _make_store_with_db(
tmp_path,
policy=SessionResetPolicy(mode="idle", idle_minutes=999999),
)
source = _make_source()
old = self._seed_stale_resume_pending(store, source)
# Simulate some conversation on the old session.
with store._lock:
old.last_prompt_tokens = 50_000
store._save()
new = store.get_or_create_session(source)
assert new.reset_had_activity is True
def test_stale_resume_pending_db_end_reason_is_specific(
self, tmp_path, monkeypatch
):
"""state.db must record end_reason='resume_pending_expired', NOT the
generic 'session_reset', so the event is auditable (#58933 fix)."""
monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
db = _make_db_mock()
store = _make_store_with_db(
tmp_path, db,
policy=SessionResetPolicy(mode="idle", idle_minutes=999999),
)
source = _make_source()
old = self._seed_stale_resume_pending(store, source)
store.get_or_create_session(source)
# Auto-reset now writes through promote_to_session_reset so an
# accidental agent_close end can't shadow the reset boundary.
db.promote_to_session_reset.assert_called_once()
ended_id, ended_reason = db.promote_to_session_reset.call_args.args
assert ended_id == old.session_id
assert ended_reason == "resume_pending_expired", (
f"expected 'resume_pending_expired', got {ended_reason!r} — "
"the DB end_reason must not be the generic 'session_reset'"
)
def test_idle_reset_db_end_reason_reflects_idle(
self, tmp_path
):
"""Regular idle auto-reset persists 'idle' as end_reason so that all
auto-reset paths are auditable (#58933 should not regress the common
idle/daily path)."""
db = _make_db_mock()
store = _make_store_with_db(
tmp_path, db, policy=SessionResetPolicy(mode="idle", idle_minutes=1)
)
source = _make_source()
entry = store.get_or_create_session(source)
# Age past idle threshold.
with store._lock:
entry.updated_at = datetime.now() - timedelta(minutes=5)
store._save()
store.get_or_create_session(source)
db.promote_to_session_reset.assert_called_once()
_, ended_reason = db.promote_to_session_reset.call_args.args
assert ended_reason == "idle"
def test_freshness_disabled_skips_resume_pending_expired(
self, tmp_path, monkeypatch
):
"""When gateway_auto_continue_freshness=0, resume_pending is never
expired — the same session is returned regardless of age."""
monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "0")
db = _make_db_mock()
store = _make_store_with_db(tmp_path, db)
source = _make_source()
old = self._seed_stale_resume_pending(store, source, freshness_seconds=999_999)
refreshed = store.get_or_create_session(source)
# Freshness disabled → same session, no DB end_session call.
assert refreshed.session_id == old.session_id
db.end_session.assert_not_called()
db.promote_to_session_reset.assert_not_called()
Reference in a new issue View git blame Copy permalink