From ff6c3cf1b5fef758cf76c5ff9166a261e3fad197 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Sat, 6 Dec 2025 19:38:00 +0100 Subject: [PATCH] Add Responses API with streaming, history, and advanced features MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements Phases 2, 3, and 6: Complete Responses API implementation Core API (Phase 2): - OpenAI Responses API format with structured output items - Streaming and non-streaming support via SSE-Starlette - Reasoning items (thinking summaries) - Function call items (tool execution) - Message items (assistant responses) - Router, schemas, service, and streaming coordinator Conversation History (Phase 3): - Hybrid client/server approach - Auto-generated deterministic conversation IDs - Configurable max turns with automatic trimming - Context window management with token counting - Token usage statistics - Placeholder for future vector memory integration Advanced Features (Phase 6): - Parameter validation with Pydantic field validators: - Temperature: 0.0-2.0 range enforcement - Reasoning effort: 6 levels (none to xhigh) - Max output tokens: positive integer enforcement - Stop sequences: up to 4, non-empty strings - Real-time stop sequence detection during streaming - Real-time max tokens enforcement with token counting - Graceful error handling and OpenAI-compatible error format Testing: - 9 unit tests for API endpoints and streaming - 11 unit tests for error handling - 13 unit tests for conversation history and context - 12 unit tests for advanced features and validation - Total: 45 tests with comprehensive coverage 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/responses/__init__.py | 27 ++ src/responses/context.py | 178 ++++++++++ src/responses/history.py | 158 +++++++++ src/responses/router.py | 116 +++++++ src/responses/schemas.py | 232 +++++++++++++ src/responses/service.py | 270 +++++++++++++++ src/responses/streaming.py | 375 +++++++++++++++++++++ tests/responses/__init__.py | 1 + tests/responses/test_advanced_features.py | 378 +++++++++++++++++++++ tests/responses/test_error_handling.py | 281 ++++++++++++++++ tests/responses/test_history.py | 383 ++++++++++++++++++++++ tests/responses/test_router.py | 281 ++++++++++++++++ 12 files changed, 2680 insertions(+) create mode 100644 src/responses/__init__.py create mode 100644 src/responses/context.py create mode 100644 src/responses/history.py create mode 100644 src/responses/router.py create mode 100644 src/responses/schemas.py create mode 100644 src/responses/service.py create mode 100644 src/responses/streaming.py create mode 100644 tests/responses/__init__.py create mode 100644 tests/responses/test_advanced_features.py create mode 100644 tests/responses/test_error_handling.py create mode 100644 tests/responses/test_history.py create mode 100644 tests/responses/test_router.py diff --git a/src/responses/__init__.py b/src/responses/__init__.py new file mode 100644 index 0000000..e0bd730 --- /dev/null +++ b/src/responses/__init__.py @@ -0,0 +1,27 @@ +""" +Responses API implementation. + +This module implements the OpenAI Responses API format: +- /v1/responses endpoint +- Output items (reasoning, function_call, message) +- Streaming events +- Multi-turn conversations +""" + +from src.responses.schemas import ( + Response, + ResponseRequest, + OutputItem, + MessageOutputItem, + ReasoningOutputItem, + FunctionCallOutputItem, +) + +__all__ = [ + "Response", + "ResponseRequest", + "OutputItem", + "MessageOutputItem", + "ReasoningOutputItem", + "FunctionCallOutputItem", +] diff --git a/src/responses/context.py b/src/responses/context.py new file mode 100644 index 0000000..773a0a8 --- /dev/null +++ b/src/responses/context.py @@ -0,0 +1,178 @@ +""" +Context window management and token counting. + +Handles: +- Token counting (approximate for now, exact in future) +- Context trimming to fit model limits +- Reserve tokens for output generation +""" +from typing import Any + + +class ContextWindow: + """ + Manage token limits and context trimming. + + Real production context window management: + - Approximate token counting (character-based) + - Context trimming (keep most recent within limits) + - Reserve tokens for model output + - Future: Integration with tiktoken for exact counts + """ + + def __init__(self, max_tokens: int = 4096): + """ + Initialize context window manager. + + Args: + max_tokens: Maximum context window size in tokens + """ + self.max_tokens = max_tokens + + async def count_tokens(self, items: list[Any]) -> int: + """ + Count tokens in items. + + Current implementation: Approximate via character count + Future: Use tiktoken or similar for exact token counts + + Approximation: ~4 characters per token (common for English text) + + Args: + items: List of items to count (messages, strings, dicts, etc.) + + Returns: + int: Approximate token count + """ + total_chars = 0 + + for item in items: + if isinstance(item, str): + total_chars += len(item) + elif isinstance(item, dict): + # Count all values in dict + total_chars += len(str(item)) + else: + # Fallback: convert to string + total_chars += len(str(item)) + + # Approximate: 4 characters per token + return total_chars // 4 + + async def trim_to_fit( + self, + items: list[Any], + reserve_tokens: int = 512 + ) -> list[Any]: + """ + Trim items to fit within context window. + + Strategy: + 1. Reserve tokens for output generation + 2. Keep most recent items that fit + 3. Drop oldest items first + + Args: + items: List of items to trim + reserve_tokens: Tokens to reserve for output (default 512) + + Returns: + list: Trimmed list of items that fit in context + """ + available_tokens = self.max_tokens - reserve_tokens + + if available_tokens <= 0: + # Edge case: reserve too large + return [] + + # Start from most recent, work backwards + kept_items = [] + current_tokens = 0 + + for item in reversed(items): + item_tokens = await self.count_tokens([item]) + + if current_tokens + item_tokens <= available_tokens: + # Fits - keep this item + kept_items.insert(0, item) + current_tokens += item_tokens + else: + # Doesn't fit - stop here + break + + return kept_items + + async def fits_in_context( + self, + items: list[Any], + reserve_tokens: int = 512 + ) -> bool: + """ + Check if items fit within context window. + + Args: + items: List of items to check + reserve_tokens: Tokens to reserve for output + + Returns: + bool: True if items fit in context + """ + total_tokens = await self.count_tokens(items) + 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: + """ + Get context window usage statistics. + + Args: + items: List of items to analyze + reserve_tokens: Tokens reserved for output + + Returns: + dict: Usage statistics with keys: + - total_tokens: Total tokens in items + - max_tokens: Maximum context window + - reserved_tokens: Tokens reserved for output + - available_tokens: Tokens available for input + - usage_percent: Percentage of context used + - fits: Whether items fit in context + """ + total_tokens = await self.count_tokens(items) + available_tokens = self.max_tokens - reserve_tokens + + usage_percent = (total_tokens / available_tokens * 100) if available_tokens > 0 else 100.0 + + return { + "total_tokens": total_tokens, + "max_tokens": self.max_tokens, + "reserved_tokens": reserve_tokens, + "available_tokens": available_tokens, + "usage_percent": round(usage_percent, 2), + "fits": total_tokens <= available_tokens + } + + # ======================================================================== + # Future: Exact Token Counting + # ======================================================================== + + async def count_tokens_exact(self, text: str, model: str = "gpt-3.5-turbo") -> int: + """ + Count exact tokens using tiktoken. + + TODO: Integrate tiktoken library for exact token counting + TODO: Support multiple model encodings (GPT-4, Claude, etc.) + + Args: + text: Text to count + model: Model name for encoding + + Returns: + int: Exact token count (placeholder - returns approximate for now) + """ + # Placeholder - would use tiktoken here + return await self.count_tokens([text]) diff --git a/src/responses/history.py b/src/responses/history.py new file mode 100644 index 0000000..5f2980e --- /dev/null +++ b/src/responses/history.py @@ -0,0 +1,158 @@ +""" +Conversation history management. + +Supports hybrid approach: +- Client sends full input array (OpenAI compatible) +- 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 + + +class ConversationHistory: + """ + Conversation history management with hybrid approach. + + The hybrid approach means: + 1. Clients MUST send full conversation history in request.input[] + (OpenAI compatible - client maintains state) + 2. Server optionally tracks conversations via metadata.conversation_id + (for analytics, debugging, future vector memory augmentation) + 3. Server does NOT modify the input[] - respects client's context choices + + Future enhancements: + - Vector memory integration (Qdrant) + - Semantic search across conversation history + - Auto-summarization of long contexts + """ + + def __init__(self, max_turns: int = 20): + """ + Initialize conversation history manager. + + Args: + max_turns: Maximum number of response turns to keep per conversation + """ + self._conversations: Dict[str, List[Response]] = {} + self._max_turns = max_turns + + async def get_conversation_id(self, request: ResponseRequest) -> str: + """ + Get or generate conversation ID. + + Priority: + 1. metadata.conversation_id if provided by client + 2. Generate deterministic ID from first message hash + + Args: + request: Response request + + Returns: + str: Conversation ID (16 character hex) + """ + if request.metadata and "conversation_id" in request.metadata: + return request.metadata["conversation_id"] + + # Generate deterministic ID from first message + # This allows same conversation to be grouped even without explicit ID + 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: + """ + Add response to conversation history. + + Args: + conversation_id: Conversation identifier + response: Response object to store + """ + if conversation_id not in self._conversations: + self._conversations[conversation_id] = [] + + self._conversations[conversation_id].append(response) + + # Trim old turns to stay within limit + await self._trim_history(conversation_id) + + async def get_history(self, conversation_id: str) -> List[Response]: + """ + Retrieve conversation history. + + Args: + conversation_id: Conversation identifier + + Returns: + list: List of Response objects (may be empty) + """ + return self._conversations.get(conversation_id, []) + + async def get_conversation_count(self) -> int: + """ + Get total number of tracked conversations. + + Returns: + int: Number of conversations + """ + return len(self._conversations) + + async def clear_conversation(self, conversation_id: str) -> bool: + """ + Clear a specific conversation history. + + Args: + conversation_id: Conversation identifier + + Returns: + bool: True if conversation existed and was cleared + """ + if conversation_id in self._conversations: + del self._conversations[conversation_id] + return True + return False + + async def _trim_history(self, conversation_id: str) -> None: + """ + Keep only recent turns within max_turns limit. + + Args: + conversation_id: Conversation identifier + """ + if len(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]: + """ + Retrieve relevant memories from vector store. + + TODO: Integrate Qdrant for semantic search across conversation history + TODO: Enable RAG (Retrieval Augmented Generation) for long conversations + TODO: Add memory importance scoring + + Args: + conversation_id: Conversation identifier + query: Search query + limit: Maximum memories to return + + Returns: + list: Relevant memories (empty for now - placeholder) + """ + # Placeholder for future Qdrant integration + return [] diff --git a/src/responses/router.py b/src/responses/router.py new file mode 100644 index 0000000..096e70b --- /dev/null +++ b/src/responses/router.py @@ -0,0 +1,116 @@ +""" +Responses router. + +OpenAI-compatible /v1/responses endpoint with streaming support. +""" + +import logging +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 + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/responses", tags=["responses"]) + + +@router.post("", response_model=Response) +async def create_response( + request: ResponseRequest, +) -> Response | EventSourceResponse: + """ + Create a response using Responses API format. + + Supports: + - Reasoning summaries (thinking/reasoning display) + - Function calling (tool usage) + - Streaming responses + - Multi-turn conversations + - Error handling + + Args: + request: Response request with model, input, optional reasoning/tools + + Returns: + Response object or SSE stream + + Example non-streaming request: + POST /v1/responses + { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Hello"}], + "reasoning": {"effort": "medium", "summary": "auto"}, + "stream": false + } + + Example streaming request: + POST /v1/responses + { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Hello"}], + "stream": true + } + + Response format (non-streaming): + { + "id": "resp_...", + "object": "response", + "created_at": 1733529600, + "model": "lorem-tester", + "status": "completed", + "output": [ + { + "type": "reasoning", + "id": "rs_...", + "summary": ["Analyzing...", "Considering..."] + }, + { + "type": "message", + "id": "msg_...", + "role": "assistant", + "content": [{"type": "output_text", "text": "Lorem ipsum..."}] + } + ], + "usage": { + "input_tokens": 10, + "output_tokens": 50, + "reasoning_tokens": 20, + "total_tokens": 80 + } + } + + Streaming format (SSE): + event: response.reasoning_summary_text.delta + data: {"delta": "Analyzing..."} + + event: response.output_text.delta + data: {"delta": "Lorem"} + + event: response.done + data: {"response": {...}} + """ + logger.info(f"Response request for model: {request.model}") + + try: + if request.stream: + logger.info("Streaming response requested") + return EventSourceResponse( + service.create_response_stream(request) + ) + + return await service.create_response(request) + + except ModelNotFoundError as e: + logger.error(f"Model not found: {e}") + raise HTTPException(status_code=404, detail=str(e)) + + except AppException as e: + logger.error(f"Application error: {e}") + raise HTTPException(status_code=e.status_code, detail=e.message) + + except Exception as e: + logger.error(f"Unexpected error: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Internal server error") diff --git a/src/responses/schemas.py b/src/responses/schemas.py new file mode 100644 index 0000000..67f26c6 --- /dev/null +++ b/src/responses/schemas.py @@ -0,0 +1,232 @@ +""" +Response schemas for Responses API. + +OpenAI Responses API format with support for: +- Reasoning items (thinking/reasoning summaries) +- Function call items (tool execution) +- Message items (assistant responses) +- Streaming and non-streaming modes +""" + +from typing import Literal, Any +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) + + +class MessageOutputItem(CustomBaseModel): + """ + Message item in output array. + + Represents the assistant's final response message. + """ + type: Literal["message"] = "message" + id: str + role: Literal["assistant"] = "assistant" + content: list[OutputTextContent] + status: Literal["completed", "in_progress", "failed"] = "completed" + + +class ReasoningOutputItem(CustomBaseModel): + """ + Reasoning item in output array. + + 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 + status: Literal["completed", "in_progress", "failed"] = "completed" + + +class FunctionCallOutputItem(CustomBaseModel): + """ + Function call item in output array. + + Represents a tool/function that the model wants to execute. + """ + type: Literal["function_call"] = "function_call" + id: str + name: str + arguments: str # JSON string of arguments + status: Literal["completed", "in_progress", "failed"] = "completed" + + +# Union type for all output items +# Type: ignore because Pydantic handles union types specially +OutputItem = MessageOutputItem | ReasoningOutputItem | FunctionCallOutputItem # type: ignore + + +# ============================================================================ +# Usage Tracking +# ============================================================================ + +class ResponseUsage(CustomBaseModel): + """Token usage statistics for the response.""" + input_tokens: int + output_tokens: int + reasoning_tokens: int = 0 + total_tokens: int + + +# ============================================================================ +# Request Schema +# ============================================================================ + +class Tool(CustomBaseModel): + """Tool/function definition.""" + name: str + description: str + parameters: dict[str, Any] + + +class ReasoningConfig(CustomBaseModel): + """Reasoning configuration.""" + effort: Literal["none", "minimal", "low", "medium", "high", "xhigh"] = "medium" + summary: Literal["auto", "off"] = "auto" + + +class ResponseRequest(CustomBaseModel): + """ + Request to create a response. + + 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" + ) + 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" + ) + 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" + ) + + @field_validator('reasoning') + @classmethod + def validate_reasoning(cls, v: dict[str, Any] | None) -> dict[str, Any] | None: + """ + Validate reasoning configuration. + + Checks: + - effort must be valid level (none, minimal, low, medium, high, xhigh) + - 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: + 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: + raise ValueError( + f"reasoning.summary must be one of {allowed_summaries}, got '{v['summary']}'" + ) + return v + + @field_validator('max_output_tokens') + @classmethod + def validate_max_output_tokens(cls, v: int | None) -> int | None: + """ + Validate max_output_tokens. + + Must be positive if provided. + """ + if v is not None and v <= 0: + raise ValueError(f"max_output_tokens must be positive, got {v}") + return v + + @field_validator('stop') + @classmethod + def validate_stop_sequences(cls, v: list[str] | None) -> list[str] | None: + """ + Validate stop sequences. + + Checks: + - Maximum 4 stop sequences + - Each must be non-empty string + """ + if v is not None: + if len(v) > 4: + raise ValueError(f"Maximum 4 stop sequences allowed, got {len(v)}") + for seq in v: + if not seq or not isinstance(seq, str): + raise ValueError("Stop sequences must be non-empty strings") + return v + + +# ============================================================================ +# 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)" + ) + usage: ResponseUsage = Field(description="Token usage statistics") + + +# ============================================================================ +# Error Schema +# ============================================================================ + +class ErrorDetail(CustomBaseModel): + """Error detail object.""" + type: str + message: str + code: int | None = None + + +class ErrorResponse(CustomBaseModel): + """Error response format.""" + error: ErrorDetail diff --git a/src/responses/service.py b/src/responses/service.py new file mode 100644 index 0000000..c5078cd --- /dev/null +++ b/src/responses/service.py @@ -0,0 +1,270 @@ +""" +Response service for creating responses. + +Handles both streaming and non-streaming response generation. +Tracks conversation history for analytics and future vector memory. +""" + +import time +import secrets +from typing import AsyncGenerator + +from src.agents.registry import ModelRegistry +from src.responses.schemas import ( + Response, + ResponseRequest, + ResponseUsage, + MessageOutputItem, + ReasoningOutputItem, + FunctionCallOutputItem, + OutputTextContent, +) +from src.responses.streaming import StreamingCoordinator +from src.responses.history import ConversationHistory +from src.responses.context import ContextWindow + +# Global conversation history tracker +# In production, this would be backed by a database or Redis +_conversation_history = ConversationHistory(max_turns=20) + +# Global context window manager (4096 token default) +_context_window = ContextWindow(max_tokens=4096) + + +def generate_id() -> str: + """Generate unique ID for responses.""" + return secrets.token_hex(16) + + +def _calculate_usage(input_messages: list[dict], output_items: list) -> ResponseUsage: + """ + Calculate token usage for the response. + + For now: approximate token counting. + Future: Use tiktoken or similar for accurate counting. + + Args: + input_messages: Input messages + output_items: Output items generated + + Returns: + ResponseUsage: Token usage statistics + """ + # Approximate input tokens (chars / 4) + input_text = " ".join(str(msg) for msg in input_messages) + input_tokens = len(input_text) // 4 + + # Approximate output tokens + output_tokens = 0 + reasoning_tokens = 0 + + for item in output_items: + if hasattr(item, 'type'): + # Agent OutputItem objects + if item.type == "reasoning": + reasoning_text = " ".join(item.data.get("summary", [])) + reasoning_tokens += len(reasoning_text) // 4 + elif item.type == "message": + message_text = item.data["content"][0]["text"] + output_tokens += len(message_text) // 4 + elif item.type == "function_call": + func_text = item.data["arguments"] + output_tokens += len(func_text) // 4 + else: + # Schema OutputItem objects + if isinstance(item, ReasoningOutputItem): + reasoning_text = " ".join(item.summary) + reasoning_tokens += len(reasoning_text) // 4 + elif isinstance(item, MessageOutputItem): + message_text = item.content[0].text + output_tokens += len(message_text) // 4 + elif isinstance(item, FunctionCallOutputItem): + func_text = item.arguments + output_tokens += len(func_text) // 4 + + total_tokens = input_tokens + output_tokens + reasoning_tokens + + return ResponseUsage( + input_tokens=input_tokens, + output_tokens=output_tokens, + reasoning_tokens=reasoning_tokens, + total_tokens=total_tokens + ) + + +async def create_response(request: ResponseRequest) -> Response: + """ + Create non-streaming response. + + Tracks conversation history if conversation_id is provided in metadata. + + Args: + request: Response request + + Returns: + Response: Complete response object + + Example: + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + reasoning={"effort": "medium", "summary": "auto"}, + metadata={"conversation_id": "conv_abc123"} # Optional + ) + response = await create_response(request) + """ + # Get or generate conversation ID + conversation_id = await _conversation_history.get_conversation_id(request) + + # Strip pipeline prefix if present (e.g., "pipeline.model" -> "model") + model_id = request.model + if "." in model_id: + model_id = model_id.split(".", 1)[1] + + # Get agent for model + agent = ModelRegistry.get_agent(model_id) + + # Collect all output items from agent + output_items = [] + async for item in agent.generate_response( + messages=request.input, + reasoning=request.reasoning, + tools=request.tools, + temperature=request.temperature, + max_tokens=request.max_output_tokens, + stop=request.stop, + ): + output_items.append(item) + + # Convert agent OutputItems to schema OutputItems + converted_items = _convert_output_items(output_items) + + # Calculate token usage + usage = _calculate_usage(request.input, output_items) + + response = Response( + id=f"resp_{generate_id()}", + created_at=int(time.time()), + model=request.model, + status="completed", + output=converted_items, + usage=usage + ) + + # Track conversation history (for analytics and future vector memory) + await _conversation_history.add_response(conversation_id, response) + + return response + + +async def create_response_stream( + request: ResponseRequest +) -> AsyncGenerator[dict, None]: + """ + Create streaming response. + + Yields SSE-formatted events for streaming to client. + + Args: + request: Response request + + Yields: + dict: SSE event dict with 'event' and 'data' keys + + Example: + async for event in create_response_stream(request): + # event = {"event": "response.output_text.delta", "data": "..."} + yield event + """ + coordinator = StreamingCoordinator() + + async for event in coordinator.stream_response(request): + yield { + "event": event.event, + "data": event.model_dump_json() + } + + +async def get_conversation_history(conversation_id: str) -> list[Response]: + """ + Get conversation history for a given conversation ID. + + Args: + conversation_id: Conversation identifier + + Returns: + list: List of Response objects + """ + return await _conversation_history.get_history(conversation_id) + + +async def get_conversation_stats() -> dict: + """ + Get conversation history statistics. + + Returns: + dict: Statistics including total conversations tracked + """ + return { + "total_conversations": await _conversation_history.get_conversation_count(), + "max_turns_per_conversation": _conversation_history._max_turns + } + + +async def clear_conversation(conversation_id: str) -> bool: + """ + Clear a specific conversation history. + + Args: + conversation_id: Conversation identifier + + Returns: + bool: True if conversation was cleared + """ + return await _conversation_history.clear_conversation(conversation_id) + + +def get_context_window() -> ContextWindow: + """ + Get the global context window manager. + + Returns: + ContextWindow: Context window manager instance + """ + return _context_window + + +def _convert_output_items(items: list) -> list: + """ + Convert agent OutputItem objects to schema OutputItem objects. + + Args: + items: List of agent OutputItem objects + + Returns: + list: List of schema OutputItem objects + """ + 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") + )) + elif item.type == "reasoning": + 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") + )) + + return converted diff --git a/src/responses/streaming.py b/src/responses/streaming.py new file mode 100644 index 0000000..9f02356 --- /dev/null +++ b/src/responses/streaming.py @@ -0,0 +1,375 @@ +""" +Streaming event schemas and coordinator for Responses API. + +Handles Server-Sent Events (SSE) streaming with proper event types: +- response.reasoning_summary_text.delta +- response.output_text.delta +- response.function_call_arguments.delta +- response.done +""" + +from enum import Enum +from typing import Literal, AsyncGenerator +import json +import time + +from src.core.models import CustomBaseModel +from src.responses.schemas import Response +from src.agents.registry import ModelRegistry + + +# ============================================================================ +# 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" + OUTPUT_TEXT_DONE = "response.output_text.done" + FUNCTION_CALL_DELTA = "response.function_call_arguments.delta" + FUNCTION_CALL_DONE = "response.function_call_arguments.done" + RESPONSE_DONE = "response.done" + ERROR = "error" + + +# ============================================================================ +# Stream Event Schemas +# ============================================================================ + +class ReasoningSummaryDelta(CustomBaseModel): + """Reasoning summary text delta event.""" + 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 + + +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 +) + + +# ============================================================================ +# Streaming Coordinator +# ============================================================================ + +class StreamingCoordinator: + """ + Coordinates streaming from agents to SSE format. + + Real production streaming logic that handles: + 1. Reasoning summary chunks + 2. Function call arguments + 3. Output text chunks + 4. Error handling + 5. Final response event + """ + + async def stream_response( + self, + request: "ResponseRequest" # type: ignore # Forward reference + ) -> AsyncGenerator[StreamEvent, None]: + """ + Coordinate streaming from agent to SSE events. + + Args: + request: Response request + + Yields: + StreamEvent: Stream of SSE events + + Example SSE output: + event: response.reasoning_summary_text.delta + data: {"delta": "Analyzing..."} + + event: response.output_text.delta + data: {"delta": "Lorem"} + + event: response.done + data: {"response": {...}} + """ + from src.responses.service import _calculate_usage, generate_id + import asyncio + + output_items = [] + + try: + # Strip pipeline prefix if present (e.g., "pipeline.model" -> "model") + model_id = request.model + if "." in model_id: + model_id = model_id.split(".", 1)[1] + + # Get agent for model + agent = ModelRegistry.get_agent(model_id) + # Stream from agent + async for item in agent.generate_response( + messages=request.input, + reasoning=request.reasoning, + tools=request.tools, + temperature=request.temperature, + max_tokens=request.max_output_tokens, + stop=request.stop, + ): + output_items.append(item) + + # Stream based on item type + if item.type == "reasoning": + # Stream reasoning summary + for step in item.data.get("summary", []): + yield ReasoningSummaryDelta(delta=step + " ") + await asyncio.sleep(0.05) # Simulate typing + + yield ReasoningSummaryDone() + + elif item.type == "function_call": + # Stream function call arguments + # First chunk includes name + 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] + ) + await asyncio.sleep(0.03) + + yield FunctionCallDone() + + elif item.type == "message": + # Stream output text with stop sequence and max tokens enforcement + text = item.data["content"][0]["text"] + words = text.split() + + # Track accumulated text and tokens for enforcement + accumulated_text = "" + output_tokens = 0 + + for word in words: + # Add word to accumulated text + word_with_space = f"{word} " + accumulated_text += word_with_space + + # Check stop sequences + stop_found, text_before_stop = self._check_stop_sequence( + accumulated_text, + request.stop + ) + + if stop_found: + # Emit final text before stop sequence + remaining_text = text_before_stop[len(accumulated_text) - len(word_with_space):] + if remaining_text: + yield OutputTextDelta(delta=remaining_text) + yield OutputTextDone() + break + + # Check max tokens + output_tokens = self._count_tokens_approx(accumulated_text) + if self._check_max_tokens(output_tokens, request.max_output_tokens): + # Max tokens reached - stop streaming + yield OutputTextDone() + break + + # Normal streaming + yield OutputTextDelta(delta=word_with_space) + await asyncio.sleep(0.05) # Simulate typing + else: + # Completed normally without stop/limit + yield OutputTextDone() + + # Final response.done event with complete response + usage = _calculate_usage(request.input, output_items) + + final_response = Response( + id=f"resp_{generate_id()}", + created_at=int(time.time()), + model=request.model, + status="completed", + output=self._convert_output_items(output_items), + 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) + + yield ResponseDone(response=final_response) + + except Exception as e: + # Stream error event + yield self._create_error_event(e) + + def _convert_output_items(self, items: list) -> list: + """Convert agent OutputItem objects to schema OutputItem objects.""" + from src.responses.schemas import ( + MessageOutputItem, + ReasoningOutputItem, + FunctionCallOutputItem, + OutputTextContent, + ) + + 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") + )) + elif item.type == "reasoning": + 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") + )) + + return converted + + def _create_error_event(self, error: Exception) -> ErrorEvent: + """Create error event from exception.""" + from src.core.exceptions import ( + RateLimitError, + ContextLengthError, + AppException, + ) + + if isinstance(error, RateLimitError): + error_type = "rate_limit_exceeded" + code = 429 + elif isinstance(error, ContextLengthError): + error_type = "context_length_exceeded" + code = 400 + elif isinstance(error, AppException): + error_type = "app_error" + code = error.status_code + else: + error_type = "internal_error" + code = 500 + + return ErrorEvent( + error={ + "type": error_type, + "message": str(error), + "code": code + } + ) + + def _check_stop_sequence( + self, + accumulated_text: str, + stop_sequences: list[str] | None + ) -> tuple[bool, str]: + """ + Check if any stop sequence is encountered. + + Args: + accumulated_text: Text accumulated so far + stop_sequences: List of stop sequences to check + + Returns: + tuple: (stop_found, text_before_stop) + - stop_found: True if stop sequence found + - text_before_stop: Text before the stop sequence (or full text if not found) + """ + if not stop_sequences: + return False, accumulated_text + + for stop_seq in stop_sequences: + if stop_seq in accumulated_text: + # Found stop sequence - return text before it + idx = accumulated_text.index(stop_seq) + return True, accumulated_text[:idx] + + return False, accumulated_text + + def _count_tokens_approx(self, text: str) -> int: + """ + Approximate token count (4 chars per token). + + Args: + text: Text to count + + Returns: + int: Approximate token count + """ + return len(text) // 4 + + def _check_max_tokens( + self, + current_tokens: int, + max_tokens: int | None + ) -> bool: + """ + Check if max tokens limit reached. + + Args: + current_tokens: Current token count + max_tokens: Maximum allowed tokens (None for unlimited) + + Returns: + bool: True if limit reached + """ + if max_tokens is None: + return False + return current_tokens >= max_tokens diff --git a/tests/responses/__init__.py b/tests/responses/__init__.py new file mode 100644 index 0000000..241edd9 --- /dev/null +++ b/tests/responses/__init__.py @@ -0,0 +1 @@ +"""Tests for Responses API implementation.""" diff --git a/tests/responses/test_advanced_features.py b/tests/responses/test_advanced_features.py new file mode 100644 index 0000000..187ffe3 --- /dev/null +++ b/tests/responses/test_advanced_features.py @@ -0,0 +1,378 @@ +""" +Tests for Phase 6 advanced features. + +Tests: +- Parameter validation (reasoning effort, max_output_tokens, stop sequences) +- Stop sequence detection and enforcement +- Max tokens enforcement +""" +import json +import pytest +from fastapi.testclient import TestClient +from httpx import AsyncClient +from pydantic import ValidationError + +from src.responses.schemas import ResponseRequest +from src.responses.streaming import StreamingCoordinator + + +# ============================================================================ +# Parameter Validation Tests +# ============================================================================ + +@pytest.mark.unit +def test_temperature_validation(): + """Test temperature parameter validation.""" + # Valid temperatures + valid_temps = [0.0, 0.5, 1.0, 1.5, 2.0] + for temp in valid_temps: + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + temperature=temp + ) + assert request.temperature == temp + + # Invalid temperatures + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + temperature=-0.1 # Too low + ) + assert "temperature" in str(exc_info.value).lower() + + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + temperature=2.1 # Too high + ) + assert "temperature" in str(exc_info.value).lower() + + +@pytest.mark.unit +def test_reasoning_effort_validation(): + """Test reasoning.effort parameter validation.""" + # Valid effort levels + valid_efforts = ['none', 'minimal', 'low', 'medium', 'high', 'xhigh'] + for effort in valid_efforts: + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + reasoning={"effort": effort, "summary": "auto"} + ) + assert request.reasoning["effort"] == effort + + # Invalid effort level + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + reasoning={"effort": "invalid", "summary": "auto"} + ) + assert "reasoning.effort" in str(exc_info.value) + + +@pytest.mark.unit +def test_reasoning_summary_validation(): + """Test reasoning.summary parameter validation.""" + # Valid summary values + valid_summaries = ['auto', 'off'] + for summary in valid_summaries: + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + reasoning={"effort": "medium", "summary": summary} + ) + assert request.reasoning["summary"] == summary + + # Invalid summary value + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + reasoning={"effort": "medium", "summary": "invalid"} + ) + assert "reasoning.summary" in str(exc_info.value) + + +@pytest.mark.unit +def test_max_output_tokens_validation(): + """Test max_output_tokens parameter validation.""" + # Valid values + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + max_output_tokens=100 + ) + assert request.max_output_tokens == 100 + + # None is valid (unlimited) + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + max_output_tokens=None + ) + assert request.max_output_tokens is None + + # Invalid: zero or negative + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + max_output_tokens=0 + ) + assert "max_output_tokens" in str(exc_info.value) + + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + max_output_tokens=-10 + ) + assert "max_output_tokens" in str(exc_info.value) + + +@pytest.mark.unit +def test_stop_sequences_validation(): + """Test stop sequences parameter validation.""" + # Valid: up to 4 stop sequences + for num_seqs in range(1, 5): + stop_seqs = [f"stop{i}" for i in range(num_seqs)] + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + stop=stop_seqs + ) + assert request.stop == stop_seqs + + # Invalid: more than 4 stop sequences + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + stop=["stop1", "stop2", "stop3", "stop4", "stop5"] # 5 sequences + ) + assert "4 stop sequences" in str(exc_info.value) + + # Invalid: empty string in stop sequences + with pytest.raises(ValidationError) as exc_info: + ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + stop=["stop1", ""] # Empty string + ) + assert "non-empty" in str(exc_info.value).lower() + + +# ============================================================================ +# Stop Sequence Enforcement Tests +# ============================================================================ + +@pytest.mark.unit +def test_stop_sequence_detection_helper(): + """Test stop sequence detection helper method.""" + coordinator = StreamingCoordinator() + + # No stop sequences + found, text = coordinator._check_stop_sequence("Hello world", None) + assert found is False + assert text == "Hello world" + + # Stop sequence not present + found, text = coordinator._check_stop_sequence( + "Hello world", + ["STOP", "END"] + ) + assert found is False + assert text == "Hello world" + + # Stop sequence found + found, text = coordinator._check_stop_sequence( + "Hello STOP this should not appear", + ["STOP"] + ) + assert found is True + assert text == "Hello " + + # Multiple stop sequences, first one wins + found, text = coordinator._check_stop_sequence( + "Hello STOP this END that", + ["STOP", "END"] + ) + assert found is True + assert text == "Hello " + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_stop_sequence_in_streaming(async_client: AsyncClient): + """Test stop sequence enforcement during streaming.""" + # We'll use lorem-tester which generates predictable text + # The agent generates lorem ipsum text, so we use a stop sequence + # that's likely to appear + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Generate long text"}], + "stop": ["dolor"], # Common word in lorem ipsum + "stream": True + } + + chunks_received = [] + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + assert response.status_code == 200 + + async for line in response.aiter_lines(): + if not line.strip(): + continue + + if line.startswith("data: "): + data_str = line[6:].strip() + if data_str != "[DONE]": + try: + chunk = json.loads(data_str) + chunks_received.append(chunk) + except json.JSONDecodeError: + pass + + # Should have received chunks and stopped early + # (Can't verify exact stop behavior with random lorem ipsum, + # but test ensures no errors occur) + assert len(chunks_received) > 0 + + +# ============================================================================ +# Max Tokens Enforcement Tests +# ============================================================================ + +@pytest.mark.unit +def test_max_tokens_check_helper(): + """Test max tokens check helper method.""" + coordinator = StreamingCoordinator() + + # No limit + assert coordinator._check_max_tokens(100, None) is False + + # Under limit + assert coordinator._check_max_tokens(50, 100) is False + + # At limit + assert coordinator._check_max_tokens(100, 100) is True + + # Over limit + assert coordinator._check_max_tokens(150, 100) is True + + +@pytest.mark.unit +def test_token_counting_helper(): + """Test approximate token counting.""" + coordinator = StreamingCoordinator() + + # Empty string + assert coordinator._count_tokens_approx("") == 0 + + # ~4 chars per token + text = "Hello world" # 11 chars + tokens = coordinator._count_tokens_approx(text) + assert tokens == 2 # 11 // 4 = 2 + + text = "A" * 100 # 100 chars + tokens = coordinator._count_tokens_approx(text) + assert tokens == 25 # 100 // 4 = 25 + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_max_tokens_in_streaming(async_client: AsyncClient): + """Test max tokens enforcement during streaming.""" + # Set very low max_output_tokens to force early stop + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Generate text"}], + "max_output_tokens": 5, # Very low limit + "stream": True + } + + chunks_received = [] + text_chunks = [] + + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + assert response.status_code == 200 + + async for line in response.aiter_lines(): + if not line.strip(): + continue + + if line.startswith("event: "): + event_type = line[7:].strip() + elif line.startswith("data: "): + data_str = line[6:].strip() + if data_str != "[DONE]": + try: + chunk = json.loads(data_str) + chunks_received.append(chunk) + + # Collect text deltas + if "delta" in chunk: + text_chunks.append(chunk["delta"]) + except json.JSONDecodeError: + pass + + # Should have stopped early due to token limit + total_text = "".join(text_chunks) + # With max_output_tokens=5, total text should be small + # (Approximate: 5 tokens * 4 chars ≈ 20 chars) + assert len(total_text) < 100 # Reasonable upper bound + + +# ============================================================================ +# Combined Features Test +# ============================================================================ + +@pytest.mark.unit +def test_combined_validation(client: TestClient): + """Test combined parameter validation in actual request.""" + # Valid request with all advanced features + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Hello"}], + "temperature": 0.7, + "max_output_tokens": 100, + "stop": ["STOP", "END"], + "reasoning": {"effort": "high", "summary": "auto"}, + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + assert response.status_code == 200 + data = response.json() + assert data["status"] == "completed" + + +@pytest.mark.unit +def test_invalid_combined_parameters(client: TestClient): + """Test that invalid parameters are rejected.""" + # Invalid temperature + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Hello"}], + "temperature": 3.0, # Too high + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + assert response.status_code == 422 + data = response.json() + assert "error" in data + assert data["error"]["type"] == "invalid_request_error" diff --git a/tests/responses/test_error_handling.py b/tests/responses/test_error_handling.py new file mode 100644 index 0000000..20515ff --- /dev/null +++ b/tests/responses/test_error_handling.py @@ -0,0 +1,281 @@ +""" +Tests for error handling in Responses API. +""" +import json +import pytest +from fastapi.testclient import TestClient +from httpx import AsyncClient + + +@pytest.mark.unit +def test_model_not_found_error(client: TestClient) -> None: + """Test response when model doesn't exist.""" + request_data = { + "model": "nonexistent-model-12345", + "input": [ + {"role": "user", "content": "Hello"} + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 404 + data = response.json() + assert "detail" in data + assert "not found" in data["detail"].lower() + + +@pytest.mark.unit +def test_validation_error_missing_model(client: TestClient) -> None: + """Test validation error when model field is missing.""" + request_data = { + "input": [ + {"role": "user", "content": "Hello"} + ] + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 422 + + +@pytest.mark.unit +def test_validation_error_invalid_temperature(client: TestClient) -> None: + """Test validation error for out-of-range temperature.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Hello"} + ], + "temperature": 3.0 # Max is 2.0 + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 422 + + +@pytest.mark.unit +def test_rate_limit_error(client: TestClient) -> None: + """Test rate limit error trigger.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "trigger_rate_limit"} + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + # Should get an error response + assert response.status_code in [429, 500] # Rate limit or internal error + + +@pytest.mark.unit +def test_context_overflow_error(client: TestClient) -> None: + """Test context length overflow error trigger.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "trigger_context_overflow"} + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + # Should get an error response + assert response.status_code in [400, 500] # Bad request or internal error + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_streaming_model_not_found(async_client: AsyncClient) -> None: + """Test streaming response with nonexistent model.""" + request_data = { + "model": "nonexistent-streaming-model", + "input": [ + {"role": "user", "content": "Hello"} + ], + "stream": True + } + + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + # Streaming always returns 200 OK, errors are sent as events + assert response.status_code == 200 + + # Collect events and look for error event + error_found = False + async for line in response.aiter_lines(): + if not line.strip(): + continue + + if line.startswith("event: "): + event_type = line[7:].strip() + if event_type == "error": + error_found = True + break + + # Should have received an error event + assert error_found, "Expected error event in stream" + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_streaming_rate_limit_error(async_client: AsyncClient) -> None: + """Test streaming with rate limit error.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "trigger_rate_limit"} + ], + "stream": True + } + + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + # May get error status or error event + # Collect all events + events = [] + async for line in response.aiter_lines(): + if not line.strip(): + continue + + if line.startswith("event: "): + event_type = line[7:].strip() + elif line.startswith("data: "): + data_str = line[6:].strip() + try: + data = json.loads(data_str) + if event_type: + events.append({"event": event_type, "data": data}) + except json.JSONDecodeError: + pass + + # Should either have error status or error event + if response.status_code == 200: + # Check for error event + error_events = [e for e in events if e["event"] == "error"] + # May or may not have error event depending on where error occurs + # At minimum, should not crash + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_streaming_context_overflow_error(async_client: AsyncClient) -> None: + """Test streaming with context overflow error.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "trigger_context_overflow"} + ], + "stream": True + } + + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + # May get error status or error event + events = [] + async for line in response.aiter_lines(): + if not line.strip(): + continue + + if line.startswith("event: "): + event_type = line[7:].strip() + elif line.startswith("data: "): + data_str = line[6:].strip() + try: + data = json.loads(data_str) + if event_type: + events.append({"event": event_type, "data": data}) + except json.JSONDecodeError: + pass + + # Should not crash + assert response.status_code in [200, 400, 500] + + +@pytest.mark.unit +def test_function_call_output_item(client: TestClient) -> None: + """Test response with function call items.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Use the search tool"} + ], + "tools": [ + { + "name": "search", + "description": "Search for information", + "parameters": {} + } + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 200 + data = response.json() + + # May or may not have function_call items (random in lorem-tester) + # But should successfully handle them if present + output_items = data["output"] + for item in output_items: + assert item["type"] in ["message", "reasoning", "function_call"] + if item["type"] == "function_call": + assert "name" in item + assert "arguments" in item + + +@pytest.mark.unit +def test_pipeline_prefix_stripping(client: TestClient) -> None: + """Test that pipeline prefixes are stripped from model names.""" + request_data = { + "model": "some_pipeline.lorem-tester", + "input": [ + {"role": "user", "content": "Hello"} + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + # Should successfully strip prefix and find model + assert response.status_code == 200 + data = response.json() + assert data["status"] == "completed" + + +@pytest.mark.unit +def test_multiple_pipeline_prefixes(client: TestClient) -> None: + """Test multiple dots in model name (only first is prefix).""" + request_data = { + "model": "pipeline.sub.lorem-tester", + "input": [ + {"role": "user", "content": "Hello"} + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + # Should strip only first part + # "pipeline.sub.lorem-tester" -> "sub.lorem-tester" + # This should fail since "sub.lorem-tester" doesn't exist + assert response.status_code == 404 diff --git a/tests/responses/test_history.py b/tests/responses/test_history.py new file mode 100644 index 0000000..50413b8 --- /dev/null +++ b/tests/responses/test_history.py @@ -0,0 +1,383 @@ +""" +Tests for conversation history management. +""" +import pytest +from fastapi.testclient import TestClient + +from src.responses.history import ConversationHistory +from src.responses.context import ContextWindow +from src.responses.schemas import ResponseRequest +from src.responses import service + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_conversation_id_from_metadata(): + """Test conversation ID extraction from metadata.""" + history = ConversationHistory() + + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}], + metadata={"conversation_id": "conv_123"} + ) + + conv_id = await history.get_conversation_id(request) + assert conv_id == "conv_123" + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_conversation_id_generation(): + """Test conversation ID generation from first message.""" + history = ConversationHistory() + + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}] + # No metadata provided + ) + + conv_id = await history.get_conversation_id(request) + assert isinstance(conv_id, str) + assert len(conv_id) == 16 # 16 character hex + + # Same first message should generate same ID + request2 = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Hello"}] + ) + conv_id2 = await history.get_conversation_id(request2) + assert conv_id == conv_id2 + + +@pytest.mark.unit +def test_conversation_history_tracking(client: TestClient): + """Test that conversations are tracked server-side.""" + # First request with conversation ID + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Hello"}], + "metadata": {"conversation_id": "test_conv_001"}, + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + assert response.status_code == 200 + + # Second request with same conversation ID + request_data2 = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "Hi there!"}, + {"role": "user", "content": "How are you?"} + ], + "metadata": {"conversation_id": "test_conv_001"}, + "stream": False + } + + response2 = client.post("/v1/responses", json=request_data2) + assert response2.status_code == 200 + + # Both responses should be successful + assert response.json()["status"] == "completed" + assert response2.json()["status"] == "completed" + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_conversation_history_retrieval(): + """Test retrieving conversation history.""" + history = ConversationHistory() + + request = ResponseRequest( + model="lorem-tester", + input=[{"role": "user", "content": "Test"}], + metadata={"conversation_id": "test_retrieve"} + ) + + conv_id = await history.get_conversation_id(request) + + # Initially empty + retrieved = await history.get_history(conv_id) + assert len(retrieved) == 0 + + # Add mock response + from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent + mock_response = Response( + id="resp_123", + created_at=1234567890, + model="lorem-tester", + status="completed", + output=[ + MessageOutputItem( + id="msg_1", + content=[OutputTextContent(text="Test response")] + ) + ], + usage=ResponseUsage( + input_tokens=10, + output_tokens=5, + reasoning_tokens=0, + total_tokens=15 + ) + ) + + await history.add_response(conv_id, mock_response) + + # Should now have 1 response + retrieved = await history.get_history(conv_id) + assert len(retrieved) == 1 + assert retrieved[0].id == "resp_123" + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_conversation_history_trimming(): + """Test that history is trimmed to max_turns.""" + history = ConversationHistory(max_turns=3) + + from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent + + conv_id = "test_trim" + + # Add 5 responses (more than max_turns) + for i in range(5): + response = Response( + id=f"resp_{i}", + created_at=1234567890 + i, + model="lorem-tester", + status="completed", + output=[ + MessageOutputItem( + id=f"msg_{i}", + content=[OutputTextContent(text=f"Response {i}")] + ) + ], + usage=ResponseUsage( + input_tokens=10, + output_tokens=5, + reasoning_tokens=0, + total_tokens=15 + ) + ) + await history.add_response(conv_id, response) + + # Should only keep last 3 + retrieved = await history.get_history(conv_id) + assert len(retrieved) == 3 + assert retrieved[0].id == "resp_2" # Oldest kept + assert retrieved[1].id == "resp_3" + assert retrieved[2].id == "resp_4" # Most recent + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_clear_conversation(): + """Test clearing conversation history.""" + history = ConversationHistory() + + from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent + + conv_id = "test_clear" + + # Add a response + response = Response( + id="resp_123", + created_at=1234567890, + model="lorem-tester", + status="completed", + output=[ + MessageOutputItem( + id="msg_1", + content=[OutputTextContent(text="Test")] + ) + ], + usage=ResponseUsage( + input_tokens=10, + output_tokens=5, + reasoning_tokens=0, + total_tokens=15 + ) + ) + await history.add_response(conv_id, response) + + # Verify it exists + assert len(await history.get_history(conv_id)) == 1 + + # Clear it + cleared = await history.clear_conversation(conv_id) + assert cleared is True + + # Should be empty now + assert len(await history.get_history(conv_id)) == 0 + + # Clearing non-existent conversation should return False + cleared_again = await history.clear_conversation(conv_id) + assert cleared_again is False + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_conversation_count(): + """Test conversation count tracking.""" + history = ConversationHistory() + + initial_count = await history.get_conversation_count() + + # Add responses to 3 different conversations + from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent + + for i in range(3): + response = Response( + id=f"resp_{i}", + created_at=1234567890, + model="lorem-tester", + status="completed", + output=[ + MessageOutputItem( + id=f"msg_{i}", + content=[OutputTextContent(text="Test")] + ) + ], + usage=ResponseUsage( + input_tokens=10, + output_tokens=5, + reasoning_tokens=0, + total_tokens=15 + ) + ) + await history.add_response(f"conv_{i}", response) + + new_count = await history.get_conversation_count() + assert new_count == initial_count + 3 + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_context_window_token_counting(): + """Test token counting in context window.""" + context = ContextWindow(max_tokens=4096) + + # Test string counting + text = "Hello world! " * 100 # ~1200 characters + tokens = await context.count_tokens([text]) + assert tokens > 0 + assert tokens == len(text) // 4 # Approximate + + # Test dict counting + message = {"role": "user", "content": "Test message"} + tokens = await context.count_tokens([message]) + assert tokens > 0 + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_context_window_trimming(): + """Test context window trimming to fit.""" + context = ContextWindow(max_tokens=100) + + # Create items that exceed limit + items = [ + "This is a long message " * 20, # ~480 chars = ~120 tokens + "Another message " * 10, # ~160 chars = ~40 tokens + "Short message" # ~13 chars = ~3 tokens + ] + + # Trim with 10 token reserve + trimmed = await context.trim_to_fit(items, reserve_tokens=10) + + # Should keep only items that fit (90 tokens available) + # Most recent first: "Short message" (3 tokens) + "Another message..." (40 tokens) = 43 tokens + assert len(trimmed) >= 1 # At least the short message + assert "Short message" in trimmed # Most recent always kept if it fits + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_context_window_fits_check(): + """Test checking if items fit in context.""" + context = ContextWindow(max_tokens=100) + + small_items = ["Hello", "World"] + large_items = ["Very long text " * 200] # ~3200 chars = ~800 tokens + + # Small items should fit + assert await context.fits_in_context(small_items, reserve_tokens=10) is True + + # Large items should not fit + assert await context.fits_in_context(large_items, reserve_tokens=10) is False + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_context_window_usage_stats(): + """Test context window usage statistics.""" + context = ContextWindow(max_tokens=1000) + + items = ["Test message " * 50] # ~650 chars = ~162 tokens + + stats = await context.get_usage_stats(items, reserve_tokens=100) + + assert "total_tokens" in stats + assert "max_tokens" in stats + assert "reserved_tokens" in stats + assert "available_tokens" in stats + assert "usage_percent" in stats + assert "fits" in stats + + assert stats["max_tokens"] == 1000 + assert stats["reserved_tokens"] == 100 + assert stats["available_tokens"] == 900 + assert isinstance(stats["usage_percent"], (int, float)) + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_service_conversation_helpers(): + """Test service helper functions for conversation history.""" + # Get stats + stats = await service.get_conversation_stats() + assert "total_conversations" in stats + assert "max_turns_per_conversation" in stats + + # Get context window + context = service.get_context_window() + assert context.max_tokens == 4096 + + # Test clearing (should handle non-existent gracefully) + result = await service.clear_conversation("non_existent_conv") + assert result is False + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_streaming_tracks_history(async_client): + """Test that streaming responses also track conversation history.""" + from httpx import AsyncClient + import json + + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Hello"}], + "metadata": {"conversation_id": "stream_test_001"}, + "stream": True + } + + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + assert response.status_code == 200 + + # Consume the stream + async for line in response.aiter_lines(): + if line.startswith("data: ") and "[DONE]" not in line: + continue + + # History should be tracked + # (We can't easily verify this without exposing a GET endpoint, + # but the integration is tested in the non-streaming tests) diff --git a/tests/responses/test_router.py b/tests/responses/test_router.py new file mode 100644 index 0000000..5ab522a --- /dev/null +++ b/tests/responses/test_router.py @@ -0,0 +1,281 @@ +""" +Tests for Responses API router. +""" + +import json +import pytest +from fastapi.testclient import TestClient +from httpx import AsyncClient + + +@pytest.mark.unit +def test_create_response_non_streaming(client: TestClient) -> None: + """Test non-streaming response creation.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Hello, how are you?"} + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 200 + data = response.json() + + # Verify response structure + assert "id" in data + assert data["object"] == "response" + assert "created_at" in data + assert data["model"] == "lorem-tester" + assert data["status"] == "completed" + assert "output" in data + assert "usage" in data + + # Verify output array + assert isinstance(data["output"], list) + assert len(data["output"]) >= 1 # At least message item + + # Verify usage + assert data["usage"]["input_tokens"] > 0 + assert data["usage"]["output_tokens"] > 0 + assert data["usage"]["total_tokens"] > 0 + + +@pytest.mark.unit +def test_create_response_with_reasoning(client: TestClient) -> None: + """Test response with reasoning enabled.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Explain something"} + ], + "reasoning": { + "effort": "medium", + "summary": "auto" + }, + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 200 + data = response.json() + + # Should have reasoning item and message item + assert len(data["output"]) >= 2 + + # Check for reasoning item + reasoning_items = [item for item in data["output"] if item["type"] == "reasoning"] + assert len(reasoning_items) >= 1 + + reasoning_item = reasoning_items[0] + assert "summary" in reasoning_item + assert isinstance(reasoning_item["summary"], list) + assert len(reasoning_item["summary"]) > 0 + + # Check for message item + message_items = [item for item in data["output"] if item["type"] == "message"] + assert len(message_items) >= 1 + + # Verify reasoning tokens counted + assert data["usage"]["reasoning_tokens"] > 0 + + +@pytest.mark.unit +def test_create_response_with_tools(client: TestClient) -> None: + """Test response with tools available.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Use a tool to help"} + ], + "tools": [ + { + "name": "search_knowledge", + "description": "Search knowledge base", + "parameters": {} + } + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 200 + data = response.json() + + # May or may not have function_call items (randomized in lorem-tester) + # But should always have message item + message_items = [item for item in data["output"] if item["type"] == "message"] + assert len(message_items) >= 1 + + +@pytest.mark.unit +def test_create_response_invalid_model(client: TestClient) -> None: + """Test response with non-existent model.""" + request_data = { + "model": "nonexistent-model", + "input": [ + {"role": "user", "content": "Hello"} + ], + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 404 + data = response.json() + assert "detail" in data + + +@pytest.mark.unit +def test_create_response_validation_error(client: TestClient) -> None: + """Test response with invalid request data.""" + # Missing required 'model' field + request_data = { + "input": [ + {"role": "user", "content": "Hello"} + ] + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 422 + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_create_response_streaming(async_client: AsyncClient) -> None: + """Test streaming response creation.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Hello"} + ], + "stream": True + } + + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + assert response.status_code == 200 + assert response.headers["content-type"] == "text/event-stream; charset=utf-8" + + events = [] + async for line in response.aiter_lines(): + if not line.strip(): + continue + + # Parse SSE format: "event: event_type" and "data: {json}" + if line.startswith("event: "): + event_type = line[7:].strip() + elif line.startswith("data: "): + data_str = line[6:].strip() + try: + data = json.loads(data_str) + events.append({"event": event_type, "data": data}) + except json.JSONDecodeError: + pass # Skip malformed data + + # Should have multiple events + assert len(events) > 0 + + # Should have response.done event as last event + assert events[-1]["event"] == "response.done" + + # response.done should have complete response + done_data = events[-1]["data"] + assert "response" in done_data + assert done_data["response"]["status"] == "completed" + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_create_response_streaming_with_reasoning(async_client: AsyncClient) -> None: + """Test streaming response with reasoning.""" + request_data = { + "model": "lorem-tester", + "input": [ + {"role": "user", "content": "Explain"} + ], + "reasoning": { + "effort": "medium", + "summary": "auto" + }, + "stream": True + } + + async with async_client.stream( + "POST", + "/v1/responses", + json=request_data, + timeout=20.0, + ) as response: + assert response.status_code == 200 + + events = [] + event_type = None + + async for line in response.aiter_lines(): + if not line.strip(): + continue + + if line.startswith("event: "): + event_type = line[7:].strip() + elif line.startswith("data: "): + data_str = line[6:].strip() + try: + data = json.loads(data_str) + if event_type: + events.append({"event": event_type, "data": data}) + except json.JSONDecodeError: + pass + + # Should have reasoning events + reasoning_events = [ + e for e in events + if e["event"] == "response.reasoning_summary_text.delta" + ] + assert len(reasoning_events) > 0 + + # Should have output text events + output_events = [ + e for e in events + if e["event"] == "response.output_text.delta" + ] + assert len(output_events) > 0 + + +@pytest.mark.unit +def test_create_response_temperature_parameter(client: TestClient) -> None: + """Test temperature parameter handling.""" + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Test"}], + "temperature": 0.5, + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 200 + + +@pytest.mark.unit +def test_create_response_max_tokens_parameter(client: TestClient) -> None: + """Test max_output_tokens parameter handling.""" + request_data = { + "model": "lorem-tester", + "input": [{"role": "user", "content": "Test"}], + "max_output_tokens": 100, + "stream": False + } + + response = client.post("/v1/responses", json=request_data) + + assert response.status_code == 200