jpmschweitzerandClaude Opus 4.5
9980e4764c
Build and Push / build (release) Successful in 1m49s
Messages in reasoning_content should be plain text, not wrapped in <think> tags. Removed wrappers from: - delegation.py household think messages - orchestration.py status messages 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
360 lines
14 KiB
Python
360 lines
14 KiB
Python
"""
|
|
Tests for delegation infrastructure.
|
|
|
|
Tests the DelegationTask dataclass and delegation wrapper functions
|
|
that implement the agent-as-tool pattern.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
|
|
from src.agents.delegation import (
|
|
ActionType,
|
|
DelegationTask,
|
|
DelegationResult,
|
|
HOUSEHOLD_THINK_MESSAGES,
|
|
STREAMING_DELEGATION_WRAPPERS,
|
|
delegate_to_librarian,
|
|
get_think_message,
|
|
_detect_action_type,
|
|
)
|
|
|
|
|
|
@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 handles Librarian errors gracefully."""
|
|
with patch(
|
|
"src.agents.librarian.agent.run_librarian",
|
|
new_callable=AsyncMock,
|
|
side_effect=Exception("Connection refused"),
|
|
):
|
|
result = await delegate_to_librarian(
|
|
task="Search for information",
|
|
)
|
|
|
|
assert isinstance(result, DelegationResult)
|
|
assert result.success is False
|
|
assert result.output == ""
|
|
assert result.error == "Connection refused"
|
|
|
|
@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 <think> 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 <think> wrappers - they go to reasoning_content field
|
|
assert "<think>" not in msg, f"{expert}/{action_type}/{phase} should not have <think> wrapper"
|
|
assert "</think>" not in msg, f"{expert}/{action_type}/{phase} should not have </think> 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 <think> wrappers - messages go to reasoning_content field
|
|
assert "<think>" 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 "<think>" 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 "<think>" 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 "<think>" 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 "<think>" not in msg
|
|
assert "unknown_expert" in msg.lower()
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestStreamingDelegationWrappers:
|
|
"""Tests for streaming delegation wrapper mapping."""
|
|
|
|
def test_streaming_wrappers_exist(self):
|
|
"""Test streaming wrappers mapping has all experts."""
|
|
assert "librarian" in STREAMING_DELEGATION_WRAPPERS
|
|
assert "biographer" in STREAMING_DELEGATION_WRAPPERS
|
|
assert "housekeeper" in STREAMING_DELEGATION_WRAPPERS
|
|
|
|
def test_streaming_wrappers_are_async_generators(self):
|
|
"""Test streaming wrappers are async generator functions."""
|
|
import inspect
|
|
for name, wrapper in STREAMING_DELEGATION_WRAPPERS.items():
|
|
assert inspect.isasyncgenfunction(wrapper), f"{name} is not an async generator"
|