""" Tests for delegation infrastructure. Tests the DelegationTask dataclass and delegation wrapper functions that implement the agent-as-tool pattern. """ from unittest.mock import AsyncMock, patch import pytest from src.agents.delegation import ( HOUSEHOLD_THINK_MESSAGES, ActionType, DelegationResult, DelegationTask, _detect_action_type, build_delegation_context, delegate_to_librarian, get_think_message, ) @pytest.mark.unit class TestBuildDelegationContext: """Tests for trimming conversation history into expert context.""" def test_empty_history_returns_empty(self): assert build_delegation_context(None) == "" assert build_delegation_context([]) == "" def test_recent_turns_are_formatted(self): history = [ {"role": "user", "content": "Tell me about Docker"}, {"role": "assistant", "content": "Docker is a container runtime."}, ] context = build_delegation_context(history) assert "Recent conversation:" in context assert "user: Tell me about Docker" in context assert "assistant: Docker is a container runtime." in context def test_only_last_max_turns_kept(self): history = [{"role": "user", "content": f"message {i}"} for i in range(10)] context = build_delegation_context(history, max_turns=6) assert "message 3" not in context assert "message 4" in context assert "message 9" in context def test_long_turns_are_truncated(self): history = [{"role": "user", "content": "x" * 2000}] context = build_delegation_context(history, max_chars_per_turn=500) assert "x" * 500 in context assert "x" * 501 not in context def test_structured_content_parts_tolerated(self): history = [{"role": "user", "content": [{"type": "text", "text": "hello there"}]}] context = build_delegation_context(history) assert "hello there" in context def test_non_dict_entries_skipped(self): history = ["garbage", {"role": "user", "content": "real message"}] context = build_delegation_context(history) assert "real message" in context assert "garbage" not in context @pytest.mark.unit class TestDelegationTask: """Tests for the DelegationTask dataclass.""" def test_delegation_task_creation(self): """Test basic DelegationTask creation.""" task = DelegationTask( expert_name="librarian", task="Create a wiki page about CI/CD", context="User is setting up a homelab", action="create", ) assert task.expert_name == "librarian" assert task.task == "Create a wiki page about CI/CD" assert task.context == "User is setting up a homelab" assert task.action == "create" def test_delegation_task_default_values(self): """Test DelegationTask default values.""" task = DelegationTask( expert_name="librarian", task="Search for Docker info", ) assert task.context == "" assert task.action == "" assert task.priority == 0 assert task.depends_on == [] assert task.result is None def test_delegation_task_auto_generates_id(self): """Test DelegationTask auto-generates unique IDs.""" task1 = DelegationTask(expert_name="librarian", task="Task 1") task2 = DelegationTask(expert_name="librarian", task="Task 2") assert task1.task_id.startswith("librarian_") assert task2.task_id.startswith("librarian_") assert task1.task_id != task2.task_id def test_delegation_task_preserves_custom_id(self): """Test DelegationTask preserves custom ID if provided.""" task = DelegationTask( expert_name="librarian", task="Custom task", task_id="custom_id_123", ) assert task.task_id == "custom_id_123" def test_delegation_task_with_dependencies(self): """Test DelegationTask with dependencies.""" task = DelegationTask( expert_name="librarian", task="Update wiki page", depends_on=["memory_abc123", "search_def456"], ) assert len(task.depends_on) == 2 assert "memory_abc123" in task.depends_on @pytest.mark.unit class TestDelegationResult: """Tests for the DelegationResult dataclass.""" def test_delegation_result_success(self): """Test successful DelegationResult.""" result = DelegationResult( expert_name="librarian", task="Search for Docker info", success=True, output="Found 5 relevant documents about Docker...", ) assert result.expert_name == "librarian" assert result.success is True assert result.output.startswith("Found") assert result.error is None def test_delegation_result_failure(self): """Test failed DelegationResult.""" result = DelegationResult( expert_name="librarian", task="Search for Docker info", success=False, output="", error="Connection timeout to library-desk API", ) assert result.success is False assert result.output == "" assert result.error == "Connection timeout to library-desk API" @pytest.mark.unit class TestDelegateToLibrarian: """Tests for the delegate_to_librarian wrapper.""" @pytest.mark.asyncio async def test_delegate_to_librarian_success(self): """Test successful delegation to Librarian.""" mock_output = "Successfully created wiki page about CI/CD pipelines..." with patch( "src.agents.librarian.agent.run_librarian", new_callable=AsyncMock, return_value=mock_output, ) as mock_run: result = await delegate_to_librarian( task="Create a wiki page about CI/CD pipelines", context="User is setting up a homelab", ) # Verify run_librarian was called correctly mock_run.assert_called_once_with( task="Create a wiki page about CI/CD pipelines", context="User is setting up a homelab", ) # Verify result assert isinstance(result, DelegationResult) assert result.expert_name == "librarian" assert result.success is True assert result.output == mock_output assert result.error is None @pytest.mark.asyncio async def test_delegate_to_librarian_without_context(self): """Test delegation to Librarian without context.""" mock_output = "Found information about Docker networking..." with patch( "src.agents.librarian.agent.run_librarian", new_callable=AsyncMock, return_value=mock_output, ) as mock_run: result = await delegate_to_librarian( task="Search for information about Docker networking", ) mock_run.assert_called_once_with( task="Search for information about Docker networking", context="", ) assert result.success is True assert result.output == mock_output @pytest.mark.asyncio async def test_delegate_to_librarian_handles_error(self): """Test delegation maps Librarian errors to a user-safe result.""" with patch( "src.agents.librarian.agent.run_librarian", new_callable=AsyncMock, side_effect=Exception("Connection refused to http://internal:8089"), ): result = await delegate_to_librarian( task="Search for information", ) assert isinstance(result, DelegationResult) assert result.success is False # Output carries a curated butler-toned sentence assert result.output == get_think_message( "librarian", "Search for information", "error" ) # Exception detail stays in logs only - never in the result assert "Connection refused" not in result.output assert result.error is not None assert "Connection refused" not in result.error assert "internal" not in result.error @pytest.mark.asyncio async def test_delegate_to_librarian_timeout(self, monkeypatch): """Delegation is capped by LIBRARIAN_TIMEOUT and fails honestly.""" import asyncio from src.core.config import config async def slow_run(task, context=""): await asyncio.sleep(5) return "too late" monkeypatch.setattr(config, "LIBRARIAN_TIMEOUT", 0.05) with patch( "src.agents.librarian.agent.run_librarian", new=slow_run, ): result = await delegate_to_librarian(task="Search for information") assert result.success is False assert "longer than expected" in result.output assert result.error is not None assert "time budget" in result.error @pytest.mark.asyncio async def test_delegate_to_librarian_preserves_task(self): """Test delegation result preserves original task.""" original_task = "Create a wiki page about Kubernetes deployments" with patch( "src.agents.librarian.agent.run_librarian", new_callable=AsyncMock, return_value="Page created", ): result = await delegate_to_librarian(task=original_task) assert result.task == original_task @pytest.mark.unit class TestActionType: """Tests for the ActionType enum.""" def test_action_type_values(self): """Test ActionType enum values.""" assert ActionType.RETRIEVE.value == "retrieve" assert ActionType.RESEARCH.value == "research" assert ActionType.CREATE.value == "create" assert ActionType.CONTROL.value == "control" assert ActionType.RECORD.value == "record" def test_action_type_is_enum(self): """Test ActionType is proper enum.""" assert len(ActionType) == 5 @pytest.mark.unit class TestHouseholdThinkMessages: """Tests for HOUSEHOLD_THINK_MESSAGES mapping.""" def test_librarian_has_messages(self): """Test librarian has think messages.""" assert "librarian" in HOUSEHOLD_THINK_MESSAGES assert ActionType.RETRIEVE in HOUSEHOLD_THINK_MESSAGES["librarian"] assert ActionType.RESEARCH in HOUSEHOLD_THINK_MESSAGES["librarian"] assert ActionType.CREATE in HOUSEHOLD_THINK_MESSAGES["librarian"] def test_biographer_has_messages(self): """Test biographer has think messages.""" assert "biographer" in HOUSEHOLD_THINK_MESSAGES assert ActionType.RETRIEVE in HOUSEHOLD_THINK_MESSAGES["biographer"] assert ActionType.RECORD in HOUSEHOLD_THINK_MESSAGES["biographer"] def test_housekeeper_has_messages(self): """Test housekeeper has think messages.""" assert "housekeeper" in HOUSEHOLD_THINK_MESSAGES assert ActionType.RETRIEVE in HOUSEHOLD_THINK_MESSAGES["housekeeper"] assert ActionType.CONTROL in HOUSEHOLD_THINK_MESSAGES["housekeeper"] def test_messages_have_phases(self): """Test each action type has start/success/error messages.""" for expert, action_types in HOUSEHOLD_THINK_MESSAGES.items(): for action_type, messages in action_types.items(): assert "start" in messages, f"{expert}/{action_type} missing 'start'" assert "success" in messages, f"{expert}/{action_type} missing 'success'" assert "error" in messages, f"{expert}/{action_type} missing 'error'" def test_messages_are_plain_text(self): """Test messages are plain text (no wrappers - those go to reasoning_content).""" for expert, action_types in HOUSEHOLD_THINK_MESSAGES.items(): for action_type, messages in action_types.items(): for phase, msg in messages.items(): # Messages should NOT have wrappers - they go to reasoning_content field assert ( "" not in msg ), f"{expert}/{action_type}/{phase} should not have wrapper" assert ( "" not in msg ), f"{expert}/{action_type}/{phase} should not have wrapper" # Messages should be non-empty strings assert isinstance(msg, str) and len(msg) > 0, f"{expert}/{action_type}/{phase}" @pytest.mark.unit class TestDetectActionType: """Tests for _detect_action_type function.""" def test_librarian_search_is_retrieve(self): """Test librarian search tasks are RETRIEVE.""" assert _detect_action_type("librarian", "search for Docker info") == ActionType.RETRIEVE assert ( _detect_action_type("librarian", "find information about CI/CD") == ActionType.RETRIEVE ) assert _detect_action_type("librarian", "look up Kubernetes docs") == ActionType.RETRIEVE def test_librarian_web_search_is_research(self): """Test librarian web search tasks are RESEARCH.""" assert _detect_action_type("librarian", "search the web for news") == ActionType.RESEARCH assert _detect_action_type("librarian", "find online resources") == ActionType.RESEARCH assert _detect_action_type("librarian", "research internet sources") == ActionType.RESEARCH def test_librarian_create_is_create(self): """Test librarian creation tasks are CREATE.""" assert _detect_action_type("librarian", "create a wiki page") == ActionType.CREATE assert _detect_action_type("librarian", "write a new article") == ActionType.CREATE assert _detect_action_type("librarian", "add a new entry") == ActionType.CREATE def test_biographer_recall_is_retrieve(self): """Test biographer recall tasks are RETRIEVE.""" assert _detect_action_type("biographer", "what car do I drive?") == ActionType.RETRIEVE assert _detect_action_type("biographer", "what is my job?") == ActionType.RETRIEVE def test_biographer_record_is_record(self): """Test biographer record tasks are RECORD.""" assert ( _detect_action_type("biographer", "remember that I work at Acme") == ActionType.RECORD ) assert _detect_action_type("biographer", "note that my car is a Tesla") == ActionType.RECORD assert ( _detect_action_type("biographer", "save my preference for dark mode") == ActionType.RECORD ) def test_housekeeper_status_is_retrieve(self): """Test housekeeper status tasks are RETRIEVE.""" assert ( _detect_action_type("housekeeper", "what devices are in the bedroom?") == ActionType.RETRIEVE ) assert ( _detect_action_type("housekeeper", "is the living room light on?") == ActionType.RETRIEVE ) def test_housekeeper_control_is_control(self): """Test housekeeper control tasks are CONTROL.""" assert _detect_action_type("housekeeper", "turn on the lights") == ActionType.CONTROL assert _detect_action_type("housekeeper", "set brightness to 50%") == ActionType.CONTROL assert _detect_action_type("housekeeper", "activate the movie scene") == ActionType.CONTROL assert _detect_action_type("housekeeper", "toggle the fan") == ActionType.CONTROL @pytest.mark.unit class TestGetThinkMessage: """Tests for get_think_message function.""" def test_librarian_retrieve_start(self): """Test getting librarian retrieve start message.""" msg = get_think_message("librarian", "search for Docker", "start") # No wrappers - messages go to reasoning_content field assert "" not in msg assert "archives" in msg.lower() or "consult" in msg.lower() def test_librarian_create_success(self): """Test getting librarian create success message.""" msg = get_think_message("librarian", "create a wiki page", "success") assert "" not in msg assert "catalogued" in msg.lower() def test_biographer_record_start(self): """Test getting biographer record start message.""" msg = get_think_message("biographer", "remember my preference", "start") assert "" not in msg assert "note" in msg.lower() or "biographer" in msg.lower() def test_housekeeper_control_success(self): """Test getting housekeeper control success message.""" msg = get_think_message("housekeeper", "turn on the lights", "success") assert "" not in msg assert "configured" in msg.lower() def test_unknown_expert_fallback(self): """Test unknown expert gets fallback message.""" msg = get_think_message("unknown_expert", "some task", "start") assert "" not in msg assert "unknown_expert" in msg.lower()