From f853db8cccd5ae96b63463c7a5f35cd85b7939a0 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Tue, 14 Jul 2026 10:12:30 +0200 Subject: [PATCH] fix(agents): make librarian failures structured and user-safe - run_librarian / run_librarian_stream raise AgentError instead of returning/yielding error text as normal output; detail stays in logs - delegate_to_* wrappers now put a curated butler-toned sentence in DelegationResult.output on failure and never expose str(e), so streaming's error branch is reachable and honest - _execute_single_delegation propagates success; direct delegation only records delegate_to_* as called when the expert actually succeeded - librarian tools return user-safe messages instead of 'Error searching: {e}' strings that leaked internal URLs into synthesis; coordination stream errors are curated as well - ruff cleanups (TYPE_CHECKING forward refs, B904, unused locals) in the touched files to keep them lint-clean Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 2 + src/agents/coordination.py | 26 +++-- src/agents/delegation.py | 31 +++-- src/agents/librarian/agent.py | 30 +++-- src/agents/librarian/tools.py | 28 ++--- src/responses/service.py | 107 ++++++++++-------- src/responses/streaming.py | 72 +++++++----- tests/agents/librarian/test_agent_failures.py | 60 ++++++++++ tests/agents/librarian/test_tools.py | 35 +++--- tests/agents/test_delegation.py | 26 +++-- 10 files changed, 277 insertions(+), 140 deletions(-) create mode 100644 tests/agents/librarian/test_agent_failures.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ff7ab0..95b5dce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **Honest expert failures** - `run_librarian`/`run_librarian_stream` now raise a structured `AgentError` instead of returning error text as if it were research output, so delegation correctly reports `success=False` and the streaming error branch is reachable. Failures surface to the user as curated butler-toned sentences; exception detail (including internal URLs) stays in the logs only. Librarian tool errors no longer leak `str(e)` into synthesis + - **HybridRAG response mapping** - The librarian client now parses the field names library-desk actually returns (`source_type`/`sources`, `rrf_score`, `context`, per-item `related_dossiers`, synonyms nested in the `keywords` dict); previously every result rendered as "unknown (score: 0.00)". Source icons now key off the per-item `sources` list. Requests no longer send zero limits (the service rejects them with 422); legs are disabled via `enable_*` flags. Pinned by a contract test against a recorded live response (`tests/agents/librarian/fixtures/`) ## [2.3.0] - 2026-07-13 diff --git a/src/agents/coordination.py b/src/agents/coordination.py index b582402..0215692 100644 --- a/src/agents/coordination.py +++ b/src/agents/coordination.py @@ -10,7 +10,8 @@ based on Steward recommendations. Handles: """ import asyncio import time -from typing import Any, AsyncGenerator, Optional +from collections.abc import AsyncGenerator +from typing import Any from src.agents.librarian import run_librarian, run_librarian_stream from src.agents.protocol import ( @@ -22,7 +23,6 @@ from src.agents.protocol import ( CoordinationResult, DelegationIntent, DelegationReason, - ToolCallRecord, ) from src.core.household_registry import get_household_registry from src.core.logging_config import get_logger @@ -87,7 +87,7 @@ class CoordinationEngine: self, intent: DelegationIntent, context: str = "", - message_history: Optional[list[Any]] = None, + message_history: list[Any] | None = None, ) -> AgentResponse: """ Execute a single delegation to an expert agent. @@ -166,7 +166,7 @@ class CoordinationEngine: duration_ms=duration_ms, ) - except asyncio.TimeoutError: + except TimeoutError as e: duration_ms = int((time.time() - start_time) * 1000) logger.error( "delegation_timeout", @@ -176,7 +176,7 @@ class CoordinationEngine: 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) @@ -198,7 +198,7 @@ class CoordinationEngine: self, intent: DelegationIntent, context: str = "", - message_history: Optional[list[Any]] = None, + message_history: list[Any] | None = None, ) -> AsyncGenerator[str, None]: """ Execute a delegation with streaming output. @@ -242,19 +242,23 @@ class CoordinationEngine: 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 f"\n\n[Error from {agent_name}: {str(e)}]" + yield "\n\n" + get_think_message(agent_name, intent.task, "error") async def coordinate( self, intents: list[DelegationIntent], context: str = "", - message_history: Optional[list[Any]] = None, + message_history: list[Any] | None = None, ) -> CoordinationResult: """ Coordinate execution of multiple delegations. @@ -329,7 +333,7 @@ class CoordinationEngine: # Global coordination engine instance -_coordination_engine: Optional[CoordinationEngine] = None +_coordination_engine: CoordinationEngine | None = None def get_coordination_engine() -> CoordinationEngine: @@ -344,7 +348,7 @@ async def delegate_to_librarian( task: str, context: str = "", reason: DelegationReason = DelegationReason.DOMAIN_EXPERTISE, - message_history: Optional[list[Any]] = None, + message_history: list[Any] | None = None, ) -> AgentResponse: """ Convenience function to delegate a task to The Librarian. @@ -377,7 +381,7 @@ async def delegate_to_librarian( async def delegate_to_librarian_stream( task: str, context: str = "", - message_history: Optional[list[Any]] = None, + message_history: list[Any] | None = None, ) -> AsyncGenerator[str, None]: """ Convenience function to delegate to Librarian with streaming. diff --git a/src/agents/delegation.py b/src/agents/delegation.py index 08d977d..462aab9 100644 --- a/src/agents/delegation.py +++ b/src/agents/delegation.py @@ -8,12 +8,12 @@ returns a structured result for synthesis. This implements the agent-as-tool pattern recommended by PydanticAI: agents call other agents via tool wrappers, keeping each agent focused. """ +from collections.abc import AsyncGenerator from dataclasses import dataclass, field from enum import Enum -from typing import AsyncGenerator, Callable, Optional, Any from src.core.logging_config import get_logger -from src.core.tracing import trace_span, SpanType +from src.core.tracing import SpanType, trace_span logger = get_logger(__name__) @@ -167,7 +167,7 @@ class DelegationTask: action: str = "" priority: int = 0 depends_on: list[str] = field(default_factory=list) - result: Optional[str] = None + result: str | None = None task_id: str = "" def __post_init__(self): @@ -186,14 +186,16 @@ class DelegationResult: expert_name: Which expert handled the task task: Original task description success: Whether the delegation succeeded - output: Expert's response/findings - error: Error message if failed + output: Expert's response/findings. On failure this holds a + curated, user-safe butler sentence (never exception detail) + error: Short user-safe error label if failed. Exception detail + stays in the logs only """ expert_name: str task: str success: bool output: str - error: Optional[str] = None + error: str | None = None async def delegate_to_librarian( @@ -285,12 +287,15 @@ async def delegate_to_librarian( span.metadata["success"] = False span.details["error"] = str(e) + # Exception detail stays in the logs; the user-facing output + # is a curated butler sentence so internals never leak into + # synthesis. return DelegationResult( expert_name="librarian", task=task, success=False, - output="", - error=str(e), + output=get_think_message("librarian", task, "error"), + error="The Librarian was unable to complete the task.", ) @@ -383,12 +388,13 @@ async def delegate_to_biographer( span.metadata["success"] = False span.details["error"] = str(e) + # Exception detail stays in the logs only. return DelegationResult( expert_name="biographer", task=task, success=False, - output="", - error=str(e), + output=get_think_message("biographer", task, "error"), + error="The Biographer was unable to complete the task.", ) @@ -480,12 +486,13 @@ async def delegate_to_housekeeper( span.metadata["success"] = False span.details["error"] = str(e) + # Exception detail stays in the logs only. return DelegationResult( expert_name="housekeeper", task=task, success=False, - output="", - error=str(e), + output=get_think_message("housekeeper", task, "error"), + error="The Housekeeper was unable to complete the task.", ) diff --git a/src/agents/librarian/agent.py b/src/agents/librarian/agent.py index 063ab6d..3e57ba6 100644 --- a/src/agents/librarian/agent.py +++ b/src/agents/librarian/agent.py @@ -7,7 +7,7 @@ the library-desk API, offering: - Wiki and document management - Semantic search and knowledge graph exploration """ -from typing import Any, Optional +from typing import Any from pydantic_ai import Agent @@ -27,7 +27,7 @@ from src.agents.librarian.tools import ( smart_create_wiki_page, update_wiki_page, ) -from src.core.config import config +from src.agents.protocol import AgentError from src.core.logging_config import get_logger logger = get_logger(__name__) @@ -138,7 +138,7 @@ If a tool fails or you cannot access a data source: """ # Lazy initialization to avoid connection issues during imports -_librarian_agent: Optional[Agent[None, str]] = None +_librarian_agent: Agent[None, str] | None = None def _create_librarian_agent() -> Agent[None, str]: @@ -204,7 +204,7 @@ def get_librarian_agent() -> Agent[None, str]: async def run_librarian( task: str, context: str = "", - message_history: Optional[list[Any]] = None, + message_history: list[Any] | None = None, ) -> str: """ Execute a research task with The Librarian. @@ -220,6 +220,10 @@ async def run_librarian( Returns: Research results and findings + Raises: + AgentError: If the research task fails. Exception detail is + logged here; callers map the failure to a user-safe message. + Example: result = await run_librarian( task="Find information about Docker networking", @@ -255,19 +259,23 @@ async def run_librarian( return result.output except Exception as e: + # Full detail stays in the logs; callers receive a structured + # failure instead of error text masquerading as research output. logger.error( "librarian_task_error", task=task[:50], error=str(e), exc_info=True, ) - return f"The Librarian encountered an error: {str(e)}" + raise AgentError( + "Research task failed", agent_name="librarian" + ) from e async def run_librarian_stream( task: str, context: str = "", - message_history: Optional[list[Any]] = None, + message_history: list[Any] | None = None, ): """ Execute a research task with streaming output. @@ -282,6 +290,10 @@ async def run_librarian_stream( 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) @@ -309,10 +321,14 @@ async def run_librarian_stream( 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, ) - yield f"\n\nThe Librarian encountered an error: {str(e)}" + raise AgentError( + "Research task failed", agent_name="librarian" + ) from e diff --git a/src/agents/librarian/tools.py b/src/agents/librarian/tools.py index a51dba7..78573a9 100644 --- a/src/agents/librarian/tools.py +++ b/src/agents/librarian/tools.py @@ -111,7 +111,7 @@ async def hybrid_search( except Exception as e: logger.error("librarian_hybrid_search_error", error=str(e), query=query) - return f"Error searching: {str(e)}" + return "I was unable to search the knowledge archives; the search service did not respond properly." # ============================================================================ @@ -159,7 +159,7 @@ async def search_wiki( except Exception as e: logger.error("librarian_wiki_search_error", error=str(e)) - return f"Error searching wiki: {str(e)}" + return "I was unable to search the wiki at this time." async def get_wiki_page( @@ -202,7 +202,7 @@ async def get_wiki_page( except Exception as e: logger.error("librarian_get_page_error", error=str(e), page_id=page_id) - return f"Error getting page {page_id}: {str(e)}" + return f"I was unable to retrieve wiki page {page_id}." async def list_dossiers() -> str: @@ -236,7 +236,7 @@ async def list_dossiers() -> str: except Exception as e: logger.error("librarian_list_dossiers_error", error=str(e)) - return f"Error listing dossiers: {str(e)}" + return "I was unable to retrieve the list of dossiers." async def get_dossier_pages( @@ -277,7 +277,7 @@ async def get_dossier_pages( except Exception as e: logger.error("librarian_get_dossier_error", error=str(e)) - return f"Error getting dossier: {str(e)}" + return f"I was unable to retrieve the dossier '{dossier_name}'." # ============================================================================ @@ -326,7 +326,7 @@ async def semantic_search( except Exception as e: logger.error("librarian_semantic_search_error", error=str(e)) - return f"Error in semantic search: {str(e)}" + return "I was unable to complete the semantic search." # ============================================================================ @@ -379,7 +379,7 @@ async def explore_knowledge_graph( except Exception as e: logger.error("librarian_explore_graph_error", error=str(e)) - return f"Error exploring knowledge graph: {str(e)}" + return "I was unable to explore the knowledge graph." async def find_related_entities( @@ -450,7 +450,7 @@ async def find_related_entities( except Exception as e: logger.error("librarian_find_related_error", error=str(e)) - return f"Error finding related entities: {str(e)}" + return f"I was unable to look up entities related to '{entity_name}'." # ============================================================================ @@ -533,7 +533,7 @@ async def search_web( except Exception as e: logger.error("librarian_web_search_error", error=str(e), query=query) - return f"Error searching web: {str(e)}" + return "I was unable to search the web at this time." async def read_url( @@ -609,7 +609,7 @@ async def read_url( except Exception as e: logger.error("librarian_read_url_error", error=str(e), url=url) - return f"Error reading URL: {str(e)}" + return f"I was unable to read the page at {url}." async def read_urls_batch( @@ -683,7 +683,7 @@ async def read_urls_batch( except Exception as e: logger.error("librarian_read_urls_batch_error", error=str(e)) - return f"Error reading URLs: {str(e)}" + return "I was unable to read the requested pages." # ============================================================================ @@ -766,7 +766,7 @@ async def update_wiki_page( except Exception as e: logger.error("librarian_update_page_error", error=str(e), page_id=page_id) - return f"Error updating page {page_id}: {str(e)}" + return f"I was unable to update wiki page {page_id}." async def create_wiki_page( @@ -841,7 +841,7 @@ async def create_wiki_page( except Exception as e: logger.error("librarian_create_page_error", error=str(e), title=title) - return f"Error creating page: {str(e)}" + return f"I was unable to create the page '{title}'." async def smart_create_wiki_page( @@ -930,7 +930,7 @@ async def smart_create_wiki_page( except Exception as e: logger.error("librarian_smart_create_error", error=str(e), topic=topic) - return f"Error creating page about '{topic}': {str(e)}" + return f"I was unable to create a page about '{topic}'." # ============================================================================ diff --git a/src/responses/service.py b/src/responses/service.py index 8a44940..59c4621 100644 --- a/src/responses/service.py +++ b/src/responses/service.py @@ -6,32 +6,32 @@ Tracks conversation history for analytics and future vector memory. Integrates with Steward preprocessing for Phase 2 two-tier architecture. """ -import time +import asyncio +import re import secrets -from typing import AsyncGenerator +import time +from collections.abc import AsyncGenerator +from src.agents.delegation import get_think_message from src.agents.registry import ModelRegistry +from src.agents.steward.schemas import StewardRecommendation +from src.core.context import current_conversation, current_user, get_default_user +from src.core.logging_config import get_logger +from src.core.preprocessing import preprocess_request +from src.core.tool_tracking import ToolCallTracker +from src.core.tracing import SpanType, end_trace, start_span, start_trace +from src.responses.context import ContextWindow +from src.responses.history import ConversationHistory from src.responses.schemas import ( + FunctionCallOutputItem, + MessageOutputItem, + OutputTextContent, + ReasoningOutputItem, Response, ResponseRequest, ResponseUsage, - MessageOutputItem, - ReasoningOutputItem, - FunctionCallOutputItem, - OutputTextContent, ) from src.responses.streaming import StreamingCoordinator -from src.responses.history import ConversationHistory -from src.responses.context import ContextWindow -from src.core.preprocessing import preprocess_request -from src.core.tool_tracking import ToolCallTracker -from src.core.logging_config import get_logger -from src.core.tracing import start_trace, end_trace, start_span, SpanType -from src.core.context import current_user, current_conversation, get_default_user -from src.agents.steward.schemas import StewardRecommendation - -import re -import asyncio logger = get_logger(__name__) @@ -63,7 +63,7 @@ async def _execute_single_delegation( agent_name: str, task: str, tracker: "ToolCallTracker", -) -> tuple[str, str]: +) -> tuple[str, str, bool]: """ Execute a single delegation to an agent. @@ -73,7 +73,8 @@ async def _execute_single_delegation( tracker: Tool call tracker Returns: - tuple: (agent_name, result_summary) + tuple: (agent_name, result_summary, success). On failure the + result summary is a curated user-safe sentence. """ import time start_time = time.time() @@ -83,24 +84,24 @@ async def _execute_single_delegation( result = await delegate_to_biographer(task=task) duration = time.time() - start_time await tracker.track_call("delegate_to_biographer", duration) - return (agent_name, result.output) + return (agent_name, result.output, result.success) elif agent_name == "librarian": from src.agents.delegation import delegate_to_librarian result = await delegate_to_librarian(task=task) duration = time.time() - start_time await tracker.track_call("delegate_to_librarian", duration) - return (agent_name, result.output) + return (agent_name, result.output, result.success) elif agent_name == "housekeeper": from src.agents.delegation import delegate_to_housekeeper result = await delegate_to_housekeeper(task=task) duration = time.time() - start_time await tracker.track_call("delegate_to_housekeeper", duration) - return (agent_name, result.output) + return (agent_name, result.output, result.success) else: - return (agent_name, f"Unknown agent: {agent_name}") + return (agent_name, f"Unknown agent: {agent_name}", False) async def _handle_text_delegation( @@ -175,13 +176,24 @@ async def _handle_text_delegation( ] results = await asyncio.gather(*tasks, return_exceptions=True) - # Combine results + # Combine results (failures carry curated user-safe sentences) summaries = [] - for agent_name, result in results: - if isinstance(result, Exception): - summaries.append(f"**{agent_name}**: Error - {result}") + for (agent, task), item in zip(matches, results, strict=False): + agent_name = agent.lower() + if isinstance(item, BaseException): + logger.error( + "delegation_failed", + agent=agent_name, + error=str(item), + conversation_id=conversation_id, + ) + summaries.append( + f"**{agent_name}**: " + f"{get_think_message(agent_name, task, 'error')}" + ) else: - summaries.append(f"**{agent_name}**: {result}") + _, output, _ = item + summaries.append(f"**{agent_name}**: {output}") return "\n\n".join(summaries) @@ -197,20 +209,19 @@ async def _handle_text_delegation( conversation_id=conversation_id, ) try: - _, result = await _execute_single_delegation( + _, output, _ = await _execute_single_delegation( agent_name, task, tracker ) - summaries.append(result) + summaries.append(output) except Exception as e: logger.error( "delegation_failed", agent=agent_name, error=str(e), conversation_id=conversation_id, + exc_info=True, ) - summaries.append( - f"I apologize, sir. Delegation to {agent_name} failed: {e}" - ) + summaries.append(get_think_message(agent_name, task, "error")) return "\n\n".join(summaries) @@ -219,8 +230,9 @@ async def _handle_text_delegation( "text_delegation_failed", error=str(e), conversation_id=conversation_id, + exc_info=True, ) - return f"I apologize, sir. I encountered an error processing delegations: {e}" + return "I apologize, sir. I was unable to complete the requested delegations." async def _direct_delegation( @@ -254,13 +266,14 @@ async def _direct_delegation( results = [] for agent in recommendation.recommended_capabilities: try: - agent_name, result = await _execute_single_delegation( + agent_name, result, success = await _execute_single_delegation( agent, user_message, tracker ) results.append(result) logger.info( "direct_delegation_complete", agent=agent_name, + success=success, result_preview=result[:100] if result else "empty", conversation_id=conversation_id, ) @@ -270,8 +283,9 @@ async def _direct_delegation( agent=agent, error=str(e), conversation_id=conversation_id, + exc_info=True, ) - results.append(f"I apologize, sir. Delegation to {agent} failed: {e}") + results.append(get_think_message(agent, user_message, "error")) return "\n\n".join(results) if results else "I apologize, sir. No delegation results available." @@ -309,15 +323,17 @@ async def _direct_delegation_with_results( for agent in recommendation.recommended_capabilities: try: - agent_name, result = await _execute_single_delegation( + agent_name, result, success = await _execute_single_delegation( agent, user_message, tracker ) expert_results[agent_name] = result - tools_called.append(f"delegate_to_{agent_name}") + if success: + tools_called.append(f"delegate_to_{agent_name}") logger.info( "direct_delegation_result", agent=agent_name, + success=success, result_preview=result[:100] if result else "empty", conversation_id=conversation_id, ) @@ -327,8 +343,9 @@ async def _direct_delegation_with_results( agent=agent, error=str(e), conversation_id=conversation_id, + exc_info=True, ) - expert_results[agent] = f"Error: {e}" + expert_results[agent] = get_think_message(agent, user_message, "error") return { "tools_called": tools_called, @@ -439,7 +456,7 @@ async def create_response(request: ResponseRequest) -> Response: user_input = _extract_user_input(request.input) # Start trace - trace = start_trace( + start_trace( conversation_id=conversation_id, user=effective_user, request={ @@ -451,7 +468,7 @@ async def create_response(request: ResponseRequest) -> Response: ) # Start service span - service_span = start_span( + start_span( "create_response", SpanType.ROUTER, metadata={"model": request.model, "user": effective_user}, @@ -509,7 +526,7 @@ async def create_response(request: ResponseRequest) -> Response: return response - except Exception as e: + except Exception: end_trace(status="error") raise @@ -550,7 +567,7 @@ async def create_response_with_steward(request: ResponseRequest) -> Response: user_input = _extract_user_input(request.input) # Start trace - trace = start_trace( + start_trace( conversation_id=conversation_id, user=effective_user, request={ @@ -562,7 +579,7 @@ async def create_response_with_steward(request: ResponseRequest) -> Response: ) # Start service span - service_span = start_span( + start_span( "create_response_with_steward", SpanType.ROUTER, metadata={"model": request.model, "user": effective_user}, @@ -706,7 +723,7 @@ async def create_response_with_steward(request: ResponseRequest) -> Response: return response - except Exception as e: + except Exception: end_trace(status="error") raise diff --git a/src/responses/streaming.py b/src/responses/streaming.py index e52c260..82791eb 100644 --- a/src/responses/streaming.py +++ b/src/responses/streaming.py @@ -8,14 +8,22 @@ Handles Server-Sent Events (SSE) streaming with proper event types: - response.done """ -from enum import Enum -from typing import Literal, AsyncGenerator -import json import time +from collections.abc import AsyncGenerator +from enum import Enum +from typing import TYPE_CHECKING, Literal +from src.agents.registry import ModelRegistry +from src.core.logging_config import get_logger from src.core.models import CustomBaseModel from src.responses.schemas import Response -from src.agents.registry import ModelRegistry + +if TYPE_CHECKING: + from src.agents.steward.schemas import StewardRecommendation + from src.core.tool_tracking import ToolCallTracker + from src.responses.schemas import ResponseRequest + +logger = get_logger(__name__) # ============================================================================ @@ -131,18 +139,17 @@ class StreamingCoordinator: Yields: StreamEvent: Stream of SSE events """ - from src.responses.service import ( - _calculate_usage, - generate_id, - _conversation_history, - _direct_delegation_with_results, - ) + import asyncio + + from src.agents.tatlock import TatlockAgent from src.core.preprocessing import preprocess_request from src.core.tool_tracking import ToolCallTracker - from src.responses.schemas import MessageOutputItem, ReasoningOutputItem, OutputTextContent - from src.agents.tatlock import TatlockAgent - from src.agents.delegation import get_think_message, STREAMING_DELEGATION_WRAPPERS - import asyncio + from src.responses.schemas import MessageOutputItem, OutputTextContent + from src.responses.service import ( + _calculate_usage, + _conversation_history, + generate_id, + ) output_items = [] @@ -280,13 +287,14 @@ class StreamingCoordinator: Returns: dict: Orchestration results with think_messages list """ + import time as time_module + from src.agents.delegation import ( - get_think_message, - delegate_to_librarian, delegate_to_biographer, delegate_to_housekeeper, + delegate_to_librarian, + get_think_message, ) - import time as time_module expert_results = {} tools_called = [] @@ -319,15 +327,26 @@ class StreamingCoordinator: success_msg = get_think_message(agent, user_message, "success") think_messages.append(success_msg + "\n") else: - error_msg = result.error if result else "Unknown error" - expert_results[agent] = f"Error: {error_msg}" - # Emit error think message + # Failed delegations carry a curated user-safe sentence + # in output; exception detail is already in the logs. error_think = get_think_message(agent, user_message, "error") + if result and result.output: + expert_results[agent] = result.output + else: + expert_results[agent] = error_think + # Emit error think message think_messages.append(error_think + "\n") except Exception as e: - expert_results[agent] = f"Error: {e}" + logger.error( + "direct_delegation_stream_error", + agent=agent, + error=str(e), + conversation_id=conversation_id, + exc_info=True, + ) error_think = get_think_message(agent, user_message, "error") + expert_results[agent] = error_think think_messages.append(error_think + "\n") return { @@ -361,9 +380,10 @@ class StreamingCoordinator: event: response.done data: {"response": {...}} """ - from src.responses.service import _calculate_usage, generate_id import asyncio + from src.responses.service import _calculate_usage, generate_id + output_items = [] last_message_text = "" # Track last streamed message text to compute deltas @@ -488,10 +508,10 @@ class StreamingCoordinator: def _convert_output_items(self, items: list) -> list: """Convert agent OutputItem objects to schema OutputItem objects.""" from src.responses.schemas import ( - MessageOutputItem, - ReasoningOutputItem, FunctionCallOutputItem, + MessageOutputItem, OutputTextContent, + ReasoningOutputItem, ) converted = [] @@ -521,9 +541,9 @@ class StreamingCoordinator: def _create_error_event(self, error: Exception) -> ErrorEvent: """Create error event from exception.""" from src.core.exceptions import ( - RateLimitError, - ContextLengthError, AppException, + ContextLengthError, + RateLimitError, ) if isinstance(error, RateLimitError): diff --git a/tests/agents/librarian/test_agent_failures.py b/tests/agents/librarian/test_agent_failures.py new file mode 100644 index 0000000..fb94139 --- /dev/null +++ b/tests/agents/librarian/test_agent_failures.py @@ -0,0 +1,60 @@ +""" +Tests for structured failure behavior of The Librarian entry points. + +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.). +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from src.agents.librarian.agent import run_librarian, run_librarian_stream +from src.agents.protocol import AgentError + + +@pytest.mark.unit +class TestRunLibrarianFailures: + """run_librarian raises structured errors instead of returning text.""" + + @pytest.mark.asyncio + async def test_run_librarian_raises_agent_error(self): + """Failures raise AgentError rather than returning error prose.""" + mock_agent = MagicMock() + mock_agent.run = AsyncMock( + side_effect=RuntimeError("Connection refused to http://internal:8089") + ) + + with patch( + "src.agents.librarian.agent.get_librarian_agent", + return_value=mock_agent, + ): + with pytest.raises(AgentError) as exc_info: + await run_librarian(task="Find Docker docs") + + assert exc_info.value.agent_name == "librarian" + # 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/librarian/test_tools.py b/tests/agents/librarian/test_tools.py index cfaad1f..032fb47 100644 --- a/tests/agents/librarian/test_tools.py +++ b/tests/agents/librarian/test_tools.py @@ -5,21 +5,20 @@ Tests the tool functions that wrap the Library-Desk API, including the new web search and content extraction tools. """ -import pytest -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, patch +import pytest + +from src.agents.librarian.client import ( + BatchExtractionResponse, + ContentExtractionResult, + WebSearchResponse, + WebSearchResult, +) from src.agents.librarian.tools import ( - search_web, read_url, read_urls_batch, - hybrid_search, - search_wiki, -) -from src.agents.librarian.client import ( - WebSearchResult, - WebSearchResponse, - ContentExtractionResult, - BatchExtractionResponse, + search_web, ) @@ -104,8 +103,10 @@ class TestSearchWeb: @pytest.mark.asyncio async def test_search_web_error_handling(self, mock_client): - """Test web search error handling.""" - mock_client.search_web.side_effect = Exception("Connection failed") + """Test web search errors return a user-safe message without internals.""" + mock_client.search_web.side_effect = Exception( + "Connection failed to http://internal-host:8089" + ) with patch( "src.agents.librarian.tools.LibraryDeskClient" @@ -115,8 +116,10 @@ class TestSearchWeb: result = await search_web("test query") - assert "Error" in result - assert "Connection failed" in result + assert "unable to search the web" in result + # Exception detail (internal URLs etc.) must not leak + assert "Connection failed" not in result + assert "internal-host" not in result @pytest.mark.asyncio async def test_search_web_with_news_type(self, mock_client): @@ -227,7 +230,7 @@ class TestReadUrl: mock_client_class.return_value.__aenter__.return_value = mock_client mock_client_class.return_value.__aexit__.return_value = None - result = await read_url("https://example.com/long", max_length=2000) + await read_url("https://example.com/long", max_length=2000) mock_client.extract_content.assert_called_with( url="https://example.com/long", diff --git a/tests/agents/test_delegation.py b/tests/agents/test_delegation.py index fb90259..489c4da 100644 --- a/tests/agents/test_delegation.py +++ b/tests/agents/test_delegation.py @@ -4,18 +4,19 @@ Tests for delegation infrastructure. Tests the DelegationTask dataclass and delegation wrapper functions that implement the agent-as-tool pattern. """ +from unittest.mock import AsyncMock, patch + import pytest -from unittest.mock import AsyncMock, patch, MagicMock from src.agents.delegation import ( - ActionType, - DelegationTask, - DelegationResult, HOUSEHOLD_THINK_MESSAGES, STREAMING_DELEGATION_WRAPPERS, + ActionType, + DelegationResult, + DelegationTask, + _detect_action_type, delegate_to_librarian, get_think_message, - _detect_action_type, ) @@ -170,11 +171,11 @@ class TestDelegateToLibrarian: @pytest.mark.asyncio async def test_delegate_to_librarian_handles_error(self): - """Test delegation handles Librarian errors gracefully.""" + """Test delegation maps Librarian errors to a user-safe result.""" with patch( "src.agents.librarian.agent.run_librarian", new_callable=AsyncMock, - side_effect=Exception("Connection refused"), + side_effect=Exception("Connection refused to http://internal:8089"), ): result = await delegate_to_librarian( task="Search for information", @@ -182,8 +183,15 @@ class TestDelegateToLibrarian: assert isinstance(result, DelegationResult) assert result.success is False - assert result.output == "" - assert result.error == "Connection refused" + # Output carries a curated butler-toned sentence + assert result.output == get_think_message( + "librarian", "Search for information", "error" + ) + # Exception detail stays in logs only - never in the result + assert "Connection refused" not in result.output + assert result.error is not None + assert "Connection refused" not in result.error + assert "internal" not in result.error @pytest.mark.asyncio async def test_delegate_to_librarian_preserves_task(self):