jpmschweitzer/tatlock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

test: add multi-expert coordination tests

Browse Source 
Comprehensive tests for Phase E multi-expert coordination:

MultiExpertResult:
- Result creation and default values
- Adding successful/failed results
- Output aggregation (excludes failed)

Sequential execution:
- All tasks succeed
- Partial failure handling
- Stop-on-failure mode

Parallel execution:
- All tasks succeed concurrently
- Partial failure handling
- Exception handling (graceful degradation)

Orchestration with think updates:
- Sequential mode think updates
- Parallel mode think updates
- Success/failure summaries
- Empty task handling

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2025-12-13 13:23:13 +01:00
co-authored by Claude Opus 4.5
parent 1e4bba2422
commit c049c1e354
1 changed files with 409 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/agents/test_orchestration.py
+409 -1
View File
@@ -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"