jpmschweitzerandClaude Opus 4.5
c049c1e354
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>
762 lines
25 KiB
Python
762 lines
25 KiB
Python
"""
|
|
Tests for orchestration module.
|
|
|
|
Tests the multi-expert coordination infrastructure including
|
|
delegation parsing, think updates, result handling, and
|
|
multi-expert sequential/parallel execution.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from src.agents.orchestration import (
|
|
OrchestrationContext,
|
|
parse_delegation_from_steward_note,
|
|
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
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestParseDelegation:
|
|
"""Tests for parsing delegation from Steward's note."""
|
|
|
|
def test_parse_librarian_create(self):
|
|
"""Test parsing librarian create delegation."""
|
|
note = """DELEGATE: librarian to create a wiki page about CI/CD pipelines
|
|
REASON: User wants to document CI/CD concepts
|
|
COMPLEXITY: moderate
|
|
CONTEXT: none"""
|
|
|
|
task = parse_delegation_from_steward_note(note)
|
|
|
|
assert task is not None
|
|
assert task.expert_name == "librarian"
|
|
assert "create a wiki page about CI/CD pipelines" in task.task
|
|
|
|
def test_parse_librarian_search(self):
|
|
"""Test parsing librarian search delegation."""
|
|
note = """DELEGATE: librarian to search for information about Docker networking
|
|
REASON: User needs Docker documentation
|
|
COMPLEXITY: simple"""
|
|
|
|
task = parse_delegation_from_steward_note(note)
|
|
|
|
assert task is not None
|
|
assert task.expert_name == "librarian"
|
|
assert "search for information about Docker networking" in task.task
|
|
|
|
def test_parse_no_delegation(self):
|
|
"""Test parsing when no delegation needed."""
|
|
note = """DELEGATE: none (conversational response only)
|
|
REASON: Simple greeting requires no tools
|
|
COMPLEXITY: simple"""
|
|
|
|
task = parse_delegation_from_steward_note(note)
|
|
|
|
assert task is None
|
|
|
|
def test_parse_tatlock_core(self):
|
|
"""Test parsing tatlock_core delegation."""
|
|
note = """DELEGATE: tatlock_core to calculate the result
|
|
REASON: Math calculation needed
|
|
COMPLEXITY: simple"""
|
|
|
|
task = parse_delegation_from_steward_note(note)
|
|
|
|
assert task is not None
|
|
assert task.expert_name == "tatlock_core"
|
|
assert "calculate the result" in task.task
|
|
|
|
def test_parse_case_insensitive(self):
|
|
"""Test parsing is case insensitive."""
|
|
note = """delegate: LIBRARIAN to search docs
|
|
reason: Research query"""
|
|
|
|
task = parse_delegation_from_steward_note(note)
|
|
|
|
assert task is not None
|
|
assert task.expert_name == "librarian"
|
|
|
|
def test_parse_missing_delegate(self):
|
|
"""Test parsing when DELEGATE line is missing."""
|
|
note = """REASON: This has no delegation
|
|
COMPLEXITY: simple"""
|
|
|
|
task = parse_delegation_from_steward_note(note)
|
|
|
|
assert task is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestExtractDelegationContext:
|
|
"""Tests for extracting context from Steward's note."""
|
|
|
|
def test_extract_all_fields(self):
|
|
"""Test extracting all context fields."""
|
|
note = """DELEGATE: librarian to create wiki page
|
|
REASON: User wants documentation
|
|
COMPLEXITY: moderate
|
|
CONTEXT: Related to previous discussion about DevOps"""
|
|
|
|
context = extract_delegation_context(note)
|
|
|
|
assert context["reason"] == "User wants documentation"
|
|
assert context["complexity"] == "moderate"
|
|
assert "Related to previous discussion" in context["context"]
|
|
|
|
def test_extract_partial_fields(self):
|
|
"""Test extracting when some fields missing."""
|
|
note = """DELEGATE: librarian to search
|
|
REASON: Research query
|
|
COMPLEXITY: simple"""
|
|
|
|
context = extract_delegation_context(note)
|
|
|
|
assert context["reason"] == "Research query"
|
|
assert context["complexity"] == "simple"
|
|
assert context["context"] == ""
|
|
|
|
def test_extract_empty_note(self):
|
|
"""Test extracting from empty note."""
|
|
context = extract_delegation_context("")
|
|
|
|
assert context["reason"] == ""
|
|
assert context["complexity"] == ""
|
|
assert context["context"] == ""
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestExecuteDelegation:
|
|
"""Tests for executing delegation tasks."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_librarian_delegation(self):
|
|
"""Test executing delegation to librarian."""
|
|
task = DelegationTask(
|
|
expert_name="librarian",
|
|
task="search for Docker docs",
|
|
context="User learning Docker",
|
|
)
|
|
|
|
mock_result = DelegationResult(
|
|
expert_name="librarian",
|
|
task="search for Docker docs",
|
|
success=True,
|
|
output="Found Docker documentation...",
|
|
)
|
|
|
|
with patch(
|
|
"src.agents.orchestration.delegate_to_librarian",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_result,
|
|
) as mock_delegate:
|
|
result = await execute_delegation(task)
|
|
|
|
mock_delegate.assert_called_once_with(
|
|
task="search for Docker docs",
|
|
context="User learning Docker",
|
|
)
|
|
|
|
assert result.success is True
|
|
assert "Docker" in result.output
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_unknown_expert(self):
|
|
"""Test executing delegation to unknown expert."""
|
|
task = DelegationTask(
|
|
expert_name="unknown_expert",
|
|
task="do something",
|
|
)
|
|
|
|
result = await execute_delegation(task)
|
|
|
|
assert result.success is False
|
|
assert "Unknown expert" in result.error
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestOrchestrateWithThinkUpdates:
|
|
"""Tests for orchestration with think updates."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orchestrate_emits_think_before_delegation(self):
|
|
"""Test that think update is emitted before delegation."""
|
|
mock_result = DelegationResult(
|
|
expert_name="librarian",
|
|
task="search docs",
|
|
success=True,
|
|
output="Found results",
|
|
)
|
|
|
|
with patch(
|
|
"src.agents.orchestration.delegate_to_librarian",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_result,
|
|
):
|
|
updates = []
|
|
async for update in orchestrate_with_think_updates(
|
|
user_message="Search for Docker info",
|
|
steward_note="DELEGATE: librarian to search for Docker info",
|
|
):
|
|
updates.append(update)
|
|
|
|
# First update should be think tag about consulting
|
|
assert any("<think>" in u and "Consulting" in u for u in updates)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orchestrate_emits_think_after_delegation(self):
|
|
"""Test that think update is emitted after delegation."""
|
|
mock_result = DelegationResult(
|
|
expert_name="librarian",
|
|
task="search docs",
|
|
success=True,
|
|
output="Found results",
|
|
)
|
|
|
|
with patch(
|
|
"src.agents.orchestration.delegate_to_librarian",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_result,
|
|
):
|
|
updates = []
|
|
async for update in orchestrate_with_think_updates(
|
|
user_message="Search for Docker info",
|
|
steward_note="DELEGATE: librarian to search for Docker info",
|
|
):
|
|
updates.append(update)
|
|
|
|
# Should have think tag about completion
|
|
assert any("<think>" in u and "completed" in u for u in updates)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orchestrate_yields_expert_output(self):
|
|
"""Test that expert output is yielded."""
|
|
mock_result = DelegationResult(
|
|
expert_name="librarian",
|
|
task="search docs",
|
|
success=True,
|
|
output="Found Docker documentation with networking details",
|
|
)
|
|
|
|
with patch(
|
|
"src.agents.orchestration.delegate_to_librarian",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_result,
|
|
):
|
|
updates = []
|
|
async for update in orchestrate_with_think_updates(
|
|
user_message="Search for Docker info",
|
|
steward_note="DELEGATE: librarian to search for Docker info",
|
|
):
|
|
updates.append(update)
|
|
|
|
# Should include expert output
|
|
all_output = "".join(updates)
|
|
assert "Docker documentation" in all_output
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orchestrate_handles_delegation_failure(self):
|
|
"""Test that delegation failure emits warning think update."""
|
|
mock_result = DelegationResult(
|
|
expert_name="librarian",
|
|
task="search docs",
|
|
success=False,
|
|
output="",
|
|
error="Connection timeout",
|
|
)
|
|
|
|
with patch(
|
|
"src.agents.orchestration.delegate_to_librarian",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_result,
|
|
):
|
|
updates = []
|
|
async for update in orchestrate_with_think_updates(
|
|
user_message="Search for info",
|
|
steward_note="DELEGATE: librarian to search",
|
|
):
|
|
updates.append(update)
|
|
|
|
# Should have warning think update
|
|
all_output = "".join(updates)
|
|
assert "⚠️" in all_output or "issue" in all_output.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orchestrate_no_delegation_returns_empty(self):
|
|
"""Test that no delegation yields nothing."""
|
|
updates = []
|
|
async for update in orchestrate_with_think_updates(
|
|
user_message="Hello",
|
|
steward_note="DELEGATE: none (conversational)",
|
|
):
|
|
updates.append(update)
|
|
|
|
assert len(updates) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orchestrate_with_preparsed_task(self):
|
|
"""Test orchestration with pre-parsed delegation task."""
|
|
task = DelegationTask(
|
|
expert_name="librarian",
|
|
task="create wiki page",
|
|
)
|
|
|
|
mock_result = DelegationResult(
|
|
expert_name="librarian",
|
|
task="create wiki page",
|
|
success=True,
|
|
output="Wiki page created",
|
|
)
|
|
|
|
with patch(
|
|
"src.agents.orchestration.delegate_to_librarian",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_result,
|
|
):
|
|
updates = []
|
|
async for update in orchestrate_with_think_updates(
|
|
user_message="Create wiki page",
|
|
steward_note="", # Empty note since task is pre-parsed
|
|
delegation_task=task,
|
|
):
|
|
updates.append(update)
|
|
|
|
assert len(updates) > 0
|
|
all_output = "".join(updates)
|
|
assert "Wiki page created" in all_output
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestOrchestrationContext:
|
|
"""Tests for OrchestrationContext dataclass."""
|
|
|
|
def test_context_creation(self):
|
|
"""Test creating orchestration context."""
|
|
ctx = OrchestrationContext(
|
|
user_message="Test message",
|
|
steward_note="Test note",
|
|
conversation_id="conv_123",
|
|
)
|
|
|
|
assert ctx.user_message == "Test message"
|
|
assert ctx.steward_note == "Test note"
|
|
assert ctx.conversation_id == "conv_123"
|
|
|
|
def test_context_defaults(self):
|
|
"""Test orchestration context default values."""
|
|
ctx = OrchestrationContext(
|
|
user_message="Test",
|
|
steward_note="Note",
|
|
)
|
|
|
|
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"
|