hermes-webui/tests/test_1058_adaptive_title_refresh.py

"""Tests for adaptive session title refresh helpers (PR #1058).

Covers all five new functions added to api/streaming.py:
  - _count_exchanges
  - _latest_exchange_snippets
  - _get_title_refresh_interval
  - _run_background_title_refresh
  - _maybe_schedule_title_refresh
"""
import sys
import os
import threading
import types
import unittest
from unittest.mock import MagicMock, patch

import pytest

# Ensure the project root is on sys.path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

from api.streaming import (
    _count_exchanges,
    _latest_exchange_snippets,
    _get_title_refresh_interval,
    _run_background_title_refresh,
    _maybe_schedule_title_refresh,
)


@pytest.fixture(autouse=True)
def _restore_auth_sessions():
    """Snapshot and restore api.auth._sessions around each test.

    Importing api.streaming can trigger api.config.load_settings() which may
    call into api.auth and create a real session token.  Without this fixture,
    that stale token leaks into test_auth_session_persistence.py tests (which
    assume _sessions starts empty) when our file runs first alphabetically.
    """
    import api.auth as _auth
    snapshot = dict(_auth._sessions)
    yield
    _auth._sessions.clear()
    _auth._sessions.update(snapshot)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _user_msg(text):
    return {'role': 'user', 'content': text}


def _asst_msg(text, tool_calls=None):
    msg = {'role': 'assistant', 'content': text}
    if tool_calls is not None:
        msg['tool_calls'] = tool_calls
    return msg


def _tool_only_asst():
    """Assistant message that only has tool_calls, no real text."""
    return {'role': 'assistant', 'content': '', 'tool_calls': [{'id': 't1', 'type': 'function'}]}


def _make_session(title='My Title', llm_title_generated=True, messages=None, session_id='sid1'):
    s = MagicMock()
    s.title = title
    s.llm_title_generated = llm_title_generated
    s.messages = messages or []
    s.session_id = session_id
    s.save = MagicMock()
    return s


# ---------------------------------------------------------------------------
# _count_exchanges
# ---------------------------------------------------------------------------

class TestCountExchanges:
    def test_empty_messages_returns_zero(self):
        assert _count_exchanges([]) == 0

    def test_none_messages_returns_zero(self):
        assert _count_exchanges(None) == 0

    def test_counts_only_user_messages(self):
        msgs = [_user_msg('hello'), _asst_msg('hi'), _user_msg('world')]
        assert _count_exchanges(msgs) == 2

    def test_skips_empty_user_messages(self):
        msgs = [_user_msg(''), _user_msg('   '), _user_msg('real question')]
        assert _count_exchanges(msgs) == 1

    def test_counts_list_content_user_messages(self):
        msgs = [
            {'role': 'user', 'content': [{'type': 'text', 'text': 'list question'}]},
        ]
        assert _count_exchanges(msgs) == 1

    def test_skips_empty_list_content(self):
        msgs = [
            {'role': 'user', 'content': [{'type': 'text', 'text': '   '}]},
        ]
        assert _count_exchanges(msgs) == 0

    def test_ignores_non_user_roles(self):
        msgs = [_asst_msg('response'), {'role': 'system', 'content': 'system prompt'}]
        assert _count_exchanges(msgs) == 0

    def test_ignores_non_dict_entries(self):
        msgs = ['not a dict', _user_msg('real'), None]
        assert _count_exchanges(msgs) == 1

    def test_five_exchanges(self):
        msgs = []
        for i in range(5):
            msgs.append(_user_msg(f'question {i}'))
            msgs.append(_asst_msg(f'answer {i}'))
        assert _count_exchanges(msgs) == 5


# ---------------------------------------------------------------------------
# _latest_exchange_snippets
# ---------------------------------------------------------------------------

