jpmschweitzerandClaude Sonnet 4.5
6eed5f4d13
Add comprehensive two-tier architecture where Steward analyzes requests and Tatlock executes with scoped tools. Includes full infrastructure for request preprocessing, tool tracking, benchmarking, and streaming. **Added:** - Steward agent for request analysis and capability recommendation - Household Registry for centralized capability management - Request preprocessing pipeline (Steward → Tatlock flow) - Tool usage tracking and benchmarking system - Streaming transparency (Steward reasoning visible in streams) - Structured logging with operation timing - Redis benchmark storage with 30-day expiry - Benchmark analysis CLI tools **Infrastructure:** - src/agents/steward/ - Steward agent implementation - src/agents/tatlock_core/ - Tatlock capability domain - src/core/preprocessing.py - Request preprocessing pipeline - src/core/tool_tracking.py - Tool call tracking - src/core/benchmarks.py - Benchmark recording system - src/core/household_registry.py - Capability registry - src/core/startup.py - Application startup coordination - src/core/logging_config.py - Structured logging setup **Integration:** - Responses API uses Steward for Tatlock requests - Chat Completions wraps Responses API for OpenAI compatibility - Streaming coordinator supports Steward + Tatlock flow - Tool scoping per request based on Steward recommendations **Testing:** - Integration tests for Steward-Tatlock flow - Benchmark and registry unit tests - Steward streaming tests See PHASE2_PLAN.md and PHASE2_COMPLETE.md for detailed documentation. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
314 lines
10 KiB
Python
314 lines
10 KiB
Python
"""
|
|
Tests for household registry.
|
|
|
|
Tests capability registration, toolset scoping, and coordination features.
|
|
"""
|
|
import pytest
|
|
from pydantic_ai.tools import Tool
|
|
|
|
from src.core.household_registry import (
|
|
HouseholdCapability,
|
|
HouseholdMember,
|
|
HouseholdRegistry,
|
|
household_registry,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def registry():
|
|
"""Create a fresh registry for each test."""
|
|
reg = HouseholdRegistry()
|
|
return reg
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_capability():
|
|
"""Sample household capability."""
|
|
return HouseholdCapability(
|
|
name="test_tools",
|
|
role="Test Tools",
|
|
category="testing",
|
|
description="Tools for testing purposes",
|
|
domains=["testing", "validation"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_tools():
|
|
"""Sample tool definitions."""
|
|
def test_function_1(x: int) -> int:
|
|
"""Test function 1."""
|
|
return x * 2
|
|
|
|
def test_function_2(x: str) -> str:
|
|
"""Test function 2."""
|
|
return x.upper()
|
|
|
|
return [
|
|
Tool(function=test_function_1, name="test_tool_1"),
|
|
Tool(function=test_function_2, name="test_tool_2"),
|
|
]
|
|
|
|
|
|
class TestHouseholdCapability:
|
|
"""Test HouseholdCapability model."""
|
|
|
|
def test_capability_creation(self, sample_capability):
|
|
"""Test creating a capability."""
|
|
assert sample_capability.name == "test_tools"
|
|
assert sample_capability.role == "Test Tools"
|
|
assert sample_capability.category == "testing"
|
|
assert "testing" in sample_capability.domains
|
|
assert sample_capability.cost == "low"
|
|
assert sample_capability.requires_network is False
|
|
|
|
def test_capability_validation(self):
|
|
"""Test capability field validation."""
|
|
# Should succeed with valid data
|
|
cap = HouseholdCapability(
|
|
name="valid",
|
|
role="Valid Role",
|
|
category="test",
|
|
description="Test description",
|
|
domains=["test"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
assert cap.name == "valid"
|
|
|
|
|
|
class TestHouseholdMember:
|
|
"""Test HouseholdMember model."""
|
|
|
|
def test_member_creation(self, sample_capability, sample_tools):
|
|
"""Test creating a household member."""
|
|
member = HouseholdMember(
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
agent=None,
|
|
)
|
|
assert member.capability.name == "test_tools"
|
|
assert len(member.tools) == 2
|
|
assert member.agent is None
|
|
|
|
def test_member_with_agent(self, sample_capability, sample_tools):
|
|
"""Test member can include an agent."""
|
|
from unittest.mock import Mock
|
|
mock_agent = Mock()
|
|
|
|
member = HouseholdMember(
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
agent=mock_agent,
|
|
)
|
|
assert member.agent is mock_agent
|
|
|
|
|
|
class TestHouseholdRegistry:
|
|
"""Test HouseholdRegistry functionality."""
|
|
|
|
def test_registry_initialization(self, registry):
|
|
"""Test registry initializes empty."""
|
|
assert len(registry) == 0
|
|
assert registry.list_members() == []
|
|
|
|
def test_register_member(self, registry, sample_capability, sample_tools):
|
|
"""Test registering a household member."""
|
|
registry.register(
|
|
name="test_tools",
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
)
|
|
|
|
assert len(registry) == 1
|
|
assert "test_tools" in registry
|
|
assert "test_tools" in registry.list_members()
|
|
|
|
def test_register_name_mismatch(self, registry, sample_capability, sample_tools):
|
|
"""Test registration fails with name mismatch."""
|
|
with pytest.raises(ValueError, match="Name mismatch"):
|
|
registry.register(
|
|
name="wrong_name",
|
|
capability=sample_capability,
|
|
tools=sample_tools,
|
|
)
|
|
|
|
def test_unregister_member(self, registry, sample_capability, sample_tools):
|
|
"""Test unregistering a member."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
assert "test_tools" in registry
|
|
|
|
registry.unregister("test_tools")
|
|
assert "test_tools" not in registry
|
|
assert len(registry) == 0
|
|
|
|
def test_get_member(self, registry, sample_capability, sample_tools):
|
|
"""Test retrieving a member."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
|
|
member = registry.get_member("test_tools")
|
|
assert member is not None
|
|
assert member.capability.name == "test_tools"
|
|
assert len(member.tools) == 2
|
|
|
|
def test_get_nonexistent_member(self, registry):
|
|
"""Test retrieving non-existent member returns None."""
|
|
member = registry.get_member("nonexistent")
|
|
assert member is None
|
|
|
|
def test_get_all_capabilities(self, registry, sample_capability, sample_tools):
|
|
"""Test retrieving all capability summaries."""
|
|
# Register multiple members
|
|
cap1 = sample_capability
|
|
cap2 = HouseholdCapability(
|
|
name="other_tools",
|
|
role="Other Tools",
|
|
category="utility",
|
|
description="Other test tools",
|
|
domains=["utility"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
|
|
registry.register("test_tools", cap1, sample_tools)
|
|
registry.register("other_tools", cap2, sample_tools[:1])
|
|
|
|
capabilities = registry.get_all_capabilities()
|
|
assert len(capabilities) == 2
|
|
assert any(cap.name == "test_tools" for cap in capabilities)
|
|
assert any(cap.name == "other_tools" for cap in capabilities)
|
|
|
|
def test_get_scoped_tools(self, registry, sample_capability, sample_tools):
|
|
"""Test creating scoped toolsets."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
|
|
# Get scoped tools
|
|
tools = registry.get_scoped_tools(["test_tools"])
|
|
assert len(tools) == 2
|
|
assert tools[0].name == "test_tool_1"
|
|
assert tools[1].name == "test_tool_2"
|
|
|
|
def test_get_scoped_tools_multiple_members(self, registry, sample_tools):
|
|
"""Test scoping with multiple members."""
|
|
cap1 = HouseholdCapability(
|
|
name="member1",
|
|
role="Member 1",
|
|
category="test",
|
|
description="First member",
|
|
domains=["test"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
cap2 = HouseholdCapability(
|
|
name="member2",
|
|
role="Member 2",
|
|
category="test",
|
|
description="Second member",
|
|
domains=["test"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
|
|
registry.register("member1", cap1, sample_tools[:1])
|
|
registry.register("member2", cap2, sample_tools[1:])
|
|
|
|
# Get combined tools
|
|
tools = registry.get_scoped_tools(["member1", "member2"])
|
|
assert len(tools) == 2
|
|
|
|
def test_get_scoped_tools_nonexistent_member(self, registry, sample_capability, sample_tools):
|
|
"""Test scoping with non-existent member logs warning."""
|
|
registry.register("test_tools", sample_capability, sample_tools)
|
|
|
|
# Request includes non-existent member
|
|
tools = registry.get_scoped_tools(["test_tools", "nonexistent"])
|
|
# Should return only existing member's tools
|
|
assert len(tools) == 2
|
|
|
|
def test_get_members_by_domain(self, registry, sample_tools):
|
|
"""Test filtering members by domain."""
|
|
cap1 = HouseholdCapability(
|
|
name="research_tools",
|
|
role="Research Tools",
|
|
category="research",
|
|
description="Research tools",
|
|
domains=["research", "analysis"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
cap2 = HouseholdCapability(
|
|
name="compute_tools",
|
|
role="Compute Tools",
|
|
category="computation",
|
|
description="Computation tools",
|
|
domains=["computation", "math"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
|
|
registry.register("research_tools", cap1, sample_tools)
|
|
registry.register("compute_tools", cap2, sample_tools)
|
|
|
|
# Filter by domain
|
|
research_caps = registry.get_members_by_domain("research")
|
|
assert len(research_caps) == 1
|
|
assert research_caps[0].name == "research_tools"
|
|
|
|
compute_caps = registry.get_members_by_domain("computation")
|
|
assert len(compute_caps) == 1
|
|
assert compute_caps[0].name == "compute_tools"
|
|
|
|
def test_get_members_by_category(self, registry, sample_tools):
|
|
"""Test filtering members by category."""
|
|
cap1 = HouseholdCapability(
|
|
name="core_tools",
|
|
role="Core Tools",
|
|
category="core",
|
|
description="Core tools",
|
|
domains=["general"],
|
|
cost="low",
|
|
requires_network=False,
|
|
)
|
|
cap2 = HouseholdCapability(
|
|
name="research_tools",
|
|
role="Research Tools",
|
|
category="research",
|
|
description="Research tools",
|
|
domains=["research"],
|
|
cost="medium",
|
|
requires_network=True,
|
|
)
|
|
|
|
registry.register("core_tools", cap1, sample_tools)
|
|
registry.register("research_tools", cap2, sample_tools)
|
|
|
|
# Filter by category
|
|
core_caps = registry.get_members_by_category("core")
|
|
assert len(core_caps) == 1
|
|
assert core_caps[0].name == "core_tools"
|
|
|
|
research_caps = registry.get_members_by_category("research")
|
|
assert len(research_caps) == 1
|
|
assert research_caps[0].name == "research_tools"
|
|
|
|
|
|
class TestGlobalRegistry:
|
|
"""Test the global registry instance."""
|
|
|
|
def test_global_registry_exists(self):
|
|
"""Test global registry is available."""
|
|
from src.core.household_registry import get_household_registry
|
|
|
|
registry = get_household_registry()
|
|
assert isinstance(registry, HouseholdRegistry)
|
|
|
|
def test_global_registry_singleton(self):
|
|
"""Test get_household_registry returns same instance."""
|
|
from src.core.household_registry import get_household_registry
|
|
|
|
reg1 = get_household_registry()
|
|
reg2 = get_household_registry()
|
|
assert reg1 is reg2
|