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

refactor(agents): delete dead coordination/streaming delegation stack

Browse Source 
One delegation implementation remains (src/agents/delegation.py).
Removed, after verifying zero live importers post-Phase-A/B:

- src/agents/coordination.py: CoordinationEngine, duplicate
  delegate_to_librarian, AGENT_EXECUTORS/AGENT_STREAM_EXECUTORS
  (only importer was its own test module)
- run_librarian_stream: documented-broken path (Ollama streaming +
  tool call bug, PydanticAI #1292/#2256), only called by the deleted
  coordination engine
- stream_delegate_to_* wrappers + STREAMING_DELEGATION_WRAPPERS and
  the never-parsed __DELEGATION_RESULT__ marker in delegation.py
- HouseholdRegistry.get_streaming_delegation_tools() (no callers)
- tests/agents/test_coordination.py and the wrapper/stream tests

Note: the STREAMING_DELEGATION_WRAPPERS import in
src/responses/streaming.py was already removed by Phase A (7ce1c1a);
nothing to delete there.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
This commit is contained in:
jpmschweitzer
2026-07-14 11:52:36 +02:00
co-authored by Claude Fable 5
parent c00224222b
commit 31b948a748
9 changed files with 9 additions and 1018 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/coordination.py
-412
View File
@@ -1,412 +0,0 @@
"""
Multi-agent coordination engine.
Orchestrates delegation from Tatlock to expert agents (Librarian, etc.)
based on Steward recommendations. Handles:
- Routing tasks to appropriate agents
- Parallel and sequential execution
- Result aggregation
- Error handling and graceful degradation
"""
import asyncio
import time
from collections.abc import AsyncGenerator
from typing import Any
from src.agents.librarian import run_librarian, run_librarian_stream
from src.agents.protocol import (
AgentError,
AgentRequest,
AgentResponse,
AgentTimeoutError,
AgentUnavailableError,
CoordinationResult,
DelegationIntent,
DelegationReason,
)
from src.core.config import config
from src.core.household_registry import get_household_registry
from src.core.logging_config import get_logger
logger = get_logger(__name__)
# Agent execution functions registry
AGENT_EXECUTORS: dict[str, Any] = {
"librarian": run_librarian,
}
AGENT_STREAM_EXECUTORS: dict[str, Any] = {
"librarian": run_librarian_stream,
}
class CoordinationEngine:
"""
Coordinates multi-agent task execution.
Routes tasks from Tatlock to appropriate expert agents,
handles execution, and aggregates results.
"""
def __init__(self):
"""Initialize the coordination engine."""
self.registry = get_household_registry()
logger.info("coordination_engine_initialized")
def get_available_agents(self) -> list[str]:
"""
Get list of available expert agents.
Returns:
List of agent names that can accept delegations
"""
available = []
for name in self.registry.list_members():
member = self.registry.get_member(name)
if member and member.agent is not None:
available.append(name)
return available
def can_delegate_to(self, agent_name: str) -> bool:
"""
Check if delegation to an agent is possible.
Args:
agent_name: Name of the target agent
Returns:
True if agent is available and can accept tasks
"""
if agent_name not in AGENT_EXECUTORS:
return False
member = self.registry.get_member(agent_name)
return member is not None and member.agent is not None
async def execute_delegation(
self,
intent: DelegationIntent,
context: str = "",
message_history: list[Any] | None = None,
) -> AgentResponse:
"""
Execute a single delegation to an expert agent.
Args:
intent: The delegation intent with task details
context: Additional context for the agent
message_history: Optional conversation history
Returns:
AgentResponse with results
Raises:
AgentUnavailableError: If agent is not available
AgentTimeoutError: If execution times out
AgentError: For other execution errors
"""
start_time = time.time()
agent_name = intent.target_agent
logger.info(
"delegation_started",
agent=agent_name,
task=intent.task[:100],
reason=intent.reason.value,
)
# Check if agent is available
if not self.can_delegate_to(agent_name):
raise AgentUnavailableError(
f"Agent '{agent_name}' is not available for delegation",
agent_name=agent_name,
)
# Get the executor
executor = AGENT_EXECUTORS.get(agent_name)
if not executor:
raise AgentUnavailableError(
f"No executor found for agent '{agent_name}'",
agent_name=agent_name,
)
try:
# Build the request
request = AgentRequest(
task=intent.task,
context=context,
delegation_reason=intent.reason,
)
# Execute with timeout (explicit request value or configured budget)
timeout = request.timeout_seconds or config.LIBRARIAN_TIMEOUT
result = await asyncio.wait_for(
executor(
task=request.task,
context=request.context,
message_history=message_history,
),
timeout=timeout,
)
duration_ms = int((time.time() - start_time) * 1000)
logger.info(
"delegation_completed",
agent=agent_name,
duration_ms=duration_ms,
output_length=len(result),
)
return AgentResponse(
success=True,
result=result,
reasoning=f"Delegated to {agent_name}: {intent.expected_outcome}",
duration_ms=duration_ms,
)
except TimeoutError as e:
duration_ms = int((time.time() - start_time) * 1000)
logger.error(
"delegation_timeout",
agent=agent_name,
duration_ms=duration_ms,
)
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)
logger.error(
"delegation_error",
agent=agent_name,
error=str(e),
duration_ms=duration_ms,
exc_info=True,
)
return AgentResponse(
success=False,
result="",
error_message=str(e),
duration_ms=duration_ms,
)
async def execute_delegation_stream(
self,
intent: DelegationIntent,
context: str = "",
message_history: list[Any] | None = None,
) -> AsyncGenerator[str, None]:
"""
Execute a delegation with streaming output.
Args:
intent: The delegation intent with task details
context: Additional context for the agent
message_history: Optional conversation history
Yields:
Text deltas from the agent
Raises:
AgentUnavailableError: If agent is not available
"""
agent_name = intent.target_agent
logger.info(
"delegation_stream_started",
agent=agent_name,
task=intent.task[:100],
)
# Check if agent is available
if agent_name not in AGENT_STREAM_EXECUTORS:
raise AgentUnavailableError(
f"Agent '{agent_name}' does not support streaming",
agent_name=agent_name,
)
executor = AGENT_STREAM_EXECUTORS[agent_name]
try:
async for delta in executor(
task=intent.task,
context=context,
message_history=message_history,
):
yield delta
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 "\n\n" + get_think_message(agent_name, intent.task, "error")
async def coordinate(
self,
intents: list[DelegationIntent],
context: str = "",
message_history: list[Any] | None = None,
) -> CoordinationResult:
"""
Coordinate execution of multiple delegations.
Handles parallel execution for independent tasks and
sequential execution for dependent tasks.
Args:
intents: List of delegation intents to execute
context: Shared context for all agents
message_history: Optional conversation history
Returns:
CoordinationResult with aggregated results
"""
start_time = time.time()
agent_responses: dict[str, AgentResponse] = {}
agents_consulted: list[str] = []
logger.info(
"coordination_started",
intent_count=len(intents),
agents=[i.target_agent for i in intents],
)
# Sort by priority
sorted_intents = sorted(intents, key=lambda x: x.priority)
# Group by dependencies (simple version: sequential for now)
# TODO: Implement parallel execution for independent tasks
for intent in sorted_intents:
try:
response = await self.execute_delegation(
intent=intent,
context=context,
message_history=message_history,
)
agent_responses[intent.target_agent] = response
if response.success:
agents_consulted.append(intent.target_agent)
except AgentError as e:
agent_responses[intent.target_agent] = AgentResponse(
success=False,
result="",
error_message=str(e),
)
# Aggregate results
successful_results = [
r.result for r in agent_responses.values() if r.success and r.result
]
final_response = "\n\n---\n\n".join(successful_results) if successful_results else ""
total_duration = int((time.time() - start_time) * 1000)
logger.info(
"coordination_completed",
total_duration_ms=total_duration,
agents_consulted=agents_consulted,
success_count=len(successful_results),
)
return CoordinationResult(
final_response=final_response,
agent_responses=agent_responses,
delegation_intents=intents,
total_duration_ms=total_duration,
agents_consulted=agents_consulted,
)
# Global coordination engine instance
_coordination_engine: CoordinationEngine | None = None
def get_coordination_engine() -> CoordinationEngine:
"""Get the global coordination engine instance."""
global _coordination_engine
if _coordination_engine is None:
_coordination_engine = CoordinationEngine()
return _coordination_engine
async def delegate_to_librarian(
task: str,
context: str = "",
reason: DelegationReason = DelegationReason.DOMAIN_EXPERTISE,
message_history: list[Any] | None = None,
) -> AgentResponse:
"""
Convenience function to delegate a task to The Librarian.
Args:
task: Research task description
context: Additional context
reason: Why delegating to Librarian
message_history: Optional conversation history
Returns:
AgentResponse with research results
"""
engine = get_coordination_engine()
intent = DelegationIntent(
target_agent="librarian",
task=task,
reason=reason,
expected_outcome="Research findings and relevant information",
)
return await engine.execute_delegation(
intent=intent,
context=context,
message_history=message_history,
)
async def delegate_to_librarian_stream(
task: str,
context: str = "",
message_history: list[Any] | None = None,
) -> AsyncGenerator[str, None]:
"""
Convenience function to delegate to Librarian with streaming.
Args:
task: Research task description
context: Additional context
message_history: Optional conversation history
Yields:
Text deltas from The Librarian
"""
engine = get_coordination_engine()
intent = DelegationIntent(
target_agent="librarian",
task=task,
reason=DelegationReason.DOMAIN_EXPERTISE,
expected_outcome="Research findings",
)
async for delta in engine.execute_delegation_stream(
intent=intent,
context=context,
message_history=message_history,
):
yield delta
src/agents/delegation.py
-98
View File
@@ -9,7 +9,6 @@ This implements the agent-as-tool pattern recommended by PydanticAI:
agents call other agents via tool wrappers, keeping each agent focused.
"""
import asyncio
from collections.abc import AsyncGenerator
from dataclasses import dataclass, field
from enum import Enum
@@ -572,103 +571,6 @@ async def delegate_to_housekeeper(
)
# =============================================================================
# Streaming Delegation Wrappers (with Think Messages)
# =============================================================================
async def stream_delegate_to_librarian(
task: str,
context: str = "",
) -> AsyncGenerator[str, None]:
"""
Stream delegation to Librarian with automatic think messages.
Yields butler-perspective think messages before and after the delegation,
allowing the UI to show progress to the user.
Args:
task: Task description
context: Additional context
Yields:
str: Think messages and final result marker
"""
# Yield start message (deterministic)
yield get_think_message("librarian", task, "start") + "\n"
# Execute delegation
result = await delegate_to_librarian(task, context)
# Yield completion message (deterministic)
if result.success:
yield get_think_message("librarian", task, "success") + "\n"
else:
yield get_think_message("librarian", task, "error") + "\n"
# Yield result marker for extraction
yield f"__DELEGATION_RESULT__:librarian:{result.output}"
async def stream_delegate_to_biographer(
task: str,
context: str = "",
) -> AsyncGenerator[str, None]:
"""
Stream delegation to Biographer with automatic think messages.
Args:
task: Task description
context: Additional context
Yields:
str: Think messages and final result marker
"""
yield get_think_message("biographer", task, "start") + "\n"
result = await delegate_to_biographer(task, context)
if result.success:
yield get_think_message("biographer", task, "success") + "\n"
else:
yield get_think_message("biographer", task, "error") + "\n"
yield f"__DELEGATION_RESULT__:biographer:{result.output}"
async def stream_delegate_to_housekeeper(
task: str,
context: str = "",
) -> AsyncGenerator[str, None]:
"""
Stream delegation to Housekeeper with automatic think messages.
Args:
task: Task description
context: Additional context
Yields:
str: Think messages and final result marker
"""
yield get_think_message("housekeeper", task, "start") + "\n"
result = await delegate_to_housekeeper(task, context)
if result.success:
yield get_think_message("housekeeper", task, "success") + "\n"
else:
yield get_think_message("housekeeper", task, "error") + "\n"
yield f"__DELEGATION_RESULT__:housekeeper:{result.output}"
# Mapping of streaming delegation wrappers
STREAMING_DELEGATION_WRAPPERS = {
"librarian": stream_delegate_to_librarian,
"biographer": stream_delegate_to_biographer,
"housekeeper": stream_delegate_to_housekeeper,
}
# Future expert delegation wrappers will be added here:
# - delegate_to_developer(task, context) -> DelegationResult
# - delegate_to_secretary(task, context) -> DelegationResult
src/agents/librarian/__init__.py
-2
View File
@@ -10,7 +10,6 @@ Connects to the library-desk API to provide:
from src.agents.librarian.agent import (
get_librarian_agent,
run_librarian,
run_librarian_stream,
)
from src.agents.librarian.capability import (
LIBRARIAN_CAPABILITY,
@@ -26,5 +25,4 @@ __all__ = [
"register_librarian",
"unregister_librarian",
"run_librarian",
"run_librarian_stream",
]
src/agents/librarian/agent.py
-64
View File
@@ -273,67 +273,3 @@ async def run_librarian(
raise AgentError(
"Research task failed", agent_name="librarian"
) from e
async def run_librarian_stream(
task: str,
context: str = "",
message_history: list[Any] | None = None,
):
"""
Execute a research task with streaming output.
Yields text deltas as The Librarian generates the response.
Args:
task: The research task or question
context: Additional context from conversation
message_history: Optional conversation history
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)
"""
agent = get_librarian_agent()
# Build prompt with context if provided
prompt = task
if context:
prompt = f"Context: {context}\n\nTask: {task}"
logger.info(
"librarian_stream_started",
task=task[:100],
)
try:
# One shared library-desk connection for all tool calls in this run
async with library_client_session():
async with agent.run_stream(
prompt,
message_history=message_history,
) as response:
async for delta in response.stream_text(delta=True):
yield delta
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,
)
raise AgentError(
"Research task failed", agent_name="librarian"
) from e
src/core/household_registry.py
-59
View File
@@ -273,65 +273,6 @@ class HouseholdRegistry:
return tools
def get_streaming_delegation_tools(self, names: list[str]) -> list[Any]:
"""
Get streaming delegation wrapper tools for specified capabilities.
Similar to get_delegation_tools() but returns streaming wrappers
that yield butler-perspective think messages during execution.
These wrappers emit think slugs like:
- "Allow me to consult the archives, sir."
- "The Librarian has compiled the relevant findings."
Args:
names: List of member names to include
Returns:
List of streaming delegation wrappers and/or raw tools
Example:
>>> tools = registry.get_streaming_delegation_tools(["librarian"])
>>> async for chunk in tools[0](task="Search for Docker"):
... print(chunk) # Yields think messages then result
"""
from src.agents.delegation import STREAMING_DELEGATION_WRAPPERS
tools = []
for name in names:
member = self._members.get(name)
if not member:
logger.warning(
"household_member_not_found",
requested_name=name,
available_names=list(self._members.keys()),
)
continue
# Check if this member has a streaming delegation wrapper
if name in STREAMING_DELEGATION_WRAPPERS and member.agent is not None:
tools.append(STREAMING_DELEGATION_WRAPPERS[name])
logger.debug(
"streaming_delegation_wrapper_added",
member=name,
)
else:
# No agent = direct tools (e.g., tatlock_core)
tools.extend(member.tools)
logger.debug(
"raw_tools_added",
member=name,
tool_count=len(member.tools),
)
logger.info(
"streaming_delegation_tools_created",
requested_members=names,
total_tools=len(tools),
)
return tools
def list_members(self) -> list[str]:
"""
List all registered member names.