class TestLatestExchangeSnippets:
    def test_empty_returns_empty_strings(self):
        u, a = _latest_exchange_snippets([])
        assert u == '' and a == ''

    def test_none_returns_empty_strings(self):
        u, a = _latest_exchange_snippets(None)
        assert u == '' and a == ''

    def test_basic_pair(self):
        msgs = [_user_msg('first q'), _asst_msg('first a'),
                _user_msg('second q'), _asst_msg('second a')]
        u, a = _latest_exchange_snippets(msgs)
        assert u == 'second q'
        assert a == 'second a'

    def test_returns_latest_not_first(self):
        msgs = [_user_msg('old q'), _asst_msg('old a'),
                _user_msg('new q'), _asst_msg('new a')]
        u, a = _latest_exchange_snippets(msgs)
        assert u == 'new q'

    def test_skips_tool_call_only_assistant(self):
        """An assistant msg with tool_calls and no real text should be skipped."""
        msgs = [_user_msg('q'), _asst_msg('real answer'),
                _user_msg('q2'), _tool_only_asst()]
        u, a = _latest_exchange_snippets(msgs)
        # _tool_only_asst should be skipped; fall back to previous real assistant
        assert a == 'real answer'
        assert u == 'q2'

    def test_truncates_long_content(self):
        long_text = 'x' * 600
        msgs = [_user_msg(long_text), _asst_msg(long_text)]
        u, a = _latest_exchange_snippets(msgs)
        assert len(u) == 500
        assert len(a) == 500

    def test_no_assistant_message(self):
        msgs = [_user_msg('q')]
        u, a = _latest_exchange_snippets(msgs)
        assert u == 'q'
        assert a == ''

    def test_no_user_message(self):
        msgs = [_asst_msg('a')]
        u, a = _latest_exchange_snippets(msgs)
        assert u == ''
        assert a == 'a'

    def test_ignores_non_dict_entries(self):
        msgs = ['noise', _user_msg('q'), None, _asst_msg('a')]
        u, a = _latest_exchange_snippets(msgs)
        assert u == 'q'
        assert a == 'a'


# ---------------------------------------------------------------------------
# _get_title_refresh_interval
# ---------------------------------------------------------------------------

class TestGetTitleRefreshInterval:
    def test_returns_int_for_valid_setting(self):
        # _get_title_refresh_interval does a local import: `from api.config import load_settings`
        # so patch the source module, not api.streaming
        with patch('api.config.load_settings', return_value={'auto_title_refresh_every': '5'}):
            assert _get_title_refresh_interval() == 5

    def test_returns_zero_for_off_setting(self):
        with patch('api.config.load_settings', return_value={'auto_title_refresh_every': '0'}):
            assert _get_title_refresh_interval() == 0

    def test_returns_zero_when_key_absent(self):
        with patch('api.config.load_settings', return_value={}):
            assert _get_title_refresh_interval() == 0

    def test_returns_zero_on_exception(self):
        with patch('api.config.load_settings', side_effect=Exception('boom')):
            assert _get_title_refresh_interval() == 0

    def test_valid_values_10_and_20(self):
        for val in ('10', '20'):
            with patch('api.config.load_settings', return_value={'auto_title_refresh_every': val}):
                assert _get_title_refresh_interval() == int(val)


# ---------------------------------------------------------------------------
# _run_background_title_refresh
# ---------------------------------------------------------------------------

class TestRunBackgroundTitleRefresh:
    def _make_put_event(self):
        events = []
        def put(name, data):
            events.append((name, data))
        return put, events

    def _make_session_obj(self, title='Old Title'):
        s = MagicMock()
        s.title = title
        s.llm_title_generated = True
        s.save = MagicMock()
        return s

    def test_skips_when_title_changed_before_call(self):
        """If the title has changed (manual rename) since the refresh was scheduled, skip."""
        put, events = self._make_put_event()
        with patch('api.streaming.get_session') as mock_get, \
             patch('api.streaming.SESSIONS', {}), \
             patch('api.streaming.LOCK', threading.Lock()):
            s = self._make_session_obj(title='Different Title')
            mock_get.return_value = s
            _run_background_title_refresh(
                'sid', 'user', 'asst', 'Old Title', put, agent=None
            )
        # No 'title' event should have been emitted
        assert not any(name == 'title' for name, _ in events)

    def test_skips_if_session_not_found(self):
        put, events = self._make_put_event()
        with patch('api.streaming.get_session', side_effect=KeyError('not found')):
            _run_background_title_refresh('sid', 'u', 'a', 'title', put)
        assert events == []

    def test_skips_when_title_is_untitled(self):
        put, events = self._make_put_event()
        with patch('api.streaming.get_session') as mock_get:
            s = self._make_session_obj(title='Untitled')
            mock_get.return_value = s
            _run_background_title_refresh('sid', 'u', 'a', 'Untitled', put)
        assert not any(name == 'title' for name, _ in events)

    def test_skips_same_title(self):
        """If the LLM generates a title identical to the current one, no event is emitted."""
        put, events = self._make_put_event()
        with patch('api.streaming.get_session') as mock_get, \
             patch('api.streaming._aux_title_configured', return_value=True), \
             patch('api.streaming._generate_llm_session_title_via_aux',
                   return_value=('Old Title', 'llm_ok', 'raw')), \
             patch('api.streaming.SESSIONS', {}), \
             patch('api.streaming.LOCK', threading.Lock()):
            s = self._make_session_obj(title='Old Title')
            mock_get.return_value = s
            _run_background_title_refresh('sid', 'u', 'a', 'Old Title', put)
        assert not any(name == 'title' for name, _ in events)

    def test_emits_title_event_on_new_title(self):
        put, events = self._make_put_event()
        s = self._make_session_obj(title='Old Title')
        # Use a real dict for SESSIONS so .get() works, pre-populated with our session
        fake_sessions = {'sid': s}
        with patch('api.streaming.get_session', return_value=s), \
             patch('api.streaming._aux_title_configured', return_value=True), \
             patch('api.streaming._generate_llm_session_title_via_aux',
                   return_value=('New Refreshed Title', 'llm_ok', 'raw')), \
             patch('api.streaming.SESSIONS', fake_sessions), \
             patch('api.streaming.LOCK', threading.Lock()):
            _run_background_title_refresh('sid', 'u', 'a', 'Old Title', put)
        title_events = [(n, d) for n, d in events if n == 'title']
        assert len(title_events) == 1
        assert title_events[0][1]['title'] == 'New Refreshed Title'

    def test_exceptions_are_silently_swallowed(self):
        """Any unexpected error inside must not propagate — it's a background daemon."""
        put, events = self._make_put_event()
        with patch('api.streaming.get_session', side_effect=RuntimeError('oops')):
            # Should not raise
            _run_background_title_refresh('sid', 'u', 'a', 'title', put)
        assert events == []


