tatlock/PHASE2_PLAN.md

Phase 2 Implementation Plan: The Steward

Status: Active Planning
Created: 2025-12-07
Estimated Duration: 4-5 weeks
Goal: Implement first-tier request analysis and household capability coordination

---

Executive Summary

Phase 2 introduces The Steward - a first-tier LLM agent that analyzes incoming requests, identifies relevant household capabilities, and provides focused recommendations to Tatlock (the Butler). This creates a two-tier architecture that prevents cognitive overload and enables efficient tool/agent coordination.

Key Deliverables

1. Household Registry: Centralized capability catalog with PydanticAI Toolsets
2. Steward Agent: Request analyzer with conversation context awareness
3. Tool Scoping: Dynamic toolset creation based on recommendations
4. Observability: Performance benchmarking and tool usage tracking via Redis
5. Integration: Full Steward → Tatlock request flow

---

Core Architectural Principles

1. Household-Based Organization

• Each expert agent owns their tools in a domain directory
• Tools organized as functional clusters around capabilities
• Example: src/agents/tatlock_core/ contains calculator, datetime, web search

2. Two-Tier Capability Abstraction

• Executive Summary: High-level capabilities for Steward/Butler coordination
• Implementation Details: Full tool specifications for household members
• Steward sees summaries, household members see full details

3. PydanticAI Native Patterns

• Use FunctionToolset and CombinedToolset for composition
• Decorator-based tool registration (@agent.tool)
• Structured outputs via Pydantic models
• Agent delegation pattern for expert agents (Phase 4)

4. Separate Registries

• Household Registry: Tools + capabilities (new in Phase 2)
• Model Registry: Agents/models (existing from Phase 1)
• Clean separation of concerns

5. Start Minimal

• Only 3 core Tatlock tools initially: calculator, datetime, web search
• No new tools until expert agents exist (Phase 4)
• Prove the pattern before expanding

---

Implementation Milestones

Milestone 1: Household Registry + Logging Infrastructure (Week 1-2)

Goal

Create a registry system that aggregates household capabilities using PydanticAI Toolsets and establish observability infrastructure.

Tasks

1.1 Create Household Registry Module

Location: src/core/household_registry.py

from pydantic import BaseModel
from pydantic_ai import FunctionToolset, CombinedToolset

class HouseholdCapability(BaseModel):
    """Executive summary of a household member's capabilities."""
    name: str  # "tatlock_core", "librarian", "developer"
    role: str  # "Butler's Core Tools", "The Librarian"
    category: str  # "core", "research", "technical"
    description: str  # One-sentence description
    domains: list[str]  # ["computation", "information", "datetime"]
    cost: str  # "low", "medium", "high"
    requires_network: bool

class HouseholdMember(BaseModel):
    """Full specification of a household member."""
    capability: HouseholdCapability
    toolset: FunctionToolset
    agent: Agent | None = None  # For expert agents in Phase 4

class HouseholdRegistry:
    """Registry of household capabilities and implementations."""
    
    def __init__(self):
        self._members: dict[str, HouseholdMember] = {}
    
    def register(
        self,
        name: str,
        capability: HouseholdCapability,
        toolset: FunctionToolset,
        agent: Agent | None = None
    ):
        """Register a household member."""
        self._members[name] = HouseholdMember(
            capability=capability,
            toolset=toolset,
            agent=agent
        )
    
    def get_all_capabilities(self) -> list[HouseholdCapability]:
        """Get executive summaries for Steward/Butler."""
        return [m.capability for m in self._members.values()]
    
    def get_scoped_toolset(self, names: list[str]) -> CombinedToolset:
        """Create combined toolset from recommended capabilities."""
        toolsets = [self._members[name].toolset for name in names]
        return CombinedToolset(toolsets)

# Global registry instance
household_registry = HouseholdRegistry()

1.2 Reorganize Tatlock Core Tools

Create domain-based organization:

