diff --git a/docs/v3/contributing/configuration.mdx b/docs/v3/contributing/configuration.mdx index 3c210e71..400224d8 100644 --- a/docs/v3/contributing/configuration.mdx +++ b/docs/v3/contributing/configuration.mdx @@ -197,6 +197,51 @@ Because provider params live on each model config, background workers such as the Deriver and Dreamer can use longer request timeouts while synchronous chat paths keep tighter defaults. +#### OpenAI Responses API mode + +The `openai` transport uses Chat Completions by default. Set +`api_mode = "responses"` on a model config to opt that feature into an +OpenAI Responses-compatible endpoint: + +```toml +[dialectic.levels.max.model_config] +transport = "openai" +model = "gpt-5.4" + +[dialectic.levels.max.model_config.overrides] +base_url = "https://api.openai.com/v1" +api_key_env = "OPENAI_API_KEY" + +[dialectic.levels.max.model_config.overrides.provider_params] +api_mode = "responses" +timeout = 120.0 +``` + +Responses mode translates Honcho's message history, tools, tool results, +reasoning items, and structured-output requests to the Responses wire format. +It is opt-in per model config: configure the Deriver, Summary, each Dialectic +level, and both Dream specialist model configs separately when they should use +Responses. A fallback also needs its own `fallback.overrides.provider_params` +setting; otherwise that fallback keeps its configured/default API mode. + +The endpoint must implement the OpenAI Responses streaming contract. Honcho +uses `responses.stream`, sends `store = false`, and reconstructs terminal text, +tool calls, reasoning details, and usage from the event stream. By default it +sends `max_output_tokens`; set `omit_max_output_tokens = true` only for a +compatible endpoint that explicitly rejects that field. + +Responses mode rejects unsupported non-null controls instead of silently +ignoring them: `stop`, `presence_penalty`, `frequency_penalty`, `seed`, and +`thinking_budget_tokens`. Use `thinking_effort` for supported OpenAI reasoning +models. + + +`api_mode = "responses"` selects a wire protocol only. It does not acquire, +store, or refresh OAuth credentials. Use the provider's documented +server-to-server credentials, or an operator-managed compatible endpoint whose +authentication and terms permit backend use. + + `timeout` gotchas: - The value is validated **at config load**: it must coerce to a positive, diff --git a/src/config.py b/src/config.py index 80827327..16ca5d88 100644 --- a/src/config.py +++ b/src/config.py @@ -115,7 +115,8 @@ class ModelOverrideSettings(BaseModel): default_factory=dict, description=( "Operator escape hatch for provider-specific request fields. " - "Three recognized keys: `extra_body` (merged into the request body), " + "`api_mode=responses` selects the OpenAI Responses wire protocol. " + "Three passthrough keys are also recognized: `extra_body` (merged into the request body), " "`extra_headers` (HTTP headers), `extra_query` (URL query params). " "OpenAI and Anthropic transports forward these as identically-named " "SDK kwargs. The Gemini transport merges `extra_body` into the " @@ -131,8 +132,11 @@ class ModelOverrideSettings(BaseModel): @field_validator("provider_params") @classmethod - def _validate_provider_timeout(cls, v: dict[str, Any]) -> dict[str, Any]: - """Reject bad `timeout` values at config load; normalize good ones to float.""" + def _validate_provider_params(cls, v: dict[str, Any]) -> dict[str, Any]: + """Reject invalid Responses modes/timeouts at config load.""" + api_mode = v.get("api_mode") + if api_mode is not None and api_mode != "responses": + raise ValueError("provider_params.api_mode must be 'responses' when set") if "timeout" not in v: return v return {**v, "timeout": coerce_provider_timeout(v["timeout"])} @@ -200,6 +204,22 @@ def _validate_structured_output_mode( ) +def _validate_api_mode_transport( + transport: ModelTransport, + provider_params: dict[str, Any], +) -> None: + """Reject invalid or unsupported Responses mode configuration.""" + api_mode = provider_params.get("api_mode") + if api_mode is None: + return + if api_mode != "responses": + raise ValueError("provider_params.api_mode must be 'responses' when set") + if transport != "openai": + raise ValueError( + "provider_params.api_mode is only supported on the 'openai' transport" + ) + + class FallbackModelSettings(BaseModel): """Independent fallback model configuration. No inheritance from primary.""" @@ -241,6 +261,7 @@ class FallbackModelSettings(BaseModel): def _validate_runtime_shape(self) -> "FallbackModelSettings": _validate_thinking_constraints(self.transport, self.thinking_budget_tokens) _validate_structured_output_mode(self.transport, self.structured_output_mode) + _validate_api_mode_transport(self.transport, self.overrides.provider_params) return self @@ -288,6 +309,7 @@ class ConfiguredModelSettings(BaseModel): def _validate_runtime_shape(self) -> "ConfiguredModelSettings": _validate_thinking_constraints(self.transport, self.thinking_budget_tokens) _validate_structured_output_mode(self.transport, self.structured_output_mode) + _validate_api_mode_transport(self.transport, self.overrides.provider_params) return self @@ -324,6 +346,11 @@ class ResolvedFallbackConfig(BaseModel): def reasoning_effort(self) -> ThinkingEffortLevel | None: return self.thinking_effort + @model_validator(mode="after") + def _validate_api_mode_on_self(self) -> "ResolvedFallbackConfig": + _validate_api_mode_transport(self.transport, self.provider_params) + return self + class ModelConfig(BaseModel): """Reusable model configuration for any non-embedding LLM caller.""" @@ -369,6 +396,7 @@ class ModelConfig(BaseModel): @model_validator(mode="after") def _validate_thinking_constraints_on_self(self) -> "ModelConfig": _validate_thinking_constraints(self.transport, self.thinking_budget_tokens) + _validate_api_mode_transport(self.transport, self.provider_params) return self def for_model( @@ -377,12 +405,13 @@ class ModelConfig(BaseModel): *, transport_override: ModelTransport | None = None, ) -> "ModelConfig": - return self.model_copy( - update={ - "model": model_override, - "transport": transport_override or self.transport, - } - ) + update = self.model_dump(mode="python", round_trip=True) + update["model"] = model_override + update["transport"] = transport_override or self.transport + # model_copy(update=...) deliberately skips Pydantic validation. A full + # re-validation is required here because changing transport can make + # retained provider params (for example api_mode=responses) invalid. + return type(self).model_validate(update) class ConfiguredEmbeddingModelSettings(BaseModel): diff --git a/src/llm/backends/openai.py b/src/llm/backends/openai.py index 5b910ae9..57735212 100644 --- a/src/llm/backends/openai.py +++ b/src/llm/backends/openai.py @@ -7,6 +7,7 @@ from collections.abc import AsyncIterator from typing import Any, cast from openai import BadRequestError, LengthFinishReasonError +from openai.lib._pydantic import to_strict_json_schema from pydantic import BaseModel, ValidationError from src.exceptions import ValidationException @@ -29,12 +30,14 @@ logger = logging.getLogger(__name__) # does not hold a reference to the keyed BaseModel so that when a dynamically created # type is no longer referenced it becomes eligible for garbage collection. This avoids a # memory leak. -_json_object_instruction_cache: weakref.WeakKeyDictionary[type[BaseModel], str] = ( - weakref.WeakKeyDictionary() -) +_json_object_instruction_cache: weakref.WeakKeyDictionary[ + type[BaseModel], dict[bool, str] +] = weakref.WeakKeyDictionary() -def _json_object_instruction(response_format: type[BaseModel]) -> str: +def _json_object_instruction( + response_format: type[BaseModel], *, tools_present: bool = False +) -> str: """Schema-injection instruction for json_object mode. The JSON schema is static per response_format class, so cache the serialized @@ -42,17 +45,25 @@ def _json_object_instruction(response_format: type[BaseModel]) -> str: it every call. """ cached = _json_object_instruction_cache.get(response_format) - if cached is not None: - return cached + if cached is not None and tools_present in cached: + return cached[tools_present] # Some OpenAI-compatible providers enforce this JSON-object precondition with # a case-sensitive substring check, so include lowercase "json" explicitly. + requirement = ( + "If not responding with a tool call, respond with" + if tools_present + else "You must respond with" + ) instruction = ( - "You must respond with a single JSON object (json) that conforms " - "exactly to the following JSON schema. Do not include any text, " - "markdown, or code fences outside the JSON object.\n\nJSON schema:\n" + f"{requirement} a single JSON object (json) that conforms exactly to " + "the following JSON schema. Do not include any text, markdown, or code " + "fences outside the JSON object.\n\nJSON schema:\n" f"{json.dumps(response_format.model_json_schema())}" ) - _json_object_instruction_cache[response_format] = instruction + if cached is None: + cached = {} + _json_object_instruction_cache[response_format] = cached + cached[tools_present] = instruction return instruction @@ -166,6 +177,21 @@ class OpenAIBackend: max_output_tokens: int | None = None, extra_params: dict[str, Any] | None = None, ) -> CompletionResult: + if extra_params and extra_params.get("api_mode") == "responses": + return await self._complete_responses( + model=model, + messages=messages, + max_tokens=max_output_tokens or max_tokens, + temperature=temperature, + stop=stop, + thinking_budget_tokens=thinking_budget_tokens, + tools=tools, + tool_choice=tool_choice, + response_format=response_format, + thinking_effort=thinking_effort, + extra_params=extra_params, + ) + params = self._build_params( model=model, messages=messages, @@ -286,6 +312,23 @@ class OpenAIBackend: max_output_tokens: int | None = None, extra_params: dict[str, Any] | None = None, ) -> AsyncIterator[StreamChunk]: + if extra_params and extra_params.get("api_mode") == "responses": + async for chunk in self._stream_responses( + model=model, + messages=messages, + max_tokens=max_output_tokens or max_tokens, + temperature=temperature, + stop=stop, + thinking_budget_tokens=thinking_budget_tokens, + tools=tools, + tool_choice=tool_choice, + response_format=response_format, + thinking_effort=thinking_effort, + extra_params=extra_params, + ): + yield chunk + return + params = self._build_params( model=model, messages=messages, @@ -335,6 +378,574 @@ class OpenAIBackend: if not usage_chunk_received and finish_reason: yield StreamChunk(is_done=True, finish_reason=finish_reason) + async def _complete_responses( + self, + *, + model: str, + messages: list[dict[str, Any]], + max_tokens: int, + temperature: float | None, + stop: list[str] | None, + thinking_budget_tokens: int | None, + tools: list[dict[str, Any]] | None, + tool_choice: str | dict[str, Any] | None, + response_format: type[BaseModel] | dict[str, Any] | None, + thinking_effort: str | None, + extra_params: dict[str, Any], + ) -> CompletionResult: + self._validate_responses_options( + stop=stop, + thinking_budget_tokens=thinking_budget_tokens, + extra_params=extra_params, + ) + params = self._build_responses_params( + model=model, + messages=messages, + max_tokens=max_tokens, + temperature=temperature, + tools=tools, + tool_choice=tool_choice, + response_format=response_format, + thinking_effort=thinking_effort, + extra_params=extra_params, + ) + text_parts: list[str] = [] + output_items: list[Any] = [] + completed_response: Any = None + async with self._client.responses.stream(**params) as stream: + async for event in stream: + event_type = self._response_value(event, "type") + if event_type == "response.output_text.delta": + delta = self._response_value(event, "delta") + if isinstance(delta, str): + text_parts.append(delta) + elif event_type == "response.output_item.done": + item = self._response_value(event, "item") + if item is not None: + output_items.append(item) + elif event_type == "response.completed": + completed_response = self._response_value(event, "response") + response = await stream.get_final_response() + # ChatGPT's Codex Responses endpoint currently omits ``output`` from + # response.completed, even though it streams all deltas/items. Rebuild + # that losslessly while retaining status/usage from the completed event. + metadata = completed_response or response + existing_output = self._response_value( + response, "output" + ) or self._response_value(metadata, "output") + existing_text = self._response_value( + response, "output_text" + ) or self._response_value(metadata, "output_text") + if (text_parts and not existing_text) or (output_items and not existing_output): + response = { + "output_text": existing_text or "".join(text_parts), + "output": existing_output or output_items, + "status": self._response_value(metadata, "status"), + "incomplete_details": self._response_value( + metadata, "incomplete_details" + ), + "refusal": self._response_value(metadata, "refusal"), + "usage": self._response_value(metadata, "usage"), + } + self._validate_responses_status(response) + return self._normalize_responses_response( + response, + response_format=response_format, + model=model, + ) + + async def _stream_responses( + self, + *, + model: str, + messages: list[dict[str, Any]], + max_tokens: int, + temperature: float | None, + stop: list[str] | None, + thinking_budget_tokens: int | None, + tools: list[dict[str, Any]] | None, + tool_choice: str | dict[str, Any] | None, + response_format: type[BaseModel] | dict[str, Any] | None, + thinking_effort: str | None, + extra_params: dict[str, Any], + ) -> AsyncIterator[StreamChunk]: + self._validate_responses_options( + stop=stop, + thinking_budget_tokens=thinking_budget_tokens, + extra_params=extra_params, + ) + params = self._build_responses_params( + model=model, + messages=messages, + max_tokens=max_tokens, + temperature=temperature, + tools=tools, + tool_choice=tool_choice, + response_format=response_format, + thinking_effort=thinking_effort, + extra_params=extra_params, + ) + terminal_response: Any = None + output_items: list[Any] = [] + async with self._client.responses.stream(**params) as stream: + async for event in stream: + event_type = self._response_value(event, "type") + if event_type == "response.output_text.delta": + delta = self._response_value(event, "delta") + if isinstance(delta, str) and delta: + yield StreamChunk(content=delta) + elif event_type == "response.output_item.done": + item = self._response_value(event, "item") + if item is not None: + output_items.append(item) + elif event_type in { + "response.completed", + "response.incomplete", + "response.failed", + "response.cancelled", + }: + terminal_response = self._response_value(event, "response") + if terminal_response is None: + terminal_response = await stream.get_final_response() + validation_response = terminal_response + if output_items and not self._response_value(terminal_response, "output"): + validation_response = { + "status": self._response_value(terminal_response, "status"), + "incomplete_details": self._response_value( + terminal_response, "incomplete_details" + ), + "refusal": self._response_value(terminal_response, "refusal"), + "output": output_items, + } + self._validate_responses_status(validation_response) + usage = self._response_value(terminal_response, "usage") + output_tokens = self._response_value(usage, "output_tokens") + yield StreamChunk( + is_done=True, + finish_reason=self._responses_finish_reason(terminal_response), + output_tokens=( + int(output_tokens) if isinstance(output_tokens, int) else None + ), + ) + + def _build_responses_params( + self, + *, + model: str, + messages: list[dict[str, Any]], + max_tokens: int, + temperature: float | None = None, + tools: list[dict[str, Any]] | None, + tool_choice: str | dict[str, Any] | None, + response_format: type[BaseModel] | dict[str, Any] | None, + thinking_effort: str | None, + extra_params: dict[str, Any], + ) -> dict[str, Any]: + instructions, response_input = self._messages_to_responses_input(messages) + if ( + isinstance(response_format, type) + and self._structured_output_mode(extra_params) == "json_object" + ): + schema_instruction = _json_object_instruction( + response_format, tools_present=bool(tools) + ) + instructions = "\n\n".join( + part for part in (instructions, schema_instruction) if part + ) + params: dict[str, Any] = { + "model": model, + "input": response_input, + "store": False, + } + if temperature is not None: + params["temperature"] = temperature + if "top_p" in extra_params and extra_params["top_p"] is not None: + params["top_p"] = extra_params["top_p"] + if not extra_params.get("omit_max_output_tokens"): + params["max_output_tokens"] = max_tokens + if instructions: + params["instructions"] = instructions + if thinking_effort: + params["reasoning"] = {"effort": thinking_effort} + if tools: + params["tools"] = self._convert_responses_tools(tools) + converted_choice = self._convert_responses_tool_choice(tool_choice) + if converted_choice is not None: + params["tool_choice"] = converted_choice + text_config = self._responses_text_config(response_format, extra_params) + if text_config: + params["text"] = text_config + apply_sdk_passthroughs(params, extra_params) + timeout = request_timeout_from_extra_params(extra_params) + if timeout is not None: + params["timeout"] = timeout + return params + + @staticmethod + def _validate_responses_options( + *, + stop: list[str] | None, + thinking_budget_tokens: int | None, + extra_params: dict[str, Any], + ) -> None: + unsupported = { + "stop": stop, + "presence_penalty": extra_params.get("presence_penalty"), + "frequency_penalty": extra_params.get("frequency_penalty"), + "seed": extra_params.get("seed"), + "thinking_budget_tokens": thinking_budget_tokens, + } + used = [name for name, value in unsupported.items() if value is not None] + if used: + raise ValidationException( + "Responses API does not support non-null provider options: " + + ", ".join(used) + ) + + @classmethod + def _convert_responses_tool_choice( + cls, + tool_choice: str | dict[str, Any] | None, + ) -> str | dict[str, Any] | None: + if tool_choice is None: + return None + if isinstance(tool_choice, dict): + name = tool_choice.get("name") + if name is None and isinstance(tool_choice.get("function"), dict): + name = tool_choice["function"].get("name") + if name is not None: + return {"type": "function", "name": name} + return tool_choice + if tool_choice in {"any", "required"}: + return "required" + if tool_choice in {"auto", "none"}: + return tool_choice + return {"type": "function", "name": tool_choice} + + @classmethod + def _messages_to_responses_input( + cls, messages: list[dict[str, Any]] + ) -> tuple[str, list[dict[str, Any]]]: + """Convert canonical and Anthropic tool history to Responses input. + + Unsupported provider-specific content blocks fail explicitly so a + fallback cannot silently discard conversation history. + """ + instructions: list[str] = [] + response_input: list[dict[str, Any]] = [] + seen_function_call_ids: set[str] = set() + for message in messages: + role = str(message.get("role") or "user") + content = message.get("content") + if role in {"system", "developer"}: + if isinstance(content, str) and content: + instructions.append(content) + elif content is not None and not isinstance(content, str): + raise ValidationException( + "Responses system/developer content must be plain text" + ) + continue + if role == "tool": + response_input.append( + { + "type": "function_call_output", + "call_id": str(message.get("tool_call_id") or ""), + "output": content + if isinstance(content, str) + else json.dumps(content), + } + ) + continue + raw_reasoning_details = message.get("reasoning_details") + if isinstance(raw_reasoning_details, list): + for raw_reasoning in cast(list[Any], raw_reasoning_details): + if isinstance(raw_reasoning, BaseModel): + reasoning = raw_reasoning.model_dump() + elif isinstance(raw_reasoning, dict): + reasoning = dict(cast(dict[str, Any], raw_reasoning)) + else: + continue + if reasoning.get("type") == "reasoning": + response_input.append(reasoning) + if isinstance(content, str): + if content: + response_input.append({"role": role, "content": content}) + elif isinstance(content, list): + for raw_block in cast(list[Any], content): + if not isinstance(raw_block, dict): + raise ValidationException( + "Unsupported Responses content block: expected an object" + ) + block = cast(dict[str, Any], raw_block) + block_type = block.get("type") + if block_type == "text": + text = block.get("text") + if isinstance(text, str) and text: + response_input.append({"role": role, "content": text}) + elif block_type == "tool_use": + call_id = str(block.get("id") or "") + tool_input = block.get("input") + if not isinstance(tool_input, dict): + raise ValidationException( + "Unsupported Responses content block: " + + "tool_use input must be an object" + ) + response_input.append( + { + "type": "function_call", + "call_id": call_id, + "name": str(block.get("name") or ""), + "arguments": json.dumps(tool_input), + } + ) + seen_function_call_ids.add(call_id) + elif block_type == "tool_result": + result = block.get("content") + response_input.append( + { + "type": "function_call_output", + "call_id": str(block.get("tool_use_id") or ""), + "output": result + if isinstance(result, str) + else json.dumps(result), + } + ) + else: + block_name = block_type or "" + raise ValidationException( + f"Unsupported Responses content block type: {block_name}" + ) + elif content is not None: + raise ValidationException( + "Unsupported Responses message content: expected text or blocks" + ) + if content is None and message.get("parts") is not None: + raise ValidationException( + "Unsupported Responses message shape: " + + "Gemini parts require provider-specific conversion" + ) + raw_tool_calls = message.get("tool_calls") + if isinstance(raw_tool_calls, list): + tool_call_values = cast(list[Any], raw_tool_calls) + for raw_tool_call in tool_call_values: + if not isinstance(raw_tool_call, dict): + continue + tool_call = cast(dict[str, Any], raw_tool_call) + call_id = str(tool_call.get("id") or "") + if call_id in seen_function_call_ids: + continue + raw_function = tool_call.get("function") + function = ( + cast(dict[str, Any], raw_function) + if isinstance(raw_function, dict) + else {} + ) + response_input.append( + { + "type": "function_call", + "call_id": call_id, + "name": str(function.get("name") or ""), + "arguments": str(function.get("arguments") or "{}"), + } + ) + seen_function_call_ids.add(call_id) + return "\n\n".join(instructions), response_input + + @staticmethod + def _convert_responses_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]: + converted: list[dict[str, Any]] = [] + for tool in tools: + function = tool.get("function") if tool.get("type") == "function" else tool + function = function or {} + converted.append( + { + "type": "function", + "name": function.get("name", ""), + "description": function.get("description", ""), + "parameters": function.get( + "parameters", + function.get( + "input_schema", {"type": "object", "properties": {}} + ), + ), + "strict": False, + } + ) + return converted + + @staticmethod + def _responses_text_config( + response_format: type[BaseModel] | dict[str, Any] | None, + extra_params: dict[str, Any], + ) -> dict[str, Any] | None: + text: dict[str, Any] = {} + verbosity = extra_params.get("verbosity") + if verbosity: + text["verbosity"] = verbosity + if isinstance(response_format, type): + if extra_params.get("structured_output_mode") == "json_object": + text["format"] = {"type": "json_object"} + else: + text["format"] = { + "type": "json_schema", + "name": response_format.__name__, + "schema": to_strict_json_schema(response_format), + "strict": True, + } + elif isinstance(response_format, dict): + if response_format.get("type") == "json_schema": + raw_schema = response_format.get("json_schema") + schema = ( + cast(dict[str, Any], raw_schema) + if isinstance(raw_schema, dict) + else {} + ) + text["format"] = { + "type": "json_schema", + "name": schema.get("name", "response"), + "schema": schema.get("schema", {}), + "strict": schema.get("strict", True), + } + elif response_format.get("type") == "json_object": + text["format"] = {"type": "json_object"} + elif extra_params.get("json_mode"): + text["format"] = {"type": "json_object"} + return text or None + + def _normalize_responses_response( + self, + response: Any, + *, + response_format: type[BaseModel] | dict[str, Any] | None, + model: str, + ) -> CompletionResult: + raw_output = self._response_value(response, "output") + output: list[Any] = ( + cast(list[Any], raw_output) if isinstance(raw_output, list) else [] + ) + raw_text = self._response_value(response, "output_text") + text = raw_text if isinstance(raw_text, str) else "" + tool_calls: list[ToolCallResult] = [] + reasoning_details: list[dict[str, Any]] = [] + for item in output: + item_type = self._response_value(item, "type") + if item_type == "function_call": + arguments = self._response_value(item, "arguments") or "{}" + try: + parsed_arguments: Any = json.loads(arguments) + except (json.JSONDecodeError, TypeError): + parsed_arguments = {} + tool_input = ( + cast(dict[str, Any], parsed_arguments) + if isinstance(parsed_arguments, dict) + else {} + ) + tool_calls.append( + ToolCallResult( + id=str( + self._response_value(item, "call_id") + or self._response_value(item, "id") + or "" + ), + name=str(self._response_value(item, "name") or ""), + input=tool_input, + ) + ) + elif item_type == "reasoning": + if isinstance(item, BaseModel): + reasoning_details.append(item.model_dump()) + elif isinstance(item, dict): + reasoning_details.append(cast(dict[str, Any], item)) + content: Any = text + if isinstance(response_format, type) and text: + try: + content = validate_structured_output(text, response_format) + except (StructuredOutputError, ValidationError): + content = repair_response_model_json(text, response_format, model) + usage = self._response_value(response, "usage") + input_tokens = self._response_value(usage, "input_tokens") or 0 + output_tokens = self._response_value(usage, "output_tokens") or 0 + details = self._response_value(usage, "input_tokens_details") + cached_tokens = self._response_value(details, "cached_tokens") or 0 + return CompletionResult( + content=content, + input_tokens=int(input_tokens), + output_tokens=int(output_tokens), + cache_read_input_tokens=int(cached_tokens), + finish_reason=self._responses_finish_reason(response), + tool_calls=tool_calls, + reasoning_details=reasoning_details, + raw_response=response, + ) + + @classmethod + def _responses_finish_reason(cls, response: Any) -> str: + status = cls._response_value(response, "status") + if status == "completed": + return "stop" + if status == "incomplete": + incomplete = cls._response_value(response, "incomplete_details") + reason = cls._response_value(incomplete, "reason") + if reason in {"max_output_tokens", "max_tokens"}: + return "length" + raise cls._responses_status_error(response) + + @classmethod + def _responses_refusal(cls, response: Any) -> str | None: + refusal = cls._response_value(response, "refusal") + if isinstance(refusal, str) and refusal: + return refusal + raw_output = cls._response_value(response, "output") + output: list[Any] = ( + cast(list[Any], raw_output) if isinstance(raw_output, list) else [] + ) + for item in output: + item_refusal = cls._response_value(item, "refusal") + if isinstance(item_refusal, str) and item_refusal: + return item_refusal + raw_content = cls._response_value(item, "content") + content: list[Any] = ( + cast(list[Any], raw_content) if isinstance(raw_content, list) else [] + ) + for part in content: + if cls._response_value(part, "type") != "refusal": + continue + part_refusal = cls._response_value(part, "refusal") + if isinstance(part_refusal, str) and part_refusal: + return part_refusal + return None + + @classmethod + def _responses_status_error(cls, response: Any) -> Exception: + status = cls._response_value(response, "status") or "unknown" + refusal = cls._responses_refusal(response) + if refusal is not None or status == "refused": + return ValidationException("Responses provider refusal") + return ValidationException( + f"Responses provider returned non-success status: {status}" + ) + + @classmethod + def _validate_responses_status(cls, response: Any) -> None: + if cls._responses_refusal(response) is not None: + raise cls._responses_status_error(response) + status = cls._response_value(response, "status") + if status == "completed": + return + if status == "incomplete": + incomplete = cls._response_value(response, "incomplete_details") + reason = cls._response_value(incomplete, "reason") + if reason in {"max_output_tokens", "max_tokens"}: + return + raise cls._responses_status_error(response) + + @staticmethod + def _response_value(value: Any, key: str) -> Any: + if isinstance(value, dict): + mapping = cast(dict[str, Any], value) + return mapping.get(key) + return getattr(value, key, None) + def _build_params( self, *, diff --git a/tests/llm/test_backends/test_openai.py b/tests/llm/test_backends/test_openai.py index b4df2e66..584ee0aa 100644 --- a/tests/llm/test_backends/test_openai.py +++ b/tests/llm/test_backends/test_openai.py @@ -9,9 +9,15 @@ from unittest.mock import AsyncMock, Mock import httpx import pytest from openai import BadRequestError +from openai.types.responses import ( + ResponseOutputMessage, + ResponseOutputRefusal, + ResponseReasoningItem, +) from pydantic import BaseModel from src.exceptions import ValidationException +from src.llm.backend import StreamChunk from src.llm.backends.openai import ( OpenAIBackend, _json_object_instruction, # pyright: ignore[reportPrivateUsage] @@ -1312,3 +1318,957 @@ async def test_openai_backend_structured_without_tools_still_uses_parse() -> Non assert result.content is parsed client.chat.completions.create.assert_not_awaited() + + +class _FakeResponsesStream: + _events: list[Any] + _final_response: Any + + def __init__(self, events: list[Any], final_response: Any) -> None: + self._events = events + self._final_response = final_response + + async def __aenter__(self) -> "_FakeResponsesStream": + return self + + async def __aexit__(self, *_args: Any) -> None: + return None + + def __aiter__(self) -> AsyncIterator[Any]: + async def _iter() -> AsyncIterator[Any]: + for event in self._events: + yield event + + return _iter() + + async def get_final_response(self) -> Any: + return self._final_response + + +def _responses_final_response( + *, + text: str = "ok", + output: list[Any] | None = None, +) -> SimpleNamespace: + return SimpleNamespace( + output_text=text, + output=output + if output is not None + else [ + SimpleNamespace( + type="message", + content=[SimpleNamespace(type="output_text", text=text)], + ) + ], + usage=SimpleNamespace( + input_tokens=11, + output_tokens=7, + input_tokens_details=SimpleNamespace(cached_tokens=3), + ), + status="completed", + ) + + +@pytest.mark.asyncio +async def test_openai_backend_responses_mode_builds_and_normalizes_completion() -> None: + client = Mock() + final = _responses_final_response(text="", output=[]) + stream = _FakeResponsesStream( + [ + SimpleNamespace( + type="response.output_text.delta", delta='{"answer":"yes"}' + ), + SimpleNamespace(type="response.completed", response=final), + ], + final, + ) + client.responses.stream = Mock(return_value=stream) + + backend = OpenAIBackend(client) + result = await backend.complete( + model="gpt-5.6-luna", + messages=[ + {"role": "system", "content": "Be precise."}, + {"role": "user", "content": "Answer."}, + ], + max_tokens=100, + response_format=_StructuredResponse, + thinking_effort="low", + extra_params={"api_mode": "responses"}, + ) + + assert result.content == _StructuredResponse(answer="yes") + assert result.input_tokens == 11 + assert result.output_tokens == 7 + assert result.cache_read_input_tokens == 3 + call = client.responses.stream.call_args.kwargs + assert call["model"] == "gpt-5.6-luna" + assert call["instructions"] == "Be precise." + assert call["input"] == [{"role": "user", "content": "Answer."}] + assert call["max_output_tokens"] == 100 + assert call["reasoning"] == {"effort": "low"} + assert call["text"]["format"]["type"] == "json_schema" + assert call["text"]["format"]["name"] == "_StructuredResponse" + assert call["text"]["format"]["schema"]["additionalProperties"] is False + assert call["store"] is False + + +@pytest.mark.asyncio +async def test_responses_sdk_reasoning_survives_completion_normalization() -> None: + client = Mock() + reasoning = ResponseReasoningItem( + id="reasoning_sdk", + summary=[], + type="reasoning", + encrypted_content="encrypted-sdk-reasoning", + status="completed", + ) + function_call = SimpleNamespace( + type="function_call", + call_id="call_sdk_reasoning", + name="lookup", + arguments='{"topic":"memory"}', + ) + final = _responses_final_response( + text="reasoned sdk answer", output=[reasoning, function_call] + ) + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + + result = await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Reason, then look it up"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + assert result.content == "reasoned sdk answer" + assert result.tool_calls[0].id == "call_sdk_reasoning" + assert result.tool_calls[0].name == "lookup" + assert result.tool_calls[0].input == {"topic": "memory"} + assert result.input_tokens == 11 + assert result.output_tokens == 7 + assert result.cache_read_input_tokens == 3 + assert result.finish_reason == "stop" + assert result.reasoning_details == [reasoning.model_dump()] + + +@pytest.mark.asyncio +async def test_responses_dict_reasoning_survives_completion_normalization() -> None: + client = Mock() + reasoning: dict[str, Any] = { + "id": "reasoning_dict", + "summary": [{"type": "summary_text", "text": "Checked memory"}], + "type": "reasoning", + "status": "completed", + } + final: dict[str, Any] = { + "output_text": "reasoned dict answer", + "output": [ + reasoning, + { + "type": "function_call", + "call_id": "call_dict_reasoning", + "name": "lookup", + "arguments": '{"topic":"memory"}', + }, + ], + "usage": { + "input_tokens": 13, + "output_tokens": 8, + "input_tokens_details": {"cached_tokens": 5}, + }, + "status": "completed", + } + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [{"type": "response.completed", "response": final}], final + ) + ) + + result = await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Reason, then look it up"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + assert result.content == "reasoned dict answer" + assert result.tool_calls[0].id == "call_dict_reasoning" + assert result.tool_calls[0].name == "lookup" + assert result.tool_calls[0].input == {"topic": "memory"} + assert result.input_tokens == 13 + assert result.output_tokens == 8 + assert result.cache_read_input_tokens == 5 + assert result.finish_reason == "stop" + assert result.reasoning_details == [reasoning] + + +@pytest.mark.asyncio +async def test_responses_reasoning_details_are_replayed_for_tool_continuation() -> None: + client = Mock() + reasoning = ResponseReasoningItem( + id="reasoning_replay", + summary=[], + type="reasoning", + encrypted_content="encrypted-replay", + status="completed", + ).model_dump() + final = _responses_final_response(text="done") + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + + await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[ + {"role": "user", "content": "Look it up"}, + { + "role": "assistant", + "content": None, + "reasoning_details": [reasoning], + "tool_calls": [ + { + "id": "call_replay", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"topic":"memory"}', + }, + } + ], + }, + { + "role": "tool", + "tool_call_id": "call_replay", + "content": "found it", + }, + ], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + response_input = client.responses.stream.call_args.kwargs["input"] + assert response_input == [ + {"role": "user", "content": "Look it up"}, + reasoning, + { + "type": "function_call", + "call_id": "call_replay", + "name": "lookup", + "arguments": '{"topic":"memory"}', + }, + { + "type": "function_call_output", + "call_id": "call_replay", + "output": "found it", + }, + ] + + +@pytest.mark.asyncio +async def test_responses_mode_supports_tools_with_structured_output() -> None: + client = Mock() + function_call = SimpleNamespace( + type="function_call", + call_id="call_structured", + name="search", + arguments='{"query":"honcho"}', + ) + final = _responses_final_response(text="", output=[]) + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [ + SimpleNamespace( + type="response.output_text.delta", delta='{"answer":"yes"}' + ), + SimpleNamespace(type="response.output_item.done", item=function_call), + SimpleNamespace(type="response.completed", response=final), + ], + final, + ) + ) + + result = await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Search and answer"}], + max_tokens=100, + tools=[AGENT_TOOL], + response_format=_StructuredResponse, + extra_params={"api_mode": "responses"}, + ) + + assert result.content == _StructuredResponse(answer="yes") + assert result.tool_calls[0].name == "search" + assert result.tool_calls[0].input == {"query": "honcho"} + call = client.responses.stream.call_args.kwargs + assert call["tools"][0]["name"] == "search" + assert call["text"]["format"]["type"] == "json_schema" + assert call["text"]["format"]["name"] == "_StructuredResponse" + + +@pytest.mark.asyncio +async def test_responses_completion_preserves_terminal_output_with_text_deltas() -> ( + None +): + client = Mock() + function_call = SimpleNamespace( + type="function_call", + call_id="call_terminal", + name="lookup", + arguments='{"topic":"terminal"}', + ) + final = _responses_final_response(text="", output=[function_call]) + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [ + SimpleNamespace(type="response.output_text.delta", delta="streamed"), + SimpleNamespace(type="response.completed", response=final), + ], + final, + ) + ) + + result = await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Look it up"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + assert result.content == "streamed" + assert result.tool_calls[0].id == "call_terminal" + assert result.tool_calls[0].name == "lookup" + assert result.tool_calls[0].input == {"topic": "terminal"} + + +@pytest.mark.asyncio +async def test_responses_dict_events_reconstruct_completion_without_usage() -> None: + client = Mock() + final = {"status": "completed", "usage": None} + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [ + {"type": "response.output_text.delta", "delta": "dict "}, + {"type": "response.output_text.delta", "delta": "ok"}, + { + "type": "response.output_item.done", + "item": { + "type": "function_call", + "call_id": "call_dict", + "name": "lookup", + "arguments": '{"topic":"memory"}', + }, + }, + {"type": "response.completed", "response": final}, + ], + final, + ) + ) + + result = await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + assert result.content == "dict ok" + assert result.tool_calls[0].id == "call_dict" + assert result.tool_calls[0].input == {"topic": "memory"} + assert result.input_tokens == 0 + assert result.output_tokens == 0 + + +@pytest.mark.asyncio +async def test_responses_dict_stream_fallback_preserves_refusal() -> None: + client = Mock() + final: dict[str, Any] = {"status": "completed", "output": [], "usage": None} + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [ + { + "type": "response.output_item.done", + "item": { + "type": "message", + "content": [{"type": "refusal", "refusal": "unsafe request"}], + }, + } + ], + final, + ) + ) + + chunks: list[StreamChunk] = [] + with pytest.raises(ValidationException, match=r"^Responses provider refusal$"): + async for chunk in OpenAIBackend(client).stream( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ): + chunks.append(chunk) + + assert not any(chunk.is_done for chunk in chunks) + + +def test_responses_converts_anthropic_tool_history_without_dropping_blocks() -> None: + instructions, response_input = OpenAIBackend._messages_to_responses_input( # pyright: ignore[reportPrivateUsage] + [ + {"role": "user", "content": "Find this."}, + { + "role": "assistant", + "content": [ + {"type": "text", "text": "I will search."}, + { + "type": "tool_use", + "id": "call_anthropic", + "name": "lookup", + "input": {"topic": "history"}, + }, + ], + }, + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "call_anthropic", + "content": "found it", + "is_error": False, + } + ], + }, + ] + ) + + assert instructions == "" + assert response_input == [ + {"role": "user", "content": "Find this."}, + {"role": "assistant", "content": "I will search."}, + { + "type": "function_call", + "call_id": "call_anthropic", + "name": "lookup", + "arguments": '{"topic": "history"}', + }, + { + "type": "function_call_output", + "call_id": "call_anthropic", + "output": "found it", + }, + ] + + +def test_responses_rejects_unsupported_list_content_instead_of_dropping_it() -> None: + with pytest.raises( + ValidationException, match="Unsupported Responses content block" + ): + OpenAIBackend._messages_to_responses_input( # pyright: ignore[reportPrivateUsage] + [{"role": "assistant", "content": [{"type": "thinking", "text": "secret"}]}] + ) + + +def test_responses_rejects_gemini_parts_instead_of_dropping_them() -> None: + with pytest.raises( + ValidationException, match="Unsupported Responses message shape" + ): + OpenAIBackend._messages_to_responses_input( # pyright: ignore[reportPrivateUsage] + [{"role": "model", "parts": [{"text": "I will search."}]}] + ) + + +def test_responses_json_object_mode_injects_schema_instructions() -> None: + backend = OpenAIBackend(Mock()) + params = backend._build_responses_params( # pyright: ignore[reportPrivateUsage] + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Answer."}], + max_tokens=100, + tools=None, + tool_choice=None, + response_format=_StructuredResponse, + thinking_effort=None, + extra_params={ + "api_mode": "responses", + "structured_output_mode": "json_object", + }, + ) + + assert params["text"]["format"] == {"type": "json_object"} + assert "answer" in params["instructions"] + + +def test_responses_json_object_mode_keeps_tool_calls_available() -> None: + backend = OpenAIBackend(Mock()) + params = backend._build_responses_params( # pyright: ignore[reportPrivateUsage] + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Search, then answer."}], + max_tokens=100, + tools=[AGENT_TOOL], + tool_choice="auto", + response_format=_StructuredResponse, + thinking_effort=None, + extra_params={ + "api_mode": "responses", + "structured_output_mode": "json_object", + }, + ) + + assert params["text"]["format"] == {"type": "json_object"} + assert params["tools"][0]["name"] == "search" + assert "If not responding with a tool call" in params["instructions"] + + +def test_openai_backend_responses_mode_can_omit_max_output_tokens() -> None: + backend = OpenAIBackend(Mock()) + params = backend._build_responses_params( # pyright: ignore[reportPrivateUsage] + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Answer."}], + max_tokens=100, + tools=None, + tool_choice=None, + response_format=None, + thinking_effort=None, + extra_params={"api_mode": "responses", "omit_max_output_tokens": True}, + ) + assert "max_output_tokens" not in params + + +@pytest.mark.asyncio +async def test_openai_backend_responses_mode_converts_tools_and_tool_history() -> None: + client = Mock() + function_call = SimpleNamespace( + type="function_call", + call_id="call_123", + name="lookup", + arguments='{"topic":"memory"}', + ) + final = _responses_final_response(text="", output=[]) + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [ + SimpleNamespace(type="response.output_item.done", item=function_call), + SimpleNamespace(type="response.completed", response=final), + ], + final, + ) + ) + + backend = OpenAIBackend(client) + result = await backend.complete( + model="gpt-5.6-luna", + messages=[ + {"role": "user", "content": "Look it up"}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call_old", + "type": "function", + "function": {"name": "lookup", "arguments": '{"topic":"old"}'}, + } + ], + }, + {"role": "tool", "tool_call_id": "call_old", "content": "old result"}, + ], + max_tokens=100, + tools=[ + { + "name": "lookup", + "description": "Look up a topic", + "input_schema": { + "type": "object", + "properties": {"topic": {"type": "string"}}, + }, + } + ], + tool_choice="required", + extra_params={"api_mode": "responses"}, + ) + + assert result.tool_calls[0].id == "call_123" + assert result.tool_calls[0].name == "lookup" + assert result.tool_calls[0].input == {"topic": "memory"} + call = client.responses.stream.call_args.kwargs + assert call["tools"][0] == { + "type": "function", + "name": "lookup", + "description": "Look up a topic", + "parameters": { + "type": "object", + "properties": {"topic": {"type": "string"}}, + }, + "strict": False, + } + assert call["tool_choice"] == "required" + assert any(item.get("type") == "function_call" for item in call["input"]) + assert any(item.get("type") == "function_call_output" for item in call["input"]) + + +@pytest.mark.asyncio +async def test_openai_backend_responses_mode_streams_text_and_done_usage() -> None: + client = Mock() + events = [ + SimpleNamespace(type="response.output_text.delta", delta="hel"), + SimpleNamespace(type="response.output_text.delta", delta="lo"), + SimpleNamespace( + type="response.completed", + response=_responses_final_response(text="hello"), + ), + ] + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + events, _responses_final_response(text="hello") + ) + ) + + backend = OpenAIBackend(client) + chunks = [ + chunk + async for chunk in backend.stream( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + ] + + assert [chunk.content for chunk in chunks if chunk.content] == ["hel", "lo"] + assert chunks[-1].is_done is True + assert chunks[-1].finish_reason == "stop" + assert chunks[-1].output_tokens == 7 + + +@pytest.mark.asyncio +async def test_responses_forwards_temperature_top_p_timeout_and_passthroughs() -> None: + client = Mock() + final = _responses_final_response() + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + temperature=0.2, + extra_params={ + "api_mode": "responses", + "top_p": 0.7, + "timeout": "12.5", + "extra_body": {"operator": "wins"}, + "extra_headers": {"x-request": "yes"}, + "extra_query": {"trace": "abc"}, + }, + ) + call = client.responses.stream.call_args.kwargs + assert call["temperature"] == 0.2 + assert call["top_p"] == 0.7 + assert call["timeout"] == 12.5 + assert call["extra_body"] == {"operator": "wins"} + assert call["extra_headers"] == {"x-request": "yes"} + assert call["extra_query"] == {"trace": "abc"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("status", "incomplete_details", "expected_finish"), + [ + ("completed", None, "stop"), + ("incomplete", SimpleNamespace(reason="max_output_tokens"), "length"), + ], + ids=["completed", "max-output-incomplete"], +) +async def test_responses_status_maps_success_and_length( + status: str, + incomplete_details: Any, + expected_finish: str, +) -> None: + client = Mock() + final = _responses_final_response() + final.status = status + final.incomplete_details = incomplete_details + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + + result = await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + assert result.finish_reason == expected_finish + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["failed", "cancelled", "in_progress"]) +async def test_responses_status_failures_raise(status: str) -> None: + client = Mock() + final = _responses_final_response() + final.status = status + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + + with pytest.raises(ValidationException, match="non-success status"): + await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + +@pytest.mark.asyncio +async def test_responses_refusal_is_explicit() -> None: + client = Mock() + final = _responses_final_response(text="") + final.status = "refused" + final.refusal = "unsafe request" + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + + with pytest.raises(ValidationException, match="provider refusal"): + await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + +@pytest.mark.asyncio +async def test_responses_completed_output_refusal_raises() -> None: + client = Mock() + refusal_message = ResponseOutputMessage( + id="msg_refusal", + content=[ResponseOutputRefusal(type="refusal", refusal="unsafe request")], + role="assistant", + status="completed", + type="message", + ) + final = _responses_final_response(text="", output=[refusal_message]) + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + + with pytest.raises(ValidationException, match=r"^Responses provider refusal$"): + await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + +@pytest.mark.asyncio +async def test_responses_refusal_error_does_not_expose_provider_payload() -> None: + client = Mock() + provider_payload = "secret-token\r\nforged-log-line" * 30 + refusal_message = ResponseOutputMessage( + id="msg_sensitive_refusal", + content=[ResponseOutputRefusal(type="refusal", refusal=provider_payload)], + role="assistant", + status="completed", + type="message", + ) + final = _responses_final_response(text="", output=[refusal_message]) + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type="response.completed", response=final)], final + ) + ) + + with pytest.raises(ValidationException) as exc_info: + await OpenAIBackend(client).complete( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + + error_text = str(exc_info.value) + assert error_text == "Responses provider refusal" + assert provider_payload not in error_text + assert "\r" not in error_text + assert "\n" not in error_text + + +@pytest.mark.asyncio +async def test_responses_stream_output_refusal_raises_without_terminal_chunk() -> None: + client = Mock() + refusal_message = ResponseOutputMessage( + id="msg_refusal", + content=[ResponseOutputRefusal(type="refusal", refusal="unsafe request")], + role="assistant", + status="completed", + type="message", + ) + # The Codex Responses endpoint can omit output from the completed response, + # so the adapter must preserve refusal output-item events for validation. + final = _responses_final_response(text="", output=[]) + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [ + SimpleNamespace(type="response.output_item.done", item=refusal_message), + SimpleNamespace(type="response.completed", response=final), + ], + final, + ) + ) + + chunks: list[StreamChunk] = [] + with pytest.raises(ValidationException, match=r"^Responses provider refusal$"): + async for chunk in OpenAIBackend(client).stream( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ): + chunks.append(chunk) + + assert not any(chunk.is_done for chunk in chunks) + + +@pytest.mark.parametrize( + ("tool_choice", "expected"), + [ + ("any", "required"), + ("required", "required"), + ("auto", "auto"), + ("none", "none"), + ( + {"type": "function", "function": {"name": "lookup"}}, + {"type": "function", "name": "lookup"}, + ), + ], + ids=["any", "required", "auto", "none", "named"], +) +def test_responses_tool_choice_uses_flat_named_function_shape( + tool_choice: Any, + expected: Any, +) -> None: + params = OpenAIBackend(Mock())._build_responses_params( # pyright: ignore[reportPrivateUsage] + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Look up"}], + max_tokens=100, + tools=[ + { + "name": "lookup", + "description": "Look up a topic", + "input_schema": {"type": "object", "properties": {}}, + } + ], + tool_choice=tool_choice, + response_format=None, + thinking_effort=None, + extra_params={"api_mode": "responses"}, + ) + assert params["tool_choice"] == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "kwargs", + [ + {"stop": ["END"]}, + {"extra_params": {"api_mode": "responses", "presence_penalty": 0.2}}, + {"extra_params": {"api_mode": "responses", "frequency_penalty": 0.2}}, + {"extra_params": {"api_mode": "responses", "seed": 7}}, + {"thinking_budget_tokens": 32}, + ], + ids=["stop", "presence-penalty", "frequency-penalty", "seed", "thinking-budget"], +) +async def test_responses_rejects_unsupported_options_before_provider_call( + kwargs: dict[str, Any], +) -> None: + client = Mock() + client.responses.stream = Mock() + call_kwargs: dict[str, Any] = { + "model": "gpt-5.6-luna", + "messages": [{"role": "user", "content": "Hello"}], + "max_tokens": 100, + "extra_params": {"api_mode": "responses"}, + } + call_kwargs.update(kwargs) + + with pytest.raises(ValidationException, match="does not support"): + await OpenAIBackend(client).complete(**call_kwargs) + client.responses.stream.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["failed", "cancelled"]) +async def test_responses_stream_status_failures_raise(status: str) -> None: + client = Mock() + final = _responses_final_response(text="partial") + final.status = status + event_type = f"response.{status}" + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [SimpleNamespace(type=event_type, response=final)], final + ) + ) + + with pytest.raises(ValidationException, match="non-success status"): + _ = [ + chunk + async for chunk in OpenAIBackend(client).stream( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + ] + + +@pytest.mark.asyncio +async def test_responses_stream_fallback_emits_one_terminal_chunk_without_usage() -> ( + None +): + client = Mock() + final = _responses_final_response(text="hello") + final.usage = None + client.responses.stream = Mock( + return_value=_FakeResponsesStream( + [ + SimpleNamespace(type="response.output_text.delta", delta="hel"), + SimpleNamespace(type="response.output_text.delta", delta="lo"), + ], + final, + ) + ) + + chunks = [ + chunk + async for chunk in OpenAIBackend(client).stream( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + extra_params={"api_mode": "responses"}, + ) + ] + + assert [chunk.content for chunk in chunks if chunk.content] == ["hel", "lo"] + terminal = [chunk for chunk in chunks if chunk.is_done] + assert len(terminal) == 1 + assert terminal[0].finish_reason == "stop" + assert terminal[0].output_tokens is None diff --git a/tests/llm/test_request_builder.py b/tests/llm/test_request_builder.py index a8355a51..3aba0cf7 100644 --- a/tests/llm/test_request_builder.py +++ b/tests/llm/test_request_builder.py @@ -4,7 +4,7 @@ from pydantic import BaseModel from src.config import ModelConfig from src.exceptions import ValidationException from src.llm.caching import PromptCachePolicy -from src.llm.request_builder import execute_completion +from src.llm.request_builder import execute_completion, execute_stream from tests.llm.conftest import FakeBackend @@ -99,6 +99,31 @@ async def test_provider_params_are_merged_into_extra_params( assert call["extra_params"]["custom_flag"] is True +async def test_provider_params_are_merged_into_stream_extra_params( + fake_backend: FakeBackend, +) -> None: + """The shared config translation also reaches the streaming backend call.""" + config = ModelConfig( + model="gpt-4.1-mini", + transport="openai", + top_p=0.9, + provider_params={"custom_flag": True}, + ) + + stream = await execute_stream( + fake_backend, + config, + messages=[{"role": "user", "content": "Hello"}], + max_tokens=100, + ) + chunks = [chunk async for chunk in stream] + + assert len(chunks) == 1 + call = fake_backend.calls[0] + assert call["extra_params"]["top_p"] == 0.9 + assert call["extra_params"]["custom_flag"] is True + + async def test_provider_timeout_is_normalized_into_extra_params( fake_backend: FakeBackend, ) -> None: diff --git a/tests/llm/test_responses_config.py b/tests/llm/test_responses_config.py new file mode 100644 index 00000000..70c09395 --- /dev/null +++ b/tests/llm/test_responses_config.py @@ -0,0 +1,90 @@ +import pytest + +from src.config import ( + ConfiguredModelSettings, + FallbackModelSettings, + ModelConfig, + ModelOverrideSettings, + ResolvedFallbackConfig, + resolve_model_config, +) + + +@pytest.mark.parametrize("api_mode", ["response", "chat_completions", True, 1]) +def test_invalid_api_mode_is_rejected_at_config_load(api_mode: object) -> None: + with pytest.raises(ValueError, match="api_mode must be 'responses'"): + ConfiguredModelSettings( + model="gpt-5.4", + transport="openai", + overrides=ModelOverrideSettings( + provider_params={"api_mode": api_mode}, + ), + ) + + +def test_responses_api_mode_is_rejected_on_non_openai_transport() -> None: + with pytest.raises(ValueError, match="api_mode is only supported"): + ConfiguredModelSettings( + model="claude-haiku-4-5", + transport="anthropic", + overrides=ModelOverrideSettings( + provider_params={"api_mode": "responses"}, + ), + ) + + +def test_responses_api_mode_is_rejected_on_non_openai_fallback() -> None: + with pytest.raises(ValueError, match="api_mode is only supported"): + FallbackModelSettings( + model="gemini-2.5-pro", + transport="gemini", + overrides=ModelOverrideSettings( + provider_params={"api_mode": "responses"}, + ), + ) + + +def test_responses_api_mode_resolves_for_openai_transport() -> None: + configured = ConfiguredModelSettings( + model="gpt-5.4", + transport="openai", + overrides=ModelOverrideSettings( + provider_params={"api_mode": "responses"}, + ), + ) + + resolved = resolve_model_config(configured) + + assert resolved.provider_params["api_mode"] == "responses" + + +def test_runtime_model_config_rejects_responses_on_non_openai_transport() -> None: + with pytest.raises(ValueError, match="api_mode is only supported"): + ModelConfig( + model="claude-haiku-4-5", + transport="anthropic", + provider_params={"api_mode": "responses"}, + ) + + +def test_resolved_fallback_rejects_responses_on_non_openai_transport() -> None: + with pytest.raises(ValueError, match="api_mode is only supported"): + ResolvedFallbackConfig( + model="gemini-2.5-pro", + transport="gemini", + provider_params={"api_mode": "responses"}, + ) + + +def test_for_model_revalidates_responses_api_mode_transport() -> None: + config = ModelConfig( + model="gpt-5.4", + transport="openai", + provider_params={"api_mode": "responses"}, + ) + + with pytest.raises(ValueError, match="api_mode is only supported"): + config.for_model( + "claude-haiku-4-5", + transport_override="anthropic", + ) diff --git a/tests/utils/test_clients.py b/tests/utils/test_clients.py index 49b2d75d..f583115e 100644 --- a/tests/utils/test_clients.py +++ b/tests/utils/test_clients.py @@ -10,6 +10,8 @@ Tests cover: """ import contextlib +from collections.abc import AsyncIterator +from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, Mock, patch @@ -49,6 +51,60 @@ class SampleTestModel(BaseModel): active: bool = Field(default=True) +class _PublicResponsesStream: + """Minimal Responses stream used to exercise the public retry boundary.""" + + _events: list[Any] + _final_response: Any + + def __init__(self, events: list[Any], final_response: Any) -> None: + self._events = events + self._final_response = final_response + + async def __aenter__(self) -> "_PublicResponsesStream": + return self + + async def __aexit__(self, *_args: Any) -> None: + return None + + def __aiter__(self) -> AsyncIterator[Any]: + async def _iter() -> AsyncIterator[Any]: + for event in self._events: + yield event + + return _iter() + + async def get_final_response(self) -> Any: + return self._final_response + + +def _public_responses_terminal( + status: str, + *, + refusal: str | None = None, + incomplete_reason: str | None = None, +) -> SimpleNamespace: + output = [] + if refusal is not None: + output = [ + SimpleNamespace( + type="message", + content=[SimpleNamespace(type="refusal", refusal=refusal)], + ) + ] + return SimpleNamespace( + status=status, + output_text="", + output=output, + incomplete_details=( + SimpleNamespace(reason=incomplete_reason) + if incomplete_reason is not None + else None + ), + usage=SimpleNamespace(input_tokens=1, output_tokens=0), + ) + + class TestLLMCallResponse: """Tests for HonchoLLMCallResponse and HonchoLLMCallStreamChunk models""" @@ -889,6 +945,230 @@ class TestMainLLMCallFunction: assert chunks[1].content == " test" assert chunks[2].is_done is True + async def test_responses_refusal_before_text_is_unretried_during_drain(self): + """Retry wraps stream setup, not an explicit refusal found while draining.""" + response = _public_responses_terminal("completed", refusal="unsafe") + mock_client = Mock() + mock_client.responses.stream = Mock( + return_value=_PublicResponsesStream( + [SimpleNamespace(type="response.completed", response=response)], + response, + ) + ) + + with patch.dict(CLIENTS, {"openai": mock_client}): + stream = await honcho_llm_call( + model_config=ModelConfig( + model="gpt-5.4-mini", + transport="openai", + provider_params={"api_mode": "responses"}, + ), + prompt="Hello", + max_tokens=100, + stream=True, + enable_retry=True, + retry_attempts=3, + ) + chunks: list[HonchoLLMCallStreamChunk] = [] + with pytest.raises(ValidationException, match="provider refusal"): + async for chunk in stream: + chunks.append(chunk) + + assert mock_client.responses.stream.call_count == 1 + assert chunks == [] + + @pytest.mark.parametrize( + ("status", "event_type", "incomplete_reason"), + [ + ("failed", "response.failed", None), + ("cancelled", "response.cancelled", None), + ("queued", "response.queued", None), + ("incomplete", "response.incomplete", "content_filter"), + ], + ) + async def test_responses_drain_status_failures_are_explicit_and_unretried( + self, + status: str, + event_type: str, + incomplete_reason: str | None, + ) -> None: + response = _public_responses_terminal( + status, incomplete_reason=incomplete_reason + ) + mock_client = Mock() + mock_client.responses.stream = Mock( + return_value=_PublicResponsesStream( + [SimpleNamespace(type=event_type, response=response)], + response, + ) + ) + + with patch.dict(CLIENTS, {"openai": mock_client}): + stream = await honcho_llm_call( + model_config=ModelConfig( + model="gpt-5.4-mini", + transport="openai", + provider_params={"api_mode": "responses"}, + ), + prompt="Hello", + max_tokens=100, + stream=True, + enable_retry=True, + retry_attempts=3, + ) + chunks: list[HonchoLLMCallStreamChunk] = [] + with pytest.raises( + ValidationException, + match=f"non-success status: {status}", + ): + async for chunk in stream: + chunks.append(chunk) + + assert mock_client.responses.stream.call_count == 1 + assert chunks == [] + + async def test_responses_refusal_after_text_never_retries_or_emits_done(self): + """Visible text cannot be retracted, so a later refusal fails in place.""" + response = _public_responses_terminal("completed", refusal="unsafe") + mock_client = Mock() + mock_client.responses.stream = Mock( + return_value=_PublicResponsesStream( + [ + SimpleNamespace(type="response.output_text.delta", delta="partial"), + SimpleNamespace(type="response.completed", response=response), + ], + response, + ) + ) + + with patch.dict(CLIENTS, {"openai": mock_client}): + stream = await honcho_llm_call( + model_config=ModelConfig( + model="gpt-5.4-mini", + transport="openai", + provider_params={"api_mode": "responses"}, + ), + prompt="Hello", + max_tokens=100, + stream=True, + enable_retry=True, + retry_attempts=3, + ) + chunks: list[HonchoLLMCallStreamChunk] = [] + with pytest.raises(ValidationException, match="provider refusal"): + async for chunk in stream: + chunks.append(chunk) + + assert mock_client.responses.stream.call_count == 1 + assert [chunk.content for chunk in chunks] == ["partial"] + assert not any(chunk.is_done for chunk in chunks) + + async def test_stream_final_only_runs_tool_turns_non_streaming(self): + """The public stream-final-only path keeps tools off the final stream.""" + from openai import AsyncOpenAI + + mock_client = AsyncMock(spec=AsyncOpenAI) + tool_response = SimpleNamespace( + choices=[ + SimpleNamespace( + finish_reason="tool_calls", + message=SimpleNamespace( + content="", + tool_calls=[ + SimpleNamespace( + id="call_1", + function=SimpleNamespace( + name="lookup", arguments='{"q":"memory"}' + ), + ) + ], + reasoning_details=[], + ), + ) + ], + usage=SimpleNamespace( + prompt_tokens=3, + completion_tokens=2, + prompt_tokens_details=None, + ), + ) + answer_response = SimpleNamespace( + choices=[ + SimpleNamespace( + finish_reason="stop", + message=SimpleNamespace( + content="The answer", + tool_calls=[], + reasoning_details=[], + ), + ) + ], + usage=SimpleNamespace( + prompt_tokens=4, + completion_tokens=3, + prompt_tokens_details=None, + ), + ) + + async def final_stream(): + yield SimpleNamespace( + choices=[ + SimpleNamespace( + delta=SimpleNamespace(content="Final answer"), + finish_reason=None, + ) + ] + ) + yield SimpleNamespace( + choices=[ + SimpleNamespace( + delta=SimpleNamespace(content=None), finish_reason="stop" + ) + ], + usage=SimpleNamespace(completion_tokens=5), + ) + + mock_client.chat.completions.create = AsyncMock( + side_effect=[tool_response, answer_response, final_stream()] + ) + executed_tools: list[tuple[str, dict[str, Any]]] = [] + + async def execute_tool(name: str, arguments: dict[str, Any]) -> str: + executed_tools.append((name, arguments)) + return "lookup result" + + with patch.dict(CLIENTS, {"openai": mock_client}): + stream = await honcho_llm_call( + model_config=ModelConfig(model="gpt-4.1", transport="openai"), + prompt="Find the memory", + max_tokens=100, + stream=True, + stream_final_only=True, + tools=[ + { + "name": "lookup", + "description": "Look up a memory", + "input_schema": { + "type": "object", + "properties": {"q": {"type": "string"}}, + }, + } + ], + tool_executor=execute_tool, + enable_retry=False, + ) + chunks = [chunk async for chunk in stream] + + assert executed_tools == [("lookup", {"q": "memory"})] + calls = mock_client.chat.completions.create.call_args_list + assert len(calls) == 3 + assert "tools" in calls[0].kwargs + assert "tools" in calls[1].kwargs + assert "tools" not in calls[2].kwargs + assert "tool_choice" not in calls[2].kwargs + assert [chunk.content for chunk in chunks if chunk.content] == ["Final answer"] + assert chunks[-1].is_done is True + async def test_retry_disabled(self): """Test that retry can be disabled"""