# ---------------------------------------------------------------------------
# _maybe_schedule_title_refresh
# ---------------------------------------------------------------------------

class TestMaybeScheduleTitleRefresh:
    def _noop_put(self, name, data):
        pass

    def test_does_nothing_when_disabled(self):
        with patch('api.streaming._get_title_refresh_interval', return_value=0):
            spawned = []
            with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
                session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 5)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
        assert spawned == []

    def test_does_nothing_when_title_is_empty(self):
        with patch('api.streaming._get_title_refresh_interval', return_value=5):
            spawned = []
            with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
                session = _make_session(title='', messages=[_user_msg('q'), _asst_msg('a')] * 5)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
        assert spawned == []

    def test_does_nothing_for_untitled(self):
        with patch('api.streaming._get_title_refresh_interval', return_value=5):
            spawned = []
            with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
                session = _make_session(title='Untitled', messages=[_user_msg('q'), _asst_msg('a')] * 5)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
        assert spawned == []

    def test_does_nothing_when_title_not_llm_generated(self):
        with patch('api.streaming._get_title_refresh_interval', return_value=5):
            spawned = []
            with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
                session = _make_session(llm_title_generated=False,
                                        messages=[_user_msg('q'), _asst_msg('a')] * 5)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
        assert spawned == []

    def test_does_nothing_when_exchange_count_not_at_interval(self):
        """Refresh only fires when exchange_count % interval == 0 (and > 0)."""
        with patch('api.streaming._get_title_refresh_interval', return_value=5):
            spawned = []
            with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
                # 4 exchanges — not a multiple of 5
                session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 4)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
        assert spawned == []

    def test_spawns_thread_at_exact_interval(self):
        """Refresh fires when exchange_count == refresh_interval."""
        with patch('api.streaming._get_title_refresh_interval', return_value=5):
            spawned = []
            with patch('threading.Thread') as mock_thread_cls:
                mock_thread = MagicMock()
                mock_thread_cls.return_value = mock_thread
                # 5 user messages = 5 exchanges
                session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 5)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
                assert mock_thread_cls.called
                assert mock_thread.start.called

    def test_spawns_thread_at_multiple_of_interval(self):
        """Refresh fires at 10 exchanges when interval is 5."""
        with patch('api.streaming._get_title_refresh_interval', return_value=5):
            with patch('threading.Thread') as mock_thread_cls:
                mock_thread = MagicMock()
                mock_thread_cls.return_value = mock_thread
                # 10 exchanges
                session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 10)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
                assert mock_thread_cls.called

    def test_does_nothing_when_no_exchange_content(self):
        """Even at interval, if both snippets are empty, don't spawn."""
        with patch('api.streaming._get_title_refresh_interval', return_value=5), \
             patch('api.streaming._latest_exchange_snippets', return_value=('', '')):
            spawned = []
            with patch('threading.Thread', side_effect=lambda **kw: spawned.append(kw) or MagicMock()):
                session = _make_session(messages=[_user_msg('q'), _asst_msg('a')] * 5)
                _maybe_schedule_title_refresh(session, self._noop_put, None)
        assert spawned == []