""" Tests for Steward service layer. Tests request analysis, logging, and benchmarking integration. """ from unittest.mock import AsyncMock, MagicMock, patch import pytest from src.agents.steward.schemas import ConversationContext, StewardRecommendation from src.agents.steward.service import ( _build_enriched_query, _extract_capabilities, analyze_request, format_steward_note, ) from src.core.startup import register_household_members @pytest.fixture(scope="module", autouse=True) def setup_household_registry(): """Initialize household registry before running tests.""" register_household_members() class TestAnalyzeRequest: """Test the analyze_request service function.""" @pytest.mark.asyncio async def test_analyze_simple_greeting(self): """Test analyzing a simple greeting.""" # Mock the Steward agent's analyze method (plain text approach) mock_agent = MagicMock() mock_agent.analyze = AsyncMock( return_value="Simple greeting requires no tools. This is a simple request." ) with patch("src.agents.steward.service.get_steward_agent", return_value=mock_agent): result = await analyze_request( "Hello!", conversation_history=[], ) assert result.recommended_capabilities == [] assert result.estimated_complexity == "simple" assert mock_agent.analyze.called @pytest.mark.asyncio async def test_analyze_math_request(self): """Test analyzing a mathematical request.""" mock_agent = MagicMock() mock_agent.analyze = AsyncMock( return_value="Mathematical calculation requires tatlock_core for solving this simple problem." ) with patch("src.agents.steward.service.get_steward_agent", return_value=mock_agent): result = await analyze_request( "What's sqrt(144)?", conversation_history=[], ) assert "tatlock_core" in result.recommended_capabilities assert result.estimated_complexity == "simple" @pytest.mark.asyncio async def test_analyze_with_conversation_history(self): """Test analyzing with previous conversation context.""" mock_agent = MagicMock() mock_agent.analyze = AsyncMock( return_value="Follow-up to previous calculation in turn 0. Requires tatlock_core. Complexity: moderate." ) conversation_history = [ {"role": "user", "content": "What's 2 + 2?"}, {"role": "assistant", "content": "4"}, ] with patch("src.agents.steward.service.get_steward_agent", return_value=mock_agent): result = await analyze_request( "And what's that times 5?", conversation_history=conversation_history, ) assert result.conversation_context.has_previous_context is True assert 0 in result.conversation_context.relevant_turns # Verify conversation history was passed call_kwargs = mock_agent.analyze.call_args.kwargs assert "conversation_history" in call_kwargs assert len(call_kwargs["conversation_history"]) == 2 @pytest.mark.asyncio async def test_analyze_with_missing_capabilities(self): """Test analyzing request that needs unavailable capabilities.""" mock_agent = MagicMock() mock_agent.analyze = AsyncMock( return_value="Image generation not available. Would be needed for this request. Complexity: simple." ) with patch("src.agents.steward.service.get_steward_agent", return_value=mock_agent): result = await analyze_request( "Generate an image of a sunset", conversation_history=[], ) assert result.missing_capabilities is not None assert "not available" in result.missing_capabilities @pytest.mark.asyncio async def test_analyze_with_conversation_id(self): """Test that analysis includes conversation ID in context.""" mock_agent = MagicMock() mock_agent.analyze = AsyncMock( return_value="This simple request requires tatlock_core to solve." ) with patch("src.agents.steward.service.get_steward_agent", return_value=mock_agent): result = await analyze_request( "Test request", conversation_history=[], conversation_id="test_conv_123", ) # Verify analysis completed successfully assert result.recommended_capabilities == ["tatlock_core"] assert result.estimated_complexity == "simple" @pytest.mark.asyncio async def test_analyze_handles_errors(self): """Test error handling in analyze_request.""" mock_agent = MagicMock() mock_agent.analyze = AsyncMock(side_effect=Exception("Test error")) with patch("src.agents.steward.service.get_steward_agent", return_value=mock_agent): with pytest.raises(Exception, match="Test error"): await analyze_request("Test", conversation_history=[]) class TestFormatStewardNote: """Test the format_steward_note function.""" @pytest.mark.asyncio async def test_format_simple_note(self): """Test formatting a simple recommendation.""" rec = StewardRecommendation( recommended_capabilities=["tatlock_core"], reasoning="Math needed", estimated_complexity="simple", conversation_context=ConversationContext(has_previous_context=False), ) note = await format_steward_note(rec) assert "📋 Steward's Analysis" in note assert "SIMPLE" in note assert "tatlock_core" in note @pytest.mark.asyncio async def test_format_note_with_context(self): """Test formatting note with conversation context.""" context = ConversationContext( has_previous_context=True, relevant_turns=[0, 1], context_summary="Previous discussion about calculations", ) rec = StewardRecommendation( recommended_capabilities=["tatlock_core"], reasoning="Follow-up calculation", estimated_complexity="moderate", conversation_context=context, ) note = await format_steward_note(rec) assert "Context:" in note assert "Previous discussion" in note @pytest.mark.asyncio async def test_format_note_with_missing_capabilities(self): """Test formatting note with missing capabilities warning.""" rec = StewardRecommendation( recommended_capabilities=[], reasoning="Not available", estimated_complexity="simple", conversation_context=ConversationContext(has_previous_context=False), missing_capabilities="Advanced research tools needed", ) note = await format_steward_note(rec) assert "⚠️ Missing:" in note assert "Advanced research" in note @pytest.mark.unit class TestBuildEnrichedQuery: """Tests for _build_enriched_query function.""" def test_no_enrichment_without_context(self): """Test no enrichment when memory context is empty.""" query = "What's the weather?" result = _build_enriched_query(query, {}) assert result == query def test_enrichment_adds_location(self): """Test location is appended for weather queries.""" query = "What's the weather?" memory_context = {"profile": {"location": "Amsterdam", "timezone": "Europe/Amsterdam"}} result = _build_enriched_query(query, memory_context) assert "location=Amsterdam" in result assert query in result assert "[User Context:" in result def test_no_location_when_specified(self): """Test location is not appended when already specified.""" query = "What's the weather in London?" memory_context = {"profile": {"location": "Amsterdam"}} result = _build_enriched_query(query, memory_context) # Should not add Amsterdam since location is specified assert result == query def test_enrichment_adds_timezone(self): """Test timezone is appended for time queries.""" query = "What time is it?" memory_context = {"profile": {"timezone": "Europe/Amsterdam"}} result = _build_enriched_query(query, memory_context) assert "timezone=Europe/Amsterdam" in result def test_no_timezone_when_specified(self): """Test timezone is not appended when already specified.""" query = "What time is it in UTC?" memory_context = {"profile": {"timezone": "Europe/Amsterdam"}} result = _build_enriched_query(query, memory_context) assert result == query def test_enrichment_adds_temperature_unit(self): """Test temperature unit is appended for weather queries.""" query = "What's the weather?" memory_context = { "profile": {"location": "Amsterdam"}, "preferences": {"temperature_unit": "celsius"}, } result = _build_enriched_query(query, memory_context) assert "temperature_unit=celsius" in result def test_multiple_context_fields(self): """Test multiple context fields are appended.""" query = "What time and weather today?" memory_context = { "profile": {"location": "Amsterdam", "timezone": "Europe/Amsterdam"}, "preferences": {"temperature_unit": "celsius"}, } result = _build_enriched_query(query, memory_context) assert "location=Amsterdam" in result assert "timezone=Europe/Amsterdam" in result assert "temperature_unit=celsius" in result def test_no_enrichment_for_unrelated_query(self): """Test no enrichment for queries that don't need context.""" query = "Tell me a joke" memory_context = {"profile": {"location": "Amsterdam", "timezone": "Europe/Amsterdam"}} result = _build_enriched_query(query, memory_context) assert result == query class TestExtractCapabilities: """Capability extraction reads the declared DELEGATE line, not free prose. The prompt tells the Steward to state its choice on a DELEGATE line and to explain itself on REASON/COMPLEXITY/CONTEXT lines. An earlier version substring-matched capability domains across the entire response, so ordinary English in the explanation routed requests: "description" contains the housekeeper domain "script", "acknowledge" contains "know". These tests pin that the explanation can no longer influence routing. """ # (prose, why it used to misroute) SUBSTRING_TRAPS = [ ("The user wants a description of the algorithm.", "script -> housekeeper"), ("I should discover what the answer is.", "cover -> housekeeper"), ("That sounds fantastic, let me compute it.", "fan -> housekeeper"), ( "I acknowledge the request to add two numbers.", "knowledge/know -> librarian, biographer", ), ("The user asks about the economy myth.", "my -> biographer"), ("Convert 98.6 Fahrenheit to Celsius.", "temperature is a housekeeper domain"), ] @pytest.mark.parametrize("prose,reason", SUBSTRING_TRAPS) def test_reason_prose_cannot_add_capabilities(self, prose, reason): """Explanatory prose must not summon agents the Steward did not request.""" text = f"DELEGATE: tatlock_core to calculate\nREASON: {prose}\nCOMPLEXITY: simple" assert _extract_capabilities(text) == ["tatlock_core"], f"regression: {reason}" def test_delegate_line_task_text_does_not_leak(self): """A domain word inside the task description must not add a capability. "home" is a housekeeper domain, but this is plainly a memory recall. """ text = "DELEGATE: biographer to recall the user's home address\nREASON: personal data" assert _extract_capabilities(text) == ["biographer"] def test_multiple_delegate_lines(self): """Each DELEGATE line contributes its capability, in order, deduplicated.""" text = ( "DELEGATE: biographer to recall the user's location\n" "DELEGATE: librarian to search_web for the forecast\n" "DELEGATE: biographer to recall preferences\n" ) assert _extract_capabilities(text) == ["biographer", "librarian"] def test_capability_named_later_on_the_line(self): """A loosely worded DELEGATE line still resolves by name.""" text = "DELEGATE: ask the librarian to search the web" assert _extract_capabilities(text) == ["librarian"] def test_domain_fallback_within_delegate_line(self): """With no capability named, domains on the DELEGATE line still resolve.""" text = "DELEGATE: turn on the lights in the kitchen" assert _extract_capabilities(text) == ["housekeeper"] def test_conversational_response_selects_nothing(self): """No DELEGATE line means no capability, which is the prompt's chat path.""" text = "This is a simple greeting. No capabilities are needed. COMPLEXITY: simple" assert _extract_capabilities(text) == [] def test_malformed_response_still_routes_by_name(self): """If the format is ignored, a named capability is still honoured.""" text = "I think the librarian should handle this research request." assert _extract_capabilities(text) == ["librarian"] def test_malformed_response_does_not_route_on_domains(self): """...but bare prose must not route on domain words alone.""" text = "The user wants a description of home automation, and I acknowledge it." assert _extract_capabilities(text) == [] def test_case_insensitive_delegate_marker(self): text = "delegate: Librarian to search_web" assert _extract_capabilities(text) == ["librarian"] def test_empty_input(self): assert _extract_capabilities("") == []