test(gateway): cover ws_orphan_reap session recovery (#63207)

Regression tests for find_latest_gateway_session_for_peer and
SessionStore stale-routing self-heal when end_reason is ws_orphan_reap.
Pin manual approval mode in blocking E2E tests so smart aux-LLM
resolution does not flake CI.

tests/gateway/test_approve_deny_commands.py

@@ -425,6 +425,16 @@ class TestBlockingApprovalE2E:
         os.environ.pop("HERMES_GATEWAY_SESSION", None)
         os.environ.pop("HERMES_EXEC_ASK", None)
         os.environ.pop("HERMES_SESSION_KEY", None)
+        # These E2E tests exercise manual gateway blocking; default config is
+        # approvals.mode=smart which may auto-approve/deny via aux LLM before
+        # notify_cb runs (flaky on CI when the LLM is slow or unavailable).
+        self._approval_mode_patch = patch(
+            "tools.approval._get_approval_mode", return_value="manual"
+        )
+        self._approval_mode_patch.start()
+
+    def teardown_method(self):
+        self._approval_mode_patch.stop()
 
     def test_blocking_approval_approve_once(self):
         """check_all_command_guards blocks until resolve_gateway_approval is called."""

tests/gateway/test_session_store_runtime_stale_guard.py

@@ -142,6 +142,24 @@ class TestRuntimeStaleGuard:
         # A brand-new session row must NOT have been created.
         db.create_session.assert_not_called()
 
+    def test_stale_ws_orphan_reap_entry_recovered_preserving_session_id(self, tmp_path):
+        """Stale ``ws_orphan_reap`` entry → recovery reopens the SAME session_id (#63207)."""
+        source = _source()
+        db = _db_returning({"sid_stale": {"end_reason": "ws_orphan_reap", "id": "sid_stale"}})
+        db.find_latest_gateway_session_for_peer.return_value = {
+            "id": "sid_stale",
+            "started_at": (datetime.now() - timedelta(hours=2)).timestamp(),
+        }
+        store = _make_store_with_db(tmp_path, db)
+        key = store._generate_session_key(source)
+        store._entries[key] = _make_entry(key, "sid_stale")
+
+        result = store.get_or_create_session(source)
+
+        assert result.session_id == "sid_stale"
+        db.reopen_session.assert_called_once_with("sid_stale")
+        db.create_session.assert_not_called()
+
     def test_stale_entry_creates_fresh_when_recovery_returns_none(self, tmp_path):
         """Stale entry, no recoverable row → brand-new session (no silent drop)."""
         source = _source()

tests/test_hermes_state.py

@@ -5627,6 +5627,34 @@ def test_gateway_session_peer_round_trip_and_recovery(db):
     assert recovered["id"] == "gw-session"
 
 
+def test_gateway_session_recovery_reopens_ws_orphan_reap_rows(db):
+    """Rows wrongly ended by the TUI ws-orphan reaper must be recoverable (#63207)."""
+    db.create_session(
+        "reaped-gw-session",
+        "telegram",
+        user_id="user-1",
+        session_key="agent:main:telegram:dm:chat-1",
+        chat_id="chat-1",
+        chat_type="dm",
+    )
+    db.append_message("reaped-gw-session", "user", "hello")
+    db.end_session("reaped-gw-session", "ws_orphan_reap")
+
+    recovered = db.find_latest_gateway_session_for_peer(
+        source="telegram",
+        user_id="user-1",
+        session_key="agent:main:telegram:dm:chat-1",
+        chat_id="chat-1",
+        chat_type="dm",
+    )
+    assert recovered["id"] == "reaped-gw-session"
+
+    db.reopen_session("reaped-gw-session")
+    row = db.get_session("reaped-gw-session")
+    assert row["ended_at"] is None
+    assert row["end_reason"] is None
+
+
 def test_gateway_session_recovery_reopens_legacy_agent_close_rows(db):
     db.create_session(
         "closed-gw-session",