diff --git a/tests/agents/test_orchestration.py b/tests/agents/test_orchestration.py index fae162b..29d2e67 100644 --- a/tests/agents/test_orchestration.py +++ b/tests/agents/test_orchestration.py @@ -2,7 +2,8 @@ Tests for orchestration module. Tests the multi-expert coordination infrastructure including -delegation parsing, think updates, and result handling. +delegation parsing, think updates, result handling, and +multi-expert sequential/parallel execution. """ import pytest from unittest.mock import AsyncMock, patch @@ -13,6 +14,12 @@ from src.agents.orchestration import ( execute_delegation, orchestrate_with_think_updates, extract_delegation_context, + ExecutionMode, + MultiExpertResult, + execute_sequential, + execute_parallel, + orchestrate_multi_expert, + _get_display_name, ) from src.agents.delegation import DelegationTask, DelegationResult @@ -351,3 +358,404 @@ class TestOrchestrationContext: ) assert ctx.conversation_id is None + + +# ============================================================================ +# Multi-Expert Coordination Tests +# ============================================================================ + +@pytest.mark.unit +class TestMultiExpertResult: + """Tests for MultiExpertResult aggregation.""" + + def test_result_creation(self): + """Test creating empty MultiExpertResult.""" + result = MultiExpertResult() + + assert result.results == {} + assert result.all_succeeded is True + assert result.failed_experts == [] + assert result.combined_output == "" + + def test_add_successful_result(self): + """Test adding a successful result.""" + result = MultiExpertResult() + + delegation_result = DelegationResult( + expert_name="librarian", + task="search docs", + success=True, + output="Found docs", + ) + result.add_result(delegation_result) + + assert "librarian" in result.results + assert result.all_succeeded is True + assert result.failed_experts == [] + + def test_add_failed_result(self): + """Test adding a failed result.""" + result = MultiExpertResult() + + delegation_result = DelegationResult( + expert_name="librarian", + task="search docs", + success=False, + output="", + error="Connection error", + ) + result.add_result(delegation_result) + + assert "librarian" in result.results + assert result.all_succeeded is False + assert "librarian" in result.failed_experts + + def test_aggregate_outputs(self): + """Test aggregating outputs from multiple experts.""" + result = MultiExpertResult() + + result.add_result(DelegationResult( + expert_name="librarian", + task="search docs", + success=True, + output="Found Docker docs", + )) + result.add_result(DelegationResult( + expert_name="memory", + task="get preferences", + success=True, + output="User prefers dark mode", + )) + + combined = result.aggregate_outputs() + + assert "Librarian" in combined + assert "Found Docker docs" in combined + assert "Memory" in combined + assert "dark mode" in combined + + def test_aggregate_excludes_failed(self): + """Test that failed results are excluded from aggregate.""" + result = MultiExpertResult() + + result.add_result(DelegationResult( + expert_name="librarian", + task="search", + success=True, + output="Success output", + )) + result.add_result(DelegationResult( + expert_name="memory", + task="get", + success=False, + output="", + error="Failed", + )) + + combined = result.aggregate_outputs() + + assert "Success output" in combined + assert "Failed" not in combined + + +@pytest.mark.unit +class TestExecuteSequential: + """Tests for sequential multi-expert execution.""" + + @pytest.mark.asyncio + async def test_sequential_all_succeed(self): + """Test sequential execution when all tasks succeed.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + ] + + mock_results = [ + DelegationResult(expert_name="librarian", task="task 1", success=True, output="Result 1"), + DelegationResult(expert_name="memory", task="task 2", success=True, output="Result 2"), + ] + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_results, + ): + result = await execute_sequential(tasks) + + assert result.all_succeeded is True + assert len(result.results) == 2 + assert result.failed_experts == [] + + @pytest.mark.asyncio + async def test_sequential_with_failure(self): + """Test sequential execution when a task fails.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + ] + + mock_results = [ + DelegationResult(expert_name="librarian", task="task 1", success=True, output="OK"), + DelegationResult(expert_name="memory", task="task 2", success=False, output="", error="Failed"), + ] + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_results, + ): + result = await execute_sequential(tasks) + + assert result.all_succeeded is False + assert len(result.results) == 2 + assert "memory" in result.failed_experts + + @pytest.mark.asyncio + async def test_sequential_stop_on_failure(self): + """Test sequential execution stops on failure when configured.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + DelegationTask(expert_name="librarian", task="task 3"), + ] + + mock_results = [ + DelegationResult(expert_name="librarian", task="task 1", success=False, output="", error="Error"), + ] + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_results, + ): + result = await execute_sequential(tasks, stop_on_failure=True) + + # Should only have 1 result (stopped after first failure) + assert len(result.results) == 1 + assert result.all_succeeded is False + + +@pytest.mark.unit +class TestExecuteParallel: + """Tests for parallel multi-expert execution.""" + + @pytest.mark.asyncio + async def test_parallel_all_succeed(self): + """Test parallel execution when all tasks succeed.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + ] + + mock_results = [ + DelegationResult(expert_name="librarian", task="task 1", success=True, output="Result 1"), + DelegationResult(expert_name="memory", task="task 2", success=True, output="Result 2"), + ] + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_results, + ): + result = await execute_parallel(tasks) + + assert result.all_succeeded is True + assert len(result.results) == 2 + + @pytest.mark.asyncio + async def test_parallel_with_failure(self): + """Test parallel execution with partial failure.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + ] + + mock_results = [ + DelegationResult(expert_name="librarian", task="task 1", success=True, output="OK"), + DelegationResult(expert_name="memory", task="task 2", success=False, output="", error="Timeout"), + ] + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_results, + ): + result = await execute_parallel(tasks) + + assert result.all_succeeded is False + assert len(result.results) == 2 + assert "memory" in result.failed_experts + + @pytest.mark.asyncio + async def test_parallel_handles_exception(self): + """Test parallel execution handles exceptions gracefully.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + ] + + async def mock_execute(task): + if task.expert_name == "memory": + raise RuntimeError("Connection lost") + return DelegationResult( + expert_name=task.expert_name, + task=task.task, + success=True, + output="OK", + ) + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_execute, + ): + result = await execute_parallel(tasks) + + assert result.all_succeeded is False + assert "memory" in result.failed_experts + assert "Connection lost" in result.results["memory"].error + + +@pytest.mark.unit +class TestOrchestrateMultiExpert: + """Tests for multi-expert orchestration with think updates.""" + + @pytest.mark.asyncio + async def test_orchestrate_sequential_emits_think_updates(self): + """Test sequential orchestration emits think updates for each task.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + ] + + mock_results = [ + DelegationResult(expert_name="librarian", task="task 1", success=True, output="Result 1"), + DelegationResult(expert_name="memory", task="task 2", success=True, output="Result 2"), + ] + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_results, + ): + updates = [] + async for update in orchestrate_multi_expert(tasks, mode=ExecutionMode.SEQUENTIAL): + updates.append(update) + + all_output = "".join(updates) + + # Should have think updates for both experts + assert "Consulting" in all_output + assert "completed" in all_output + assert "Librarian" in all_output + + @pytest.mark.asyncio + async def test_orchestrate_parallel_emits_think_updates(self): + """Test parallel orchestration emits appropriate think updates.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + DelegationTask(expert_name="memory", task="task 2"), + ] + + mock_results = [ + DelegationResult(expert_name="librarian", task="task 1", success=True, output="Result 1"), + DelegationResult(expert_name="memory", task="task 2", success=True, output="Result 2"), + ] + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + side_effect=mock_results, + ): + updates = [] + async for update in orchestrate_multi_expert(tasks, mode=ExecutionMode.PARALLEL): + updates.append(update) + + all_output = "".join(updates) + + # Should mention parallel execution + assert "parallel" in all_output + + @pytest.mark.asyncio + async def test_orchestrate_empty_tasks_yields_nothing(self): + """Test orchestration with empty tasks yields nothing.""" + updates = [] + async for update in orchestrate_multi_expert([]): + updates.append(update) + + assert len(updates) == 0 + + @pytest.mark.asyncio + async def test_orchestrate_success_summary(self): + """Test orchestration emits success summary when all succeed.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + ] + + mock_result = DelegationResult( + expert_name="librarian", + task="task 1", + success=True, + output="Done", + ) + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + return_value=mock_result, + ): + updates = [] + async for update in orchestrate_multi_expert(tasks): + updates.append(update) + + all_output = "".join(updates) + + # Should have success message + assert "🎉" in all_output or "successfully" in all_output.lower() + + @pytest.mark.asyncio + async def test_orchestrate_failure_summary(self): + """Test orchestration emits failure summary when some fail.""" + tasks = [ + DelegationTask(expert_name="librarian", task="task 1"), + ] + + mock_result = DelegationResult( + expert_name="librarian", + task="task 1", + success=False, + output="", + error="Failed", + ) + + with patch( + "src.agents.orchestration.execute_delegation", + new_callable=AsyncMock, + return_value=mock_result, + ): + updates = [] + async for update in orchestrate_multi_expert(tasks): + updates.append(update) + + all_output = "".join(updates) + + # Should mention failure + assert "⚠️" in all_output or "failed" in all_output.lower() + + +@pytest.mark.unit +class TestGetDisplayName: + """Tests for _get_display_name helper.""" + + def test_librarian_display_name(self): + """Test librarian gets 'The Librarian' display name.""" + assert _get_display_name("librarian") == "The Librarian" + + def test_memory_display_name(self): + """Test memory gets 'Memory' display name.""" + assert _get_display_name("memory") == "Memory" + + def test_unknown_expert_title_case(self): + """Test unknown expert gets title-cased name.""" + assert _get_display_name("some_expert") == "Some_Expert" + assert _get_display_name("newagent") == "Newagent"