feat(responses): pass conversation context and stream thinks in real time

- delegate_to_* now receives a trimmed conversation history (last ~6
  turns, 500 chars/turn) as context on both live direct-delegation
  paths (streaming and steward non-streaming), via new
  build_delegation_context helper
- _stream_direct_delegation restructured as an async generator: the
  butler 'start' think message streams BEFORE the expert runs and the
  success/error message right after it finishes, instead of all
  messages arriving after the research completed

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Conversation context for experts + real-time think messages** - direct delegation (streaming and non-streaming) now passes a trimmed conversation history (last 6 turns) as expert context, so follow-up questions keep their referent; `_stream_direct_delegation` is now an async generator, so butler think messages ("Allow me to consult the archives, sir.") stream BEFORE the research runs instead of after it completes
 - **Bounded retries and connection reuse for library-desk** - GETs and the read-only `POST /query/*` and `POST /rag/search` endpoints retry once (2 attempts, short backoff) on transport errors and retryable 5xx; wiki writes are never retried. The client now honors `LIBRARY_DESK_TIMEOUT` instead of hardcoded 60s/30s, a librarian run holds one shared HTTP connection instead of constructing a client per tool call, and read tools raise `ModelRetry` on transient HTTP errors so the agent's retry budget engages
 - **One librarian timeout budget** - new `LIBRARIAN_TIMEOUT` (default 180s) enforced with `asyncio.wait_for` inside `delegate_to_librarian`, capping the previously uncapped live paths (steward direct delegation and streaming). The Ollama provider's AsyncOpenAI client now carries an explicit `OLLAMA_TIMEOUT` instead of the SDK's ~600s default, and the contradictory unused 60s default in `AgentRequest.timeout_seconds` was removed (None defers to the configured budget)
 - **Search degradation signaling** - The librarian client parses `source_counts` (plus the additive `source_status`/`degraded` fields when a newer library-desk sends them; absence is tolerated), and `hybrid_search` appends a one-line coverage note when a search is degraded or an enabled source leg contributed nothing, so outages are visible to the model and the user

src/agents/delegation.py

@@ -128,6 +128,50 @@ def _detect_action_type(expert: str, task: str) -> ActionType:
     return ActionType.RETRIEVE
 
 
+def build_delegation_context(
+    conversation_history: list[dict] | None,
+    max_turns: int = 6,
+    max_chars_per_turn: int = 500,
+) -> str:
+    """
+    Format the most recent conversation turns as delegation context.
+
+    Experts accept a context string but the live paths never passed the
+    in-scope conversation history; this trims it to the last few turns
+    so follow-up questions ("and what about X?") keep their referent.
+
+    Args:
+        conversation_history: Prior messages as {"role", "content"} dicts
+        max_turns: How many trailing turns to include
+        max_chars_per_turn: Truncation limit per turn
+
+    Returns:
+        str: Newline-joined "role: content" lines ("" when no history)
+    """
+    if not conversation_history:
+        return ""
+
+    lines = []
+    for msg in conversation_history[-max_turns:]:
+        if not isinstance(msg, dict):
+            continue
+        role = msg.get("role", "user")
+        content = msg.get("content", "")
+        if isinstance(content, list):
+            # Tolerate structured content parts
+            content = " ".join(
+                part.get("text", "") if isinstance(part, dict) else str(part)
+                for part in content
+            )
+        content = str(content).strip()
+        if content:
+            lines.append(f"{role}: {content[:max_chars_per_turn]}")
+
+    if not lines:
+        return ""
+    return "Recent conversation:\n" + "\n".join(lines)
+
+
 def get_think_message(expert: str, task: str, phase: str) -> str:
     """
     Get the appropriate think message for an expert delegation.

src/responses/service.py

@@ -12,7 +12,7 @@ import secrets
 import time
 from collections.abc import AsyncGenerator
 
-from src.agents.delegation import get_think_message
+from src.agents.delegation import build_delegation_context, get_think_message
 from src.agents.registry import ModelRegistry
 from src.agents.steward.schemas import StewardRecommendation
 from src.core.context import current_conversation, current_user, get_default_user
@@ -63,6 +63,7 @@ async def _execute_single_delegation(
     agent_name: str,
     task: str,
     tracker: "ToolCallTracker",
+    context: str = "",
 ) -> tuple[str, str, bool]:
     """
     Execute a single delegation to an agent.
