hermes-webui/tests/test_title_aux_routing.py

"""Regression tests for auxiliary title-generation config routing.

Covers:
  - _aux_title_configured() broad detection (provider, model, base_url)
  - generate_title_raw_via_aux() reads timeout from config instead of hardcoding 15.0
  - aux→agent fallback triggers on 'llm_invalid_aux' status
  - _aux_title_timeout rejects zero, negative, and non-numeric values
"""
import os
import sys
import types
import unittest
from unittest.mock import MagicMock, patch

# Stub agent.auxiliary_client so it is importable in the test environment
# (the real package lives in hermes-agent, which is not installed here).
_agent_stub = types.ModuleType('agent')
_aux_stub = types.ModuleType('agent.auxiliary_client')
sys.modules.setdefault('agent', _agent_stub)
sys.modules.setdefault('agent.auxiliary_client', _aux_stub)
_agent_stub.auxiliary_client = _aux_stub


def _patch_tg_config(config_dict):
    """Return a patch context manager that makes _get_auxiliary_task_config return config_dict."""
    return patch('agent.auxiliary_client._get_auxiliary_task_config', return_value=config_dict, create=True)


class TestAuxTitleConfigured(unittest.TestCase):
    def _call(self, tg_config):
        from api.streaming import _aux_title_configured
        with _patch_tg_config(tg_config):
            return _aux_title_configured()

    def test_model_set_returns_true(self):
        self.assertTrue(self._call({'provider': '', 'model': 'gpt-4o-mini', 'base_url': ''}))

    def test_base_url_set_returns_true(self):
        self.assertTrue(self._call({'provider': '', 'model': '', 'base_url': 'http://localhost:1234'}))

    def test_provider_set_non_auto_returns_true(self):
        self.assertTrue(self._call({'provider': 'openai', 'model': '', 'base_url': ''}))

    def test_provider_auto_returns_false(self):
        self.assertFalse(self._call({'provider': 'auto', 'model': '', 'base_url': ''}))

    def test_provider_auto_case_insensitive_returns_false(self):
        self.assertFalse(self._call({'provider': 'AUTO', 'model': '', 'base_url': ''}))

    def test_all_empty_returns_false(self):
        self.assertFalse(self._call({'provider': '', 'model': '', 'base_url': ''}))

    def test_empty_dict_returns_false(self):
        self.assertFalse(self._call({}))

    def test_provider_configured_model_blank_returns_true(self):
        """Regression: provider set + blank model must still be treated as configured."""
        self.assertTrue(self._call({'provider': 'anthropic', 'model': '', 'base_url': ''}))

    def test_base_url_only_returns_true(self):
        """Regression: base_url alone (no model) must still be treated as configured."""
        self.assertTrue(self._call({'provider': '', 'model': '', 'base_url': 'https://api.example.com'}))

    def test_import_error_returns_false(self):
        from api.streaming import _aux_title_configured
        with patch('agent.auxiliary_client._get_auxiliary_task_config', side_effect=ImportError("no module"), create=True):
            self.assertFalse(_aux_title_configured())


