RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge 88e56540db into 699c99368d

This commit is contained in:
Elkim 2026-09-04 08:28:48 +10:00 committed by GitHub
parent 699c99368d 88e56540db
commit 4b39f0641c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 2060 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
docs/v3/contributing/configuration.mdx
View File

@ -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.
<Note>
`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.
</Note>
`timeout` gotchas:
- The value is validated **at config load**: it must coerce to a positive,

47
src/config.py
View File

@ -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):

631
src/llm/backends/openai.py
View File

@ -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 "<missing>"
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,
*,

960
tests/llm/test_backends/test_openai.py
View File

@ -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

27
tests/llm/test_request_builder.py
View File

@ -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:

90
tests/llm/test_responses_config.py Normal file
View File

@ -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",
)

280
tests/utils/test_clients.py
View File

@ -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"""