@@ -71,6 +72,7 @@ async def _execute_single_delegation(
         agent_name: Name of agent (biographer, librarian, housekeeper)
         task: Task description
         tracker: Tool call tracker
+        context: Trimmed conversation context for the expert
 
     Returns:
         tuple: (agent_name, result_summary, success). On failure the
@@ -81,21 +83,21 @@ async def _execute_single_delegation(
 
     if agent_name == "biographer":
         from src.agents.delegation import delegate_to_biographer
-        result = await delegate_to_biographer(task=task)
+        result = await delegate_to_biographer(task=task, context=context)
         duration = time.time() - start_time
         await tracker.track_call("delegate_to_biographer", duration)
         return (agent_name, result.output, result.success)
 
     elif agent_name == "librarian":
         from src.agents.delegation import delegate_to_librarian
-        result = await delegate_to_librarian(task=task)
+        result = await delegate_to_librarian(task=task, context=context)
         duration = time.time() - start_time
         await tracker.track_call("delegate_to_librarian", duration)
         return (agent_name, result.output, result.success)
 
     elif agent_name == "housekeeper":
         from src.agents.delegation import delegate_to_housekeeper
-        result = await delegate_to_housekeeper(task=task)
+        result = await delegate_to_housekeeper(task=task, context=context)
         duration = time.time() - start_time
         await tracker.track_call("delegate_to_housekeeper", duration)
         return (agent_name, result.output, result.success)
@@ -295,6 +297,7 @@ async def _direct_delegation_with_results(
     recommendation: "StewardRecommendation",
     tracker: "ToolCallTracker",
     conversation_id: str,
+    conversation_history: list | None = None,
 ) -> dict:
     """
     Directly delegate to expert agents and return structured results.
@@ -308,6 +311,7 @@ async def _direct_delegation_with_results(
         recommendation: Steward's recommendation
         tracker: Tool call tracker
         conversation_id: Conversation ID
+        conversation_history: Prior turns, trimmed into expert context
 
     Returns:
         dict: Orchestration results with expert_results, tool_outputs, etc.
@@ -320,11 +324,12 @@ async def _direct_delegation_with_results(
 
     expert_results = {}
     tools_called = []
+    context = build_delegation_context(conversation_history)
 
     for agent in recommendation.recommended_capabilities:
         try:
             agent_name, result, success = await _execute_single_delegation(
-                agent, user_message, tracker
+                agent, user_message, tracker, context=context
             )
             expert_results[agent_name] = result
             if success:
@@ -634,7 +639,11 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
         if delegation_only:
             # Direct delegation path - collect results then synthesize
             orchestration_results = await _direct_delegation_with_results(
-                effective_query, enriched.recommendation, tracker, conversation_id
+                effective_query,
+                enriched.recommendation,
+                tracker,
+                conversation_id,
+                conversation_history=conversation_history,
             )
         else:
             # Phase 1: Orchestrate tool calls

src/responses/streaming.py

@@ -189,20 +189,18 @@ class StreamingCoordinator:
             tatlock = TatlockAgent()
 
             if delegation_only:
-                # Direct delegation path with streaming think slugs
+                # Direct delegation path - think slugs stream in real time,
-                orchestration_results = await self._stream_direct_delegation(
+                # BEFORE and after each expert runs (not after the fact)
+                orchestration_results: dict = {}
+                async for event in self._stream_direct_delegation(
                     user_message=user_message,
                     recommendation=enriched.recommendation,
                     tracker=tracker,
                     conversation_id=conversation_id,
-                )
+                    conversation_history=conversation_history,
-
+                    results=orchestration_results,
-                # Stream think slugs that were collected during delegation
+                ):
-                # Each think message is complete, so we signal done after each
+                    yield event
-                for think_msg in orchestration_results.get("think_messages", []):
-                    yield ReasoningSummaryDelta(delta=think_msg)
-                    yield ReasoningSummaryDone()
-                    await asyncio.sleep(0.05)
 
             else:
                 # Phase 1: Orchestrate tool calls
@@ -272,24 +270,33 @@ class StreamingCoordinator:
         recommendation: "StewardRecommendation",  # type: ignore
         tracker: "ToolCallTracker",  # type: ignore
         conversation_id: str,
-    ) -> dict:
+        conversation_history: list | None = None,
+        results: dict | None = None,
+    ) -> AsyncGenerator[StreamEvent, None]:
         """
-        Execute direct delegation with streaming think messages.
+        Execute direct delegation, streaming think messages in real time.
 
-        Collects think messages as delegations execute for streaming to client.
+        An async generator: the "start" think message for each expert is
+        yielded BEFORE its research runs (so the user sees 'Allow me to
+        consult the archives, sir.' while waiting), and the success/error
+        message right after it finishes.
 
         Args:
             user_message: User's request
             recommendation: Steward's recommendation
             tracker: Tool call tracker
             conversation_id: Conversation ID
+            conversation_history: Prior turns, trimmed into expert context
+            results: Mutable dict populated with orchestration results
+                (expert_results, tools_called, think_messages, ...)
 
-        Returns:
+        Yields:
-            dict: Orchestration results with think_messages list
+            StreamEvent: Reasoning summary events as delegation progresses
         """
         import time as time_module
 
         from src.agents.delegation import (
+            build_delegation_context,
             delegate_to_biographer,
             delegate_to_housekeeper,
             delegate_to_librarian,
@@ -299,21 +306,30 @@ class StreamingCoordinator:
         expert_results = {}
         tools_called = []
         think_messages = []
+        context = build_delegation_context(conversation_history)
 
         for agent in recommendation.recommended_capabilities:
-            # Emit start think message
+            # Emit start think message BEFORE the expert runs
             start_msg = get_think_message(agent, user_message, "start")
             think_messages.append(start_msg + "\n")
+            yield ReasoningSummaryDelta(delta=start_msg + "\n")
+            yield ReasoningSummaryDone()
 
             start_time = time_module.time()
             try:
                 # Execute delegation
                 if agent == "librarian":
-                    result = await delegate_to_librarian(task=user_message)
+                    result = await delegate_to_librarian(
+                        task=user_message, context=context
+                    )
                 elif agent == "biographer":
-                    result = await delegate_to_biographer(task=user_message)
+                    result = await delegate_to_biographer(
+                        task=user_message, context=context
+                    )
                 elif agent == "housekeeper":
-                    result = await delegate_to_housekeeper(task=user_message)
+                    result = await delegate_to_housekeeper(
+                        task=user_message, context=context
+                    )
                 else:
                     result = None
 
@@ -324,18 +340,15 @@ class StreamingCoordinator:
                     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")
+                    phase_msg = get_think_message(agent, user_message, "success")
-                    think_messages.append(success_msg + "\n")
                 else:
                     # Failed delegations carry a curated user-safe sentence
                     # in output; exception detail is already in the logs.
-                    error_think = get_think_message(agent, user_message, "error")
+                    phase_msg = get_think_message(agent, user_message, "error")
                     if result and result.output:
                         expert_results[agent] = result.output
                     else:
-                        expert_results[agent] = error_think
+                        expert_results[agent] = phase_msg
-                    # Emit error think message
-                    think_messages.append(error_think + "\n")
 
             except Exception as e:
                 logger.error(
@@ -345,17 +358,21 @@ class StreamingCoordinator:
                     conversation_id=conversation_id,
                     exc_info=True,
                 )
-                error_think = get_think_message(agent, user_message, "error")
+                phase_msg = get_think_message(agent, user_message, "error")
-                expert_results[agent] = error_think
+                expert_results[agent] = phase_msg
-                think_messages.append(error_think + "\n")
 
-        return {
+            think_messages.append(phase_msg + "\n")
-            "tools_called": tools_called,
+            yield ReasoningSummaryDelta(delta=phase_msg + "\n")
-            "expert_results": expert_results,
+            yield ReasoningSummaryDone()
-            "tool_outputs": {},
+
-            "raw_output": "",
+        if results is not None:
-            "think_messages": think_messages,
+            results.update({
-        }
+                "tools_called": tools_called,
+                "expert_results": expert_results,
+                "tool_outputs": {},
+                "raw_output": "",
+                "think_messages": think_messages,
+            })
 
     async def stream_response(
         self,

tests/agents/test_delegation.py

@@ -15,11 +15,69 @@ from src.agents.delegation import (
     DelegationResult,
     DelegationTask,
     _detect_action_type,
+    build_delegation_context,
     delegate_to_librarian,
     get_think_message,
 )
 
 
+@pytest.mark.unit
+class TestBuildDelegationContext:
+    """Tests for trimming conversation history into expert context."""
+
+    def test_empty_history_returns_empty(self):
+        assert build_delegation_context(None) == ""
+        assert build_delegation_context([]) == ""
+
+    def test_recent_turns_are_formatted(self):
+        history = [
+            {"role": "user", "content": "Tell me about Docker"},
+            {"role": "assistant", "content": "Docker is a container runtime."},
+        ]
+
+        context = build_delegation_context(history)
+
+        assert "Recent conversation:" in context
+        assert "user: Tell me about Docker" in context
+        assert "assistant: Docker is a container runtime." in context
+
+    def test_only_last_max_turns_kept(self):
+        history = [
+            {"role": "user", "content": f"message {i}"} for i in range(10)
+        ]
+
+        context = build_delegation_context(history, max_turns=6)
+
+        assert "message 3" not in context
+        assert "message 4" in context
+        assert "message 9" in context
+
+    def test_long_turns_are_truncated(self):
+        history = [{"role": "user", "content": "x" * 2000}]
+
+        context = build_delegation_context(history, max_chars_per_turn=500)
+
+        assert "x" * 500 in context
+        assert "x" * 501 not in context
+
+    def test_structured_content_parts_tolerated(self):
+        history = [
+            {"role": "user", "content": [{"type": "text", "text": "hello there"}]}
+        ]
+
+        context = build_delegation_context(history)
+
+        assert "hello there" in context
+
+    def test_non_dict_entries_skipped(self):
+        history = ["garbage", {"role": "user", "content": "real message"}]
+
+        context = build_delegation_context(history)
+
+        assert "real message" in context
+        assert "garbage" not in context
+
+
 @pytest.mark.unit
 class TestDelegationTask:
     """Tests for the DelegationTask dataclass."""

tests/responses/test_streaming_delegation.py

@@ -0,0 +1,185 @@
+"""
+Tests for real-time think message streaming and context plumbing in
+the direct delegation paths.
+
+_stream_direct_delegation must be an async generator that yields the
+"start" think message BEFORE the expert runs (so 'Allow me to consult
+the archives, sir.' streams while research is in flight), and both
+direct delegation paths must pass trimmed conversation history as
+expert context.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from src.agents.delegation import DelegationResult
+from src.responses.streaming import (
+    ReasoningSummaryDelta,
+    ReasoningSummaryDone,
+    StreamingCoordinator,
+)
+
+HISTORY = [
+    {"role": "user", "content": "Tell me about my homelab wiki"},
+    {"role": "assistant", "content": "It documents your services, sir."},
+]
+
+
+def _librarian_result(output: str = "Findings.") -> DelegationResult:
+    return DelegationResult(
+        expert_name="librarian",
+        task="task",
+        success=True,
+        output=output,
+    )
+
+
+@pytest.mark.unit
+class TestStreamDirectDelegation:
+    """Real-time streaming behavior of _stream_direct_delegation."""
+
+    @pytest.mark.asyncio
+    async def test_start_think_streams_before_research_runs(self):
+        coordinator = StreamingCoordinator()
+        tracker = AsyncMock()
+        results: dict = {}
+
+        with patch(
+            "src.agents.delegation.delegate_to_librarian",
+            new_callable=AsyncMock,
+            return_value=_librarian_result(),
+        ) as mock_delegate:
+            gen = coordinator._stream_direct_delegation(
+                user_message="Search for Docker info",
+                recommendation=SimpleNamespace(
+                    recommended_capabilities=["librarian"]
+                ),
+                tracker=tracker,
+                conversation_id="conv_1",
+                conversation_history=HISTORY,
+                results=results,
+            )
+
+            # First event: the start think message, BEFORE any research
+            first = await gen.__anext__()
+            assert isinstance(first, ReasoningSummaryDelta)
+            assert first.delta.strip() != ""
+            assert mock_delegate.await_count == 0, (
+                "start think message must stream before the expert runs"
+            )
+
+            second = await gen.__anext__()
+            assert isinstance(second, ReasoningSummaryDone)
+            assert mock_delegate.await_count == 0
+
+            # Third event: completion message - research has now run
+            third = await gen.__anext__()
+            assert isinstance(third, ReasoningSummaryDelta)
+            assert mock_delegate.await_count == 1
+
+            remaining = [event async for event in gen]
+            assert any(isinstance(e, ReasoningSummaryDone) for e in remaining)
+
+        # Results dict is populated for Phase 2 synthesis
+        assert results["expert_results"] == {"librarian": "Findings."}
+        assert results["tools_called"] == ["delegate_to_librarian"]
+        assert len(results["think_messages"]) == 2
+
+    @pytest.mark.asyncio
+    async def test_conversation_history_passed_as_context(self):
+        coordinator = StreamingCoordinator()
+        tracker = AsyncMock()
+
+        with patch(
+            "src.agents.delegation.delegate_to_librarian",
+            new_callable=AsyncMock,
+            return_value=_librarian_result(),
+        ) as mock_delegate:
+            events = [
+                event
+                async for event in coordinator._stream_direct_delegation(
+                    user_message="And what services does it list?",
+                    recommendation=SimpleNamespace(
+                        recommended_capabilities=["librarian"]
+                    ),
+                    tracker=tracker,
+                    conversation_id="conv_1",
+                    conversation_history=HISTORY,
+                    results={},
+                )
+            ]
+
+        assert events, "generator must yield think events"
+        context = mock_delegate.await_args.kwargs["context"]
+        assert "Tell me about my homelab wiki" in context
+        assert "It documents your services, sir." in context
+
+    @pytest.mark.asyncio
+    async def test_failed_delegation_streams_error_think(self):
+        coordinator = StreamingCoordinator()
+        tracker = AsyncMock()
+        results: dict = {}
+
+        failed = DelegationResult(
+            expert_name="librarian",
+            task="task",
+            success=False,
+            output="I'm afraid the archives proved difficult to access.",
+            error="The Librarian was unable to complete the task.",
+        )
+
+        with patch(
+            "src.agents.delegation.delegate_to_librarian",
+            new_callable=AsyncMock,
+            return_value=failed,
+        ):
+            events = [
+                event
+                async for event in coordinator._stream_direct_delegation(
+                    user_message="Search for Docker info",
+                    recommendation=SimpleNamespace(
+                        recommended_capabilities=["librarian"]
+                    ),
+                    tracker=tracker,
+                    conversation_id="conv_1",
+                    results=results,
+                )
+            ]
+
+        assert results["tools_called"] == []
+        # Expert result carries the curated user-safe sentence
+        assert "archives" in results["expert_results"]["librarian"]
+        deltas = [e.delta for e in events if isinstance(e, ReasoningSummaryDelta)]
+        assert len(deltas) == 2  # start + error think messages
+
+
+@pytest.mark.unit
+class TestServiceDelegationContext:
+    """The non-streaming direct delegation path passes trimmed history."""
+
+    @pytest.mark.asyncio
+    async def test_direct_delegation_with_results_passes_context(self):
+        from src.responses.service import _direct_delegation_with_results
+
+        tracker = AsyncMock()
+
+        with patch(
+            "src.agents.delegation.delegate_to_librarian",
+            new_callable=AsyncMock,
+            return_value=_librarian_result(),
+        ) as mock_delegate:
+            results = await _direct_delegation_with_results(
+                user_message="And what services does it list?",
+                recommendation=SimpleNamespace(
+                    recommended_capabilities=["librarian"]
+                ),
+                tracker=tracker,
+                conversation_id="conv_1",
+                conversation_history=HISTORY,
+            )
+
+        context = mock_delegate.await_args.kwargs["context"]
+        assert "Tell me about my homelab wiki" in context
+        assert results["expert_results"]["librarian"] == "Findings."