class TestGenerateTitleRawViaAuxTimeout(unittest.TestCase):
    """Verify generate_title_raw_via_aux() reads timeout from config rather than hardcoding 15.0."""

    def _run_with_config(self, tg_config, expected_timeout):
        from api.streaming import generate_title_raw_via_aux

        mock_resp = types.SimpleNamespace(
            choices=[
                types.SimpleNamespace(
                    message=types.SimpleNamespace(content='Test Title'),
                    finish_reason='stop',
                )
            ]
        )

        captured = {}

        def fake_call_llm(**kwargs):
            captured['timeout'] = kwargs.get('timeout')
            return mock_resp

        with _patch_tg_config(tg_config):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='What is the weather?',
                    assistant_text='It is sunny.',
                )

        self.assertEqual(result, 'Test Title')
        self.assertAlmostEqual(captured['timeout'], expected_timeout)

    def test_default_timeout_when_not_set(self):
        """No timeout in config → uses 15.0 default."""
        self._run_with_config({'provider': '', 'model': 'gpt-4o', 'base_url': ''}, 15.0)

    def test_custom_timeout_from_config(self):
        """Regression: timeout set in config must be used instead of hardcoded 15.0."""
        self._run_with_config(
            {'provider': '', 'model': 'gpt-4o', 'base_url': '', 'timeout': 30.0},
            30.0,
        )

    def test_webui_prefixed_model_id_is_stripped_before_aux_call(self):
        """Regression: @provider:model picker ids must not reach provider APIs verbatim."""
        from api.streaming import generate_title_raw_via_aux

        mock_resp = types.SimpleNamespace(
            choices=[
                types.SimpleNamespace(
                    message=types.SimpleNamespace(content='Gemini Title'),
                    finish_reason='stop',
                )
            ]
        )
        captured = {}

        def fake_call_llm(**kwargs):
            captured.update(kwargs)
            return mock_resp

        tg_config = {
            'provider': 'gemini',
            'model': '@gemini:gemini-3.1-flash-lite',
            'base_url': '',
        }
        with _patch_tg_config(tg_config):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Как настроить title generation?',
                    assistant_text='Нужно указать auxiliary.title_generation в config.yaml.',
                )

        self.assertEqual(result, 'Gemini Title')
        self.assertEqual(status, 'llm_aux')
        self.assertEqual(captured.get('provider'), 'gemini')
        self.assertEqual(captured.get('model'), 'gemini-3.1-flash-lite')

    def test_configured_provider_model_and_base_url_are_passed_to_aux_client(self):
        """Regression for #2235: task config must select the first title model.

        If generate_title_raw_via_aux leaves provider/model/base_url as None,
        agent.auxiliary_client.call_llm can fall back to the main chat model and
        WebUI titles look like local first-message placeholders or unrelated
        chat-model output instead of using auxiliary.title_generation.model.
        """
        from api.streaming import generate_title_raw_via_aux

        mock_resp = types.SimpleNamespace(
            choices=[
                types.SimpleNamespace(
                    message=types.SimpleNamespace(content='Configured Model Title'),
                    finish_reason='stop',
                )
            ]
        )
        captured = {}

        def fake_call_llm(**kwargs):
            captured.update(kwargs)
            return mock_resp

        tg_config = {
            'provider': 'openrouter',
            'model': 'anthropic/claude-haiku-title',
            'base_url': 'https://openrouter.ai/api/v1',
            'timeout': 22.0,
        }
        with _patch_tg_config(tg_config):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Explain why the moon has phases.',
                    assistant_text='The moon has phases because we see different sunlit portions.',
                )

        self.assertEqual(result, 'Configured Model Title')
        self.assertEqual(status, 'llm_aux')
        self.assertEqual(captured.get('task'), 'title_generation')
        self.assertEqual(captured.get('provider'), 'openrouter')
        self.assertEqual(captured.get('model'), 'anthropic/claude-haiku-title')
        self.assertEqual(captured.get('base_url'), 'https://openrouter.ai/api/v1')
        self.assertEqual(captured.get('timeout'), 22.0)

    def test_configured_api_key_is_passed_to_aux_client(self):
        """Regression: explicit title_generation api_key must be forwarded.

        Hermes Agent does not fall back to auxiliary.title_generation.api_key
        once WebUI passes explicit provider/model/base_url values, so WebUI
        must forward the configured task api_key with the rest of the route.
        """
        from api.streaming import generate_title_raw_via_aux

        mock_resp = types.SimpleNamespace(
            choices=[
                types.SimpleNamespace(
                    message=types.SimpleNamespace(content='Configured Key Title'),
                    finish_reason='stop',
                )
            ]
        )
        captured = {}

        def fake_call_llm(**kwargs):
            captured.update(kwargs)
            return mock_resp

        tg_config = {
            'provider': '',
            'model': 'gemma-4-31b-it',
            'base_url': 'http://openrouter:4000/v1',
            'api_key': 'test-title-api-key',
        }
        with _patch_tg_config(tg_config):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Summarize this title routing bug.',
                    assistant_text='The title model route needs its configured key.',
                )

        self.assertEqual(result, 'Configured Key Title')
        self.assertEqual(status, 'llm_aux')
        self.assertEqual(captured.get('task'), 'title_generation')
        self.assertIsNone(captured.get('provider'))
        self.assertEqual(captured.get('model'), 'gemma-4-31b-it')
        self.assertEqual(captured.get('base_url'), 'http://openrouter:4000/v1')
        self.assertEqual(captured.get('api_key'), 'test-title-api-key')

    def test_title_prompt_requires_matching_user_language(self):
        """Conversation starts should get a language-neutral match-language instruction."""
        from api.streaming import generate_title_raw_via_aux

        mock_resp = types.SimpleNamespace(
            choices=[
                types.SimpleNamespace(
                    message=types.SimpleNamespace(content='Alte Session Bilder'),
                    finish_reason='stop',
                )
            ]
        )
        captured = {}

        def fake_call_llm(**kwargs):
            captured.update(kwargs)
            return mock_resp

        with _patch_tg_config({'provider': '', 'model': 'title-model', 'base_url': ''}):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Warum werden hier die Bilder der alten Session nicht mehr angezeigt?',
                    assistant_text='Ich prüfe die Attachment-Pfade im WebUI.',
                )

        self.assertEqual(result, 'Alte Session Bilder')
        self.assertEqual(status, 'llm_aux')
        messages = captured.get('messages') or []
        self.assertIn('Match the language of the user question', messages[0]['content'])
        self.assertNotIn('If the user writes German', messages[0]['content'])
        self.assertNotIn('German good:', messages[0]['content'])

    def test_title_prompt_language_rule_is_same_for_supported_locales(self):
        from api.streaming import _title_prompt_language_rule

        expected = "Match the language of the user question.\n"
        examples = [
            'Warum werden hier die Bilder nicht angezeigt?',
            'Pourquoi les images ne sont-elles pas affichées ?',
            '¿Por qué no se muestran las imágenes?',
            '为什么图片没有显示？',
            'Why are the images not displayed?',
        ]
        for text in examples:
            with self.subTest(text=text):
                self.assertEqual(_title_prompt_language_rule(text), expected)

    def test_title_language_detection_avoids_english_tech_false_positives(self):
        """English tech/jargon text must not be classified as German by shared tokens."""
        from api.streaming import _detect_title_language

        examples = [
            'Why did the session die after the DAS storage failover?',
            'The session can die when DAS storage disconnects.',
            'Debug the session and DER certificate import failure.',
        ]
        for text in examples:
            with self.subTest(text=text):
                self.assertEqual(_detect_title_language(text), '')

    def test_title_language_detection_keeps_german_without_umlaut(self):
        """German without umlauts still needs a language hint when evidence is specific."""
        from api.streaming import _detect_title_language

        self.assertEqual(
            _detect_title_language('Warum werden hier die Bilder der alten Session nicht angezeigt?'),
            'de',
        )

    def test_german_source_rejects_english_aux_title(self):
        """Regression: an English aux title must not overwrite a German conversation."""
        from api.streaming import _generate_llm_session_title_via_aux

        mock_resp = types.SimpleNamespace(
            choices=[
                types.SimpleNamespace(
                    message=types.SimpleNamespace(content='Old Session Image Display Issue'),
                    finish_reason='stop',
                )
            ]
        )

        with _patch_tg_config({'provider': '', 'model': 'title-model', 'base_url': ''}):
            with patch('agent.auxiliary_client.call_llm', return_value=mock_resp, create=True):
                title, status, raw_preview = _generate_llm_session_title_via_aux(
                    'Warum werden hier die Bilder der alten Session nicht mehr angezeigt?',
                    'Ich prüfe die Attachment-Pfade im WebUI.',
                )

        self.assertIsNone(title)
        self.assertEqual(status, 'llm_language_mismatch_aux')
        self.assertEqual(raw_preview, 'Old Session Image Display Issue')

    def test_german_fallback_uses_generic_topic_extraction_without_literal_override(self):
        from api.streaming import _fallback_title_from_exchange

        title = _fallback_title_from_exchange(
            'Warum werden hier die Bilder der alten Session nicht mehr angezeigt?',
            'Ich prüfe die Rendering- und Attachment-Pfade im WebUI.',
        )

        self.assertIsNotNone(title)
        self.assertIsInstance(title, str)
        self.assertNotEqual(title, 'Alte Session Bilder')
        self.assertNotEqual(title, 'Session Bilder')
        self.assertIn('Warum', title)

    def test_code_only_first_message_does_not_trigger_german_language_guard(self):
        """Code-only starts should fall through to the neutral/default title path."""
        from api.streaming import _detect_title_language, _title_language_mismatch, _title_prompt_language_rule

        code_only = "print('hello')\nfor i in range(3):\n    print(i)"

        self.assertEqual(_detect_title_language(code_only), '')
        self.assertEqual(_title_prompt_language_rule(code_only), 'Match the language of the user question.\n')
        self.assertFalse(_title_language_mismatch(code_only, 'Python Hello Loop'))

    def test_configured_api_key_is_not_sent_to_caller_supplied_route(self):
        """Regression: title task keys must not leak to explicit fallback routes.

        Some callers pass the active agent route into title generation. In that
        path WebUI should still use title-generation prompts/timeouts, but it
        must not attach an unrelated auxiliary.title_generation api_key to the
        caller-supplied provider/model/base_url.
        """
        from api.streaming import generate_title_raw_via_aux

        mock_resp = types.SimpleNamespace(
            choices=[
                types.SimpleNamespace(
                    message=types.SimpleNamespace(content='Agent Route Title'),
                    finish_reason='stop',
                )
            ]
        )
        captured = {}

        def fake_call_llm(**kwargs):
            captured.update(kwargs)
            return mock_resp

        tg_config = {
            'provider': '',
            'model': 'gemma-4-31b-it',
            'base_url': 'http://openrouter:4000/v1',
            'api_key': 'test-title-api-key',
        }
        with _patch_tg_config(tg_config):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Summarize this title routing bug.',
                    assistant_text='The caller route should not receive the task key.',
                    provider='custom:agent-route',
                    model='agent-title-model',
                    base_url='https://agent-route.example/v1',
                )

        self.assertEqual(result, 'Agent Route Title')
        self.assertEqual(status, 'llm_aux')
        self.assertEqual(captured.get('task'), 'title_generation')
        self.assertEqual(captured.get('provider'), 'custom:agent-route')
        self.assertEqual(captured.get('model'), 'agent-title-model')
        self.assertEqual(captured.get('base_url'), 'https://agent-route.example/v1')
        self.assertIsNone(captured.get('api_key'))

    def test_integer_timeout_from_config(self):
        """Config timeout as int is coerced to float."""
        self._run_with_config(
            {'provider': '', 'model': 'gpt-4o', 'base_url': '', 'timeout': 5},
            5.0,
        )

    def test_timeout_none_in_config_falls_back_to_default(self):
        """Explicit None in config falls back to 15.0."""
        self._run_with_config(
            {'provider': '', 'model': 'gpt-4o', 'base_url': '', 'timeout': None},
            15.0,
        )


