From 31b948a748c683612ce462aa06c561d990013c34 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Tue, 14 Jul 2026 11:52:36 +0200 Subject: [PATCH] refactor(agents): delete dead coordination/streaming delegation stack One delegation implementation remains (src/agents/delegation.py). Removed, after verifying zero live importers post-Phase-A/B: - src/agents/coordination.py: CoordinationEngine, duplicate delegate_to_librarian, AGENT_EXECUTORS/AGENT_STREAM_EXECUTORS (only importer was its own test module) - run_librarian_stream: documented-broken path (Ollama streaming + tool call bug, PydanticAI #1292/#2256), only called by the deleted coordination engine - stream_delegate_to_* wrappers + STREAMING_DELEGATION_WRAPPERS and the never-parsed __DELEGATION_RESULT__ marker in delegation.py - HouseholdRegistry.get_streaming_delegation_tools() (no callers) - tests/agents/test_coordination.py and the wrapper/stream tests Note: the STREAMING_DELEGATION_WRAPPERS import in src/responses/streaming.py was already removed by Phase A (7ce1c1a); nothing to delete there. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L --- CHANGELOG.md | 4 + src/agents/coordination.py | 412 ------------------ src/agents/delegation.py | 98 ----- src/agents/librarian/__init__.py | 2 - src/agents/librarian/agent.py | 64 --- src/core/household_registry.py | 59 --- tests/agents/librarian/test_agent_failures.py | 31 +- tests/agents/test_coordination.py | 339 -------------- tests/agents/test_delegation.py | 18 - 9 files changed, 9 insertions(+), 1018 deletions(-) delete mode 100644 src/agents/coordination.py delete mode 100644 tests/agents/test_coordination.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9dbae26..1f05398 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Removed + +- **Dead delegation stack** - deleted the duplicate, never-wired coordination layer so exactly ONE delegation implementation remains (`src/agents/delegation.py`): `src/agents/coordination.py` (`CoordinationEngine`, its own `delegate_to_librarian`, `AGENT_EXECUTORS`/`AGENT_STREAM_EXECUTORS`), the broken-by-design `run_librarian_stream` path it used (Ollama streaming + tool call bug), the `stream_delegate_to_*` wrappers with their never-parsed `__DELEGATION_RESULT__` marker, and `HouseholdRegistry.get_streaming_delegation_tools()` (no callers) + ### Added - **Test-suite tenant guard** - `tests/conftest.py` hard-fails the whole pytest session (exit code 1, zero tests run) if the effective tenant resolves to the production tenant `jpmschweitzer`, mirroring the guard library-desk applies on its side. Suite-level assertions pin that the session runs under `llm_tester` namespaces (Qdrant `memories_llm_tester`, Redis `session:llm_tester:*`), and the e2e isolation constants now derive from the shared `TEST_TENANT`/`PRODUCTION_TENANT` config constants instead of string literals diff --git a/src/agents/coordination.py b/src/agents/coordination.py deleted file mode 100644 index 1039712..0000000 --- a/src/agents/coordination.py +++ /dev/null @@ -1,412 +0,0 @@ -""" -Multi-agent coordination engine. - -Orchestrates delegation from Tatlock to expert agents (Librarian, etc.) -based on Steward recommendations. Handles: -- Routing tasks to appropriate agents -- Parallel and sequential execution -- Result aggregation -- Error handling and graceful degradation -""" -import asyncio -import time -from collections.abc import AsyncGenerator -from typing import Any - -from src.agents.librarian import run_librarian, run_librarian_stream -from src.agents.protocol import ( - AgentError, - AgentRequest, - AgentResponse, - AgentTimeoutError, - AgentUnavailableError, - CoordinationResult, - DelegationIntent, - DelegationReason, -) -from src.core.config import config -from src.core.household_registry import get_household_registry -from src.core.logging_config import get_logger - -logger = get_logger(__name__) - - -# Agent execution functions registry -AGENT_EXECUTORS: dict[str, Any] = { - "librarian": run_librarian, -} - -AGENT_STREAM_EXECUTORS: dict[str, Any] = { - "librarian": run_librarian_stream, -} - - -class CoordinationEngine: - """ - Coordinates multi-agent task execution. - - Routes tasks from Tatlock to appropriate expert agents, - handles execution, and aggregates results. - """ - - def __init__(self): - """Initialize the coordination engine.""" - self.registry = get_household_registry() - logger.info("coordination_engine_initialized") - - def get_available_agents(self) -> list[str]: - """ - Get list of available expert agents. - - Returns: - List of agent names that can accept delegations - """ - available = [] - for name in self.registry.list_members(): - member = self.registry.get_member(name) - if member and member.agent is not None: - available.append(name) - return available - - def can_delegate_to(self, agent_name: str) -> bool: - """ - Check if delegation to an agent is possible. - - Args: - agent_name: Name of the target agent - - Returns: - True if agent is available and can accept tasks - """ - if agent_name not in AGENT_EXECUTORS: - return False - - member = self.registry.get_member(agent_name) - return member is not None and member.agent is not None - - async def execute_delegation( - self, - intent: DelegationIntent, - context: str = "", - message_history: list[Any] | None = None, - ) -> AgentResponse: - """ - Execute a single delegation to an expert agent. - - Args: - intent: The delegation intent with task details - context: Additional context for the agent - message_history: Optional conversation history - - Returns: - AgentResponse with results - - Raises: - AgentUnavailableError: If agent is not available - AgentTimeoutError: If execution times out - AgentError: For other execution errors - """ - start_time = time.time() - agent_name = intent.target_agent - - logger.info( - "delegation_started", - agent=agent_name, - task=intent.task[:100], - reason=intent.reason.value, - ) - - # Check if agent is available - if not self.can_delegate_to(agent_name): - raise AgentUnavailableError( - f"Agent '{agent_name}' is not available for delegation", - agent_name=agent_name, - ) - - # Get the executor - executor = AGENT_EXECUTORS.get(agent_name) - if not executor: - raise AgentUnavailableError( - f"No executor found for agent '{agent_name}'", - agent_name=agent_name, - ) - - try: - # Build the request - request = AgentRequest( - task=intent.task, - context=context, - delegation_reason=intent.reason, - ) - - # Execute with timeout (explicit request value or configured budget) - timeout = request.timeout_seconds or config.LIBRARIAN_TIMEOUT - - result = await asyncio.wait_for( - executor( - task=request.task, - context=request.context, - message_history=message_history, - ), - timeout=timeout, - ) - - duration_ms = int((time.time() - start_time) * 1000) - - logger.info( - "delegation_completed", - agent=agent_name, - duration_ms=duration_ms, - output_length=len(result), - ) - - return AgentResponse( - success=True, - result=result, - reasoning=f"Delegated to {agent_name}: {intent.expected_outcome}", - duration_ms=duration_ms, - ) - - except TimeoutError as e: - duration_ms = int((time.time() - start_time) * 1000) - logger.error( - "delegation_timeout", - agent=agent_name, - duration_ms=duration_ms, - ) - raise AgentTimeoutError( - f"Agent '{agent_name}' timed out after {duration_ms}ms", - agent_name=agent_name, - ) from e - - except Exception as e: - duration_ms = int((time.time() - start_time) * 1000) - logger.error( - "delegation_error", - agent=agent_name, - error=str(e), - duration_ms=duration_ms, - exc_info=True, - ) - return AgentResponse( - success=False, - result="", - error_message=str(e), - duration_ms=duration_ms, - ) - - async def execute_delegation_stream( - self, - intent: DelegationIntent, - context: str = "", - message_history: list[Any] | None = None, - ) -> AsyncGenerator[str, None]: - """ - Execute a delegation with streaming output. - - Args: - intent: The delegation intent with task details - context: Additional context for the agent - message_history: Optional conversation history - - Yields: - Text deltas from the agent - - Raises: - AgentUnavailableError: If agent is not available - """ - agent_name = intent.target_agent - - logger.info( - "delegation_stream_started", - agent=agent_name, - task=intent.task[:100], - ) - - # Check if agent is available - if agent_name not in AGENT_STREAM_EXECUTORS: - raise AgentUnavailableError( - f"Agent '{agent_name}' does not support streaming", - agent_name=agent_name, - ) - - executor = AGENT_STREAM_EXECUTORS[agent_name] - - try: - async for delta in executor( - task=intent.task, - context=context, - message_history=message_history, - ): - yield delta - - logger.info("delegation_stream_completed", agent=agent_name) - - except Exception as e: - # Exception detail stays in the logs; yield a curated - # user-safe sentence instead of leaking internals. - from src.agents.delegation import get_think_message - - logger.error( - "delegation_stream_error", - agent=agent_name, - error=str(e), - exc_info=True, - ) - yield "\n\n" + get_think_message(agent_name, intent.task, "error") - - async def coordinate( - self, - intents: list[DelegationIntent], - context: str = "", - message_history: list[Any] | None = None, - ) -> CoordinationResult: - """ - Coordinate execution of multiple delegations. - - Handles parallel execution for independent tasks and - sequential execution for dependent tasks. - - Args: - intents: List of delegation intents to execute - context: Shared context for all agents - message_history: Optional conversation history - - Returns: - CoordinationResult with aggregated results - """ - start_time = time.time() - agent_responses: dict[str, AgentResponse] = {} - agents_consulted: list[str] = [] - - logger.info( - "coordination_started", - intent_count=len(intents), - agents=[i.target_agent for i in intents], - ) - - # Sort by priority - sorted_intents = sorted(intents, key=lambda x: x.priority) - - # Group by dependencies (simple version: sequential for now) - # TODO: Implement parallel execution for independent tasks - for intent in sorted_intents: - try: - response = await self.execute_delegation( - intent=intent, - context=context, - message_history=message_history, - ) - agent_responses[intent.target_agent] = response - if response.success: - agents_consulted.append(intent.target_agent) - - except AgentError as e: - agent_responses[intent.target_agent] = AgentResponse( - success=False, - result="", - error_message=str(e), - ) - - # Aggregate results - successful_results = [ - r.result for r in agent_responses.values() if r.success and r.result - ] - - final_response = "\n\n---\n\n".join(successful_results) if successful_results else "" - - total_duration = int((time.time() - start_time) * 1000) - - logger.info( - "coordination_completed", - total_duration_ms=total_duration, - agents_consulted=agents_consulted, - success_count=len(successful_results), - ) - - return CoordinationResult( - final_response=final_response, - agent_responses=agent_responses, - delegation_intents=intents, - total_duration_ms=total_duration, - agents_consulted=agents_consulted, - ) - - -# Global coordination engine instance -_coordination_engine: CoordinationEngine | None = None - - -def get_coordination_engine() -> CoordinationEngine: - """Get the global coordination engine instance.""" - global _coordination_engine - if _coordination_engine is None: - _coordination_engine = CoordinationEngine() - return _coordination_engine - - -async def delegate_to_librarian( - task: str, - context: str = "", - reason: DelegationReason = DelegationReason.DOMAIN_EXPERTISE, - message_history: list[Any] | None = None, -) -> AgentResponse: - """ - Convenience function to delegate a task to The Librarian. - - Args: - task: Research task description - context: Additional context - reason: Why delegating to Librarian - message_history: Optional conversation history - - Returns: - AgentResponse with research results - """ - engine = get_coordination_engine() - - intent = DelegationIntent( - target_agent="librarian", - task=task, - reason=reason, - expected_outcome="Research findings and relevant information", - ) - - return await engine.execute_delegation( - intent=intent, - context=context, - message_history=message_history, - ) - - -async def delegate_to_librarian_stream( - task: str, - context: str = "", - message_history: list[Any] | None = None, -) -> AsyncGenerator[str, None]: - """ - Convenience function to delegate to Librarian with streaming. - - Args: - task: Research task description - context: Additional context - message_history: Optional conversation history - - Yields: - Text deltas from The Librarian - """ - engine = get_coordination_engine() - - intent = DelegationIntent( - target_agent="librarian", - task=task, - reason=DelegationReason.DOMAIN_EXPERTISE, - expected_outcome="Research findings", - ) - - async for delta in engine.execute_delegation_stream( - intent=intent, - context=context, - message_history=message_history, - ): - yield delta diff --git a/src/agents/delegation.py b/src/agents/delegation.py index c538839..8b3ef2a 100644 --- a/src/agents/delegation.py +++ b/src/agents/delegation.py @@ -9,7 +9,6 @@ This implements the agent-as-tool pattern recommended by PydanticAI: agents call other agents via tool wrappers, keeping each agent focused. """ import asyncio -from collections.abc import AsyncGenerator from dataclasses import dataclass, field from enum import Enum @@ -572,103 +571,6 @@ async def delegate_to_housekeeper( ) -# ============================================================================= -# Streaming Delegation Wrappers (with Think Messages) -# ============================================================================= - -async def stream_delegate_to_librarian( - task: str, - context: str = "", -) -> AsyncGenerator[str, None]: - """ - Stream delegation to Librarian with automatic think messages. - - Yields butler-perspective think messages before and after the delegation, - allowing the UI to show progress to the user. - - Args: - task: Task description - context: Additional context - - Yields: - str: Think messages and final result marker - """ - # Yield start message (deterministic) - yield get_think_message("librarian", task, "start") + "\n" - - # Execute delegation - result = await delegate_to_librarian(task, context) - - # Yield completion message (deterministic) - if result.success: - yield get_think_message("librarian", task, "success") + "\n" - else: - yield get_think_message("librarian", task, "error") + "\n" - - # Yield result marker for extraction - yield f"__DELEGATION_RESULT__:librarian:{result.output}" - - -async def stream_delegate_to_biographer( - task: str, - context: str = "", -) -> AsyncGenerator[str, None]: - """ - Stream delegation to Biographer with automatic think messages. - - Args: - task: Task description - context: Additional context - - Yields: - str: Think messages and final result marker - """ - yield get_think_message("biographer", task, "start") + "\n" - - result = await delegate_to_biographer(task, context) - - if result.success: - yield get_think_message("biographer", task, "success") + "\n" - else: - yield get_think_message("biographer", task, "error") + "\n" - - yield f"__DELEGATION_RESULT__:biographer:{result.output}" - - -async def stream_delegate_to_housekeeper( - task: str, - context: str = "", -) -> AsyncGenerator[str, None]: - """ - Stream delegation to Housekeeper with automatic think messages. - - Args: - task: Task description - context: Additional context - - Yields: - str: Think messages and final result marker - """ - yield get_think_message("housekeeper", task, "start") + "\n" - - result = await delegate_to_housekeeper(task, context) - - if result.success: - yield get_think_message("housekeeper", task, "success") + "\n" - else: - yield get_think_message("housekeeper", task, "error") + "\n" - - yield f"__DELEGATION_RESULT__:housekeeper:{result.output}" - - -# Mapping of streaming delegation wrappers -STREAMING_DELEGATION_WRAPPERS = { - "librarian": stream_delegate_to_librarian, - "biographer": stream_delegate_to_biographer, - "housekeeper": stream_delegate_to_housekeeper, -} - - # Future expert delegation wrappers will be added here: # - delegate_to_developer(task, context) -> DelegationResult # - delegate_to_secretary(task, context) -> DelegationResult diff --git a/src/agents/librarian/__init__.py b/src/agents/librarian/__init__.py index 2a0d6ed..d5f18d0 100644 --- a/src/agents/librarian/__init__.py +++ b/src/agents/librarian/__init__.py @@ -10,7 +10,6 @@ Connects to the library-desk API to provide: from src.agents.librarian.agent import ( get_librarian_agent, run_librarian, - run_librarian_stream, ) from src.agents.librarian.capability import ( LIBRARIAN_CAPABILITY, @@ -26,5 +25,4 @@ __all__ = [ "register_librarian", "unregister_librarian", "run_librarian", - "run_librarian_stream", ] diff --git a/src/agents/librarian/agent.py b/src/agents/librarian/agent.py index a926629..f3d4b14 100644 --- a/src/agents/librarian/agent.py +++ b/src/agents/librarian/agent.py @@ -273,67 +273,3 @@ async def run_librarian( raise AgentError( "Research task failed", agent_name="librarian" ) from e - - -async def run_librarian_stream( - task: str, - context: str = "", - message_history: list[Any] | None = None, -): - """ - Execute a research task with streaming output. - - Yields text deltas as The Librarian generates the response. - - Args: - task: The research task or question - context: Additional context from conversation - message_history: Optional conversation history - - Yields: - str: Text deltas from the response - - Raises: - AgentError: If the research task fails. Exception detail is - logged here; callers map the failure to a user-safe message. - - Example: - async for delta in run_librarian_stream("Find Docker docs"): - print(delta, end="", flush=True) - """ - agent = get_librarian_agent() - - # Build prompt with context if provided - prompt = task - if context: - prompt = f"Context: {context}\n\nTask: {task}" - - logger.info( - "librarian_stream_started", - task=task[:100], - ) - - try: - # One shared library-desk connection for all tool calls in this run - async with library_client_session(): - async with agent.run_stream( - prompt, - message_history=message_history, - ) as response: - async for delta in response.stream_text(delta=True): - yield delta - - logger.info("librarian_stream_completed", task=task[:50]) - - except Exception as e: - # Full detail stays in the logs; raise instead of yielding error - # text into the stream as if it were research output. - logger.error( - "librarian_stream_error", - task=task[:50], - error=str(e), - exc_info=True, - ) - raise AgentError( - "Research task failed", agent_name="librarian" - ) from e diff --git a/src/core/household_registry.py b/src/core/household_registry.py index ddf1c88..780dc52 100644 --- a/src/core/household_registry.py +++ b/src/core/household_registry.py @@ -273,65 +273,6 @@ class HouseholdRegistry: return tools - def get_streaming_delegation_tools(self, names: list[str]) -> list[Any]: - """ - Get streaming delegation wrapper tools for specified capabilities. - - Similar to get_delegation_tools() but returns streaming wrappers - that yield butler-perspective think messages during execution. - - These wrappers emit think slugs like: - - "Allow me to consult the archives, sir." - - "The Librarian has compiled the relevant findings." - - Args: - names: List of member names to include - - Returns: - List of streaming delegation wrappers and/or raw tools - - Example: - >>> tools = registry.get_streaming_delegation_tools(["librarian"]) - >>> async for chunk in tools[0](task="Search for Docker"): - ... print(chunk) # Yields think messages then result - """ - from src.agents.delegation import STREAMING_DELEGATION_WRAPPERS - - tools = [] - for name in names: - member = self._members.get(name) - if not member: - logger.warning( - "household_member_not_found", - requested_name=name, - available_names=list(self._members.keys()), - ) - continue - - # Check if this member has a streaming delegation wrapper - if name in STREAMING_DELEGATION_WRAPPERS and member.agent is not None: - tools.append(STREAMING_DELEGATION_WRAPPERS[name]) - logger.debug( - "streaming_delegation_wrapper_added", - member=name, - ) - else: - # No agent = direct tools (e.g., tatlock_core) - tools.extend(member.tools) - logger.debug( - "raw_tools_added", - member=name, - tool_count=len(member.tools), - ) - - logger.info( - "streaming_delegation_tools_created", - requested_members=names, - total_tools=len(tools), - ) - - return tools - def list_members(self) -> list[str]: """ List all registered member names. diff --git a/tests/agents/librarian/test_agent_failures.py b/tests/agents/librarian/test_agent_failures.py index fb94139..5d8fb47 100644 --- a/tests/agents/librarian/test_agent_failures.py +++ b/tests/agents/librarian/test_agent_failures.py @@ -1,16 +1,16 @@ """ -Tests for structured failure behavior of The Librarian entry points. +Tests for structured failure behavior of The Librarian entry point. -run_librarian and run_librarian_stream must raise AgentError on failure -instead of returning/yielding error text as if it were research output, -and the raised error must not leak exception detail (internal URLs etc.). +run_librarian must raise AgentError on failure instead of returning +error text as if it were research output, and the raised error must +not leak exception detail (internal URLs etc.). """ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from src.agents.librarian.agent import run_librarian, run_librarian_stream +from src.agents.librarian.agent import run_librarian from src.agents.protocol import AgentError @@ -37,24 +37,3 @@ class TestRunLibrarianFailures: # Exception detail stays in logs only assert "internal" not in str(exc_info.value) assert "Connection refused" not in str(exc_info.value) - - @pytest.mark.asyncio - async def test_run_librarian_stream_raises_agent_error(self): - """Streaming failures raise instead of yielding error text.""" - mock_agent = MagicMock() - mock_agent.run_stream = MagicMock( - side_effect=RuntimeError("Connection refused to http://internal:8089") - ) - - with patch( - "src.agents.librarian.agent.get_librarian_agent", - return_value=mock_agent, - ): - collected: list[str] = [] - with pytest.raises(AgentError) as exc_info: - async for delta in run_librarian_stream(task="Find Docker docs"): - collected.append(delta) - - assert exc_info.value.agent_name == "librarian" - assert collected == [], "no error text may be yielded as output" - assert "internal" not in str(exc_info.value) diff --git a/tests/agents/test_coordination.py b/tests/agents/test_coordination.py deleted file mode 100644 index e057ac0..0000000 --- a/tests/agents/test_coordination.py +++ /dev/null @@ -1,339 +0,0 @@ -""" -Tests for multi-agent coordination engine. -""" - -import pytest -from unittest.mock import AsyncMock, MagicMock, patch - -from src.agents.coordination import ( - CoordinationEngine, - get_coordination_engine, - delegate_to_librarian, -) -from src.agents.protocol import ( - AgentResponse, - AgentUnavailableError, - DelegationIntent, - DelegationReason, -) - - -@pytest.fixture -def coordination_engine(): - """Create a fresh coordination engine for testing.""" - return CoordinationEngine() - - -@pytest.fixture -def mock_registry(): - """Mock the household registry.""" - with patch("src.agents.coordination.get_household_registry") as mock: - registry = MagicMock() - mock.return_value = registry - yield registry - - -@pytest.fixture -def librarian_intent(): - """Create a standard librarian delegation intent.""" - return DelegationIntent( - target_agent="librarian", - task="Find information about Docker networking", - reason=DelegationReason.DOMAIN_EXPERTISE, - expected_outcome="Documentation and examples", - ) - - -@pytest.mark.unit -class TestCoordinationEngine: - """Tests for CoordinationEngine class.""" - - def test_initialization(self, coordination_engine): - """Test engine initializes correctly.""" - assert coordination_engine is not None - assert coordination_engine.registry is not None - - def test_get_available_agents_empty(self, mock_registry): - """Test getting available agents when none have agents.""" - mock_registry.list_members.return_value = ["tatlock_core"] - mock_member = MagicMock() - mock_member.agent = None # No agent - mock_registry.get_member.return_value = mock_member - - engine = CoordinationEngine() - available = engine.get_available_agents() - - assert available == [] - - def test_get_available_agents_with_librarian(self, mock_registry): - """Test getting available agents with librarian registered.""" - mock_registry.list_members.return_value = ["tatlock_core", "librarian"] - - # tatlock_core has no agent - core_member = MagicMock() - core_member.agent = None - - # librarian has an agent - librarian_member = MagicMock() - librarian_member.agent = MagicMock() - - def get_member_side_effect(name): - if name == "tatlock_core": - return core_member - elif name == "librarian": - return librarian_member - return None - - mock_registry.get_member.side_effect = get_member_side_effect - - engine = CoordinationEngine() - available = engine.get_available_agents() - - assert "librarian" in available - assert "tatlock_core" not in available - - def test_can_delegate_to_unknown_agent(self, mock_registry): - """Test checking delegation to unknown agent.""" - mock_registry.get_member.return_value = None - - engine = CoordinationEngine() - - assert engine.can_delegate_to("unknown_agent") is False - - def test_can_delegate_to_librarian(self, mock_registry): - """Test checking delegation to librarian.""" - mock_member = MagicMock() - mock_member.agent = MagicMock() # Has an agent - mock_registry.get_member.return_value = mock_member - - engine = CoordinationEngine() - - assert engine.can_delegate_to("librarian") is True - - -@pytest.mark.unit -class TestDelegationExecution: - """Tests for delegation execution.""" - - @pytest.mark.asyncio - async def test_execute_delegation_unavailable_agent( - self, mock_registry, librarian_intent - ): - """Test delegation fails for unavailable agent.""" - mock_registry.get_member.return_value = None - - engine = CoordinationEngine() - - with pytest.raises(AgentUnavailableError) as exc_info: - await engine.execute_delegation(librarian_intent) - - assert "librarian" in str(exc_info.value) - - @pytest.mark.asyncio - async def test_execute_delegation_success( - self, mock_registry, librarian_intent - ): - """Test successful delegation execution.""" - # Setup mock member with agent - mock_member = MagicMock() - mock_member.agent = MagicMock() - mock_registry.get_member.return_value = mock_member - - # Mock the executor - with patch( - "src.agents.coordination.AGENT_EXECUTORS", - {"librarian": AsyncMock(return_value="Research results here")}, - ): - engine = CoordinationEngine() - response = await engine.execute_delegation(librarian_intent) - - assert response.success is True - assert response.result == "Research results here" - # Duration might be 0 for very fast mock execution - assert response.duration_ms >= 0 - - @pytest.mark.asyncio - async def test_execute_delegation_error( - self, mock_registry, librarian_intent - ): - """Test delegation handles executor errors.""" - mock_member = MagicMock() - mock_member.agent = MagicMock() - mock_registry.get_member.return_value = mock_member - - # Mock executor that raises - async def failing_executor(**kwargs): - raise ValueError("API connection failed") - - with patch( - "src.agents.coordination.AGENT_EXECUTORS", - {"librarian": failing_executor}, - ): - engine = CoordinationEngine() - response = await engine.execute_delegation(librarian_intent) - - assert response.success is False - assert "API connection failed" in response.error_message - - -@pytest.mark.unit -class TestCoordinate: - """Tests for multi-agent coordination.""" - - @pytest.mark.asyncio - async def test_coordinate_single_intent(self, mock_registry, librarian_intent): - """Test coordinating a single delegation.""" - mock_member = MagicMock() - mock_member.agent = MagicMock() - mock_registry.get_member.return_value = mock_member - - with patch( - "src.agents.coordination.AGENT_EXECUTORS", - {"librarian": AsyncMock(return_value="Found docs")}, - ): - engine = CoordinationEngine() - result = await engine.coordinate([librarian_intent]) - - assert result.final_response == "Found docs" - assert "librarian" in result.agents_consulted - # Duration might be 0 for very fast mock execution - assert result.total_duration_ms >= 0 - - @pytest.mark.asyncio - async def test_coordinate_empty_intents(self, mock_registry): - """Test coordinating with no intents.""" - engine = CoordinationEngine() - result = await engine.coordinate([]) - - assert result.final_response == "" - assert result.agents_consulted == [] - - @pytest.mark.asyncio - async def test_coordinate_multiple_intents(self, mock_registry): - """Test coordinating multiple delegations.""" - mock_member = MagicMock() - mock_member.agent = MagicMock() - mock_registry.get_member.return_value = mock_member - - intents = [ - DelegationIntent( - target_agent="librarian", - task="Task 1", - reason=DelegationReason.DOMAIN_EXPERTISE, - expected_outcome="Result 1", - priority=1, - ), - DelegationIntent( - target_agent="librarian", - task="Task 2", - reason=DelegationReason.DOMAIN_EXPERTISE, - expected_outcome="Result 2", - priority=2, - ), - ] - - call_count = 0 - - async def mock_executor(**kwargs): - nonlocal call_count - call_count += 1 - return f"Result {call_count}" - - with patch( - "src.agents.coordination.AGENT_EXECUTORS", - {"librarian": mock_executor}, - ): - engine = CoordinationEngine() - result = await engine.coordinate(intents) - - # Both intents were executed (check agents_consulted count) - assert len(result.agents_consulted) == 2 - # Current implementation replaces same-agent responses in dict - # So final_response has the last result (or combined if different agents) - assert len(result.final_response) > 0 - - -@pytest.mark.unit -class TestDelegateToLibrarian: - """Tests for convenience delegation function.""" - - @pytest.mark.asyncio - async def test_delegate_to_librarian(self, mock_registry): - """Test the delegate_to_librarian helper.""" - mock_member = MagicMock() - mock_member.agent = MagicMock() - mock_registry.get_member.return_value = mock_member - - with patch( - "src.agents.coordination.AGENT_EXECUTORS", - {"librarian": AsyncMock(return_value="Wiki search results")}, - ): - # Reset global engine - with patch( - "src.agents.coordination._coordination_engine", - None, - ): - response = await delegate_to_librarian( - task="Search for Docker docs", - context="Setting up homelab", - ) - - assert response.success is True - assert response.result == "Wiki search results" - - -@pytest.mark.unit -class TestGetCoordinationEngine: - """Tests for engine singleton.""" - - def test_get_coordination_engine_singleton(self): - """Test engine is singleton.""" - with patch("src.agents.coordination._coordination_engine", None): - engine1 = get_coordination_engine() - engine2 = get_coordination_engine() - - # Should be same instance - assert engine1 is engine2 - - -@pytest.mark.unit -class TestDelegationStreaming: - """Tests for streaming delegation.""" - - @pytest.mark.asyncio - async def test_execute_delegation_stream_unavailable( - self, mock_registry, librarian_intent - ): - """Test streaming fails for unavailable agent.""" - engine = CoordinationEngine() - - # Change target to an agent that doesn't have a stream executor - librarian_intent.target_agent = "nonexistent_agent" - - with pytest.raises(AgentUnavailableError): - async for _ in engine.execute_delegation_stream(librarian_intent): - pass - - @pytest.mark.asyncio - async def test_execute_delegation_stream_success( - self, mock_registry, librarian_intent - ): - """Test successful streaming delegation.""" - mock_member = MagicMock() - mock_member.agent = MagicMock() - mock_registry.get_member.return_value = mock_member - - async def mock_stream(**kwargs): - yield "Hello " - yield "world" - - with patch( - "src.agents.coordination.AGENT_STREAM_EXECUTORS", - {"librarian": mock_stream}, - ): - engine = CoordinationEngine() - chunks = [] - async for chunk in engine.execute_delegation_stream(librarian_intent): - chunks.append(chunk) - - assert chunks == ["Hello ", "world"] diff --git a/tests/agents/test_delegation.py b/tests/agents/test_delegation.py index 3e1c4eb..e0e6d4d 100644 --- a/tests/agents/test_delegation.py +++ b/tests/agents/test_delegation.py @@ -10,7 +10,6 @@ import pytest from src.agents.delegation import ( HOUSEHOLD_THINK_MESSAGES, - STREAMING_DELEGATION_WRAPPERS, ActionType, DelegationResult, DelegationTask, @@ -430,20 +429,3 @@ class TestGetThinkMessage: msg = get_think_message("unknown_expert", "some task", "start") assert "" 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"