diff --git a/CHANGELOG.md b/CHANGELOG.md index 0442070..72c75fd 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/agents/delegation.py b/src/agents/delegation.py index 8bd82b0..c538839 100644 --- a/src/agents/delegation.py +++ b/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. diff --git a/src/responses/service.py b/src/responses/service.py index 59c4621..00c46f7 100644 --- a/src/responses/service.py +++ b/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 diff --git a/src/responses/streaming.py b/src/responses/streaming.py index 82791eb..0db4ec8 100644 --- a/src/responses/streaming.py +++ b/src/responses/streaming.py @@ -189,20 +189,18 @@ class StreamingCoordinator: tatlock = TatlockAgent() if delegation_only: - # Direct delegation path with streaming think slugs - orchestration_results = await self._stream_direct_delegation( + # Direct delegation path - think slugs stream in real time, + # 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, - ) - - # Stream think slugs that were collected during delegation - # Each think message is complete, so we signal done after each - for think_msg in orchestration_results.get("think_messages", []): - yield ReasoningSummaryDelta(delta=think_msg) - yield ReasoningSummaryDone() - await asyncio.sleep(0.05) + conversation_history=conversation_history, + results=orchestration_results, + ): + yield event 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: - dict: Orchestration results with think_messages list + Yields: + 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") - think_messages.append(success_msg + "\n") + phase_msg = get_think_message(agent, user_message, "success") 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 - # Emit error think message - think_messages.append(error_think + "\n") + expert_results[agent] = phase_msg 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") - expert_results[agent] = error_think - think_messages.append(error_think + "\n") + phase_msg = get_think_message(agent, user_message, "error") + expert_results[agent] = phase_msg - return { - "tools_called": tools_called, - "expert_results": expert_results, - "tool_outputs": {}, - "raw_output": "", - "think_messages": think_messages, - } + think_messages.append(phase_msg + "\n") + yield ReasoningSummaryDelta(delta=phase_msg + "\n") + yield ReasoningSummaryDone() + + if results is not None: + results.update({ + "tools_called": tools_called, + "expert_results": expert_results, + "tool_outputs": {}, + "raw_output": "", + "think_messages": think_messages, + }) async def stream_response( self, diff --git a/tests/agents/test_delegation.py b/tests/agents/test_delegation.py index a8a1703..3e1c4eb 100644 --- a/tests/agents/test_delegation.py +++ b/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.""" diff --git a/tests/responses/test_streaming_delegation.py b/tests/responses/test_streaming_delegation.py new file mode 100644 index 0000000..111536a --- /dev/null +++ b/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."