class TestReasoningModelTitleGeneration(unittest.TestCase):
    """Regression coverage for reasoning models that spend output budget on reasoning."""

    def test_title_budget_defaults_to_reasoning_safe_value(self):
        """Title generation should not use a tiny output cap that starves final content."""
        from api.streaming import _title_completion_budget, _title_retry_completion_budget

        self.assertEqual(_title_completion_budget(), 512)
        self.assertEqual(_title_retry_completion_budget(), 1024)

    def test_aux_short_circuits_on_empty_reasoning_without_retrying(self):
        """Regression for #2083: reasoning models that emit only hidden
        reasoning tokens (no visible content) must NOT trigger a budget-doubling
        retry — the second call invariably produces the same empty-reasoning
        shape and just doubles the GPU/credit burn.  Short-circuit to the local
        fallback path instead."""
        from api.streaming import generate_title_raw_via_aux

        call_count = [0]

        def fake_call_llm(**kwargs):
            call_count[0] += 1
            return {
                'choices': [
                    {
                        'message': {'content': '', 'reasoning': 'long hidden reasoning'},
                        'finish_reason': 'length',
                    }
                ]
            }

        with _patch_tg_config({'provider': 'ollama', 'model': 'kimi-k2.6', 'base_url': 'https://ollama.com/v1'}):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Hey nur ein kurzer Test',
                    assistant_text='Alles klar, ich helfe dir dabei.',
                )

        self.assertIsNone(result)
        self.assertEqual(status, 'llm_empty_reasoning_aux')
        # One call per prompt at the base budget — no retry on prompt 0, no
        # second-prompt attempt either (short-circuited).
        self.assertEqual(call_count[0], 1)

    def test_aux_still_retries_finish_length_without_reasoning(self):
        """Length-truncated responses WITHOUT reasoning tokens still get the
        budget-doubling retry — those are legitimately recoverable by giving
        the model more headroom."""
        from api.streaming import generate_title_raw_via_aux

        responses = [
            {'choices': [{'message': {'content': ''}, 'finish_reason': 'length'}]},
            {'choices': [{'message': {'content': 'Useful Session Title'}, 'finish_reason': 'stop'}]},
        ]
        captured_budgets = []

        def fake_call_llm(**kwargs):
            captured_budgets.append(kwargs.get('max_tokens'))
            return responses.pop(0)

        with _patch_tg_config({'provider': 'ollama', 'model': 'kimi-k2.6', 'base_url': 'https://ollama.com/v1'}):
            with patch('agent.auxiliary_client.call_llm', side_effect=fake_call_llm, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Hey nur ein kurzer Test',
                    assistant_text='Alles klar, ich helfe dir dabei.',
                )

        self.assertEqual(result, 'Useful Session Title')
        self.assertEqual(status, 'llm_aux_retry')
        self.assertEqual(captured_budgets, [512, 1024])

    def test_aux_returns_specific_status_when_reasoning_retry_still_empty(self):
        """Diagnostics should expose the provider failure mode instead of generic llm_error_aux."""
        from api.streaming import generate_title_raw_via_aux

        def empty_length_response(**kwargs):
            return {
                'choices': [
                    {
                        'message': {'content': '', 'reasoning': 'still reasoning'},
                        'finish_reason': 'length',
                    }
                ]
            }

        with _patch_tg_config({'provider': 'ollama', 'model': 'kimi-k2.6', 'base_url': 'https://ollama.com/v1'}):
            with patch('agent.auxiliary_client.call_llm', side_effect=empty_length_response, create=True):
                result, status = generate_title_raw_via_aux(
                    user_text='Hey nur ein kurzer Test',
                    assistant_text='Alles klar, ich helfe dir dabei.',
                )

        self.assertIsNone(result)
        self.assertEqual(status, 'llm_empty_reasoning_aux')

    def test_agent_route_short_circuits_on_empty_reasoning_without_retrying(self):
        """Regression for #2083 on the active-agent route: empty-reasoning
        responses must NOT trigger a budget-doubling retry."""
        from api.streaming import generate_title_raw_via_agent

        call_count = [0]

        def fake_create(**kwargs):
            call_count[0] += 1
            return {
                'choices': [
                    {
                        'message': {'content': '', 'reasoning': 'long hidden reasoning'},
                        'finish_reason': 'length',
                    }
                ]
            }

        client = types.SimpleNamespace(
            chat=types.SimpleNamespace(
                completions=types.SimpleNamespace(create=fake_create)
            )
        )
        agent = MagicMock()
        agent.api_mode = 'openai'
        agent.provider = 'ollama'
        agent.model = 'kimi-k2.6'
        agent.base_url = 'https://ollama.com/v1'
        agent.reasoning_config = None
        agent._build_api_kwargs.return_value = {}
        agent._ensure_primary_openai_client.return_value = client

        result, status = generate_title_raw_via_agent(
            agent,
            user_text='Hey nur ein kurzer Test',
            assistant_text='Alles klar, ich helfe dir dabei.',
        )

        self.assertIsNone(result)
        self.assertEqual(status, 'llm_empty_reasoning')
        # One call per prompt at base budget — no retry, no second-prompt attempt.
        self.assertEqual(call_count[0], 1)
        self.assertIsNone(agent.reasoning_config)

    def test_agent_route_still_retries_finish_length_without_reasoning(self):
        """The active-agent route should preserve retry-on-length-no-reasoning."""
        from api.streaming import generate_title_raw_via_agent

        responses = [
            {'choices': [{'message': {'content': ''}, 'finish_reason': 'length'}]},
            {'choices': [{'message': {'content': 'Agent Session Title'}, 'finish_reason': 'stop'}]},
        ]
        captured_budgets = []

        def fake_create(**kwargs):
            captured_budgets.append(kwargs.get('max_tokens') or kwargs.get('max_completion_tokens'))
            return responses.pop(0)

        client = types.SimpleNamespace(
            chat=types.SimpleNamespace(
                completions=types.SimpleNamespace(create=fake_create)
            )
        )
        agent = MagicMock()
        agent.api_mode = 'openai'
        agent.provider = 'ollama'
        agent.model = 'kimi-k2.6'
        agent.base_url = 'https://ollama.com/v1'
        agent.reasoning_config = None
        agent._build_api_kwargs.return_value = {}
        agent._ensure_primary_openai_client.return_value = client

        result, status = generate_title_raw_via_agent(
            agent,
            user_text='Hey nur ein kurzer Test',
            assistant_text='Alles klar, ich helfe dir dabei.',
        )

        self.assertEqual(result, 'Agent Session Title')
        self.assertEqual(status, 'llm_retry')
        self.assertEqual(captured_budgets, [512, 1024])
        self.assertIsNone(agent.reasoning_config)

    @patch('api.streaming._aux_title_configured', return_value=True)
    @patch('api.streaming._generate_llm_session_title_via_aux')
    @patch('api.streaming.get_session')
    def test_fallback_title_status_keeps_underlying_llm_reason(
        self, mock_get_session, mock_aux_title, mock_configured,
    ):
        """Local fallback should not hide that the LLM failed because it hit length."""
        from api.streaming import _run_background_title_update

        mock_session = MagicMock()
        mock_session.title = 'Untitled'
        mock_session.llm_title_generated = False
        mock_session.messages = [
            {'role': 'user', 'content': 'Hey nur ein kurzer Test'},
            {'role': 'assistant', 'content': 'Alles klar, ich helfe dir dabei.'},
        ]
        mock_get_session.return_value = mock_session
        mock_aux_title.return_value = (None, 'llm_length_aux', '')
        events = []

        _run_background_title_update(
            session_id='reasoning-title-session',
            user_text='Hey nur ein kurzer Test',
            assistant_text='Alles klar, ich helfe dir dabei.',
            placeholder_title='Untitled',
            put_event=lambda event_type, data: events.append((event_type, data)),
            agent=None,
        )

        title_status = [data for event_type, data in events if event_type == 'title_status']
        self.assertTrue(title_status)
        self.assertEqual(title_status[0]['status'], 'fallback')
        self.assertEqual(title_status[0]['reason'], 'local_summary:llm_length_aux')

    @patch('api.streaming._aux_title_configured', return_value=True)
    @patch('api.streaming._generate_llm_session_title_via_aux')
    @patch('api.streaming.get_session')
    def test_generic_fallback_title_is_not_persisted(
        self, mock_get_session, mock_aux_title, mock_configured,
    ):
        """A generic local fallback is worse than the provisional first-message title."""
        from api.streaming import _run_background_title_update

        provisional_title = '\u5e2e\u6211\u53bb\u627e\u4e00\u672c\u300a\u7ea2\u697c\u68a6\u300b\u7535\u5b50\u4e66'
        first_user_text = provisional_title + '\u3002'
        mock_session = MagicMock()
        mock_session.title = provisional_title
        mock_session.llm_title_generated = False
        mock_session.messages = [
            {'role': 'user', 'content': first_user_text},
            {'role': 'assistant', 'content': ''},
        ]
        mock_get_session.return_value = mock_session
        mock_aux_title.return_value = (None, 'llm_error_aux', '')
        events = []

        _run_background_title_update(
            session_id='generic-title-session',
            user_text=first_user_text,
            assistant_text='',
            placeholder_title=provisional_title,
            put_event=lambda event_type, data: events.append((event_type, data)),
            agent=None,
        )

        title_events = [data for event_type, data in events if event_type == 'title']
        title_status = [data for event_type, data in events if event_type == 'title_status']
        self.assertEqual(title_events, [])
        self.assertTrue(title_status)
        self.assertEqual(title_status[0]['status'], 'skipped')
        self.assertEqual(title_status[0]['reason'], 'llm_error_aux')
        self.assertEqual(title_status[0]['title'], provisional_title)
        self.assertEqual(mock_session.title, provisional_title)
        self.assertFalse(mock_session.llm_title_generated)
        mock_session.save.assert_not_called()


