diff --git a/gateway/session.py b/gateway/session.py index 5b71e1581f26..b33b87a9384d 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1587,12 +1587,15 @@ class SessionStore: # a durable ``session_reset`` end_reason, later agent cleanup can # close the row as ``agent_close``; stale-route recovery treats # that as resumable and resurrects the expired full history. - # Reopen first because SessionDB.end_session is first-writer-wins. - self._db.reopen_session(entry.session_id) - self._db.end_session(entry.session_id, "session_reset") + # + # promote_to_session_reset is conditional: it only promotes + # live rows or rows ended with ``agent_close``. Explicit + # boundaries (compression, session_reset, new_command, etc.) + # are preserved — the first writer wins. + self._db.promote_to_session_reset(entry.session_id) except Exception as exc: logger.debug( - "Session DB end_session(session_reset) failed for %s: %s", + "Session DB promote_to_session_reset failed for %s: %s", entry.session_id, exc, ) diff --git a/hermes_state.py b/hermes_state.py index 4c90d5b0ee1d..4df2e31278fa 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -2286,6 +2286,36 @@ class SessionDB: ) self._execute_write(_do) + def promote_to_session_reset(self, session_id: str) -> bool: + """Mark a session as ended by session_reset — but only when safe. + + Promotes *only* live rows (``ended_at IS NULL``) or rows ended with + ``agent_close``. Explicit conversation boundaries such as + ``compression``, ``session_reset``, ``new_command``, etc. are + preserved — the first writer wins for those, and a later expiry + finalization must not silently overwrite them. + + Returns ``True`` when the row was promoted, ``False`` when skipped + (already has a different explicit end_reason, or row not found). + """ + if not session_id: + return False + now = time.time() + + def _do(conn): + cursor = conn.execute( + "UPDATE sessions SET ended_at = ?, end_reason = 'session_reset' " + "WHERE id = ? AND (ended_at IS NULL OR end_reason = 'agent_close')", + (now, session_id), + ) + return cursor.rowcount + + try: + rows = self._execute_write(_do) + return bool(rows) + except Exception: + return False + def update_session_cwd( self, session_id: str, cwd: str, git_branch: str = None, git_repo_root: str = None ) -> None: diff --git a/tests/gateway/test_session_store_expiry_finalized.py b/tests/gateway/test_session_store_expiry_finalized.py index f777eca2ebb5..424f625fcc46 100644 --- a/tests/gateway/test_session_store_expiry_finalized.py +++ b/tests/gateway/test_session_store_expiry_finalized.py @@ -5,63 +5,180 @@ then agent cleanup can close it as ``agent_close``. Stale routing recovery treat ``agent_close`` as recoverable, so expired sessions were reopened with full history unless expiry finalization also persisted the real conversation boundary as ``end_reason='session_reset'``. + +These tests use a real ``SessionDB`` (in-memory) to verify the actual recovery +contract in ``find_latest_gateway_session_for_peer`` — not just call counts on +a MagicMock. """ -from datetime import datetime, timedelta -from unittest.mock import MagicMock, patch +from __future__ import annotations -from gateway.config import GatewayConfig, Platform, SessionResetPolicy -from gateway.session import SessionEntry, SessionStore +import time +from pathlib import Path + +import pytest + +from hermes_state import SessionDB -def _make_store_with_db(tmp_path, db_mock) -> SessionStore: - config = GatewayConfig(default_reset_policy=SessionResetPolicy(mode="daily")) - with patch("gateway.session.SessionStore._ensure_loaded"): - store = SessionStore(sessions_dir=tmp_path, config=config) - store._db = db_mock - store._loaded = True - return store +@pytest.fixture +def db(tmp_path: Path) -> SessionDB: + return SessionDB(tmp_path / "state.db") -def _entry(session_id: str = "sid-expired") -> SessionEntry: - now = datetime.now() - return SessionEntry( - session_key="agent:main:telegram:dm:8494508720", - session_id=session_id, - created_at=now - timedelta(days=1), - updated_at=now - timedelta(days=1), - platform=Platform.TELEGRAM, - chat_type="dm", - model_override={"provider": "openrouter", "model": "test/model"}, - ) +_SESSION_KEY = "agent:main:telegram:dm:8494508720" +_SOURCE = "telegram" +_USER_ID = "8494508720" -def test_set_expiry_finalized_persists_session_reset_boundary(tmp_path): - db = MagicMock() - db.set_expiry_finalized.return_value = None - db.reopen_session.return_value = None - db.end_session.return_value = None - store = _make_store_with_db(tmp_path, db) - entry = _entry() +# ------------------------------------------------------------------ +# promote_to_session_reset — unit tests on real DB +# ------------------------------------------------------------------ - store.set_expiry_finalized(entry) +class TestPromoteToSessionReset: + """promote_to_session_reset promotes only safe rows.""" - assert entry.expiry_finalized is True - assert entry.model_override is None - db.set_expiry_finalized.assert_called_once_with("sid-expired", True) - db.reopen_session.assert_called_once_with("sid-expired") - db.end_session.assert_called_once_with("sid-expired", "session_reset") + def test_promotes_live_row(self, db: SessionDB) -> None: + """A live row (ended_at IS NULL) is promoted to session_reset.""" + db.create_session( + "sid-live", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + # Seed a message so recovery would match + db.append_message("sid-live", "user", "hello") + + assert db.promote_to_session_reset("sid-live") is True + row = db.get_session("sid-live") + assert row["end_reason"] == "session_reset" + assert row["ended_at"] is not None + + def test_promotes_agent_close_row(self, db: SessionDB) -> None: + """A row ended with agent_close is promoted to session_reset.""" + db.create_session( + "sid-ac", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + db.append_message("sid-ac", "user", "hello") + db.end_session("sid-ac", "agent_close") + + assert db.promote_to_session_reset("sid-ac") is True + row = db.get_session("sid-ac") + assert row["end_reason"] == "session_reset" + + def test_does_not_overwrite_compression(self, db: SessionDB) -> None: + """An existing compression boundary must not be overwritten.""" + db.create_session( + "sid-comp", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + db.end_session("sid-comp", "compression") + + assert db.promote_to_session_reset("sid-comp") is False + row = db.get_session("sid-comp") + assert row["end_reason"] == "compression" + + def test_does_not_overwrite_existing_session_reset(self, db: SessionDB) -> None: + """Already-promoted rows are idempotently skipped.""" + db.create_session( + "sid-reset", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + db.end_session("sid-reset", "session_reset") + + # Should be a no-op (rowcount = 0) + assert db.promote_to_session_reset("sid-reset") is False + row = db.get_session("sid-reset") + assert row["end_reason"] == "session_reset" + + def test_does_not_overwrite_new_command(self, db: SessionDB) -> None: + """A /new-command boundary is preserved.""" + db.create_session( + "sid-new", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + db.end_session("sid-new", "new_command") + + assert db.promote_to_session_reset("sid-new") is False + row = db.get_session("sid-new") + assert row["end_reason"] == "new_command" + + def test_noop_on_missing_session(self, db: SessionDB) -> None: + """Non-existent session_id returns False without error.""" + assert db.promote_to_session_reset("nonexistent") is False + + def test_noop_on_empty_session_id(self, db: SessionDB) -> None: + assert db.promote_to_session_reset("") is False -def test_set_expiry_finalized_still_sets_flag_if_end_session_fails(tmp_path): - db = MagicMock() - db.set_expiry_finalized.return_value = None - db.reopen_session.side_effect = RuntimeError("database locked") - store = _make_store_with_db(tmp_path, db) - entry = _entry() +# ------------------------------------------------------------------ +# Integration: promotion blocks stale-route recovery +# ------------------------------------------------------------------ - store.set_expiry_finalized(entry) +class TestPromotionBlocksRecovery: + """After promotion, find_latest_gateway_session_for_peer must NOT + recover the session — it is now ended with session_reset, which + the recovery query excludes (only live/agent_close are recoverable). + """ - assert entry.expiry_finalized is True - db.set_expiry_finalized.assert_called_once_with("sid-expired", True) - db.end_session.assert_not_called() + def test_live_session_recoverable_before_promotion(self, db: SessionDB) -> None: + db.create_session( + "sid-pre", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + db.append_message("sid-pre", "user", "hello") + + recovered = db.find_latest_gateway_session_for_peer( + source=_SOURCE, session_key=_SESSION_KEY, + user_id=_USER_ID, chat_id="8494508720", chat_type="dm", + ) + assert recovered is not None + assert recovered["id"] == "sid-pre" + + def test_promoted_session_not_recoverable(self, db: SessionDB) -> None: + db.create_session( + "sid-post", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + db.append_message("sid-post", "user", "hello") + db.promote_to_session_reset("sid-post") + + recovered = db.find_latest_gateway_session_for_peer( + source=_SOURCE, session_key=_SESSION_KEY, + user_id=_USER_ID, chat_id="8494508720", chat_type="dm", + ) + # session_reset rows are not in the recovery set + assert recovered is None + + def test_agent_close_session_recoverable_but_not_after_promotion( + self, db: SessionDB + ) -> None: + db.create_session( + "sid-ac-rec", _SOURCE, + user_id=_USER_ID, session_key=_SESSION_KEY, + chat_id="8494508720", chat_type="dm", + ) + db.append_message("sid-ac-rec", "user", "hello") + db.end_session("sid-ac-rec", "agent_close") + + # agent_close is recoverable + recovered = db.find_latest_gateway_session_for_peer( + source=_SOURCE, session_key=_SESSION_KEY, + user_id=_USER_ID, chat_id="8494508720", chat_type="dm", + ) + assert recovered is not None + assert recovered["id"] == "sid-ac-rec" + + # After promotion, it is no longer recoverable + db.promote_to_session_reset("sid-ac-rec") + recovered2 = db.find_latest_gateway_session_for_peer( + source=_SOURCE, session_key=_SESSION_KEY, + user_id=_USER_ID, chat_id="8494508720", chat_type="dm", + ) + assert recovered2 is None