fix: conditional promote + real SessionDB tests

Address review feedback on #63068:

1. Replace unconditional reopen_session() + end_session() with a
   conditional promote_to_session_reset() method in SessionDB.
   The new method only promotes live rows or rows ended with
   agent_close — explicit boundaries (compression, session_reset,
   new_command) are preserved via first-writer-wins semantics.

2. Rewrite tests to use real SessionDB instead of MagicMock:
   - 7 unit tests for promote_to_session_reset edge cases
   - 3 integration tests verifying the actual recovery contract
     in find_latest_gateway_session_for_peer after promotion

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