class TestBackgroundTitleProfileRouting(unittest.TestCase):
    def test_profile_env_context_logs_fail_open_resolution_errors(self):
        """Profile env setup failures should be diagnosable without breaking workers."""
        import api.profiles as profiles

        session = types.SimpleNamespace(profile='work')
        captured = {}

        with patch.object(
            profiles,
            'get_hermes_home_for_profile',
            side_effect=RuntimeError('profile lookup failed'),
        ):
            with patch.dict(os.environ, {'HERMES_HOME': 'default-home'}, clear=False):
                with self.assertLogs('api.profiles', level='DEBUG') as logs:
                    with profiles.profile_env_for_background_worker(session, 'background title'):
                        captured['HERMES_HOME'] = os.environ.get('HERMES_HOME')

        message_found = any(
            'Failed to resolve profile env for background title profile work' in record.getMessage()
            for record in logs.records
        )
        self.assertEqual(captured['HERMES_HOME'], 'default-home')
        self.assertTrue(message_found)
        self.assertTrue(any(record.exc_info for record in logs.records))

    def test_skill_home_snapshot_removes_modules_imported_during_context(self):
        """Modules first imported inside a temporary profile context must not leak."""
        import api.profiles as profiles

        original_parent = sys.modules.get('tools')
        original_skill_module = sys.modules.get('tools.skills_tool')
        original_manager_module = sys.modules.get('tools.skill_manager_tool')

        sys.modules.pop('tools.skills_tool', None)
        sys.modules.pop('tools.skill_manager_tool', None)
        tools_parent = types.ModuleType('tools')
        sys.modules['tools'] = tools_parent
        try:
            snapshot = profiles.snapshot_skill_home_modules()

            imported_during_context = types.ModuleType('tools.skills_tool')
            setattr(imported_during_context, 'HERMES_HOME', 'profile-home')
            setattr(imported_during_context, 'SKILLS_DIR', 'profile-home/skills')
            sys.modules['tools.skills_tool'] = imported_during_context
            setattr(tools_parent, 'skills_tool', imported_during_context)

            profiles.restore_skill_home_modules(snapshot)

            self.assertNotIn('tools.skills_tool', sys.modules)
            self.assertFalse(hasattr(tools_parent, 'skills_tool'))
        finally:
            sys.modules.pop('tools.skills_tool', None)
            sys.modules.pop('tools.skill_manager_tool', None)
            if original_parent is None:
                sys.modules.pop('tools', None)
            else:
                sys.modules['tools'] = original_parent
            if original_skill_module is not None:
                sys.modules['tools.skills_tool'] = original_skill_module
            if original_manager_module is not None:
                sys.modules['tools.skill_manager_tool'] = original_manager_module

    @patch('api.streaming._aux_title_configured', return_value=True)
    @patch('api.streaming.get_session')
    def test_background_title_generation_uses_session_profile_home(
        self, mock_get_session, mock_configured,
    ):
        """A background title worker for a non-default profile must resolve aux config from that profile."""
        from api.streaming import _run_background_title_update

        mock_session = MagicMock()
        mock_session.title = 'Untitled'
        mock_session.profile = 'work'
        mock_session.llm_title_generated = False
        mock_session.messages = [
            {'role': 'user', 'content': 'This is a test message'},
            {'role': 'assistant', 'content': 'Received.'},
        ]
        mock_get_session.return_value = mock_session

        captured = {}

        original_skill_module = sys.modules.get('tools.skills_tool')
        fake_skill_module = types.ModuleType('tools.skills_tool')
        setattr(fake_skill_module, 'HERMES_HOME', 'default-home')
        setattr(fake_skill_module, 'SKILLS_DIR', 'default-home/skills')
        sys.modules['tools.skills_tool'] = fake_skill_module

        def fake_aux_title(*args, **kwargs):
            captured['hermes_home'] = os.environ.get('HERMES_HOME')
            captured['skill_module_home'] = getattr(fake_skill_module, 'HERMES_HOME')
            captured['skill_module_dir'] = getattr(fake_skill_module, 'SKILLS_DIR')
            return ('Profile Routed Title', 'llm_aux', '')

        events = []
        try:
            with patch('api.profiles.get_hermes_home_for_profile', return_value='profile-home'):
                with patch('api.streaming._generate_llm_session_title_via_aux', side_effect=fake_aux_title):
                    with patch.dict(os.environ, {'HERMES_HOME': 'default-home'}, clear=False):
                        _run_background_title_update(
                            session_id='profile-title-session',
                            user_text='This is a test message',
                            assistant_text='Received.',
                            placeholder_title='Untitled',
                            put_event=lambda event_type, data: events.append((event_type, data)),
                            agent=None,
                        )
                        captured['restored_hermes_home'] = os.environ.get('HERMES_HOME')
        finally:
            if original_skill_module is None:
                sys.modules.pop('tools.skills_tool', None)
            else:
                sys.modules['tools.skills_tool'] = original_skill_module

        self.assertEqual(captured.get('hermes_home'), 'profile-home')
        self.assertEqual(str(captured.get('skill_module_home')), 'profile-home')
        self.assertEqual(str(captured.get('skill_module_dir')), 'profile-home/skills')
        self.assertEqual(captured.get('restored_hermes_home'), 'default-home')
        self.assertEqual(getattr(fake_skill_module, 'HERMES_HOME'), 'default-home')
        self.assertEqual(getattr(fake_skill_module, 'SKILLS_DIR'), 'default-home/skills')
        self.assertEqual(mock_session.title, 'Profile Routed Title')

    def test_background_profile_env_routes_load_config_and_provider_credentials(self):
        """Hybrid worker env must satisfy config and os.getenv provider-key readers."""
        import tempfile

        import pytest

        import api.profiles as profiles
        from api.config import _thread_ctx
        try:
            from hermes_cli import config as hermes_config
        except ModuleNotFoundError:
            pytest.skip('hermes_cli is not installed in this CI environment')

        session = types.SimpleNamespace(profile='work')
        captured = {}

        with tempfile.TemporaryDirectory() as tmp:
            default_home = os.path.join(tmp, 'default-home')
            profile_home = os.path.join(tmp, 'profile-home')
            os.makedirs(default_home, exist_ok=True)
            os.makedirs(profile_home, exist_ok=True)
            with open(os.path.join(default_home, 'config.yaml'), 'w', encoding='utf-8') as f:
                f.write('model:\n  provider: default-provider\n  default: default-model\n')
            with open(os.path.join(profile_home, 'config.yaml'), 'w', encoding='utf-8') as f:
                f.write('model:\n  provider: profile-provider\n  default: profile-model\n')

            with patch('api.profiles.get_hermes_home_for_profile', return_value=profile_home):
                runtime_env = {
                    'PROFILE_ONLY_KEY': 'profile-only',
                    'OPENROUTER_API_KEY': 'profile-openrouter-key',
                }
                with patch('api.profiles.get_profile_runtime_env', return_value=runtime_env):
                    with patch.dict(os.environ, {'HERMES_HOME': default_home, 'OPENROUTER_API_KEY': 'default-openrouter-key'}, clear=False):
                        os.environ.pop('PROFILE_ONLY_KEY', None)
                        hermes_config._LOAD_CONFIG_CACHE.clear()
                        with profiles.profile_env_for_background_worker(session, 'background title'):
                            loaded = hermes_config.load_config()
                            captured['loaded_provider'] = loaded.get('model', {}).get('provider')
                            captured['process_home'] = os.environ.get('HERMES_HOME')
                            captured['process_runtime_key'] = os.environ.get('PROFILE_ONLY_KEY')
                            captured['provider_credential'] = os.getenv('OPENROUTER_API_KEY')
                            captured['thread_home'] = getattr(_thread_ctx, 'env', {}).get('HERMES_HOME')
                            captured['thread_runtime_key'] = getattr(_thread_ctx, 'env', {}).get('PROFILE_ONLY_KEY')
                        captured['restored_home'] = os.environ.get('HERMES_HOME')
                        captured['restored_runtime_key'] = os.environ.get('PROFILE_ONLY_KEY')
                        captured['restored_provider_credential'] = os.environ.get('OPENROUTER_API_KEY')
                        hermes_config._LOAD_CONFIG_CACHE.clear()

        self.assertEqual(captured['loaded_provider'], 'profile-provider')
        self.assertEqual(captured['process_home'], profile_home)
        self.assertEqual(captured['process_runtime_key'], 'profile-only')
        self.assertEqual(captured['provider_credential'], 'profile-openrouter-key')
        self.assertEqual(captured['thread_home'], profile_home)
        self.assertEqual(captured['thread_runtime_key'], 'profile-only')
        self.assertEqual(captured['restored_home'], default_home)
        self.assertIsNone(captured['restored_runtime_key'])
        self.assertEqual(captured['restored_provider_credential'], 'default-openrouter-key')