src/agents/tatlock_core/
├── __init__.py
├── tools.py          # Tool implementations (moved from src/agents/tools.py)
├── toolset.py        # PydanticAI toolset registration  
└── capability.py     # Executive summary for registry

1.3 Create Logging Infrastructure

Location: src/core/logging_config.py

• Structured logging with structlog
• JSON format for machine parsing
• Operation timing and metadata tracking
• Context manager for automatic timing

1.4 Create Redis Benchmark Storage

Location: src/core/benchmarks.py

Features:

• Performance benchmark recording (Steward analysis, tool calls)
• Cross-session persistence via Redis
• Time-series storage with automatic expiry (30 days)
• Queryable metrics for analysis

Benchmark schema:

class PerformanceBenchmark(BaseModel):
    timestamp: datetime
    operation: str  # "steward_analysis", "tool_call"
    duration_seconds: float
    success: bool
    
    # Steward-specific
    recommendation_count: Optional[int]
    confidence: Optional[float]
    
    # Tool-specific
    tool_name: Optional[str]
    was_recommended: Optional[bool]
    was_actually_used: Optional[bool]
    
    # Context
    conversation_id: Optional[str]
    metadata: dict

1.5 Testing

• Test household registry registration and retrieval
• Test Toolset composition
• Test benchmark recording to Redis
• Test structured logging output

Success Criteria

• ✅ Household registry operational
• ✅ Tatlock core tools organized in domain directory
• ✅ Redis benchmarks working
• ✅ Structured logging functional
• ✅ Tests pass and maintain 80%+ coverage

---

Milestone 2: Minimal Steward Agent with Context Analysis (Week 3-4)

Goal

Create a Steward agent that analyzes requests with full conversation context and recommends relevant household capabilities.

Tasks

2.1 Create Steward Agent

Location: src/agents/steward/agent.py

Structured output schema:

class ConversationContext(BaseModel):
    """Contextual information from conversation history."""
    has_previous_context: bool
    relevant_turns: list[int]  # 0-indexed turn numbers
    context_summary: str  # Summary for Butler

class StewardRecommendation(BaseModel):
    """Structured recommendation from Steward analysis."""
    recommended_capabilities: list[str]
    reasoning: str
    estimated_complexity: Literal["simple", "moderate", "complex"]
    conversation_context: ConversationContext
    missing_capabilities: Optional[str] = None

Key features:

• Uses same model as Tatlock (ollama:mistral-nemo) for VRAM efficiency
• Receives FULL conversation history
• Queries household registry via tool
• Conservative recommendations (avoid over-inclusion)
• Explicit handling of missing capabilities

2.2 Steward System Prompt

Responsibilities:

1. Capability Recommendation: Query registry, recommend only necessary tools
2. Conversation Analysis: Identify references to previous topics
3. Complexity Assessment: Simple/moderate/complex classification
4. Missing Capability Detection: Suggest what's needed if no tools available

2.3 Steward Service Layer with Logging

Location: src/agents/steward/service.py

async def analyze_request(
    user_request: str,
    conversation_history: list[dict]  # FULL conversation
) -> StewardRecommendation:
    """Analyze request with full conversation context."""
    
    async with log_operation("steward_analysis", {...}) as log_ctx:
        result = await steward_agent.run(
            user_request,
            message_history=convert_to_pydantic_history(conversation_history),
            usage_limits=UsageLimits(request_limit=3)
        )
        
        # Log and benchmark
        log_ctx["recommendation_count"] = len(result.data.recommended_capabilities)
        await benchmark_store.record(...)
        
        return result.data

2.4 Testing

Test scenarios:

• Calculator request → recommends tatlock_core
• Simple greeting → recommends []
• Web search request → recommends tatlock_core
• Request referencing previous turn → identifies context
• Impossible request → returns missing_capabilities

Success Criteria

