tatlock/src/agents/protocol.py

"""
Agent communication protocol for multi-agent coordination.

Defines standardized request/response formats for communication between:
- Steward (request analysis) → Tatlock (coordination)
- Tatlock (coordination) → Expert agents (Librarian, Developer, etc.)
"""
from enum import Enum
from typing import Any, Optional

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: Optional[int] = Field(
        default=None,
        description="Optional token limit for response"
    )
    timeout_seconds: Optional[int] = Field(
        default=60,
        description="Maximum time for task completion"
    )


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: Optional[str] = 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):
    """Base exception for agent errors."""

    def __init__(self, message: str, agent_name: str = "unknown"):
        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