refactor(agents): reduce protocol.py to the live AgentError
Post-coordination-removal sweep: the coordination wire protocol (AgentRequest, AgentResponse, DelegationIntent, CoordinationResult, DelegationReason, TaskComplexity, ToolCallRecord, AgentTimeoutError, AgentUnavailableError, DelegationError) had zero importers left in src/ - only its own test module. AgentError stays (raised by run_librarian, mapped to user-safe failures by delegation.py). Also drops the stale coordination.py line from the README tree. Import-cycle sanity: python -c 'import src.main' passes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
This commit is contained in:
Claude Fable 5
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+5
-192
@@ -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
|
||||
|
||||
+15
-241
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user