jpmschweitzer/tatlock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

style: apply ruff's automatic fixes and formatter

Browse Source 
Mechanical only, and separated from the judgment calls that follow so the
reviewable changes are not buried in a 98-file whitespace diff.

227 automatic fixes: 60 blank lines carrying whitespace, 60 unsorted import
blocks, 34 Optional[X] to X | None, 28 unused imports, 16 deprecated typing
imports, 12 datetime.timezone.utc to datetime.UTC, and assorted smaller
modernisations. Then `ruff format` over src and tests: 98 files reformatted,
35 already conforming.

No file among the unused-import findings defines __all__ or is an __init__.py,
so nothing here removes a re-export.

`make test`: 658 passed, unchanged from HEAD.

Two things observed while verifying, neither addressed here:

`pytest tests/` cannot collect — tests/e2e/test_orchestration_e2e.py uses an
`e2e` marker that is not registered, and the config is strict about markers.
This fails identically at HEAD, so it predates this change; `make test` passes
because it ignores tests/e2e, tests/integration and tests/contracts.

test_tatlock_tool_call_logging_calculator is flaky. It failed once in a full run
with these changes and passed on the next, passes in isolation with them, and
fails in isolation at HEAD. It is order- or timing-dependent, not a regression
from this commit — established by running the full suite both ways rather than
by reasoning about which change could have caused it.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-08-11 17:25:18 +02:00
co-authored by Claude
parent 57fa6c13fc
commit 78066fab1b
103 changed files with 1601 additions and 1749 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/responses/__init__.py
+4 -4
View File
@@ -9,12 +9,12 @@ This module implements the OpenAI Responses API format:
"""
from src.responses.schemas import (
FunctionCallOutputItem,
MessageOutputItem,
OutputItem,
ReasoningOutputItem,
Response,
ResponseRequest,
OutputItem,
MessageOutputItem,
ReasoningOutputItem,
FunctionCallOutputItem,
)
__all__ = [
src/responses/context.py
+5 -16
View File
@@ -6,6 +6,7 @@ Handles:
- Context trimming to fit model limits
- Reserve tokens for output generation
"""
from typing import Any
@@ -59,11 +60,7 @@ class ContextWindow:
# Approximate: 4 characters per token
return total_chars // 4
async def trim_to_fit(
self,
items: list[Any],
reserve_tokens: int = 512
) -> list[Any]:
async def trim_to_fit(self, items: list[Any], reserve_tokens: int = 512) -> list[Any]:
"""
Trim items to fit within context window.
@@ -102,11 +99,7 @@ class ContextWindow:
return kept_items
async def fits_in_context(
self,
items: list[Any],
reserve_tokens: int = 512
) -> bool:
async def fits_in_context(self, items: list[Any], reserve_tokens: int = 512) -> bool:
"""
Check if items fit within context window.
@@ -121,11 +114,7 @@ class ContextWindow:
available_tokens = self.max_tokens - reserve_tokens
return total_tokens <= available_tokens
async def get_usage_stats(
self,
items: list[Any],
reserve_tokens: int = 512
) -> dict:
async def get_usage_stats(self, items: list[Any], reserve_tokens: int = 512) -> dict:
"""
Get context window usage statistics.
@@ -153,7 +142,7 @@ class ContextWindow:
"reserved_tokens": reserve_tokens,
"available_tokens": available_tokens,
"usage_percent": round(usage_percent, 2),
"fits": total_tokens <= available_tokens
"fits": total_tokens <= available_tokens,
}
# ========================================================================
src/responses/history.py
+9 -16
View File
@@ -6,8 +6,8 @@ Supports hybrid approach:
- Optional conversation_id in metadata for server-side grouping
- Server can augment with vector memories (future)
"""
import hashlib
from typing import Dict, List
from src.responses.schemas import Response, ResponseRequest
@@ -36,7 +36,7 @@ class ConversationHistory:
Args:
max_turns: Maximum number of response turns to keep per conversation
"""
self._conversations: Dict[str, List[Response]] = {}
self._conversations: dict[str, list[Response]] = {}
self._max_turns = max_turns
async def get_conversation_id(self, request: ResponseRequest) -> str:
@@ -61,11 +61,7 @@ class ConversationHistory:
first_msg = str(request.input[0]) if request.input else ""
return hashlib.sha256(first_msg.encode()).hexdigest()[:16]
async def add_response(
self,
conversation_id: str,
response: Response
) -> None:
async def add_response(self, conversation_id: str, response: Response) -> None:
"""
Add response to conversation history.
@@ -81,7 +77,7 @@ class ConversationHistory:
# Trim old turns to stay within limit
await self._trim_history(conversation_id)
async def get_history(self, conversation_id: str) -> List[Response]:
async def get_history(self, conversation_id: str) -> list[Response]:
"""
Retrieve conversation history.
@@ -125,20 +121,17 @@ class ConversationHistory:
conversation_id: Conversation identifier
"""
if len(self._conversations[conversation_id]) > self._max_turns:
self._conversations[conversation_id] = (
self._conversations[conversation_id][-self._max_turns:]
)
self._conversations[conversation_id] = self._conversations[conversation_id][
-self._max_turns :
]
# ========================================================================
# Future: Vector Memory Integration
# ========================================================================
async def get_relevant_memories(
self,
conversation_id: str,
query: str,
limit: int = 5
) -> List[dict]:
self, conversation_id: str, query: str, limit: int = 5
) -> list[dict]:
"""
Retrieve relevant memories from vector store.
src/responses/router.py
+6 -9
View File
@@ -7,10 +7,10 @@ OpenAI-compatible /v1/responses endpoint with streaming support.
from fastapi import APIRouter, HTTPException
from sse_starlette.sse import EventSourceResponse
from src.responses import service
from src.responses.schemas import ResponseRequest, Response
from src.core.exceptions import ModelNotFoundError, AppException
from src.core.exceptions import AppException, ModelNotFoundError
from src.core.logging_config import get_logger
from src.responses import service
from src.responses.schemas import Response, ResponseRequest
logger = get_logger(__name__)
@@ -58,14 +58,11 @@ async def create_response(
if use_steward:
logger.info("Streaming with Steward preprocessing for Tatlock request")
from src.responses.streaming import StreamingCoordinator
coordinator = StreamingCoordinator()
return EventSourceResponse(
coordinator.stream_response_with_steward(request)
)
return EventSourceResponse(coordinator.stream_response_with_steward(request))
else:
return EventSourceResponse(
service.create_response_stream(request)
)
return EventSourceResponse(service.create_response_stream(request))
# Non-streaming response
if use_steward:
src/responses/schemas.py
+37 -45
View File
@@ -8,18 +8,20 @@ OpenAI Responses API format with support for:
- Streaming and non-streaming modes
"""
from typing import Literal, Any
from typing import Any, Literal
from pydantic import Field, field_validator
from src.core.models import CustomBaseModel
# ============================================================================
# Output Item Schemas (appear in response.output array)
# ============================================================================
class OutputTextContent(CustomBaseModel):
"""Text content in message output."""
type: Literal["output_text"] = "output_text"
text: str
annotations: list[dict] = Field(default_factory=list)
@@ -31,6 +33,7 @@ class MessageOutputItem(CustomBaseModel):
Represents the assistant's final response message.
"""
type: Literal["message"] = "message"
id: str
role: Literal["assistant"] = "assistant"
@@ -45,6 +48,7 @@ class ReasoningOutputItem(CustomBaseModel):
Represents the model's thinking/reasoning process.
Displayed separately from the final answer.
"""
type: Literal["reasoning"] = "reasoning"
id: str
summary: list[str] # List of reasoning steps
@@ -57,6 +61,7 @@ class FunctionCallOutputItem(CustomBaseModel):
Represents a tool/function that the model wants to execute.
"""
type: Literal["function_call"] = "function_call"
id: str
name: str
@@ -73,8 +78,10 @@ OutputItem = MessageOutputItem | ReasoningOutputItem | FunctionCallOutputItem #
# Usage Tracking
# ============================================================================
class ResponseUsage(CustomBaseModel):
"""Token usage statistics for the response."""
input_tokens: int
output_tokens: int
reasoning_tokens: int = 0
@@ -85,8 +92,10 @@ class ResponseUsage(CustomBaseModel):
# Request Schema
# ============================================================================
class Tool(CustomBaseModel):
"""Tool/function definition."""
name: str
description: str
parameters: dict[str, Any]
@@ -94,6 +103,7 @@ class Tool(CustomBaseModel):
class ReasoningConfig(CustomBaseModel):
"""Reasoning configuration."""
effort: Literal["none", "minimal", "low", "medium", "high", "xhigh"] = "medium"
summary: Literal["auto", "off"] = "auto"
@@ -104,46 +114,25 @@ class ResponseRequest(CustomBaseModel):
OpenAI Responses API format with optional extensions.
"""
model: str = Field(description="Model ID to use")
input: list[dict] = Field(
description="Input messages or previous responses"
)
input: list[dict] = Field(description="Input messages or previous responses")
reasoning: dict[str, Any] | None = Field(
default=None,
description="Reasoning configuration: {effort: 'medium', summary: 'auto'}"
)
tools: list[dict] | None = Field(
default=None,
description="Available tools/functions"
default=None, description="Reasoning configuration: {effort: 'medium', summary: 'auto'}"
)
tools: list[dict] | None = Field(default=None, description="Available tools/functions")
metadata: dict[str, Any] | None = Field(
default=None,
description="Custom metadata (e.g., conversation_id for server-side tracking)"
)
stream: bool = Field(
default=False,
description="Enable streaming mode"
)
max_output_tokens: int | None = Field(
default=None,
description="Maximum tokens to generate"
)
temperature: float = Field(
default=1.0,
ge=0.0,
le=2.0,
description="Sampling temperature"
)
stop: list[str] | None = Field(
default=None,
description="Stop sequences"
default=None, description="Custom metadata (e.g., conversation_id for server-side tracking)"
)
stream: bool = Field(default=False, description="Enable streaming mode")
max_output_tokens: int | None = Field(default=None, description="Maximum tokens to generate")
temperature: float = Field(default=1.0, ge=0.0, le=2.0, description="Sampling temperature")
stop: list[str] | None = Field(default=None, description="Stop sequences")
user: str | None = Field(
default=None,
description="Unique identifier for end-user (OpenAI standard)"
default=None, description="Unique identifier for end-user (OpenAI standard)"
)
@field_validator('reasoning')
@field_validator("reasoning")
@classmethod
def validate_reasoning(cls, v: dict[str, Any] | None) -> dict[str, Any] | None:
"""
@@ -154,21 +143,21 @@ class ResponseRequest(CustomBaseModel):
- summary must be 'auto' or 'off'
"""
if v is not None:
if 'effort' in v:
allowed_efforts = ['none', 'minimal', 'low', 'medium', 'high', 'xhigh']
if v['effort'] not in allowed_efforts:
if "effort" in v:
allowed_efforts = ["none", "minimal", "low", "medium", "high", "xhigh"]
if v["effort"] not in allowed_efforts:
raise ValueError(
f"reasoning.effort must be one of {allowed_efforts}, got '{v['effort']}'"
)
if 'summary' in v:
allowed_summaries = ['auto', 'off']
if v['summary'] not in allowed_summaries:
if "summary" in v:
allowed_summaries = ["auto", "off"]
if v["summary"] not in allowed_summaries:
raise ValueError(
f"reasoning.summary must be one of {allowed_summaries}, got '{v['summary']}'"
)
return v
@field_validator('max_output_tokens')
@field_validator("max_output_tokens")
@classmethod
def validate_max_output_tokens(cls, v: int | None) -> int | None:
"""
@@ -180,7 +169,7 @@ class ResponseRequest(CustomBaseModel):
raise ValueError(f"max_output_tokens must be positive, got {v}")
return v
@field_validator('stop')
@field_validator("stop")
@classmethod
def validate_stop_sequences(cls, v: list[str] | None) -> list[str] | None:
"""
@@ -203,20 +192,20 @@ class ResponseRequest(CustomBaseModel):
# Response Schema
# ============================================================================
class Response(CustomBaseModel):
"""
Complete response object.
Contains output array with reasoning, function calls, and messages.
"""
id: str = Field(description="Unique response ID")
object: Literal["response"] = "response"
created_at: int = Field(description="Unix timestamp")
model: str = Field(description="Model used")
status: Literal["completed", "in_progress", "failed", "cancelled"]
output: list[OutputItem] = Field(
description="Output items (reasoning, function_call, message)"
)
output: list[OutputItem] = Field(description="Output items (reasoning, function_call, message)")
usage: ResponseUsage = Field(description="Token usage statistics")
@@ -224,8 +213,10 @@ class Response(CustomBaseModel):
# Error Schema
# ============================================================================
class ErrorDetail(CustomBaseModel):
"""Error detail object."""
type: str
message: str
code: int | None = None
@@ -233,4 +224,5 @@ class ErrorDetail(CustomBaseModel):
class ErrorResponse(CustomBaseModel):
"""Error response format."""
error: ErrorDetail
src/responses/service.py
+67 -64
View File
@@ -52,9 +52,9 @@ def _extract_response_preview(response: Response) -> str:
"""Extract response preview text for tracing."""
if response.output:
for item in response.output:
if hasattr(item, 'content'):
if hasattr(item, "content"):
for content in item.content:
if hasattr(content, 'text'):
if hasattr(content, "text"):
return content.text[:200]
return ""
@@ -79,10 +79,12 @@ async def _execute_single_delegation(
result summary is a curated user-safe sentence.
"""
import time
start_time = time.time()
if agent_name == "biographer":
from src.agents.delegation import delegate_to_biographer
result = await delegate_to_biographer(task=task, context=context)
duration = time.time() - start_time
await tracker.track_call("delegate_to_biographer", duration)
@@ -90,6 +92,7 @@ async def _execute_single_delegation(
elif agent_name == "librarian":
from src.agents.delegation import delegate_to_librarian
result = await delegate_to_librarian(task=task, context=context)
duration = time.time() - start_time
await tracker.track_call("delegate_to_librarian", duration)
@@ -97,6 +100,7 @@ async def _execute_single_delegation(
elif agent_name == "housekeeper":
from src.agents.delegation import delegate_to_housekeeper
result = await delegate_to_housekeeper(task=task, context=context)
duration = time.time() - start_time
await tracker.track_call("delegate_to_housekeeper", duration)
@@ -107,9 +111,7 @@ async def _execute_single_delegation(
async def _handle_text_delegation(
response: str,
tracker: "ToolCallTracker",
conversation_id: str
response: str, tracker: "ToolCallTracker", conversation_id: str
) -> str:
"""
Handle text-based delegation fallback.
@@ -173,8 +175,7 @@ async def _handle_text_delegation(
conversation_id=conversation_id,
)
tasks = [
_execute_single_delegation(agent.lower(), task, tracker)
for agent, task in matches
_execute_single_delegation(agent.lower(), task, tracker) for agent, task in matches
]
results = await asyncio.gather(*tasks, return_exceptions=True)
@@ -188,10 +189,7 @@ async def _handle_text_delegation(
got=len(results),
conversation_id=conversation_id,
)
return (
"I apologize, sir. I was unable to complete the "
"requested delegations."
)
return "I apologize, sir. I was unable to complete the " "requested delegations."
# Combine results (failures carry curated user-safe sentences)
summaries = []
@@ -205,8 +203,7 @@ async def _handle_text_delegation(
conversation_id=conversation_id,
)
summaries.append(
f"**{agent_name}**: "
f"{get_think_message(agent_name, task, 'error')}"
f"**{agent_name}**: " f"{get_think_message(agent_name, task, 'error')}"
)
else:
_, output, _ = item
@@ -226,9 +223,7 @@ async def _handle_text_delegation(
conversation_id=conversation_id,
)
try:
_, output, _ = await _execute_single_delegation(
agent_name, task, tracker
)
_, output, _ = await _execute_single_delegation(agent_name, task, tracker)
summaries.append(output)
except Exception as e:
logger.error(
@@ -421,7 +416,7 @@ def _calculate_usage(input_messages: list[dict], output_items: list) -> Response
elif isinstance(item, FunctionCallOutputItem):
func_text = item.arguments
output_tokens += len(func_text) // 4
elif hasattr(item, 'type'):
elif hasattr(item, "type"):
# Agent OutputItem objects (backward compatibility)
if item.type == "reasoning":
reasoning_text = " ".join(item.data.get("summary", []))
@@ -439,7 +434,7 @@ def _calculate_usage(input_messages: list[dict], output_items: list) -> Response
input_tokens=input_tokens,
output_tokens=output_tokens,
reasoning_tokens=reasoning_tokens,
total_tokens=total_tokens
total_tokens=total_tokens,
)
@@ -527,7 +522,7 @@ async def create_response(request: ResponseRequest) -> Response:
model=request.model,
status="completed",
output=converted_items,
usage=usage
usage=usage,
)
# Track conversation history (for analytics and future vector memory)
@@ -640,12 +635,15 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
# 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 delegation_agents
for cap in enriched.recommendation.recommended_capabilities
) and enriched.recommendation.recommended_capabilities
delegation_only = (
all(
cap in delegation_agents for cap in enriched.recommendation.recommended_capabilities
)
and enriched.recommendation.recommended_capabilities
)
from src.agents.tatlock import TatlockAgent
tatlock = TatlockAgent()
# Use enriched query (with location/timezone context) if available
@@ -677,7 +675,9 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
)
# Add text delegation results to expert_results
if text_delegation_results != orchestration_results["raw_output"]:
orchestration_results["expert_results"]["text_delegation"] = text_delegation_results
orchestration_results["expert_results"]["text_delegation"] = (
text_delegation_results
)
# Phase 2: Synthesize butler-toned response from all results
tatlock_response = await tatlock.synthesize_from_results(
@@ -694,23 +694,25 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
# Add Steward reasoning as reasoning output
if enriched.steward_reasoning:
output_items.append(ReasoningOutputItem(
id=f"rs_{generate_id()}",
summary=[enriched.steward_reasoning],
status="completed"
))
output_items.append(
ReasoningOutputItem(
id=f"rs_{generate_id()}",
summary=[enriched.steward_reasoning],
status="completed",
)
)
# Add Tatlock's message
output_items.append(MessageOutputItem(
id=f"msg_{generate_id()}",
role="assistant",
content=[OutputTextContent(
type="output_text",
text=tatlock_response,
annotations=[]
)],
status="completed"
))
output_items.append(
MessageOutputItem(
id=f"msg_{generate_id()}",
role="assistant",
content=[
OutputTextContent(type="output_text", text=tatlock_response, annotations=[])
],
status="completed",
)
)
# Calculate usage (approximate)
usage = _calculate_usage(request.input, output_items)
@@ -721,7 +723,7 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
model=request.model,
status="completed",
output=output_items,
usage=usage
usage=usage,
)
# Track conversation history
@@ -752,9 +754,7 @@ async def create_response_with_steward(request: ResponseRequest) -> Response:
raise
async def create_response_stream(
request: ResponseRequest
) -> AsyncGenerator[dict, None]:
async def create_response_stream(request: ResponseRequest) -> AsyncGenerator[dict, None]:
"""
Create streaming response.
@@ -774,10 +774,7 @@ async def create_response_stream(
coordinator = StreamingCoordinator()
async for event in coordinator.stream_response(request):
yield {
"event": event.event,
"data": event.model_dump_json()
}
yield {"event": event.event, "data": event.model_dump_json()}
async def get_conversation_history(conversation_id: str) -> list[Response]:
@@ -802,7 +799,7 @@ async def get_conversation_stats() -> dict:
"""
return {
"total_conversations": await _conversation_history.get_conversation_count(),
"max_turns_per_conversation": _conversation_history._max_turns
"max_turns_per_conversation": _conversation_history._max_turns,
}
@@ -843,23 +840,29 @@ def _convert_output_items(items: list) -> list:
for item in items:
if item.type == "message":
converted.append(MessageOutputItem(
id=item.id,
content=[OutputTextContent(**c) for c in item.data["content"]],
status=item.data.get("status", "completed")
))
converted.append(
MessageOutputItem(
id=item.id,
content=[OutputTextContent(**c) for c in item.data["content"]],
status=item.data.get("status", "completed"),
)
)
elif item.type == "reasoning":
converted.append(ReasoningOutputItem(
id=item.id,
summary=item.data["summary"],
status=item.data.get("status", "completed")
))
converted.append(
ReasoningOutputItem(
id=item.id,
summary=item.data["summary"],
status=item.data.get("status", "completed"),
)
)
elif item.type == "function_call":
converted.append(FunctionCallOutputItem(
id=item.id,
name=item.data["name"],
arguments=item.data["arguments"],
status=item.data.get("status", "completed")
))
converted.append(
FunctionCallOutputItem(
id=item.id,
name=item.data["name"],
arguments=item.data["arguments"],
status=item.data.get("status", "completed"),
)
)
return converted
src/responses/streaming.py
+83 -83
View File
@@ -30,8 +30,10 @@ logger = get_logger(__name__)
# Stream Event Types
# ============================================================================
class StreamEventType(str, Enum):
"""Streaming event types for Responses API."""
REASONING_SUMMARY_DELTA = "response.reasoning_summary_text.delta"
REASONING_SUMMARY_DONE = "response.reasoning_summary_text.done"
OUTPUT_TEXT_DELTA = "response.output_text.delta"
@@ -46,30 +48,38 @@ class StreamEventType(str, Enum):
# Stream Event Schemas
# ============================================================================
class ReasoningSummaryDelta(CustomBaseModel):
"""Reasoning summary text delta event."""
event: Literal[StreamEventType.REASONING_SUMMARY_DELTA] = StreamEventType.REASONING_SUMMARY_DELTA
event: Literal[StreamEventType.REASONING_SUMMARY_DELTA] = (
StreamEventType.REASONING_SUMMARY_DELTA
)
delta: str
class ReasoningSummaryDone(CustomBaseModel):
"""Reasoning summary completion event."""
event: Literal[StreamEventType.REASONING_SUMMARY_DONE] = StreamEventType.REASONING_SUMMARY_DONE
class OutputTextDelta(CustomBaseModel):
"""Output text delta event."""
event: Literal[StreamEventType.OUTPUT_TEXT_DELTA] = StreamEventType.OUTPUT_TEXT_DELTA
delta: str
class OutputTextDone(CustomBaseModel):
"""Output text completion event."""
event: Literal[StreamEventType.OUTPUT_TEXT_DONE] = StreamEventType.OUTPUT_TEXT_DONE
class FunctionCallDelta(CustomBaseModel):
"""Function call arguments delta event."""
event: Literal[StreamEventType.FUNCTION_CALL_DELTA] = StreamEventType.FUNCTION_CALL_DELTA
delta: str
name: str | None = None # Only in first chunk
@@ -77,31 +87,34 @@ class FunctionCallDelta(CustomBaseModel):
class FunctionCallDone(CustomBaseModel):
"""Function call completion event."""
event: Literal[StreamEventType.FUNCTION_CALL_DONE] = StreamEventType.FUNCTION_CALL_DONE
class ResponseDone(CustomBaseModel):
"""Response completion event with full response."""
event: Literal[StreamEventType.RESPONSE_DONE] = StreamEventType.RESPONSE_DONE
response: Response
class ErrorEvent(CustomBaseModel):
"""Error event."""
event: Literal[StreamEventType.ERROR] = StreamEventType.ERROR
error: dict
# Union type for all stream events
StreamEvent = (
ReasoningSummaryDelta |
ReasoningSummaryDone |
OutputTextDelta |
OutputTextDone |
FunctionCallDelta |
FunctionCallDone |
ResponseDone |
ErrorEvent
ReasoningSummaryDelta
| ReasoningSummaryDone
| OutputTextDelta
| OutputTextDone
| FunctionCallDelta
| FunctionCallDone
| ResponseDone
| ErrorEvent
)
@@ -109,6 +122,7 @@ StreamEvent = (
# Streaming Coordinator
# ============================================================================
class StreamingCoordinator:
"""
Coordinates streaming from agents to SSE format.
@@ -123,7 +137,7 @@ class StreamingCoordinator:
async def stream_response_with_steward(
self,
request: "ResponseRequest" # type: ignore # Forward reference
request: "ResponseRequest", # type: ignore # Forward reference
) -> AsyncGenerator[StreamEvent, None]:
"""
Stream response with Steward preprocessing and two-phase Tatlock execution.
@@ -181,10 +195,13 @@ class StreamingCoordinator:
# 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
delegation_only = (
all(
cap in delegation_agents
for cap in enriched.recommendation.recommended_capabilities
)
and enriched.recommendation.recommended_capabilities
)
tatlock = TatlockAgent()
@@ -222,7 +239,7 @@ class StreamingCoordinator:
# 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])
yield OutputTextDelta(delta=tatlock_response[i : i + chunk_size])
await asyncio.sleep(0.02)
yield OutputTextDone()
@@ -231,12 +248,10 @@ class StreamingCoordinator:
message_item = MessageOutputItem(
id=f"msg_{generate_id()}",
role="assistant",
content=[OutputTextContent(
type="output_text",
text=tatlock_response,
annotations=[]
)],
status="completed"
content=[
OutputTextContent(type="output_text", text=tatlock_response, annotations=[])
],
status="completed",
)
output_items.append(message_item)
@@ -252,7 +267,7 @@ class StreamingCoordinator:
model=request.model,
status="completed",
output=output_items,
usage=usage
usage=usage,
)
# Track conversation history
@@ -319,17 +334,11 @@ class StreamingCoordinator:
try:
# Execute delegation
if agent == "librarian":
result = await delegate_to_librarian(
task=user_message, context=context
)
result = await delegate_to_librarian(task=user_message, context=context)
elif agent == "biographer":
result = await delegate_to_biographer(
task=user_message, context=context
)
result = await delegate_to_biographer(task=user_message, context=context)
elif agent == "housekeeper":
result = await delegate_to_housekeeper(
task=user_message, context=context
)
result = await delegate_to_housekeeper(task=user_message, context=context)
else:
result = None
@@ -366,17 +375,19 @@ class StreamingCoordinator:
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,
})
results.update(
{
"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
request: "ResponseRequest", # type: ignore # Forward reference
) -> AsyncGenerator[StreamEvent, None]:
"""
Coordinate streaming from agent to SSE events.
@@ -435,18 +446,13 @@ class StreamingCoordinator:
elif item.type == "function_call":
# Stream function call arguments
# First chunk includes name
yield FunctionCallDelta(
name=item.data["name"],
delta=""
)
yield FunctionCallDelta(name=item.data["name"], delta="")
# Stream arguments in chunks
args = item.data["arguments"]
chunk_size = 20
for i in range(0, len(args), chunk_size):
yield FunctionCallDelta(
delta=args[i:i+chunk_size]
)
yield FunctionCallDelta(delta=args[i : i + chunk_size])
await asyncio.sleep(0.03)
yield FunctionCallDone()
@@ -458,7 +464,7 @@ class StreamingCoordinator:
# Only stream the NEW text (delta) since last update
if current_text.startswith(last_message_text):
# Extract only the new portion
delta_text = current_text[len(last_message_text):]
delta_text = current_text[len(last_message_text) :]
if delta_text:
# Stream the delta text in chunks while preserving formatting
@@ -466,17 +472,16 @@ class StreamingCoordinator:
chunk_size = 50 # characters per chunk
for i in range(0, len(delta_text), chunk_size):
chunk = delta_text[i:i+chunk_size]
chunk = delta_text[i : i + chunk_size]
# Check stop sequences on full accumulated text
stop_found, text_before_stop = self._check_stop_sequence(
current_text,
request.stop
current_text, request.stop
)
if stop_found:
# Only emit remaining delta before stop
remaining = text_before_stop[len(last_message_text):]
remaining = text_before_stop[len(last_message_text) :]
if remaining:
yield OutputTextDelta(delta=remaining)
yield OutputTextDone()
@@ -508,11 +513,12 @@ class StreamingCoordinator:
model=request.model,
status="completed",
output=self._convert_output_items(output_items),
usage=usage
usage=usage,
)
# Track conversation history (import here to avoid circular dependency)
from src.responses.service import _conversation_history
conversation_id = await _conversation_history.get_conversation_id(request)
await _conversation_history.add_response(conversation_id, final_response)
@@ -534,24 +540,30 @@ class StreamingCoordinator:
converted = []
for item in items:
if item.type == "message":
converted.append(MessageOutputItem(
id=item.id,
content=[OutputTextContent(**c) for c in item.data["content"]],
status=item.data.get("status", "completed")
))
converted.append(
MessageOutputItem(
id=item.id,
content=[OutputTextContent(**c) for c in item.data["content"]],
status=item.data.get("status", "completed"),
)
)
elif item.type == "reasoning":
converted.append(ReasoningOutputItem(
id=item.id,
summary=item.data["summary"],
status=item.data.get("status", "completed")
))
converted.append(
ReasoningOutputItem(
id=item.id,
summary=item.data["summary"],
status=item.data.get("status", "completed"),
)
)
elif item.type == "function_call":
converted.append(FunctionCallOutputItem(
id=item.id,
name=item.data["name"],
arguments=item.data["arguments"],
status=item.data.get("status", "completed")
))
converted.append(
FunctionCallOutputItem(
id=item.id,
name=item.data["name"],
arguments=item.data["arguments"],
status=item.data.get("status", "completed"),
)
)
return converted
@@ -576,18 +588,10 @@ class StreamingCoordinator:
error_type = "internal_error"
code = 500
return ErrorEvent(
error={
"type": error_type,
"message": str(error),
"code": code
}
)
return ErrorEvent(error={"type": error_type, "message": str(error), "code": code})
def _check_stop_sequence(
self,
accumulated_text: str,
stop_sequences: list[str] | None
self, accumulated_text: str, stop_sequences: list[str] | None
) -> tuple[bool, str]:
"""
Check if any stop sequence is encountered.
@@ -624,11 +628,7 @@ class StreamingCoordinator:
"""
return len(text) // 4
def _check_max_tokens(
self,
current_tokens: int,
max_tokens: int | None
) -> bool:
def _check_max_tokens(self, current_tokens: int, max_tokens: int | None) -> bool:
"""
Check if max tokens limit reached.