• ✅ Steward queries household registry successfully
• ✅ Produces structured recommendations
• ✅ Analyzes full conversation context
• ✅ Handles missing capabilities gracefully
• ✅ Conservative recommendations (> 90% accuracy)
• ✅ Benchmarks recorded to Redis

---

Milestone 3: Request Preprocessing & Tool Tracking (Week 5-6)

Goal

Wire Steward into request flow, implement tool scoping, and track tool usage.

Tasks

3.1 Create Preprocessing Pipeline

Location: src/core/preprocessing.py

@dataclass
class EnrichedRequest:
    """Request enriched with Steward's analysis."""
    original_request: str
    steward_note: str  # Formatted note for Tatlock
    scoped_toolset: CombinedToolset  # Only recommended tools
    recommendation: StewardRecommendation
    steward_reasoning_output: str  # For streaming to user

async def preprocess_request(
    user_request: str,
    conversation_history: list[dict]  # FULL conversation
) -> EnrichedRequest:
    """Analyze via Steward and prepare scoped context."""
    # Call Steward with full conversation
    recommendation = await analyze_request(user_request, conversation_history)
    
    # Format note to Tatlock (includes conversation context)
    steward_note = format_steward_note(recommendation)
    
    # Create scoped toolset
    scoped_toolset = household_registry.get_scoped_toolset(
        recommendation.recommended_capabilities
    )
    
    return EnrichedRequest(...)

Note formatting:

• Includes conversation context summary
• Highlights missing capabilities if applicable
• Provides complexity estimate

3.2 Tool Usage Tracking

Location: src/core/tool_tracking.py

class ToolCallTracker:
    """Tracks tool calls for benchmarking."""
    
    def __init__(self, recommended_tools: list[str]):
        self.recommended_tools = set(recommended_tools)
        self.actual_calls: dict[str, list[float]] = {}
    
    async def track_call(self, tool_name: str, duration: float):
        """Record a tool call with timing."""
        # Log if tool wasn't recommended
        if tool_name not in self.recommended_tools:
            logger.warning("tool_call_not_recommended", ...)
        
        # Record benchmark to Redis
        await benchmark_store.record(...)
    
    async def finalize(self):
        """Log unused recommended tools."""
        unused = self.recommended_tools - set(self.actual_calls.keys())
        # Record benchmarks for unused tools

3.3 Integrate with Responses API

Modify src/responses/service.py:

async def generate_response(request: ResponseRequest) -> ResponseOutput:
    # Preprocess via Steward (with full conversation)
    enriched = await preprocess_request(
        user_message,
        conversation_history=request.input[:-1]
    )
    
    # Run Tatlock with scoped tools and tracker
    result = await run_tatlock_with_scoped_tools(
        enriched.original_request,
        enriched.steward_note,
        enriched.scoped_toolset,
        enriched.recommendation.recommended_capabilities,  # For tracking
        message_history,
        usage_tracker
    )
    
    # Build response with Steward reasoning
    return build_response_with_steward_reasoning(...)

3.4 Update Tatlock Agent

Location: src/agents/tatlock.py

async def run_tatlock_with_scoped_tools(
    user_request: str,
    steward_note: str,
    scoped_toolset: CombinedToolset,
    recommended_tools: list[str],
    message_history: list[dict],
    usage: UsageeLimits
):
    # Initialize tracker
    tracker = ToolCallTracker(recommended_tools)
    
    # Prepend Steward's note (invisible to user, visible to Tatlock)
    enriched_prompt = f"{steward_note}\n\n{user_request}"
    
    # Run with ONLY scoped tools
    result = await tatlock_agent.run(
        enriched_prompt,
        message_history=convert_to_pydantic_history(message_history),
        toolsets=[scoped_toolset],  # Tool scoping enforced
        deps=tracker,  # For tracking
        usage=usage
    )
    
    # Finalize tracking
    await tracker.finalize()
    
    return result

3.5 Add Streaming Transparency

Modify src/responses/streaming.py:

• Stream Steward's reasoning first
• Then stream Tatlock's response
• Include conversation context notes
• Format missing capabilities warnings

3.6 Testing

Integration tests:

• Full Steward → Tatlock flow
• Tool scoping enforcement (can't use non-recommended tools)
• Tool usage tracking (recommended vs. actual)
• Conversation context propagation
• Missing capabilities handling

Success Criteria

• ✅ Full request flow working (User → Steward → Tatlock)
• ✅ Steward reasoning visible in output stream
• ✅ Tool scoping enforced (only recommended tools available)
• ✅ Tool usage tracked and logged to Redis
• ✅ Conversation context passed through pipeline
• ✅ Integration tests pass end-to-end

---

Milestone 4: Testing, Benchmarking & Refinement (Week 7)

Goal

Validate the system, optimize performance, refine prompts, and establish monitoring.

Tasks

4.1 Comprehensive Testing

Test categories:

• End-to-end integration tests (full request flow)
• Performance benchmarks (latency targets)
• Prompt refinement (recommendation accuracy)
• Edge cases (errors, timeouts, missing capabilities)
• Conversation context accuracy

4.2 Performance Validation

Targets:

• Steward analysis: < 2 seconds
• Total added latency: < 3 seconds
• Model stays hot in VRAM (no reload delays)
• Tool recommendation accuracy: > 90%

4.3 Benchmark Analysis Tools

Create scripts/benchmark_analysis.py:

# View Steward performance over last 24 hours
python scripts/benchmark_analysis.py --operation steward_analysis --hours 24

# Analyze tool recommendation accuracy
python scripts/benchmark_analysis.py --tool-accuracy --days 7

Metrics to track:

• Average Steward analysis time
• Recommendation count distribution
• Tool accuracy (recommended & used, recommended but unused, not recommended but used)
• Recommendation precision percentage

4.4 Prompt Engineering

Iterate on Steward system prompt:

• Test with diverse request types
• Tune conservativeness (balance false positives/negatives)
• Validate conversation context analysis
• Test missing capability detection

4.5 Documentation

Update documentation:

• README.md: Steward explanation and examples
• AGENTS.md: Household registration pattern
• IMPLEMENTATION_ROADMAP.md: Mark Phase 2 complete
• Add benchmark analysis guide

Success Criteria

• ✅ < 3 seconds added latency for Steward analysis
• ✅ > 90% recommendation accuracy (manual evaluation)
• ✅ All integration tests pass
• ✅ Benchmark tools functional
• ✅ Documentation complete and accurate
• ✅ Ready for Phase 3/4 (expert agents)

---

Architecture Diagram

User Request
    ↓
Orchestrator (FastAPI)
    ↓
Preprocessing Pipeline
    ├─→ Steward Agent
    │   ├─ Receives: FULL conversation history
    │   ├─ Analyzes: Context, references, requirements
    │   ├─ Queries: Household registry (capabilities)
    │   ├─ Outputs: StewardRecommendation
    │   │   ├─ recommended_capabilities: list[str]
    │   │   ├─ conversation_context: ConversationContext
    │   │   ├─ missing_capabilities: str | None
    │   │   └─ reasoning: str
    │   └─ Logs: Performance benchmarks → Redis
    │
    ├─→ Create Scoped Toolset
    │   └─ CombinedToolset from recommended capabilities
    │
    └─→ Format Steward Note
        └─ Includes conversation context for Tatlock
    ↓
Tatlock Agent (with scoped tools)
    ├─ Receives: Enriched request + Steward note
    ├─ Has access to: ONLY recommended tools
    ├─ Tool calls tracked: ToolCallTracker
    └─ Logs: Tool usage benchmarks → Redis
    ↓
Response to User
    ├─ Steward's reasoning (streamed first)
    └─ Tatlock's response (streamed second)

Background:
    └─ Redis: Performance benchmarks, tool usage analysis

---

Design Decisions Summary

1. Logging & Performance Benchmarks

Decision: Full observability with Redis-backed benchmark storage

Rationale:

• Track Steward recommendations vs. Tatlock's actual tool usage
• Measure performance metrics (latency, token usage)
• Cross-session analysis for optimization
• Identify recommendation accuracy over time

2. Steward Fallback Behavior

Decision: Explicit missing capability communication

Rationale:

• No suitable tools → Steward states "missing capabilities" with description
• Can suggest what type of tool would be helpful
• Code errors → standard exception handlers (don't suppress real errors)
• Better UX than silent failures or defaulting to all tools

3. Conversation History for Steward

Decision: Steward sees FULL conversation, not just current turn

Rationale:

• Can identify references to previous topics
• Provides contextual notes to Butler
• "Two sets of eyes" on conversation
• Example: "User mentioned Python debugging in turn 3, relevant details: async code"

4. Registry Pattern

Decision: Separate Household Registry from Model Registry

Rationale:

• Tools belong to household members, not models
• Clean separation of concerns
• Executive summaries for coordination, details for execution

5. Tool Composition

Decision: PydanticAI FunctionToolset + CombinedToolset

Rationale:

• Native PydanticAI pattern
• Clean composition and filtering
• Dynamic scoping per request

6. Tool Scoping

Decision: Compile-time scoping via toolset creation

Rationale:

• Tools not even visible to LLM
• Cleaner than runtime permission checks
• Enforced at PydanticAI level

7. Organization

Decision: Domain-based household directories

Rationale:

• Each household member owns their tools
• Clear bounded contexts
• Example: src/agents/tatlock_core/, src/agents/librarian/ (future)

---

Infrastructure Requirements

Redis Setup

Development (quick start):

# Docker (recommended)
docker run -d -p 6379:6379 --name tatlock-redis redis:7-alpine

# Or local installation
# macOS: brew install redis && brew services start redis
# Linux: sudo apt install redis-server && sudo systemctl start redis

Production (docker-compose.yml):

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes

volumes:
  redis_data:

Dependencies Update

Add to requirements.txt:

redis[hiredis]>=5.0.0,<6.0.0
structlog>=24.1.0,<25.0.0

Configuration

Add to .env:

# Redis Configuration
REDIS_URL=redis://localhost:6379/1

# Logging
LOG_LEVEL=INFO
LOG_FORMAT=json
ENABLE_BENCHMARKS=true

---

Timeline

Week 1-2: Household Registry + Logging Infrastructure

• Household registry with Toolsets
• Structured logging with structlog
• Redis benchmark storage
• Tatlock core reorganization
• Tests: Registry + benchmarking

Week 3-4: Steward Agent with Context Analysis

• Steward agent with conversation context
• ConversationContext in recommendations
• Missing capabilities handling
• Tests: Context analysis, missing capabilities

Week 5-6: Integration + Tool Tracking

• Request preprocessing with full conversation
• Tool usage tracking middleware
• Scoped toolset creation
• Streaming transparency
• Tests: Full flow + tool tracking

Week 7: Testing, Benchmarking & Refinement

• End-to-end integration tests
• Benchmark analysis tools
• Prompt refinement
• Performance validation
• Documentation updates

Total: 4-5 weeks (core implementation complete in 6 weeks, polish in week 7)

---

Success Metrics

Technical

• ✅ Household registry operational with executive summaries
• ✅ Steward produces accurate recommendations (> 90%)
• ✅ Steward analyzes full conversation context
• ✅ Tool scoping enforced (Tatlock can't use non-recommended tools)
• ✅ Model efficiency preserved (no reload delays)
• ✅ Added latency < 3 seconds
• ✅ Performance benchmarks recorded to Redis
• ✅ Tool usage tracking (recommended vs. actual)

Observability

• ✅ Structured logging (JSON format)
• ✅ Benchmark analysis tools available
• ✅ Tool recommendation accuracy measurable
• ✅ Cross-session performance trends visible

Error Handling

• ✅ Missing capabilities explicitly communicated
• ✅ Steward can guide user toward needed resources
• ✅ Code errors properly surfaced (not suppressed)

Architectural

• ✅ PydanticAI patterns followed (Toolsets, decorators, structured outputs)
• ✅ Clean separation: registry vs. agents vs. tools
• ✅ Two-tier abstraction working (summaries vs. details)
• ✅ Future-proof for expert agents (Phase 4)

Testing

• ✅ Maintain 80%+ test coverage
• ✅ Integration tests for full flow
• ✅ Performance benchmarks established

---

Future-Proofing for Phase 4

Expert Agent Pattern (Template)

When adding The Librarian, The Developer, etc., follow this structure:

src/agents/librarian/
├── __init__.py
├── agent.py          # Librarian PydanticAI agent
├── tools.py          # Librarian-specific tools (wiki, research, etc.)
├── toolset.py        # PydanticAI toolset creation
└── capability.py     # Executive summary for registry

Example capability registration:

# capability.py
LIBRARIAN_CAPABILITY = HouseholdCapability(
    name="librarian",
    role="The Librarian",
    category="research",
    description="Research assistance, knowledge management, and information synthesis",
    domains=["research", "knowledge_base", "documentation"],
    cost="medium",
    requires_network=True
)

def register_librarian():
    household_registry.register(
        name="librarian",
        capability=LIBRARIAN_CAPABILITY,
        toolset=librarian_toolset,
        agent=librarian_agent  # Expert agent for delegation
    )

Tatlock delegation pattern (Phase 4):

@tatlock_agent.tool
async def consult_librarian(
    ctx: RunContext[None],
    research_query: str
) -> str:
    """Consult the Librarian for research assistance."""
    from src.agents.librarian.agent import librarian_agent
    
    result = await librarian_agent.run(
        research_query,
        usage=ctx.usage  # Aggregate usage
    )
    return result.data

---

Risk Mitigation

Identified Risks

1. Steward recommendations too broad

   • Mitigation: Conservative prompt engineering, benchmark tracking, iterate based on false positives

2. Added latency unacceptable

   • Mitigation: Stream Steward reasoning for transparency, optimize prompt, use same base model

3. Tool registry becomes unwieldy

   • Mitigation: Good categorization, semantic search (future), regular pruning

4. Model VRAM competition

   • Mitigation: Use same base model for Steward and Tatlock, sequential calls

5. Redis dependency

   • Mitigation: Make benchmarking optional, graceful degradation if Redis unavailable

---

Open Questions - RESOLVED

All major design questions have been resolved. See "Design Decisions Summary" section above.

---

Next Steps

Immediate (Today/This Week)

1. Set up Redis (Docker or local)
2. Create src/core/logging_config.py with structured logging
3. Create src/core/benchmarks.py with Redis storage
4. Add redis and structlog to requirements.txt
5. Create household registry skeleton

Week 1-2

1. Complete household registry with Toolset integration
2. Reorganize Tatlock core tools into domain directory
3. Implement logging infrastructure
4. Write tests for registry + benchmarking

Week 3-4

1. Create Steward agent with conversation context
2. Implement missing capabilities handling
3. Test context analysis accuracy
4. Iterate on system prompt

Week 5-6

1. Build preprocessing pipeline
2. Integrate with Responses API
3. Implement tool tracking
4. Add streaming transparency

Week 7

1. End-to-end testing
2. Benchmark analysis
3. Performance optimization
4. Documentation updates

---

Document Status

Status: Active Planning Document
Created: 2025-12-07
Last Updated: 2025-12-07
Version: 1.0
Next Review: After Milestone 1 completion

---

Reference Documents:

• PHILOSOPHY.md - System vision and architecture
• IMPLEMENTATION_ROADMAP.md - Full project roadmap
• AGENTS.md - Agent development guidelines
• README.md - User documentation