class TestAuxTitleTimeoutEdgeCases(unittest.TestCase):
    """_aux_title_timeout must reject zero, negative, and non-numeric values."""

    def _call(self, tg_config, default=15.0):
        from api.streaming import _aux_title_timeout
        with _patch_tg_config(tg_config):
            return _aux_title_timeout(default=default)

    def test_timeout_zero_falls_back_to_default(self):
        """timeout: 0 is not strictly positive → fall back to default."""
        result = self._call({'timeout': 0}, default=15.0)
        self.assertEqual(result, 15.0)

    def test_timeout_negative_falls_back_to_default(self):
        """timeout: -1 is not strictly positive → fall back to default."""
        result = self._call({'timeout': -1}, default=15.0)
        self.assertEqual(result, 15.0)

    def test_timeout_non_numeric_string_falls_back_to_default(self):
        """timeout: 'abc' cannot be coerced to float → fall back to default."""
        result = self._call({'timeout': 'abc'}, default=15.0)
        self.assertEqual(result, 15.0)

    def test_timeout_empty_string_falls_back_to_default(self):
        """timeout: '' cannot be coerced to a positive float → fall back to default."""
        result = self._call({'timeout': ''}, default=15.0)
        self.assertEqual(result, 15.0)

    def test_timeout_positive_passes_through(self):
        """A valid positive timeout is returned as-is."""
        result = self._call({'timeout': 25.0}, default=15.0)
        self.assertEqual(result, 25.0)

    def test_custom_default_used_on_invalid(self):
        """When the value is invalid, the caller-supplied *default* is returned."""
        result = self._call({'timeout': 0}, default=20.0)
        self.assertEqual(result, 20.0)


