diff --git a/CHANGELOG.md b/CHANGELOG.md index 23d10b4..d7fcb26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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) +- **Orphaned agent protocol models** - `src/agents/protocol.py` now contains only the live `AgentError`; the coordination wire protocol it carried (`AgentRequest`, `AgentResponse`, `DelegationIntent`, `CoordinationResult`, `DelegationReason`, `TaskComplexity`, `ToolCallRecord`, `AgentTimeoutError`, `AgentUnavailableError`, `DelegationError`) had no importer left outside its own tests after the coordination stack removal ### Added diff --git a/README.md b/README.md index b4bd78c..c47563a 100644 --- a/README.md +++ b/README.md @@ -384,9 +384,8 @@ tatlock/ │ │ ├── steward/ # The Steward - request analysis │ │ ├── tatlock_core/ # Core butler tools │ │ ├── tatlock.py # Tatlock PydanticAI agent -│ │ ├── coordination.py # Multi-agent coordination │ │ ├── delegation.py # Expert delegation wrappers -│ │ └── protocol.py # Agent communication protocol +│ │ └── protocol.py # Agent error protocol │ ├── responses/ # Responses API (primary endpoint) │ ├── chat/ # Chat Completions wrapper │ ├── models/ # Models listing diff --git a/src/agents/protocol.py b/src/agents/protocol.py index 6161a5a..783c78d 100644 --- a/src/agents/protocol.py +++ b/src/agents/protocol.py @@ -1,183 +1,11 @@ """ -Agent communication protocol for multi-agent coordination. +Agent error protocol. -Defines standardized request/response formats for communication between: -- Steward (request analysis) → Tatlock (coordination) -- Tatlock (coordination) → Expert agents (Librarian, Developer, etc.) +Structured exceptions raised by expert agents (e.g. The Librarian) so +callers - the delegation wrappers in src/agents/delegation.py - can +report success=False and map failures to curated user-safe messages +while exception detail stays in the logs. """ -from enum import Enum -from typing import Any - -from pydantic import BaseModel, Field - - -class DelegationReason(str, Enum): - """Why a task is being delegated to an expert agent.""" - DOMAIN_EXPERTISE = "domain_expertise" # Expert has specialized knowledge - TOOL_ACCESS = "tool_access" # Expert has required tools - RESOURCE_EFFICIENCY = "resource_efficiency" # Better handled by specialist - USER_PREFERENCE = "user_preference" # User requested specific agent - - -class TaskComplexity(str, Enum): - """Complexity estimate for task execution.""" - SIMPLE = "simple" # Single tool call, fast - MODERATE = "moderate" # Multiple steps, moderate time - COMPLEX = "complex" # Multi-agent, significant processing - - -class AgentRequest(BaseModel): - """ - Request to an expert agent. - - Contains everything the agent needs to execute a task, - including context from the conversation and delegation intent. - """ - task: str = Field( - ..., - description="Clear description of what the agent should do" - ) - context: str = Field( - default="", - description="Relevant context from conversation history" - ) - constraints: list[str] = Field( - default_factory=list, - description="Any constraints or requirements for the task" - ) - delegation_reason: DelegationReason = Field( - default=DelegationReason.DOMAIN_EXPERTISE, - description="Why this task was delegated to this agent" - ) - user_id: str = Field( - default="default", - description="User identifier for multi-tenant operations" - ) - max_tokens: int | None = Field( - default=None, - description="Optional token limit for response" - ) - timeout_seconds: int | None = Field( - default=None, - description=( - "Maximum time for task completion; None uses the configured " - "expert budget (LIBRARIAN_TIMEOUT)" - ) - ) - - -class ToolCallRecord(BaseModel): - """Record of a tool call made during execution.""" - tool_name: str - arguments: dict[str, Any] - result: str - duration_ms: int - - -class AgentResponse(BaseModel): - """ - Response from an expert agent. - - Contains the result, reasoning, and metadata about execution. - """ - success: bool = Field( - ..., - description="Whether the task completed successfully" - ) - result: str = Field( - ..., - description="The main output/answer from the agent" - ) - reasoning: str = Field( - default="", - description="Agent's reasoning process (for transparency)" - ) - tool_calls: list[ToolCallRecord] = Field( - default_factory=list, - description="Tools called during execution" - ) - confidence: float = Field( - default=1.0, - ge=0.0, - le=1.0, - description="Agent's confidence in the result (0.0-1.0)" - ) - sources: list[str] = Field( - default_factory=list, - description="Sources or references used" - ) - error_message: str | None = Field( - default=None, - description="Error details if success=False" - ) - duration_ms: int = Field( - default=0, - description="Total execution time in milliseconds" - ) - - -class DelegationIntent(BaseModel): - """ - Intent to delegate a task to an expert agent. - - Created by Tatlock when deciding to delegate, based on - Steward's recommendations. - """ - target_agent: str = Field( - ..., - description="Name of the expert agent to delegate to" - ) - task: str = Field( - ..., - description="Task description for the agent" - ) - reason: DelegationReason = Field( - default=DelegationReason.DOMAIN_EXPERTISE, - description="Why delegating to this agent" - ) - expected_outcome: str = Field( - default="", - description="What we expect the agent to provide" - ) - priority: int = Field( - default=1, - ge=1, - le=10, - description="Priority (1=highest, 10=lowest)" - ) - depends_on: list[str] = Field( - default_factory=list, - description="Other delegation IDs this depends on (for sequencing)" - ) - - -class CoordinationResult(BaseModel): - """ - Result of multi-agent coordination. - - Aggregates results from multiple expert agents into - a single coherent response. - """ - final_response: str = Field( - ..., - description="Synthesized response from all agents" - ) - agent_responses: dict[str, AgentResponse] = Field( - default_factory=dict, - description="Individual responses keyed by agent name" - ) - delegation_intents: list[DelegationIntent] = Field( - default_factory=list, - description="All delegations that were executed" - ) - total_duration_ms: int = Field( - default=0, - description="Total coordination time" - ) - agents_consulted: list[str] = Field( - default_factory=list, - description="Names of agents that contributed" - ) class AgentError(Exception): @@ -187,18 +15,3 @@ class AgentError(Exception): self.message = message self.agent_name = agent_name super().__init__(f"[{agent_name}] {message}") - - -class AgentTimeoutError(AgentError): - """Agent execution timed out.""" - pass - - -class AgentUnavailableError(AgentError): - """Agent is not available or registered.""" - pass - - -class DelegationError(AgentError): - """Error during task delegation.""" - pass diff --git a/tests/agents/test_protocol.py b/tests/agents/test_protocol.py index f8a762e..6ada5b1 100644 --- a/tests/agents/test_protocol.py +++ b/tests/agents/test_protocol.py @@ -1,257 +1,31 @@ """ -Tests for agent communication protocol. +Tests for the agent error protocol. """ - import pytest -from src.agents.protocol import ( - AgentError, - AgentRequest, - AgentResponse, - AgentTimeoutError, - AgentUnavailableError, - CoordinationResult, - DelegationIntent, - DelegationReason, - ToolCallRecord, -) +from src.agents.protocol import AgentError @pytest.mark.unit -class TestAgentRequest: - """Tests for AgentRequest model.""" +class TestAgentError: + """Tests for the AgentError exception.""" - def test_basic_request(self): - """Test creating a basic agent request.""" - request = AgentRequest(task="Find information about Docker") - - assert request.task == "Find information about Docker" - assert request.context == "" - # No hardcoded default - None defers to the configured budget - assert request.timeout_seconds is None - - def test_request_with_context(self): - """Test request with additional context.""" - request = AgentRequest( - task="Find Docker networking docs", - context="User is setting up a homelab", - delegation_reason=DelegationReason.DOMAIN_EXPERTISE, - ) - - assert request.task == "Find Docker networking docs" - assert request.context == "User is setting up a homelab" - assert request.delegation_reason == DelegationReason.DOMAIN_EXPERTISE - - def test_request_serialization(self): - """Test request can be serialized to dict.""" - request = AgentRequest( - task="Research task", - context="Some context", - ) - - data = request.model_dump() - - assert data["task"] == "Research task" - assert data["context"] == "Some context" - - -@pytest.mark.unit -class TestAgentResponse: - """Tests for AgentResponse model.""" - - def test_successful_response(self): - """Test creating a successful response.""" - response = AgentResponse( - success=True, - result="Here are the findings...", - reasoning="Searched wiki and found relevant docs", - duration_ms=1500, - ) - - assert response.success is True - assert response.result == "Here are the findings..." - assert response.reasoning == "Searched wiki and found relevant docs" - assert response.duration_ms == 1500 - assert response.error_message is None - - def test_failed_response(self): - """Test creating a failed response.""" - response = AgentResponse( - success=False, - result="", - error_message="Connection timeout", - duration_ms=30000, - ) - - assert response.success is False - assert response.result == "" - assert response.error_message == "Connection timeout" - - def test_response_with_tool_calls(self): - """Test response tracking tool calls.""" - tool_call = ToolCallRecord( - tool_name="hybrid_search", - arguments={"query": "Docker networking"}, - result="Found 5 results", - duration_ms=500, - ) - - response = AgentResponse( - success=True, - result="Based on search...", - tool_calls=[tool_call], - ) - - assert len(response.tool_calls) == 1 - assert response.tool_calls[0].tool_name == "hybrid_search" - - -@pytest.mark.unit -class TestDelegationIntent: - """Tests for DelegationIntent model.""" - - def test_basic_intent(self): - """Test creating a basic delegation intent.""" - intent = DelegationIntent( - target_agent="librarian", - task="Research Docker networking", - reason=DelegationReason.DOMAIN_EXPERTISE, - expected_outcome="Documentation and examples", - ) - - assert intent.target_agent == "librarian" - assert intent.task == "Research Docker networking" - assert intent.reason == DelegationReason.DOMAIN_EXPERTISE - assert intent.priority == 1 # Default - - def test_intent_with_priority(self): - """Test intent with custom priority.""" - intent = DelegationIntent( - target_agent="librarian", - task="Urgent research", - reason=DelegationReason.RESOURCE_EFFICIENCY, - expected_outcome="Quick answer", - priority=1, - ) - - assert intent.priority == 1 - - -@pytest.mark.unit -class TestDelegationReason: - """Tests for DelegationReason enum.""" - - def test_all_reasons_have_values(self): - """Test all delegation reasons are defined.""" - reasons = list(DelegationReason) - - assert DelegationReason.DOMAIN_EXPERTISE in reasons - assert DelegationReason.TOOL_ACCESS in reasons - assert DelegationReason.RESOURCE_EFFICIENCY in reasons - assert DelegationReason.USER_PREFERENCE in reasons - - -@pytest.mark.unit -class TestCoordinationResult: - """Tests for CoordinationResult model.""" - - def test_single_agent_result(self): - """Test coordination with single agent.""" - agent_response = AgentResponse( - success=True, - result="Research findings", - duration_ms=1000, - ) - - intent = DelegationIntent( - target_agent="librarian", - task="Research task", - reason=DelegationReason.DOMAIN_EXPERTISE, - expected_outcome="Findings", - ) - - result = CoordinationResult( - final_response="Research findings", - agent_responses={"librarian": agent_response}, - delegation_intents=[intent], - total_duration_ms=1200, - agents_consulted=["librarian"], - ) - - assert result.final_response == "Research findings" - assert len(result.agent_responses) == 1 - assert result.agents_consulted == ["librarian"] - - def test_empty_result(self): - """Test coordination with no delegations.""" - result = CoordinationResult( - final_response="", - agent_responses={}, - delegation_intents=[], - total_duration_ms=0, - agents_consulted=[], - ) - - assert result.final_response == "" - assert len(result.agents_consulted) == 0 - - -@pytest.mark.unit -class TestAgentErrors: - """Tests for agent error types.""" - - def test_agent_error(self): - """Test base AgentError.""" + def test_agent_error_defaults(self): + """Test base AgentError with default agent name.""" error = AgentError("Something went wrong") assert "Something went wrong" in str(error) assert error.agent_name == "unknown" - def test_agent_timeout_error(self): - """Test AgentTimeoutError.""" - error = AgentTimeoutError( - "Timed out after 60s", - agent_name="librarian", - ) + def test_agent_error_carries_agent_name(self): + """Agent name is stored and prefixed into the message.""" + error = AgentError("Research task failed", agent_name="librarian") - assert "Timed out" in str(error) assert error.agent_name == "librarian" + assert str(error) == "[librarian] Research task failed" + assert error.message == "Research task failed" - def test_agent_unavailable_error(self): - """Test AgentUnavailableError.""" - error = AgentUnavailableError( - "Agent not registered", - agent_name="unknown_agent", - ) - - assert "not registered" in str(error) - assert error.agent_name == "unknown_agent" - - -@pytest.mark.unit -class TestToolCallRecord: - """Tests for ToolCallRecord model.""" - - def test_tool_call_record(self): - """Test creating a tool call record.""" - record = ToolCallRecord( - tool_name="semantic_search", - arguments={"query": "networking concepts", "limit": 10}, - result="Found 10 relevant documents", - duration_ms=250, - ) - - assert record.tool_name == "semantic_search" - assert record.arguments["query"] == "networking concepts" - assert record.duration_ms == 250 - - def test_tool_call_with_empty_result(self): - """Test tool call with empty result.""" - record = ToolCallRecord( - tool_name="query_graph", - arguments={"cypher": "MATCH (n) RETURN n"}, - result="", - duration_ms=100, - ) - - assert record.result == "" + def test_agent_error_is_catchable_as_exception(self): + """AgentError participates in normal exception handling.""" + with pytest.raises(AgentError): + raise AgentError("boom", agent_name="librarian")