diff --git a/CHANGELOG.md b/CHANGELOG.md
index 290294f..d69ef48 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,55 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
+## [1.6.0] - 2025-12-15
+
+### Added
+
+#### Two-Phase Tatlock Execution
+- **Phase 1: Orchestration** - Executes tool calls and expert delegations, returns structured results
+- **Phase 2: Synthesis** - Synthesizes butler-toned response from gathered results
+- `orchestrate_tool_calls()` method in TatlockAgent for coordination phase
+- `synthesize_from_results()` method in TatlockAgent for synthesis phase
+- Guarantees butler personality in all responses by separating coordination from response generation
+
+#### Automatic Think Slugs
+- **Deterministic butler-perspective messages** during expert delegation (no LLM involved)
+- `ActionType` enum: RETRIEVE, RESEARCH, CREATE, CONTROL, RECORD
+- `HOUSEHOLD_THINK_MESSAGES` mapping with butler-perspective messages for all experts:
+ - Librarian: "Allow me to consult the archives, sir." / "I'm having the Librarian prepare a new entry."
+ - Biographer: "Let me consult the household records." / "I've asked the Biographer to take note, sir."
+ - Housekeeper: "I'm instructing the household staff now, sir." / "Allow me to inquire with the household staff."
+- `_detect_action_type()` function for keyword-based action detection
+- `get_think_message()` helper for retrieving appropriate messages
+- Streaming delegation wrappers: `stream_delegate_to_librarian()`, `stream_delegate_to_biographer()`, `stream_delegate_to_housekeeper()`
+- `STREAMING_DELEGATION_WRAPPERS` mapping in delegation.py
+- `get_streaming_delegation_tools()` method in HouseholdRegistry
+
+#### Steward Query Enrichment
+- **Auto-fill user context** (location, timezone) when not specified in query
+- `_build_enriched_query()` function in steward service
+- Regex word boundary matching for accurate location detection (avoids false positives)
+- `enriched_query` field added to `StewardRecommendation` schema
+- Automatic enrichment for weather queries (location), time queries (timezone), temperature preferences
+
+#### Documentation
+- **ORCHESTRATION_SCENARIOS.md** completely rewritten with:
+ - Mermaid flow diagrams for two-phase execution
+ - 4 new Housekeeper scenarios (light control, device status, parallel delegation)
+ - Biographer memory recording scenario
+ - Complete think slug reference tables
+ - Action type detection tables
+ - Updated architecture mindmap
+- **TESTING_IMPROVEMENTS.md** - LLM testing best practices for future implementation
+
+### Changed
+
+- `create_response_with_steward()` now uses two-phase execution
+- `_direct_delegation()` routes through synthesis phase for consistent butler tone
+- `_execute_single_delegation()` now supports housekeeper
+- Streaming response handler integrated with think slug system
+- All 326 unit tests passing
+
## [1.5.0] - 2025-12-15
### Added
@@ -608,7 +657,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- CORS middleware
- Exception handlers (OpenAI-compatible error format)
-[Unreleased]: https://git.schweitz.net/jpmschweitzer/tatlock/compare/v1.4.0...main
+[Unreleased]: https://git.schweitz.net/jpmschweitzer/tatlock/compare/v1.6.0...main
+[1.6.0]: https://git.schweitz.net/jpmschweitzer/tatlock/compare/v1.5.0...v1.6.0
+[1.5.0]: https://git.schweitz.net/jpmschweitzer/tatlock/compare/v1.4.0...v1.5.0
[1.4.0]: https://git.schweitz.net/jpmschweitzer/tatlock/compare/v1.3.3...v1.4.0
[1.3.3]: https://git.schweitz.net/jpmschweitzer/tatlock/compare/v1.3.2...v1.3.3
[1.3.2]: https://git.schweitz.net/jpmschweitzer/tatlock/compare/v1.3.1...v1.3.2
diff --git a/ORCHESTRATION_SCENARIOS.md b/ORCHESTRATION_SCENARIOS.md
index 78d8bfd..884f6c6 100644
--- a/ORCHESTRATION_SCENARIOS.md
+++ b/ORCHESTRATION_SCENARIOS.md
@@ -1,256 +1,349 @@
# Orchestration Scenarios and Tool Flows
-This document outlines example scenarios of varying complexity to illustrate the desired orchestration patterns between Tatlock (Butler/Coordinator), expert agents (The Librarian, etc.), and the user.
+This document outlines example scenarios of varying complexity to illustrate the desired orchestration patterns between Tatlock (Butler/Coordinator), expert agents (The Librarian, Housekeeper, Biographer), and the user.
## Architecture Overview
+### Two-Phase Execution Model
+
+Tatlock operates in two distinct phases to maintain consistent butler personality:
+
+```mermaid
+flowchart TB
+ subgraph Phase1["Phase 1: Coordination"]
+ direction TB
+ U[User Request] --> S[Steward]
+ S -->|"Analyzes & routes"| T1[Tatlock Orchestration]
+ T1 -->|"delegate_to_*"| E1[Expert Agents]
+ T1 -->|"tool calls"| TC[tatlock_core Tools]
+ E1 -->|"structured results"| R1[Results Collection]
+ TC -->|"tool outputs"| R1
+ end
+
+ subgraph Phase2["Phase 2: Synthesis"]
+ direction TB
+ R1 --> T2[Tatlock Synthesis]
+ T2 -->|"butler-toned response"| U2[User Response]
+ end
+
+ style Phase1 fill:#1a1a2e,stroke:#4a4a6a
+ style Phase2 fill:#16213e,stroke:#4a4a6a
```
-User Request
- ↓
-[Steward] → Analyzes request, has visibility into ALL capabilities
- → Makes routing decision: which experts needed
- → Passes simplified instruction to Tatlock (not raw tool schemas)
- ↓
-[Tatlock/Butler] → Coordinator, receives "use Librarian for wiki creation"
- → Calls expert agents as tools
- → Synthesizes responses into butler-voice answer
- ↓
-[Expert Agents] → The Librarian, Home Automation, Memory, etc.
- → Each has their own specialized tools
- → Return structured results to Tatlock
- ↓
-[External APIs] → library-desk, home-assistant, user-db, etc.
+
+### Component Flow
+
+```mermaid
+flowchart LR
+ subgraph Steward
+ S1[Request Analysis]
+ S2[Query Enrichment]
+ S3[Capability Routing]
+ end
+
+ subgraph Tatlock
+ T1[Phase 1: Orchestrate]
+ T2[Phase 2: Synthesize]
+ end
+
+ subgraph Experts
+ L[Librarian]
+ H[Housekeeper]
+ B[Biographer]
+ end
+
+ subgraph External
+ LD[library-desk]
+ HA[Home Assistant]
+ DB[user-db]
+ end
+
+ S1 --> S2 --> S3 --> T1
+ T1 --> L & H & B
+ L --> LD
+ H --> HA
+ B --> DB
+ L & H & B --> T2
+ T2 --> Response
```
**Key Principles**:
1. **Steward sees everything** - Has access to all capability descriptions to make informed routing decisions
-2. **Simplified passthrough** - Tatlock receives "delegate to Librarian for research" not 16 tool schemas
-3. **Expert agents are tools** - Tatlock calls `librarian_agent(task)`, not `hybrid_search()` directly
-4. **Each expert owns their tools** - Librarian has wiki tools, Home Automation has device tools
-5. **Results flow up** - Tatlock synthesizes all expert responses into coherent butler answer
+2. **Query enrichment** - Steward auto-fills user context (location, timezone) when not specified
+3. **Two-phase execution** - Coordination (tool calls) separated from synthesis (butler response)
+4. **Expert agents are tools** - Tatlock calls `delegate_to_librarian(task)`, not `hybrid_search()` directly
+5. **Each expert owns their tools** - Librarian has wiki tools, Housekeeper has device tools, Biographer has memory tools
+6. **Think slugs for transparency** - Deterministic butler-perspective messages during expert delegation
+7. **Results flow up** - Phase 2 synthesizes all expert responses into coherent butler answer
---
-## Scenario 1: Weather Check (Multi-Step with Memory Lookup)
+## Scenario 1: Weather Check (With Query Enrichment)
**User**: "What's the weather like?"
### Complexity Analysis
-This seemingly simple request requires:
-1. **Location determination** - Where does the user want weather for?
-2. **Memory/database lookup** - Retrieve user's home location or current location
-3. **Weather data fetch** - Search for weather at determined location
+This seemingly simple request is now streamlined by query enrichment:
+1. **Steward enrichment** - Auto-fills location from user profile
+2. **Weather data fetch** - Search for weather at enriched location
+
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant TC as tatlock_core
+
+ U->>S: "What's the weather like?"
+ S->>S: Prefetch memory context
+ Note over S: location=Amsterdam
timezone=Europe/Amsterdam
+ S->>T: Enriched query + routing
+ T->>TC: search_web("weather Amsterdam")
+ TC-->>T: "12°C, light rain"
+ T->>U: Butler-toned response
+```
### Flow
```
1. Steward Analysis
- → Capabilities needed: memory (user context), tatlock_core (web search)
- → Complexity: moderate
- → Note: Location must be determined before weather lookup
+ → Pre-fetch: memory_context = {profile: {location: "Amsterdam"}}
+ → Enriched query: "What's the weather like?\n\n[User Context: location=Amsterdam]"
+ → Capabilities needed: tatlock_core (web search)
+ → Complexity: simple
-2. Tatlock Execution - Step 1
- User asked about weather but didn't specify location.
- Checking user profile for home location...
- → Calls: memory_agent(task: "get user home location")
- → Memory queries user database
- → Returns: "User home location: Amsterdam, Netherlands"
-
-3. Tatlock Execution - Step 2
- User is based in Amsterdam. Fetching current weather...
+2. Phase 1: Tatlock Orchestration
→ Calls: search_web("current weather Amsterdam Netherlands")
→ Receives: "Amsterdam: 12°C, light rain, humidity 78%"
+ → Results: {tool_outputs: {search_web: "12°C, light rain..."}}
+
+3. Phase 2: Tatlock Synthesis
+ → Input: user message, tool results, enriched context
+ → Output: Butler-toned response
4. Response
"Currently 12°C with light rain in Amsterdam, sir. You might want
to grab an umbrella if you're heading out."
```
-### Intra-System Prompts
-
-**Steward → Tatlock Note**:
-```
-Weather query - location not specified.
-1. First: Query memory for user's location (home or current)
-2. Then: Search weather for that location
-Capabilities: memory, tatlock_core
-Complexity: moderate
-```
-
-**Tatlock → Memory Agent**:
-```
-Task: Retrieve user's location for weather query.
-Context: User asked about weather without specifying location.
-Action required: Return user's home location or current known location.
-
-Reference (user's original request): "What's the weather like?"
-```
-
-**Memory Agent → Tatlock Response**:
-```
-User location retrieved:
-- Home location: Amsterdam, Netherlands
-- Last known location: Amsterdam (home)
-- Location confidence: high
-- Source: user profile settings
-```
-
-### Alternative Flow: Location Ambiguity
-
-If user has multiple locations or is traveling:
+### Steward Note Format
```
-Memory Agent → Tatlock Response:
- User has multiple locations:
- - Home: Amsterdam, Netherlands
- - Office: Rotterdam, Netherlands
- - Currently traveling: Unknown
-
- Recommendation: Ask user to clarify or use home location as default.
+📋 Steward's Analysis
+========================================
+Complexity: SIMPLE
+Recommended tools: tatlock_core
+----------------------------------------
+User Context:
+ • location: Amsterdam
+ • timezone: Europe/Amsterdam
+ • preferences: temperature_unit=celsius
+========================================
```
-Tatlock could then either:
-- Ask user: "Shall I check the weather in Amsterdam, sir, or elsewhere?"
-- Default to home: Use Amsterdam and mention the assumption
+### Alternative Flow: Location Specified
+
+If user specifies location, no enrichment occurs:
+
+**User**: "What's the weather in London?"
+
+```
+Steward Analysis:
+ → Location specified in query ("in London")
+ → No enrichment needed
+ → Direct routing to tatlock_core
+```
---
-## Scenario 2: Adjust Temperature Based on Weather (Conditional Multi-Expert)
+## Scenario 2: Adjust Temperature Based on Weather (Conditional with Housekeeper)
**User**: "Check the weather and if it's cold, turn up the heating"
### Complexity Analysis
This requires:
-1. **Location lookup** - Where to check weather (implicit: user's home)
+1. **Query enrichment** - Location auto-filled from profile
2. **Weather fetch** - Get current outdoor temperature
3. **Conditional evaluation** - Is it "cold"? (requires threshold judgment)
-4. **Home automation** - Adjust heating if condition met
+4. **Housekeeper delegation** - Adjust heating if condition met
-### Flow
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant TC as tatlock_core
+ participant H as Housekeeper
+ participant HA as Home Assistant
+
+ U->>S: "Check weather, if cold turn up heating"
+ S->>S: Enrich with location
+ S->>T: Route to tatlock_core + housekeeper
+
+ rect rgb(40, 40, 60)
+ Note over T: Phase 1: Orchestration
+ T->>TC: search_web("weather Amsterdam")
+ TC-->>T: "8°C, cloudy"
+ T->>T: Evaluate: 8°C < 15°C = cold
+ T-->>U: I'm instructing the household staff now, sir.
+ T->>H: delegate_to_housekeeper(task)
+ H->>HA: climate.set_temperature(21)
+ HA-->>H: Success
+ H-->>T: "Thermostat set to 21°C"
+ T-->>U: The household has been configured as requested.
+ end
+
+ rect rgb(30, 50, 70)
+ Note over T: Phase 2: Synthesis
+ T->>U: Butler-toned response
+ end
+```
+
+### Flow with Think Slugs
```
1. Steward Analysis
- → Capabilities needed: memory, tatlock_core, home_automation
- → Complexity: moderate
- → Note: Conditional logic - heating only if cold
- → Sequence: location → weather → evaluate → (maybe) heating
+ → Enriched query: [User Context: location=Amsterdam]
+ → Capabilities needed: tatlock_core, housekeeper
+ → Complexity: moderate (conditional)
-2. Tatlock Execution - Step 1
- Need to check weather at user's location first...
- → Calls: memory_agent(task: "get user home location")
- → Returns: "Amsterdam, Netherlands"
+2. Phase 1: Tatlock Orchestration
+ → Calls: search_web("current weather Amsterdam")
+ → Receives: "8°C, cloudy, wind 15km/h"
+ → Evaluates: 8°C is cold (< 15°C threshold)
-3. Tatlock Execution - Step 2
- Fetching weather for Amsterdam...
- → Calls: search_web("current weather Amsterdam Netherlands")
- → Receives: "Current temperature: 8°C, cloudy, wind 15km/h"
+ → Emits think slug (streamed to user):
+ I'm instructing the household staff now, sir.
-4. Tatlock Evaluation
- Temperature is 8°C - that's cold by most standards.
- User requested heating adjustment if cold. Will proceed...
+ → Calls: delegate_to_housekeeper(
+ task="Set thermostat to 21°C - cold weather detected (8°C outside)"
+ )
+ → Housekeeper executes: climate.set_temperature(entity_id, 21)
+ → Returns: {success: true, output: "Thermostat set to 21°C"}
-5. Tatlock Execution - Step 3
- Delegating heating adjustment to Home Automation...
- → Calls: home_automation_agent(task)
- → Home Automation executes: set_thermostat(temperature=21)
- → Receives: "Thermostat set to 21°C"
+ → Emits think slug (streamed to user):
+ The household has been configured as requested.
-6. Response
+ → Results: {
+ tool_outputs: {search_web: "8°C..."},
+ expert_results: {housekeeper: "Thermostat set to 21°C"}
+ }
+
+3. Phase 2: Tatlock Synthesis
+ → Input: user message, weather data, housekeeper result
+ → Output: Butler-toned response
+
+4. Response
"It's rather brisk outside at 8°C, sir. I've taken the liberty of raising
the heating to a comfortable 21°C. The house should warm up shortly."
```
-### Intra-System Prompts
+### Think Slug Mapping (Housekeeper)
-**Steward → Tatlock Note**:
-```
-Conditional weather-to-heating request.
-1. Get user location from memory
-2. Check weather at location
-3. IF cold (suggest: below 15°C): delegate to home_automation to increase heating
-4. IF not cold: inform user, no action needed
-Capabilities: memory, tatlock_core, home_automation
-Complexity: moderate (conditional)
-```
-
-**Tatlock → Home Automation Agent**:
-```
-Task: Increase heating due to cold weather conditions.
-Context: Current outdoor temperature is 8°C in Amsterdam. User requested
-heating adjustment if cold - condition is met.
-Action required: Set thermostat to comfortable temperature (recommend 21°C).
-
-Reference (user's original request): "Check the weather and if it's cold, turn up the heating"
-```
-
-**Home Automation → Tatlock Response**:
-```
-Action completed: Thermostat adjusted
-- Previous setting: 18°C
-- New setting: 21°C
-- Zone: All zones (whole house)
-- Estimated time to reach target: 15 minutes
-- Note: Eco mode disabled temporarily for faster heating
-```
+| Action Type | Phase | Message |
+|-------------|-------|---------|
+| CONTROL | start | `I'm instructing the household staff now, sir.` |
+| CONTROL | success | `The household has been configured as requested.` |
+| CONTROL | error | `I'm afraid the staff reports an issue with that request.` |
+| RETRIEVE | start | `Allow me to inquire with the household staff.` |
+| RETRIEVE | success | `The staff reports the current status, sir.` |
### Alternative Flow: Not Cold
```
Weather returns: "Current temperature: 18°C, partly sunny"
-Tatlock Evaluation:
-18°C isn't cold - no heating adjustment needed.
-Will inform user that conditions don't warrant heating...
+Phase 1 Evaluation:
+ → 18°C is not cold (>= 15°C)
+ → No housekeeper delegation needed
+ → Results: {tool_outputs: {search_web: "18°C..."}, expert_results: {}}
-Response:
-"It's a pleasant 18°C outside, sir - not particularly cold.
-I don't believe the heating requires adjustment, but do let
-me know if you'd like it warmer regardless."
+Phase 2 Synthesis:
+ "It's a pleasant 18°C outside, sir - not particularly cold.
+ I don't believe the heating requires adjustment, but do let
+ me know if you'd like it warmer regardless."
```
---
-## Scenario 3: Adjust Lights Based on Time (Conditional Logic)
+## Scenario 3: Adjust Lights Based on Time (Housekeeper Scene Control)
**User**: "Set the lights appropriately for the time of day"
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant TC as tatlock_core
+ participant H as Housekeeper
+ participant HA as Home Assistant
+
+ U->>S: "Set lights for time of day"
+ S->>S: Enrich with timezone
+ S->>T: Route to tatlock_core + housekeeper
+
+ rect rgb(40, 40, 60)
+ Note over T: Phase 1: Orchestration
+ T->>TC: get_current_datetime()
+ TC-->>T: "2024-12-11 21:45 Europe/Amsterdam"
+ T->>T: Evaluate: 21:45 = evening
+ T-->>U: I'm instructing the household staff now, sir.
+ T->>H: delegate_to_housekeeper(task)
+ H->>HA: scene.turn_on("evening_relaxation")
+ HA-->>H: Scene activated
+ H-->>T: "Scene activated: warm white 40%"
+ T-->>U: The household has been configured as requested.
+ end
+
+ rect rgb(30, 50, 70)
+ Note over T: Phase 2: Synthesis
+ T->>U: Butler-toned response
+ end
+```
+
### Flow
```
1. Steward Analysis
- → Capabilities needed: tatlock_core (datetime), home_automation
+ → Enriched query: [User Context: timezone=Europe/Amsterdam]
+ → Capabilities needed: tatlock_core, housekeeper
→ Complexity: moderate
-2. Tatlock Execution - Step 1
- Checking current time to determine appropriate lighting...
+2. Phase 1: Tatlock Orchestration
→ Calls: get_current_datetime()
- → Receives: "2024-12-11 21:45"
+ → Receives: "2024-12-11 21:45 Europe/Amsterdam"
+ → Evaluates: 21:45 = evening, warm dim lighting appropriate
-3. Tatlock Evaluation
- It's evening (21:45). Will set warm, dimmed lighting for wind-down time...
+ → Emits: I'm instructing the household staff now, sir.
-4. Tatlock Execution - Step 2
- → Calls: home_automation_agent(task)
- → Home Automation executes: set_scene("evening_relaxation")
- → Receives: "Scene 'evening_relaxation' activated - warm white 40%"
+ → Calls: delegate_to_housekeeper(
+ task="Activate evening lighting scene - it's 21:45"
+ )
+ → Housekeeper executes: scene.turn_on("evening_relaxation")
+ → Returns: "Scene 'evening_relaxation' activated - warm white 40%"
-5. Response
+ → Emits: The household has been configured as requested.
+
+3. Phase 2: Tatlock Synthesis
+ → Input: time context, scene activation result
+ → Output: Butler-toned response with proactive offer
+
+4. Response
"It's quarter to ten in the evening, sir. I've set the lights to a warm,
subdued glow - ideal for winding down. Shall I also draw the curtains?"
```
-### Intra-System Prompts
+### Housekeeper Task Format
-**Tatlock → Home Automation Agent**:
```
-Task: Set lighting appropriate for current time of day.
-Context: Current time is 21:45 (evening). User wants lights adjusted automatically.
-Action required: Activate appropriate lighting scene for evening/night.
+Task: Activate evening lighting scene - it's 21:45
+Context: User wants lights appropriate for time of day.
+Action required: Activate scene suitable for late evening/wind-down.
-Reference (user's original request): "Set the lights appropriately for the time of day"
+Reference: "Set the lights appropriately for the time of day"
```
---
@@ -332,59 +425,96 @@ Reference (user's original request): "Schedule the lights to turn on at 09:00 ev
---
-## Scenario 6: Create Wiki Page About Topic (Expert with Research)
+## Scenario 6: Create Wiki Page About Topic (Librarian with Research)
**User**: "Create a wiki page about CI/CD"
-### Flow
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant L as Librarian
+ participant LD as library-desk
+
+ U->>S: "Create a wiki page about CI/CD"
+ S->>T: Route to librarian (CREATE action)
+
+ rect rgb(40, 40, 60)
+ Note over T: Phase 1: Orchestration
+ T-->>U: I'm having the Librarian prepare a new entry.
+ T->>L: delegate_to_librarian(task)
+
+ rect rgb(50, 50, 80)
+ Note over L: Librarian Internal
+ L->>LD: hybrid_search("CI/CD")
+ LD-->>L: 12 sources found
+ L->>LD: create_wiki_page(...)
+ LD-->>L: Page created
+ end
+
+ L-->>T: Structured result
+ T-->>U: The new material has been properly catalogued, sir.
+ end
+
+ rect rgb(30, 50, 70)
+ Note over T: Phase 2: Synthesis
+ T->>U: Butler-toned response
+ end
+```
+
+### Flow with Think Slugs
```
1. Steward Analysis
→ Capabilities needed: librarian
→ Complexity: complex
- → Specific action: smart_create (research-backed page creation)
+ → Action detected: CREATE (create, write, add keywords)
-2. Tatlock Execution
- Delegating to The Librarian for research and wiki page creation...
- → Calls: librarian_agent(task)
+2. Phase 1: Tatlock Orchestration
+ → Emits: I'm having the Librarian prepare a new entry.
-3. Librarian Internal Flow
- → hybrid_search("CI/CD continuous integration deployment")
- → Synthesizes content from wiki, graph, and web sources
- → create_wiki_page(title, path, content, tags)
- → Returns summary to Tatlock
+ → Calls: delegate_to_librarian(
+ task="Create a comprehensive wiki page about CI/CD"
+ )
-4. Tatlock Receives
- "Page created: CI/CD
- Path: /technology/cicd
- Sources used: 12 (3 wiki, 4 graph, 5 web)
- Forward links: 8, Backward links: 3"
+ → Librarian Internal Flow:
+ - hybrid_search("CI/CD continuous integration deployment")
+ - Synthesizes content from wiki, graph, and web sources
+ - create_wiki_page(title, path, content, tags)
-5. Response
+ → Returns: {
+ success: true,
+ output: "Page created: CI/CD, Path: /technology/cicd, Sources: 12"
+ }
+
+ → Emits: The new material has been properly catalogued, sir.
+
+3. Phase 2: Tatlock Synthesis
+ → Input: user request, librarian result with page details
+ → Output: Butler-toned response with specifics
+
+4. Response
"The wiki page on CI/CD has been created, sir. The Librarian has compiled
information from 12 sources, including our existing documentation on DevOps
practices. The page is now available at /technology/cicd and has been
cross-linked with related topics like Jenkins, GitHub Actions, and Docker."
```
-### Intra-System Prompts
+### Think Slug Mapping (Librarian)
-**Tatlock → Librarian Agent**:
-```
-Task: Create a comprehensive wiki page about CI/CD.
-Context: User wants a new wiki page created. Use research capabilities to
-gather information and create a well-sourced page.
+| Action Type | Phase | Message |
+|-------------|-------|---------|
+| CREATE | start | `I'm having the Librarian prepare a new entry.` |
+| CREATE | success | `The new material has been properly catalogued, sir.` |
+| CREATE | error | `I'm afraid there was difficulty filing the entry.` |
+| RETRIEVE | start | `Allow me to consult the archives, sir.` |
+| RETRIEVE | success | `The Librarian has compiled the relevant findings.` |
+| RESEARCH | start | `I've dispatched the Librarian to conduct some fresh research.` |
+| RESEARCH | success | `The Librarian has returned with findings, sir.` |
-Action required:
-1. Research the topic using hybrid search (wiki, knowledge graph, web)
-2. Create wiki page with synthesized content
-3. Apply appropriate tags (suggest: technology, devops, automation)
-4. Report back with page details and research summary
+### Librarian Response Format
-Reference (user's original request): "Create a wiki page about CI/CD"
-```
-
-**Librarian → Tatlock Response**:
```
Task completed: Wiki page created
@@ -402,13 +532,6 @@ Research Summary:
Knowledge Graph Updates:
- Forward links created: 8 (to existing related pages)
- Backward links created: 3 (updated existing pages to reference new page)
-
-Content includes sections on:
-- Definition and core concepts
-- CI vs CD distinction
-- Popular tools comparison
-- Best practices
-- Integration with our documented infrastructure
```
---
@@ -601,27 +724,316 @@ Reference: Part of user's request to track and highlight near-term deadlines.
---
+## Scenario 9: Simple Light Control (Direct Housekeeper Delegation)
+
+**User**: "Turn on the living room lights"
+
+This is a simple direct delegation scenario - no tool calls needed, just Housekeeper.
+
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant H as Housekeeper
+ participant HA as Home Assistant
+
+ U->>S: "Turn on the living room lights"
+ S->>S: Detect: housekeeper only, CONTROL action
+ S->>T: Direct delegation recommended
+
+ T-->>U: I'm instructing the household staff now, sir.
+ T->>H: delegate_to_housekeeper(task)
+ H->>HA: light.turn_on("light.living_room")
+ HA-->>H: Success
+ H-->>T: "Living room lights turned on"
+ T-->>U: The household has been configured as requested.
+
+ Note over T: Phase 2: Synthesis
+ T->>U: "Very good, sir. The living room lights are now on."
+```
+
+### Flow
+
+```
+1. Steward Analysis
+ → Capabilities needed: housekeeper
+ → Action detected: CONTROL (turn on/off, set, toggle)
+ → Complexity: simple
+ → Direct delegation: Yes (single expert, no tools needed)
+
+2. Direct Delegation (bypasses full orchestration)
+ → Emits: I'm instructing the household staff now, sir.
+ → Calls: delegate_to_housekeeper(task="Turn on the living room lights")
+ → Housekeeper executes: light.turn_on("light.living_room")
+ → Returns: {success: true, output: "Living room lights turned on"}
+ → Emits: The household has been configured as requested.
+
+3. Phase 2: Synthesis (still runs for butler tone)
+ → Input: simple action result
+ → Output: Concise butler acknowledgment
+
+4. Response
+ "Very good, sir. The living room lights are now on."
+```
+
+---
+
+## Scenario 10: Device Status Query (Housekeeper RETRIEVE)
+
+**User**: "What devices are on in the bedroom?"
+
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant H as Housekeeper
+ participant HA as Home Assistant
+
+ U->>S: "What devices are on in the bedroom?"
+ S->>T: Route to housekeeper (RETRIEVE action)
+
+ T-->>U: Allow me to inquire with the household staff.
+ T->>H: delegate_to_housekeeper(task)
+ H->>HA: get_area_devices("bedroom", state="on")
+ HA-->>H: Device list
+ H-->>T: Structured result
+ T-->>U: The staff reports the current status, sir.
+
+ T->>U: Butler-toned response
+```
+
+### Flow
+
+```
+1. Steward Analysis
+ → Capabilities needed: housekeeper
+ → Action detected: RETRIEVE (what, which, status, is)
+ → Complexity: simple
+
+2. Phase 1: Tatlock Orchestration
+ → Emits: Allow me to inquire with the household staff.
+ → Calls: delegate_to_housekeeper(task="List active devices in the bedroom")
+ → Returns: {
+ success: true,
+ output: "Bedroom: Ceiling light (on, 75%), Bedside lamp (on, 30%)"
+ }
+ → Emits: The staff reports the current status, sir.
+
+3. Phase 2: Tatlock Synthesis
+ → Output: Butler-toned summary
+
+4. Response
+ "In the bedroom, sir, we have the ceiling light at 75% brightness
+ and the bedside lamp at a gentle 30%. Would you like me to adjust
+ either of them?"
+```
+
+---
+
+## Scenario 11: Biographer Memory Recording
+
+**User**: "Remember that I prefer my coffee black"
+
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant B as Biographer
+ participant DB as user-db
+
+ U->>S: "Remember that I prefer my coffee black"
+ S->>T: Route to biographer (RECORD action)
+
+ T-->>U: I've asked the Biographer to take note of this, sir.
+ T->>B: delegate_to_biographer(task)
+ B->>DB: store_preference(category="food", key="coffee", value="black")
+ DB-->>B: Stored
+ B-->>T: "Preference recorded"
+ T-->>U: The household records have been updated accordingly.
+
+ T->>U: Butler-toned confirmation
+```
+
+### Flow
+
+```
+1. Steward Analysis
+ → Capabilities needed: biographer
+ → Action detected: RECORD (remember, note, save, record)
+ → Complexity: simple
+
+2. Phase 1: Tatlock Orchestration
+ → Emits: I've asked the Biographer to take note of this, sir.
+ → Calls: delegate_to_biographer(task="Record preference: coffee black")
+ → Returns: {success: true, output: "Preference recorded: coffee=black"}
+ → Emits: The household records have been updated accordingly.
+
+3. Phase 2: Tatlock Synthesis
+
+4. Response
+ "Noted, sir. I'll remember you prefer your coffee black."
+```
+
+### Think Slug Mapping (Biographer)
+
+| Action Type | Phase | Message |
+|-------------|-------|---------|
+| RECORD | start | `I've asked the Biographer to take note of this, sir.` |
+| RECORD | success | `The household records have been updated accordingly.` |
+| RECORD | error | `I'm afraid there was difficulty recording the entry.` |
+| RETRIEVE | start | `Let me consult the household records.` |
+| RETRIEVE | success | `The Biographer has located the relevant information, sir.` |
+
+---
+
+## Scenario 12: Multi-Expert Morning Routine (Parallel Delegation)
+
+**User**: "Good morning! What's the weather, and set the house for daytime"
+
+This scenario demonstrates parallel expert delegation with multiple think slugs.
+
+```mermaid
+sequenceDiagram
+ participant U as User
+ participant S as Steward
+ participant T as Tatlock
+ participant TC as tatlock_core
+ participant H as Housekeeper
+ participant HA as Home Assistant
+
+ U->>S: "Good morning! Weather + daytime settings"
+ S->>S: Enrich with location, timezone
+ S->>T: Route to tatlock_core + housekeeper
+
+ rect rgb(40, 40, 60)
+ Note over T: Phase 1: Parallel Orchestration
+ par Weather Check
+ T->>TC: search_web("weather Amsterdam")
+ TC-->>T: "14°C, sunny"
+ and Home Setup
+ T-->>U: I'm instructing the household staff now, sir.
+ T->>H: delegate_to_housekeeper(task)
+ H->>HA: scene.turn_on("daytime")
+ HA-->>H: Scene activated
+ H-->>T: "Daytime scene activated"
+ T-->>U: The household has been configured as requested.
+ end
+ end
+
+ rect rgb(30, 50, 70)
+ Note over T: Phase 2: Synthesis
+ T->>U: Combined butler greeting
+ end
+```
+
+### Flow
+
+```
+1. Steward Analysis
+ → Enriched query: [User Context: location=Amsterdam, timezone=Europe/Amsterdam]
+ → Capabilities needed: tatlock_core, housekeeper
+ → Complexity: moderate
+ → Parallel execution possible: weather + home setup independent
+
+2. Phase 1: Tatlock Orchestration (parallel)
+ → Calls in parallel:
+ - search_web("current weather Amsterdam")
+ - delegate_to_housekeeper(task="Set house for daytime")
+
+ → Think slugs emitted as housekeeper delegation starts:
+ I'm instructing the household staff now, sir.
+
+ → Results collected:
+ {
+ tool_outputs: {search_web: "14°C, sunny, light breeze"},
+ expert_results: {housekeeper: "Daytime scene activated: blinds open, lights off"}
+ }
+
+ → Think slug on completion:
+ The household has been configured as requested.
+
+3. Phase 2: Tatlock Synthesis
+ → Input: weather data, home status, morning context
+ → Output: Warm morning greeting with combined info
+
+4. Response
+ "Good morning, sir. It's a lovely day outside - 14°C and sunny with a
+ light breeze. I've opened the blinds and set the house for daytime.
+ Is there anything else you'd like to start your day?"
+```
+
+---
+
## Response Pattern Guidelines
-### Tatlock's Think Updates (Streaming to User)
+### Think Slugs (Deterministic, No LLM)
-During multi-step operations, Tatlock should emit `` updates to keep the user informed:
+Think slugs are **deterministic messages** emitted during expert delegation. They are:
+- Generated from `HOUSEHOLD_THINK_MESSAGES` mapping (no LLM involved)
+- Based on expert name + detected action type
+- Written from butler's perspective, addressed to user
+- Streamed in real-time as experts are called
-```
-Analyzing your request...
-Searching for deadline information in the wiki...
-Found the deadline - December 15th. Creating reminder...
-Updating the project page with status note...
-All tasks complete. Composing response...
+```mermaid
+flowchart LR
+ subgraph Detection
+ T[Task Description] --> D[_detect_action_type]
+ D --> AT[ActionType]
+ end
+
+ subgraph Lookup
+ AT --> M[HOUSEHOLD_THINK_MESSAGES]
+ E[Expert Name] --> M
+ M --> MSG[Think Message]
+ end
+
+ subgraph Stream
+ MSG --> S["..."]
+ S --> U[User]
+ end
```
-### Tatlock's Final Response Pattern
+### Action Type Detection
+| Expert | Keywords → Action Type |
+|--------|----------------------|
+| Librarian | search, find, look up → RETRIEVE |
+| Librarian | web, online, research → RESEARCH |
+| Librarian | create, write, add → CREATE |
+| Biographer | remember, note, save → RECORD |
+| Biographer | what, recall, who → RETRIEVE |
+| Housekeeper | turn, set, activate, toggle → CONTROL |
+| Housekeeper | what, which, status, is → RETRIEVE |
+
+### Complete Think Slug Reference
+
+| Expert | Action | Start | Success | Error |
+|--------|--------|-------|---------|-------|
+| **Librarian** | RETRIEVE | Allow me to consult the archives, sir. | The Librarian has compiled the relevant findings. | I'm afraid the archives proved difficult to access. |
+| **Librarian** | RESEARCH | I've dispatched the Librarian to conduct some fresh research. | The Librarian has returned with findings, sir. | The research proved inconclusive, I'm afraid. |
+| **Librarian** | CREATE | I'm having the Librarian prepare a new entry. | The new material has been properly catalogued, sir. | I'm afraid there was difficulty filing the entry. |
+| **Biographer** | RETRIEVE | Let me consult the household records. | The Biographer has located the relevant information, sir. | I'm unable to locate those particular records. |
+| **Biographer** | RECORD | I've asked the Biographer to take note of this, sir. | The household records have been updated accordingly. | I'm afraid there was difficulty recording the entry. |
+| **Housekeeper** | RETRIEVE | Allow me to inquire with the household staff. | The staff reports the current status, sir. | The household staff is momentarily unavailable, I'm afraid. |
+| **Housekeeper** | CONTROL | I'm instructing the household staff now, sir. | The household has been configured as requested. | I'm afraid the staff reports an issue with that request. |
+
+### Tatlock's Final Response Pattern (Phase 2 Synthesis)
+
+Phase 2 synthesis ensures butler tone by receiving:
+- Original user message
+- All expert results
+- All tool outputs
+- Conversation history
+
+Response structure:
1. **Acknowledge** - Confirm understanding of the request
2. **Summarize actions** - What was done, by whom (implicitly)
3. **Key details** - Important information the user should know
4. **Proactive offer** - Suggest related actions or follow-ups
-5. **Butler voice** - Formal but warm, with personality
+5. **Butler voice** - Formal but warm, with personality ("sir", formal language)
### Expert Agent Response Pattern
@@ -670,10 +1082,59 @@ set up the calendar connection?"
## Summary: Key Design Principles
-1. **Tatlock is the orchestrator** - Never exposes raw tool complexity to users
-2. **Expert agents are tools** - Tatlock calls them, they return structured responses
-3. **Context flows down** - Each expert gets only what they need to complete their task
-4. **Results flow up** - Tatlock synthesizes all responses into coherent butler-voice answer
-5. **Think updates maintain engagement** - User sees progress during complex operations
-6. **Errors are handled gracefully** - Tatlock explains and offers alternatives
-7. **Proactive suggestions** - Tatlock anticipates follow-up needs
+```mermaid
+mindmap
+ root((Tatlock))
+ Orchestration
+ Two-phase execution
+ Phase 1: Tool calls
+ Phase 2: Synthesis
+ Experts
+ Librarian
+ Archives
+ Research
+ Wiki creation
+ Housekeeper
+ Device control
+ Scenes
+ Status queries
+ Biographer
+ Memory
+ Preferences
+ User context
+ UX
+ Think slugs
+ Butler tone
+ Proactive offers
+ Context
+ Query enrichment
+ Steward routing
+ History awareness
+```
+
+### Core Principles
+
+1. **Two-phase execution** - Coordination (tool calls) separated from synthesis (butler response)
+2. **Expert agents as tools** - Tatlock calls `delegate_to_*()`, they return structured responses
+3. **Deterministic think slugs** - Butler-perspective messages during delegation (no LLM)
+4. **Query enrichment** - Steward auto-fills user context (location, timezone) when not specified
+5. **Results flow up** - Phase 2 synthesizes all expert responses into coherent butler answer
+6. **Butler tone guaranteed** - Phase 2 always produces formal, warm butler voice
+7. **Errors handled gracefully** - Butler explains issues and offers alternatives
+8. **Proactive suggestions** - Tatlock anticipates follow-up needs
+
+### Expert Responsibilities
+
+| Expert | Domains | External Service |
+|--------|---------|------------------|
+| **Librarian** | Wiki, knowledge graph, web research | library-desk |
+| **Housekeeper** | Lights, climate, scenes, device status | Home Assistant |
+| **Biographer** | User memory, preferences, profile | user-db |
+
+### Action Types by Expert
+
+| Expert | RETRIEVE | RESEARCH | CREATE | CONTROL | RECORD |
+|--------|:--------:|:--------:|:------:|:-------:|:------:|
+| Librarian | ✓ | ✓ | ✓ | | |
+| Housekeeper | ✓ | | | ✓ | |
+| Biographer | ✓ | | | | ✓ |
diff --git a/TESTING_IMPROVEMENTS.md b/TESTING_IMPROVEMENTS.md
new file mode 100644
index 0000000..ad09492
--- /dev/null
+++ b/TESTING_IMPROVEMENTS.md
@@ -0,0 +1,105 @@
+# Testing Improvements for LLM Outputs
+
+## Problem
+
+LLM outputs are non-deterministic. Tests checking for exact string matches fail when the LLM writes "thirty-seven" instead of "37".
+
+## Proposed Solutions
+
+### 1. LLM-as-Judge Pattern
+
+Use a smaller/faster model to evaluate semantic correctness:
+
+```python
+async def llm_judge(output: str, criteria: str) -> bool:
+ """Use LLM to evaluate if output meets criteria."""
+ prompt = f"""
+ Evaluate if this output is correct:
+ Output: {output}
+ Criteria: {criteria}
+ Answer only YES or NO.
+ """
+ result = await judge_model.run(prompt)
+ return "YES" in result.output.upper()
+
+# Usage in test:
+assert await llm_judge(
+ response,
+ "The answer correctly states that sqrt(144) + 25 = 37"
+)
+```
+
+### 2. Fuzzy/Regex Matching
+
+For numeric answers, accept multiple representations:
+
+```python
+import re
+
+def contains_number(text: str, number: int) -> bool:
+ """Check if text contains number in any form."""
+ patterns = [
+ rf'\b{number}\b', # Digit form
+ number_to_words(number), # Word form
+ ]
+ return any(re.search(p, text, re.I) for p in patterns)
+
+# Usage:
+assert contains_number(response, 37) # Matches "37" or "thirty-seven"
+```
+
+### 3. DeepEval Framework
+
+```python
+from deepeval.metrics import AnswerRelevancyMetric
+from deepeval.test_case import LLMTestCase
+
+def test_calculation():
+ test_case = LLMTestCase(
+ input="What is sqrt(144) + 25?",
+ actual_output=response,
+ expected_output="37"
+ )
+ metric = AnswerRelevancyMetric(threshold=0.7)
+ assert metric.measure(test_case)
+```
+
+### 4. pytest-evals Plugin
+
+Minimal pytest plugin for LLM testing with metrics collection.
+
+```bash
+pip install pytest-evals
+```
+
+### 5. Multiple Runs with Threshold
+
+Run flaky tests multiple times and require majority pass:
+
+```python
+@pytest.mark.flaky(reruns=3, reruns_delay=1)
+def test_llm_response():
+ ...
+```
+
+Or custom:
+
+```python
+@pytest.mark.parametrize("run", range(3))
+def test_llm_response(run):
+ ...
+ # Aggregate results across runs
+```
+
+## Resources
+
+- [DeepEval](https://github.com/confident-ai/deepeval) - LLM evaluation framework
+- [pytest-evals](https://github.com/AlmogBaku/pytest-evals) - pytest plugin for LLM evals
+- [LLM Testing Guide 2025](https://www.confident-ai.com/blog/llm-testing-in-2024-top-methods-and-strategies)
+- [Testing LLM Applications - Langfuse](https://langfuse.com/blog/2025-10-21-testing-llm-applications)
+
+## Implementation Priority
+
+1. Add fuzzy number matching helper (quick win)
+2. Evaluate DeepEval for complex output testing
+3. Consider LLM-as-judge for semantic correctness
diff --git a/pyproject.toml b/pyproject.toml
index 1afd813..18994e3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "tatlock"
-version = "1.5.0"
+version = "1.6.0"
description = "OpenAI-compatible API with Ollama backend"
requires-python = ">=3.12"
dependencies = []
diff --git a/src/agents/delegation.py b/src/agents/delegation.py
index eb6f781..cb11720 100644
--- a/src/agents/delegation.py
+++ b/src/agents/delegation.py
@@ -9,13 +9,136 @@ This implements the agent-as-tool pattern recommended by PydanticAI:
agents call other agents via tool wrappers, keeping each agent focused.
"""
from dataclasses import dataclass, field
-from typing import Callable, Optional, Any
+from enum import Enum
+from typing import AsyncGenerator, Callable, Optional, Any
from src.core.logging_config import get_logger
logger = get_logger(__name__)
+# =============================================================================
+# Action Types for Think Slug Selection
+# =============================================================================
+
+class ActionType(Enum):
+ """
+ Categories of actions for selecting appropriate think messages.
+
+ Each expert has different action types that warrant different
+ butler-perspective messages to the user.
+ """
+ RETRIEVE = "retrieve" # Looking up existing information
+ RESEARCH = "research" # Conducting new research (web search, etc.)
+ CREATE = "create" # Creating new content (pages, notes)
+ CONTROL = "control" # Controlling devices/automations
+ RECORD = "record" # Recording memories/notes
+
+
+# =============================================================================
+# Household Think Messages (Butler's Perspective)
+# =============================================================================
+
+HOUSEHOLD_THINK_MESSAGES: dict[str, dict[ActionType, dict[str, str]]] = {
+ "librarian": {
+ ActionType.RETRIEVE: {
+ "start": "Allow me to consult the archives, sir.",
+ "success": "The Librarian has compiled the relevant findings.",
+ "error": "I'm afraid the archives proved difficult to access.",
+ },
+ ActionType.RESEARCH: {
+ "start": "I've dispatched the Librarian to conduct some fresh research.",
+ "success": "The Librarian has returned with findings, sir.",
+ "error": "The research proved inconclusive, I'm afraid.",
+ },
+ ActionType.CREATE: {
+ "start": "I'm having the Librarian prepare a new entry.",
+ "success": "The new material has been properly catalogued, sir.",
+ "error": "I'm afraid there was difficulty filing the entry.",
+ },
+ },
+ "biographer": {
+ ActionType.RETRIEVE: {
+ "start": "Let me consult the household records.",
+ "success": "The Biographer has located the relevant information, sir.",
+ "error": "I'm unable to locate those particular records.",
+ },
+ ActionType.RECORD: {
+ "start": "I've asked the Biographer to take note of this, sir.",
+ "success": "The household records have been updated accordingly.",
+ "error": "I'm afraid there was difficulty recording the entry.",
+ },
+ },
+ "housekeeper": {
+ ActionType.RETRIEVE: {
+ "start": "Allow me to inquire with the household staff.",
+ "success": "The staff reports the current status, sir.",
+ "error": "The household staff is momentarily unavailable, I'm afraid.",
+ },
+ ActionType.CONTROL: {
+ "start": "I'm instructing the household staff now, sir.",
+ "success": "The household has been configured as requested.",
+ "error": "I'm afraid the staff reports an issue with that request.",
+ },
+ },
+}
+
+
+def _detect_action_type(expert: str, task: str) -> ActionType:
+ """
+ Detect action type from expert name and task description.
+
+ Used to select appropriate butler-perspective think messages.
+
+ Args:
+ expert: Name of the expert (librarian, biographer, housekeeper)
+ task: Task description
+
+ Returns:
+ ActionType: Detected action type for message selection
+ """
+ task_lower = task.lower()
+
+ if expert == "librarian":
+ if any(w in task_lower for w in ["search", "find", "look up", "research"]):
+ if any(w in task_lower for w in ["web", "online", "internet"]):
+ return ActionType.RESEARCH
+ return ActionType.RETRIEVE
+ if any(w in task_lower for w in ["create", "write", "add", "make", "new"]):
+ return ActionType.CREATE
+ return ActionType.RETRIEVE
+
+ elif expert == "biographer":
+ if any(w in task_lower for w in ["remember", "note", "record", "save", "store"]):
+ return ActionType.RECORD
+ return ActionType.RETRIEVE
+
+ elif expert == "housekeeper":
+ if any(w in task_lower for w in ["turn", "set", "activate", "enable", "disable", "toggle"]):
+ return ActionType.CONTROL
+ return ActionType.RETRIEVE
+
+ return ActionType.RETRIEVE
+
+
+def get_think_message(expert: str, task: str, phase: str) -> str:
+ """
+ Get the appropriate think message for an expert delegation.
+
+ Args:
+ expert: Name of the expert
+ task: Task description (used to detect action type)
+ phase: One of "start", "success", "error"
+
+ Returns:
+ str: Butler-perspective think message
+ """
+ action_type = _detect_action_type(expert, task)
+ expert_messages = HOUSEHOLD_THINK_MESSAGES.get(expert, {})
+ action_messages = expert_messages.get(action_type, expert_messages.get(ActionType.RETRIEVE, {}))
+ return action_messages.get(phase, f"Consulting {expert}...")
+
+
@dataclass
class DelegationTask:
"""
@@ -301,6 +424,103 @@ 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
diff --git a/src/agents/steward/schemas.py b/src/agents/steward/schemas.py
index a84a95f..377ce58 100644
--- a/src/agents/steward/schemas.py
+++ b/src/agents/steward/schemas.py
@@ -60,6 +60,10 @@ class StewardRecommendation(BaseModel):
default_factory=dict,
description="Pre-fetched user context from memory (profile, preferences)"
)
+ enriched_query: str = Field(
+ default="",
+ description="User query with auto-filled context (location, timezone) when not specified"
+ )
def format_for_butler(self) -> str:
"""
diff --git a/src/agents/steward/service.py b/src/agents/steward/service.py
index 2db83fa..b152b92 100644
--- a/src/agents/steward/service.py
+++ b/src/agents/steward/service.py
@@ -149,6 +149,68 @@ def _extract_missing_capabilities(text: str) -> Optional[str]:
return None
+def _build_enriched_query(user_request: str, memory_context: dict[str, Any]) -> str:
+ """
+ Build an enriched query by appending user context when not specified.
+
+ When the user asks location-dependent questions (weather, nearby, etc.)
+ without specifying a location, this appends their known location.
+ Similarly for timezone-dependent queries.
+
+ Args:
+ user_request: The user's original request
+ memory_context: Pre-fetched memory context with profile/preferences
+
+ Returns:
+ str: Query with context appended, or original query if no enrichment needed
+
+ Example:
+ >>> query = _build_enriched_query(
+ ... "What's the weather?",
+ ... {"profile": {"location": "Amsterdam", "timezone": "Europe/Amsterdam"}}
+ ... )
+ >>> query
+ "What's the weather?\n\n[User Context: location=Amsterdam, timezone=Europe/Amsterdam]"
+ """
+ if not memory_context:
+ return user_request
+
+ request_lower = user_request.lower()
+ profile = memory_context.get("profile", {})
+ preferences = memory_context.get("preferences", {})
+
+ context_parts = []
+
+ # Check if location is needed and not specified
+ location_keywords = ["weather", "temperature", "forecast", "nearby", "local", "here"]
+ # Use word boundary pattern to avoid false positives like "at" in "what"
+ location_prepositions = [r'\bin\b', r'\bat\b', r'\bnear\b', r'\baround\b', r'\bfor\b']
+ location_specified = any(re.search(p, request_lower) for p in location_prepositions)
+
+ if any(word in request_lower for word in location_keywords):
+ if not location_specified and profile.get("location"):
+ context_parts.append(f"location={profile['location']}")
+
+ # Check if timezone is needed and not specified
+ time_keywords = ["time", "schedule", "meeting", "appointment", "when", "today", "tomorrow"]
+ timezone_specified = any(word in request_lower for word in ["timezone", "tz", "utc", "gmt"])
+
+ if any(word in request_lower for word in time_keywords):
+ if not timezone_specified and profile.get("timezone"):
+ context_parts.append(f"timezone={profile['timezone']}")
+
+ # Add preferences if relevant
+ if preferences.get("temperature_unit") and "weather" in request_lower:
+ context_parts.append(f"temperature_unit={preferences['temperature_unit']}")
+
+ # Build enriched query
+ if context_parts:
+ context_str = ", ".join(context_parts)
+ return f"{user_request}\n\n[User Context: {context_str}]"
+
+ return user_request
+
+
async def _prefetch_memory_context(user_request: str) -> dict[str, Any]:
"""
Pre-fetch user context that might be needed for this request.
@@ -277,6 +339,9 @@ async def analyze_request(
context = _extract_conversation_context(analysis_text, conversation_history)
missing = _extract_missing_capabilities(analysis_text)
+ # Build enriched query with auto-filled context
+ enriched_query = _build_enriched_query(user_request, memory_context)
+
recommendation = StewardRecommendation(
recommended_capabilities=capabilities,
reasoning=analysis_text,
@@ -284,6 +349,7 @@ async def analyze_request(
conversation_context=context,
missing_capabilities=missing,
memory_context=memory_context,
+ enriched_query=enriched_query,
)
# Update log context with results
diff --git a/src/agents/tatlock.py b/src/agents/tatlock.py
index a3d8752..6944fce 100644
--- a/src/agents/tatlock.py
+++ b/src/agents/tatlock.py
@@ -630,6 +630,241 @@ class TatlockAgent(AgentInterface):
logger.info("tatlock_scoped_run_complete")
+ async def orchestrate_tool_calls(
+ self,
+ user_message: str,
+ steward_note: str,
+ scoped_tools: list[Any],
+ message_history: list[dict],
+ tool_tracker: Any = None,
+ ) -> dict[str, Any]:
+ """
+ Phase 1: Execute tool calls and delegations, return structured results.
+
+ This is the coordination phase where Tatlock orchestrates tool calls
+ and expert delegations. The raw output is captured for Phase 2 synthesis.
+
+ Args:
+ user_message: The user's original message
+ steward_note: Note from Steward (invisible to user)
+ scoped_tools: List of tool definitions from household registry
+ message_history: Conversation history
+ tool_tracker: Optional tool call tracker for benchmarking
+
+ Returns:
+ dict with:
+ - tools_called: List of tool names that were called
+ - expert_results: Dict mapping expert names to their outputs
+ - tool_outputs: Dict mapping tool names to their outputs
+ - raw_output: The agent's raw text output
+ """
+ from pydantic_ai.models.openai import OpenAIChatModel
+ from pydantic_ai.providers.ollama import OllamaProvider
+ from pydantic_ai.settings import ModelSettings
+ from pydantic_ai.messages import (
+ ModelRequest,
+ ModelResponse,
+ UserPromptPart,
+ TextPart,
+ ToolCallPart,
+ ToolReturnPart,
+ )
+
+ logger.info(
+ "tatlock_orchestrate_tool_calls",
+ user_message_preview=user_message[:100],
+ scoped_tool_count=len(scoped_tools),
+ history_length=len(message_history),
+ )
+
+ # Create a fresh agent instance with scoped tools only
+ clean_host = self.ollama_host.rstrip('/')
+ base_url = f"{clean_host}/v1"
+
+ ollama_model = OpenAIChatModel(
+ model_name=self.model_name,
+ provider=OllamaProvider(base_url=base_url)
+ )
+
+ # Create agent with scoped tools
+ scoped_agent = Agent(
+ ollama_model,
+ system_prompt=TATLOCK_SYSTEM_PROMPT,
+ tools=scoped_tools,
+ )
+
+ # Prepend Steward's note to the request
+ enriched_message = f"{steward_note}\n\n{user_message}"
+
+ # Convert message history to PydanticAI format
+ pydantic_history = []
+ for msg in message_history:
+ role = msg.get("role")
+ content = msg.get("content", "")
+
+ if not content or not content.strip():
+ continue
+
+ if role == "user":
+ pydantic_history.append(
+ ModelRequest(parts=[UserPromptPart(content=content)])
+ )
+ elif role == "assistant":
+ pydantic_history.append(
+ ModelResponse(parts=[TextPart(content=content)])
+ )
+
+ # Run with scoped tools and tracker
+ result = await scoped_agent.run(
+ enriched_message,
+ message_history=pydantic_history if pydantic_history else None,
+ deps=tool_tracker,
+ model_settings=ModelSettings(extra_body={"tool_choice": "required"})
+ )
+
+ # Extract tool calls and results from the agent's messages
+ tools_called = []
+ expert_results = {}
+ tool_outputs = {}
+
+ # Parse through new messages to find tool calls and returns
+ for msg in result.new_messages():
+ if isinstance(msg, ModelResponse):
+ for part in msg.parts:
+ if isinstance(part, ToolCallPart):
+ tools_called.append(part.tool_name)
+ elif isinstance(msg, ModelRequest):
+ for part in msg.parts:
+ if isinstance(part, ToolReturnPart):
+ tool_name = part.tool_name
+ content = part.content
+
+ # Categorize as expert result or tool output
+ if tool_name.startswith("delegate_to_"):
+ expert_name = tool_name.replace("delegate_to_", "")
+ expert_results[expert_name] = content
+ else:
+ tool_outputs[tool_name] = content
+
+ logger.info(
+ "tatlock_orchestration_complete",
+ tools_called=tools_called,
+ expert_count=len(expert_results),
+ tool_output_count=len(tool_outputs),
+ )
+
+ return {
+ "tools_called": tools_called,
+ "expert_results": expert_results,
+ "tool_outputs": tool_outputs,
+ "raw_output": result.output,
+ }
+
+ async def synthesize_from_results(
+ self,
+ user_message: str,
+ orchestration_results: dict[str, Any],
+ message_history: list[dict],
+ ) -> str:
+ """
+ Phase 2: Synthesize butler-toned response from gathered results.
+
+ This is the synthesis phase where Tatlock takes the coordination
+ results and produces a properly butler-toned response.
+
+ Args:
+ user_message: The user's original message
+ orchestration_results: Results from orchestrate_tool_calls()
+ message_history: Conversation history
+
+ Returns:
+ str: Butler-toned response synthesized from all results
+ """
+ from pydantic_ai.models.openai import OpenAIChatModel
+ from pydantic_ai.providers.ollama import OllamaProvider
+ from pydantic_ai.messages import ModelRequest, ModelResponse, UserPromptPart, TextPart
+
+ logger.info(
+ "tatlock_synthesize_from_results",
+ user_message_preview=user_message[:100],
+ expert_count=len(orchestration_results.get("expert_results", {})),
+ tool_count=len(orchestration_results.get("tool_outputs", {})),
+ )
+
+ # Build synthesis prompt with all available information
+ synthesis_parts = []
+ synthesis_parts.append(f"The user asked: {user_message}")
+ synthesis_parts.append("")
+
+ # Add expert findings if any
+ if orchestration_results.get("expert_results"):
+ synthesis_parts.append("Expert findings:")
+ for expert, result in orchestration_results["expert_results"].items():
+ synthesis_parts.append(f"- {expert.title()}: {result}")
+ synthesis_parts.append("")
+
+ # Add tool outputs if any
+ if orchestration_results.get("tool_outputs"):
+ synthesis_parts.append("Tool results:")
+ for tool, result in orchestration_results["tool_outputs"].items():
+ synthesis_parts.append(f"- {tool}: {result}")
+ synthesis_parts.append("")
+
+ synthesis_parts.append(
+ "Based on this information, provide a response to the user. "
+ "Maintain your butler personality - address them as 'sir', "
+ "use formal but personable language, and be helpful."
+ )
+
+ synthesis_prompt = "\n".join(synthesis_parts)
+
+ # Create synthesis agent (no tools needed)
+ clean_host = self.ollama_host.rstrip('/')
+ base_url = f"{clean_host}/v1"
+
+ ollama_model = OpenAIChatModel(
+ model_name=self.model_name,
+ provider=OllamaProvider(base_url=base_url)
+ )
+
+ # Synthesis agent uses butler prompt but no tools
+ synthesis_agent = Agent(
+ ollama_model,
+ system_prompt=TATLOCK_SYSTEM_PROMPT,
+ # No tools for synthesis phase
+ )
+
+ # Convert message history to PydanticAI format
+ pydantic_history = []
+ for msg in message_history:
+ role = msg.get("role")
+ content = msg.get("content", "")
+
+ if not content or not content.strip():
+ continue
+
+ if role == "user":
+ pydantic_history.append(
+ ModelRequest(parts=[UserPromptPart(content=content)])
+ )
+ elif role == "assistant":
+ pydantic_history.append(
+ ModelResponse(parts=[TextPart(content=content)])
+ )
+
+ # Run synthesis
+ result = await synthesis_agent.run(
+ synthesis_prompt,
+ message_history=pydantic_history if pydantic_history else None,
+ )
+
+ logger.info(
+ "tatlock_synthesis_complete",
+ response_preview=result.output[:100],
+ )
+
+ return result.output
+
async def get_capabilities(self) -> dict:
"""Return current capabilities."""
return {
diff --git a/src/core/household_registry.py b/src/core/household_registry.py
index 780dc52..ddf1c88 100644
--- a/src/core/household_registry.py
+++ b/src/core/household_registry.py
@@ -273,6 +273,65 @@ 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.
diff --git a/src/responses/service.py b/src/responses/service.py
index 19b6cd4..bc8eea0 100644
--- a/src/responses/service.py
+++ b/src/responses/service.py
@@ -43,7 +43,7 @@ async def _execute_single_delegation(
Execute a single delegation to an agent.
Args:
- agent_name: Name of agent (biographer, librarian)
+ agent_name: Name of agent (biographer, librarian, housekeeper)
task: Task description
tracker: Tool call tracker
@@ -67,6 +67,13 @@ async def _execute_single_delegation(
await tracker.track_call("delegate_to_librarian", duration)
return (agent_name, result.output)
+ elif agent_name == "housekeeper":
+ from src.agents.delegation import delegate_to_housekeeper
+ result = await delegate_to_housekeeper(task=task)
+ duration = time.time() - start_time
+ await tracker.track_call("delegate_to_housekeeper", duration)
+ return (agent_name, result.output)
+
else:
return (agent_name, f"Unknown agent: {agent_name}")
@@ -244,6 +251,68 @@ async def _direct_delegation(
return "\n\n".join(results) if results else "I apologize, sir. No delegation results available."
+async def _direct_delegation_with_results(
+ user_message: str,
+ recommendation: "StewardRecommendation",
+ tracker: "ToolCallTracker",
+ conversation_id: str,
+) -> dict:
+ """
+ Directly delegate to expert agents and return structured results.
+
+ This is the Phase 1 variant of direct delegation that returns results
+ in the same format as TatlockAgent.orchestrate_tool_calls() for
+ consistent Phase 2 synthesis.
+
+ Args:
+ user_message: User's request
+ recommendation: Steward's recommendation
+ tracker: Tool call tracker
+ conversation_id: Conversation ID
+
+ Returns:
+ dict: Orchestration results with expert_results, tool_outputs, etc.
+ """
+ logger.info(
+ "direct_delegation_with_results",
+ agents=recommendation.recommended_capabilities,
+ conversation_id=conversation_id,
+ )
+
+ expert_results = {}
+ tools_called = []
+
+ for agent in recommendation.recommended_capabilities:
+ try:
+ agent_name, result = await _execute_single_delegation(
+ agent, user_message, tracker
+ )
+ expert_results[agent_name] = result
+ tools_called.append(f"delegate_to_{agent_name}")
+
+ logger.info(
+ "direct_delegation_result",
+ agent=agent_name,
+ result_preview=result[:100] if result else "empty",
+ conversation_id=conversation_id,
+ )
+ except Exception as e:
+ logger.error(
+ "direct_delegation_failed",
+ agent=agent,
+ error=str(e),
+ conversation_id=conversation_id,
+ )
+ expert_results[agent] = f"Error: {e}"
+
+ return {
+ "tools_called": tools_called,
+ "expert_results": expert_results,
+ "tool_outputs": {}, # No tool outputs for direct delegation
+ "raw_output": "", # No raw output for direct delegation
+ }
+
+
# Global conversation history tracker
# In production, this would be backed by a database or Redis
_conversation_history = ConversationHistory(max_turns=20)
@@ -379,12 +448,13 @@ async def create_response(request: ResponseRequest) -> Response:
async def create_response_with_steward(request: ResponseRequest) -> Response:
"""
- Create response using Steward preprocessing (Phase 2 flow).
+ Create response using Steward preprocessing and two-phase Tatlock execution.
- This is the two-tier architecture where:
+ This is the two-tier architecture with two-phase synthesis:
1. Steward analyzes the request and recommends capabilities
- 2. Tatlock runs with scoped tools based on recommendations
- 3. Tool usage is tracked for benchmarking
+ 2. Phase 1: Tatlock orchestrates tool calls and expert delegations
+ 3. Phase 2: Tatlock synthesizes butler-toned response from results
+ 4. Tool usage is tracked for benchmarking
Args:
request: Response request
@@ -420,37 +490,39 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
conversation_id=conversation_id,
)
- # Phase 1: Steward preprocessing
+ # Steward preprocessing
enriched = await preprocess_request(
user_message,
conversation_history=conversation_history,
conversation_id=conversation_id,
)
- # Phase 2: Initialize tool tracker
+ # Initialize tool tracker
tracker = ToolCallTracker(
recommended_capabilities=enriched.recommendation.recommended_capabilities,
conversation_id=conversation_id,
)
- # Phase 3: Check if direct delegation is recommended
- # If Steward recommends ONLY delegation agents (biographer/librarian),
- # skip Tatlock and delegate directly
+ # Check if direct delegation is recommended
+ # If Steward recommends ONLY delegation agents (biographer/librarian/housekeeper),
+ # we still use two-phase but delegate directly in Phase 1
+ delegation_agents = {"biographer", "librarian", "housekeeper"}
delegation_only = all(
- cap in ("biographer", "librarian")
+ cap in delegation_agents
for cap in enriched.recommendation.recommended_capabilities
) and enriched.recommendation.recommended_capabilities
+ from src.agents.tatlock import TatlockAgent
+ tatlock = TatlockAgent()
+
if delegation_only:
- tatlock_response = await _direct_delegation(
+ # Direct delegation path - collect results then synthesize
+ orchestration_results = await _direct_delegation_with_results(
user_message, enriched.recommendation, tracker, conversation_id
)
else:
- # Phase 3a: Run Tatlock with scoped tools
- from src.agents.tatlock import TatlockAgent
- tatlock = TatlockAgent()
-
- tatlock_response = await tatlock.run_with_scoped_tools(
+ # Phase 1: Orchestrate tool calls
+ orchestration_results = await tatlock.orchestrate_tool_calls(
user_message=user_message,
steward_note=enriched.steward_note,
scoped_tools=enriched.scoped_tools,
@@ -458,14 +530,23 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
tool_tracker=tracker,
)
- # Phase 3b: Check for text-based delegation fallback
- # If Tatlock outputs [DELEGATE:...] instead of calling the function,
- # we parse and execute it here
- tatlock_response = await _handle_text_delegation(
- tatlock_response, tracker, conversation_id
- )
+ # Handle text-based delegation fallback if present
+ if "[DELEGATE:" in orchestration_results.get("raw_output", ""):
+ text_delegation_results = await _handle_text_delegation(
+ orchestration_results["raw_output"], tracker, conversation_id
+ )
+ # Add text delegation results to expert_results
+ if text_delegation_results != orchestration_results["raw_output"]:
+ orchestration_results["expert_results"]["text_delegation"] = text_delegation_results
- # Phase 4: Finalize tool tracking
+ # Phase 2: Synthesize butler-toned response from all results
+ tatlock_response = await tatlock.synthesize_from_results(
+ user_message=user_message,
+ orchestration_results=orchestration_results,
+ message_history=conversation_history,
+ )
+
+ # Finalize tool tracking
await tracker.finalize()
# Build response output items
diff --git a/src/responses/streaming.py b/src/responses/streaming.py
index beaadb8..0987d75 100644
--- a/src/responses/streaming.py
+++ b/src/responses/streaming.py
@@ -118,11 +118,12 @@ class StreamingCoordinator:
request: "ResponseRequest" # type: ignore # Forward reference
) -> AsyncGenerator[StreamEvent, None]:
"""
- Stream response with Steward preprocessing (Phase 2 flow).
+ Stream response with Steward preprocessing and two-phase Tatlock execution.
Streams in order:
1. Steward's analysis as reasoning summary
- 2. Tatlock's response as output text
+ 2. Think slugs during expert delegation (butler-perspective messages)
+ 3. Synthesized butler-toned response as output text
Args:
request: Response request
@@ -130,11 +131,17 @@ class StreamingCoordinator:
Yields:
StreamEvent: Stream of SSE events
"""
- from src.responses.service import _calculate_usage, generate_id, _conversation_history
+ from src.responses.service import (
+ _calculate_usage,
+ generate_id,
+ _conversation_history,
+ _direct_delegation_with_results,
+ )
from src.core.preprocessing import preprocess_request
from src.core.tool_tracking import ToolCallTracker
from src.responses.schemas import MessageOutputItem, ReasoningOutputItem, OutputTextContent
from src.agents.tatlock import TatlockAgent
+ from src.agents.delegation import get_think_message, STREAMING_DELEGATION_WRAPPERS
import asyncio
output_items = []
@@ -152,7 +159,7 @@ class StreamingCoordinator:
conversation_history = request.input[:-1] if len(request.input) > 1 else []
- # Phase 1: Steward preprocessing
+ # Steward preprocessing
enriched = await preprocess_request(
user_message,
conversation_history=conversation_history,
@@ -179,31 +186,60 @@ class StreamingCoordinator:
)
output_items.append(reasoning_item)
- # Phase 2: Initialize tool tracker
+ # Initialize tool tracker
tracker = ToolCallTracker(
recommended_capabilities=enriched.recommendation.recommended_capabilities,
conversation_id=conversation_id,
)
- # Phase 3: Stream Tatlock's response with scoped tools
- tatlock = TatlockAgent()
- tatlock_response_parts = []
+ # Check if direct delegation is recommended
+ delegation_agents = {"biographer", "librarian", "housekeeper"}
+ delegation_only = all(
+ cap in delegation_agents
+ for cap in enriched.recommendation.recommended_capabilities
+ ) and enriched.recommendation.recommended_capabilities
- async for chunk in tatlock.run_with_scoped_tools_stream(
+ tatlock = TatlockAgent()
+
+ if delegation_only:
+ # Direct delegation path with streaming think slugs
+ orchestration_results = await self._stream_direct_delegation(
+ user_message=user_message,
+ recommendation=enriched.recommendation,
+ tracker=tracker,
+ conversation_id=conversation_id,
+ )
+
+ # Stream think slugs that were collected during delegation
+ for think_msg in orchestration_results.get("think_messages", []):
+ yield ReasoningSummaryDelta(delta=think_msg)
+ await asyncio.sleep(0.05)
+
+ else:
+ # Phase 1: Orchestrate tool calls
+ orchestration_results = await tatlock.orchestrate_tool_calls(
+ user_message=user_message,
+ steward_note=enriched.steward_note,
+ scoped_tools=enriched.scoped_tools,
+ message_history=conversation_history,
+ tool_tracker=tracker,
+ )
+
+ # Phase 2: Synthesize butler-toned response
+ tatlock_response = await tatlock.synthesize_from_results(
user_message=user_message,
- steward_note=enriched.steward_note,
- scoped_tools=enriched.scoped_tools,
+ orchestration_results=orchestration_results,
message_history=conversation_history,
- tool_tracker=tracker,
- ):
- tatlock_response_parts.append(chunk)
- yield OutputTextDelta(delta=chunk)
+ )
+
+ # Stream the synthesized response
+ chunk_size = 50
+ for i in range(0, len(tatlock_response), chunk_size):
+ yield OutputTextDelta(delta=tatlock_response[i:i + chunk_size])
+ await asyncio.sleep(0.02)
yield OutputTextDone()
- # Combine response for output item
- tatlock_response = "".join(tatlock_response_parts)
-
# Add Tatlock message to output items
message_item = MessageOutputItem(
id=f"msg_{generate_id()}",
@@ -217,7 +253,7 @@ class StreamingCoordinator:
)
output_items.append(message_item)
- # Phase 4: Finalize tool tracking
+ # Finalize tool tracking
await tracker.finalize()
# Calculate usage and build final response
@@ -241,6 +277,85 @@ class StreamingCoordinator:
# Stream error event
yield self._create_error_event(e)
+ async def _stream_direct_delegation(
+ self,
+ user_message: str,
+ recommendation: "StewardRecommendation", # type: ignore
+ tracker: "ToolCallTracker", # type: ignore
+ conversation_id: str,
+ ) -> dict:
+ """
+ Execute direct delegation with streaming think messages.
+
+ Collects think messages as delegations execute for streaming to client.
+
+ Args:
+ user_message: User's request
+ recommendation: Steward's recommendation
+ tracker: Tool call tracker
+ conversation_id: Conversation ID
+
+ Returns:
+ dict: Orchestration results with think_messages list
+ """
+ from src.agents.delegation import (
+ get_think_message,
+ delegate_to_librarian,
+ delegate_to_biographer,
+ delegate_to_housekeeper,
+ )
+ import time as time_module
+
+ expert_results = {}
+ tools_called = []
+ think_messages = []
+
+ for agent in recommendation.recommended_capabilities:
+ # Emit start think message
+ start_msg = get_think_message(agent, user_message, "start")
+ think_messages.append(start_msg + "\n")
+
+ start_time = time_module.time()
+ try:
+ # Execute delegation
+ if agent == "librarian":
+ result = await delegate_to_librarian(task=user_message)
+ elif agent == "biographer":
+ result = await delegate_to_biographer(task=user_message)
+ elif agent == "housekeeper":
+ result = await delegate_to_housekeeper(task=user_message)
+ else:
+ result = None
+
+ duration = time_module.time() - start_time
+ await tracker.track_call(f"delegate_to_{agent}", duration)
+
+ if result and result.success:
+ expert_results[agent] = result.output
+ tools_called.append(f"delegate_to_{agent}")
+ # Emit success think message
+ success_msg = get_think_message(agent, user_message, "success")
+ think_messages.append(success_msg + "\n")
+ else:
+ error_msg = result.error if result else "Unknown error"
+ expert_results[agent] = f"Error: {error_msg}"
+ # Emit error think message
+ error_think = get_think_message(agent, user_message, "error")
+ think_messages.append(error_think + "\n")
+
+ except Exception as e:
+ expert_results[agent] = f"Error: {e}"
+ error_think = get_think_message(agent, user_message, "error")
+ think_messages.append(error_think + "\n")
+
+ return {
+ "tools_called": tools_called,
+ "expert_results": expert_results,
+ "tool_outputs": {},
+ "raw_output": "",
+ "think_messages": think_messages,
+ }
+
async def stream_response(
self,
request: "ResponseRequest" # type: ignore # Forward reference
diff --git a/tests/agents/steward/test_steward_service.py b/tests/agents/steward/test_steward_service.py
index 7b87496..59f5e25 100644
--- a/tests/agents/steward/test_steward_service.py
+++ b/tests/agents/steward/test_steward_service.py
@@ -8,7 +8,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.agents.steward.schemas import ConversationContext, StewardRecommendation
-from src.agents.steward.service import analyze_request, format_steward_note
+from src.agents.steward.service import analyze_request, format_steward_note, _build_enriched_query
from src.core.startup import initialize_application
@@ -199,3 +199,102 @@ class TestFormatStewardNote:
assert "⚠️ Missing:" in note
assert "Advanced research" in note
+
+
+@pytest.mark.unit
+class TestBuildEnrichedQuery:
+ """Tests for _build_enriched_query function."""
+
+ def test_no_enrichment_without_context(self):
+ """Test no enrichment when memory context is empty."""
+ query = "What's the weather?"
+ result = _build_enriched_query(query, {})
+
+ assert result == query
+
+ def test_enrichment_adds_location(self):
+ """Test location is appended for weather queries."""
+ query = "What's the weather?"
+ memory_context = {
+ "profile": {"location": "Amsterdam", "timezone": "Europe/Amsterdam"}
+ }
+
+ result = _build_enriched_query(query, memory_context)
+
+ assert "location=Amsterdam" in result
+ assert query in result
+ assert "[User Context:" in result
+
+ def test_no_location_when_specified(self):
+ """Test location is not appended when already specified."""
+ query = "What's the weather in London?"
+ memory_context = {
+ "profile": {"location": "Amsterdam"}
+ }
+
+ result = _build_enriched_query(query, memory_context)
+
+ # Should not add Amsterdam since location is specified
+ assert result == query
+
+ def test_enrichment_adds_timezone(self):
+ """Test timezone is appended for time queries."""
+ query = "What time is it?"
+ memory_context = {
+ "profile": {"timezone": "Europe/Amsterdam"}
+ }
+
+ result = _build_enriched_query(query, memory_context)
+
+ assert "timezone=Europe/Amsterdam" in result
+
+ def test_no_timezone_when_specified(self):
+ """Test timezone is not appended when already specified."""
+ query = "What time is it in UTC?"
+ memory_context = {
+ "profile": {"timezone": "Europe/Amsterdam"}
+ }
+
+ result = _build_enriched_query(query, memory_context)
+
+ assert result == query
+
+ def test_enrichment_adds_temperature_unit(self):
+ """Test temperature unit is appended for weather queries."""
+ query = "What's the weather?"
+ memory_context = {
+ "profile": {"location": "Amsterdam"},
+ "preferences": {"temperature_unit": "celsius"}
+ }
+
+ result = _build_enriched_query(query, memory_context)
+
+ assert "temperature_unit=celsius" in result
+
+ def test_multiple_context_fields(self):
+ """Test multiple context fields are appended."""
+ query = "What time and weather today?"
+ memory_context = {
+ "profile": {
+ "location": "Amsterdam",
+ "timezone": "Europe/Amsterdam"
+ },
+ "preferences": {"temperature_unit": "celsius"}
+ }
+
+ result = _build_enriched_query(query, memory_context)
+
+ assert "location=Amsterdam" in result
+ assert "timezone=Europe/Amsterdam" in result
+ assert "temperature_unit=celsius" in result
+
+ def test_no_enrichment_for_unrelated_query(self):
+ """Test no enrichment for queries that don't need context."""
+ query = "Tell me a joke"
+ memory_context = {
+ "profile": {"location": "Amsterdam", "timezone": "Europe/Amsterdam"}
+ }
+
+ result = _build_enriched_query(query, memory_context)
+
+ assert result == query
diff --git a/tests/agents/test_delegation.py b/tests/agents/test_delegation.py
index 0ad87e2..1abe358 100644
--- a/tests/agents/test_delegation.py
+++ b/tests/agents/test_delegation.py
@@ -8,9 +8,14 @@ import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from src.agents.delegation import (
+ ActionType,
DelegationTask,
DelegationResult,
+ HOUSEHOLD_THINK_MESSAGES,
+ STREAMING_DELEGATION_WRAPPERS,
delegate_to_librarian,
+ get_think_message,
+ _detect_action_type,
)
@@ -193,3 +198,158 @@ class TestDelegateToLibrarian:
result = await delegate_to_librarian(task=original_task)
assert result.task == original_task
+
+
+@pytest.mark.unit
+class TestActionType:
+ """Tests for the ActionType enum."""
+
+ def test_action_type_values(self):
+ """Test ActionType enum values."""
+ assert ActionType.RETRIEVE.value == "retrieve"
+ assert ActionType.RESEARCH.value == "research"
+ assert ActionType.CREATE.value == "create"
+ assert ActionType.CONTROL.value == "control"
+ assert ActionType.RECORD.value == "record"
+
+ def test_action_type_is_enum(self):
+ """Test ActionType is proper enum."""
+ assert len(ActionType) == 5
+
+
+@pytest.mark.unit
+class TestHouseholdThinkMessages:
+ """Tests for HOUSEHOLD_THINK_MESSAGES mapping."""
+
+ def test_librarian_has_messages(self):
+ """Test librarian has think messages."""
+ assert "librarian" in HOUSEHOLD_THINK_MESSAGES
+ assert ActionType.RETRIEVE in HOUSEHOLD_THINK_MESSAGES["librarian"]
+ assert ActionType.RESEARCH in HOUSEHOLD_THINK_MESSAGES["librarian"]
+ assert ActionType.CREATE in HOUSEHOLD_THINK_MESSAGES["librarian"]
+
+ def test_biographer_has_messages(self):
+ """Test biographer has think messages."""
+ assert "biographer" in HOUSEHOLD_THINK_MESSAGES
+ assert ActionType.RETRIEVE in HOUSEHOLD_THINK_MESSAGES["biographer"]
+ assert ActionType.RECORD in HOUSEHOLD_THINK_MESSAGES["biographer"]
+
+ def test_housekeeper_has_messages(self):
+ """Test housekeeper has think messages."""
+ assert "housekeeper" in HOUSEHOLD_THINK_MESSAGES
+ assert ActionType.RETRIEVE in HOUSEHOLD_THINK_MESSAGES["housekeeper"]
+ assert ActionType.CONTROL in HOUSEHOLD_THINK_MESSAGES["housekeeper"]
+
+ def test_messages_have_phases(self):
+ """Test each action type has start/success/error messages."""
+ for expert, action_types in HOUSEHOLD_THINK_MESSAGES.items():
+ for action_type, messages in action_types.items():
+ assert "start" in messages, f"{expert}/{action_type} missing 'start'"
+ assert "success" in messages, f"{expert}/{action_type} missing 'success'"
+ assert "error" in messages, f"{expert}/{action_type} missing 'error'"
+
+ def test_messages_are_think_tags(self):
+ """Test messages are wrapped in tags."""
+ for expert, action_types in HOUSEHOLD_THINK_MESSAGES.items():
+ for action_type, messages in action_types.items():
+ for phase, msg in messages.items():
+ assert msg.startswith(""), f"{expert}/{action_type}/{phase}"
+ assert msg.endswith(""), f"{expert}/{action_type}/{phase}"
+
+
+@pytest.mark.unit
+class TestDetectActionType:
+ """Tests for _detect_action_type function."""
+
+ def test_librarian_search_is_retrieve(self):
+ """Test librarian search tasks are RETRIEVE."""
+ assert _detect_action_type("librarian", "search for Docker info") == ActionType.RETRIEVE
+ assert _detect_action_type("librarian", "find information about CI/CD") == ActionType.RETRIEVE
+ assert _detect_action_type("librarian", "look up Kubernetes docs") == ActionType.RETRIEVE
+
+ def test_librarian_web_search_is_research(self):
+ """Test librarian web search tasks are RESEARCH."""
+ assert _detect_action_type("librarian", "search the web for news") == ActionType.RESEARCH
+ assert _detect_action_type("librarian", "find online resources") == ActionType.RESEARCH
+ assert _detect_action_type("librarian", "research internet sources") == ActionType.RESEARCH
+
+ def test_librarian_create_is_create(self):
+ """Test librarian creation tasks are CREATE."""
+ assert _detect_action_type("librarian", "create a wiki page") == ActionType.CREATE
+ assert _detect_action_type("librarian", "write a new article") == ActionType.CREATE
+ assert _detect_action_type("librarian", "add a new entry") == ActionType.CREATE
+
+ def test_biographer_recall_is_retrieve(self):
+ """Test biographer recall tasks are RETRIEVE."""
+ assert _detect_action_type("biographer", "what car do I drive?") == ActionType.RETRIEVE
+ assert _detect_action_type("biographer", "what is my job?") == ActionType.RETRIEVE
+
+ def test_biographer_record_is_record(self):
+ """Test biographer record tasks are RECORD."""
+ assert _detect_action_type("biographer", "remember that I work at Acme") == ActionType.RECORD
+ assert _detect_action_type("biographer", "note that my car is a Tesla") == ActionType.RECORD
+ assert _detect_action_type("biographer", "save my preference for dark mode") == ActionType.RECORD
+
+ def test_housekeeper_status_is_retrieve(self):
+ """Test housekeeper status tasks are RETRIEVE."""
+ assert _detect_action_type("housekeeper", "what devices are in the bedroom?") == ActionType.RETRIEVE
+ assert _detect_action_type("housekeeper", "is the living room light on?") == ActionType.RETRIEVE
+
+ def test_housekeeper_control_is_control(self):
+ """Test housekeeper control tasks are CONTROL."""
+ assert _detect_action_type("housekeeper", "turn on the lights") == ActionType.CONTROL
+ assert _detect_action_type("housekeeper", "set brightness to 50%") == ActionType.CONTROL
+ assert _detect_action_type("housekeeper", "activate the movie scene") == ActionType.CONTROL
+ assert _detect_action_type("housekeeper", "toggle the fan") == ActionType.CONTROL
+
+
+@pytest.mark.unit
+class TestGetThinkMessage:
+ """Tests for get_think_message function."""
+
+ def test_librarian_retrieve_start(self):
+ """Test getting librarian retrieve start message."""
+ msg = get_think_message("librarian", "search for Docker", "start")
+ assert "" in msg
+ assert "" in msg
+
+ def test_librarian_create_success(self):
+ """Test getting librarian create success message."""
+ msg = get_think_message("librarian", "create a wiki page", "success")
+ assert "" in msg
+ assert "catalogued" in msg.lower()
+
+ def test_biographer_record_start(self):
+ """Test getting biographer record start message."""
+ msg = get_think_message("biographer", "remember my preference", "start")
+ assert "" in msg
+ assert "note" in msg.lower() or "biographer" in msg.lower()
+
+ def test_housekeeper_control_success(self):
+ """Test getting housekeeper control success message."""
+ msg = get_think_message("housekeeper", "turn on the lights", "success")
+ assert "" in msg
+ assert "configured" in msg.lower()
+
+ def test_unknown_expert_fallback(self):
+ """Test unknown expert gets fallback message."""
+ msg = get_think_message("unknown_expert", "some task", "start")
+ assert "" in msg
+ assert "unknown_expert" in msg.lower()
+
+
+@pytest.mark.unit
+class TestStreamingDelegationWrappers:
+ """Tests for streaming delegation wrapper mapping."""
+
+ def test_streaming_wrappers_exist(self):
+ """Test streaming wrappers mapping has all experts."""
+ assert "librarian" in STREAMING_DELEGATION_WRAPPERS
+ assert "biographer" in STREAMING_DELEGATION_WRAPPERS
+ assert "housekeeper" in STREAMING_DELEGATION_WRAPPERS
+
+ def test_streaming_wrappers_are_async_generators(self):
+ """Test streaming wrappers are async generator functions."""
+ import inspect
+ for name, wrapper in STREAMING_DELEGATION_WRAPPERS.items():
+ assert inspect.isasyncgenfunction(wrapper), f"{name} is not an async generator"