class TestAuxInvalidAuxTriggersAgentFallback(unittest.TestCase):
    """When aux returns llm_invalid_aux, the agent route must be tried as fallback.

    Pins the behaviour so the fallback tuple in _run_background_title_update
    stays synchronised with the statuses that _generate_llm_session_title_via_aux
    actually emits.
    """

    @patch('api.streaming._aux_title_configured', return_value=True)
    @patch('api.streaming._generate_llm_session_title_via_aux')
    @patch('api.streaming._generate_llm_session_title_for_agent')
    @patch('api.streaming.get_session')
    def test_llm_invalid_aux_triggers_agent_fallback(
        self, mock_get_session, mock_agent_title, mock_aux_title, mock_configured,
    ):
        """Simulate aux returning (None, 'llm_invalid_aux', '...') and verify agent fallback fires."""
        from api.streaming import _run_background_title_update

        # Build a mock session that passes all the pre-checks
        mock_session = MagicMock()
        mock_session.title = 'Untitled'
        mock_session.llm_title_generated = False
        mock_session.messages = [
            {'role': 'user', 'content': 'What is the weather?'},
            {'role': 'assistant', 'content': 'It is sunny and warm.'},
        ]
        mock_get_session.return_value = mock_session

        # aux route returns invalid title
        mock_aux_title.return_value = (None, 'llm_invalid_aux', 'bad thinking preamble')

        # agent route succeeds
        mock_agent_title.return_value = ('Weather Report', 'llm', '')

        events = []

        def fake_put_event(event_type, data):
            events.append((event_type, data))

        _run_background_title_update(
            session_id='test-session',
            user_text='What is the weather?',
            assistant_text='It is sunny and warm.',
            placeholder_title='Untitled',
            put_event=fake_put_event,
            agent=MagicMock(),
        )

        # The agent fallback must have been invoked
        mock_agent_title.assert_called_once()

        # A title must have been produced via the agent route
        title_events = [(e, d) for e, d in events if e == 'title']
        self.assertTrue(len(title_events) > 0, "Expected a 'title' event to be emitted")
        self.assertEqual(title_events[0][1]['title'], 'Weather Report')

    @patch('api.streaming._aux_title_configured', return_value=True)
    @patch('api.streaming._generate_llm_session_title_via_aux')
    @patch('api.streaming._generate_llm_session_title_for_agent')
    @patch('api.streaming.get_session')
    def test_llm_error_aux_triggers_agent_fallback(
        self, mock_get_session, mock_agent_title, mock_aux_title, mock_configured,
    ):
        """Simulate aux returning (None, 'llm_error_aux', '') and verify agent fallback fires."""
        from api.streaming import _run_background_title_update

        mock_session = MagicMock()
        mock_session.title = 'Untitled'
        mock_session.llm_title_generated = False
        mock_session.messages = [
            {'role': 'user', 'content': 'Tell me a joke.'},
            {'role': 'assistant', 'content': 'Why did the chicken cross the road?'},
        ]
        mock_get_session.return_value = mock_session

        mock_aux_title.return_value = (None, 'llm_error_aux', '')
        mock_agent_title.return_value = ('Chicken Joke', 'llm', '')

        events = []

        def fake_put_event(event_type, data):
            events.append((event_type, data))

        _run_background_title_update(
            session_id='test-session-2',
            user_text='Tell me a joke.',
            assistant_text='Why did the chicken cross the road?',
            placeholder_title='Untitled',
            put_event=fake_put_event,
            agent=MagicMock(),
        )

        mock_agent_title.assert_called_once()

    @patch('api.streaming._aux_title_configured', return_value=True)
    @patch('api.streaming._generate_llm_session_title_via_aux')
    @patch('api.streaming._generate_llm_session_title_for_agent')
    @patch('api.streaming.get_session')
    def test_success_status_does_not_trigger_agent_fallback(
        self, mock_get_session, mock_agent_title, mock_aux_title, mock_configured,
    ):
        """When aux succeeds, the agent route must NOT be called."""
        from api.streaming import _run_background_title_update

        mock_session = MagicMock()
        mock_session.title = 'Untitled'
        mock_session.llm_title_generated = False
        mock_session.messages = [
            {'role': 'user', 'content': 'Hello'},
            {'role': 'assistant', 'content': 'Hi there'},
        ]
        mock_get_session.return_value = mock_session

        # aux succeeds on first try
        mock_aux_title.return_value = ('Greeting', 'llm_aux', '')

        events = []

        def fake_put_event(event_type, data):
            events.append((event_type, data))

        _run_background_title_update(
            session_id='test-session-3',
            user_text='Hello',
            assistant_text='Hi there',
            placeholder_title='Untitled',
            put_event=fake_put_event,
            agent=MagicMock(),
        )

        # Agent route must NOT have been invoked
        mock_agent_title.assert_not_called()


if __name__ == '__main__':
    unittest.main()