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

fix(agents): make librarian failures structured and user-safe

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-07-14 10:12:30 +02:00
co-authored by Claude Fable 5
parent 59f5b54ac9
commit f853db8ccc
10 changed files with 277 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/responses/service.py
+62 -45
View File
@@ -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
src/responses/streaming.py
+46 -26
View File
@@ -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):