# SPDX-License-Identifier: Apache-2.0 """Conversion utilities for the OpenAI Responses API.""" import copy import json import logging import uuid from collections import OrderedDict from pathlib import Path from typing import Any, Dict, List, Optional, Union from .responses_models import ( InputItem, InputTokensDetails, OutputContent, OutputItem, ResponsesTool, ResponseUsage, ) from .shared_models import IDPrefix, generate_id logger = logging.getLogger(__name__) class ResponseStateError(RuntimeError): """Base error for persisted Responses API conversation state.""" class ResponseStateNotFoundError(ResponseStateError): """Raised when the requested response state does not exist.""" class ResponseStateCorruptError(ResponseStateError): """Raised when a stored response chain is incomplete or invalid.""" def _try_parse_json(s: str): """Try to parse a string as JSON dict/list, return original string on failure.""" if not isinstance(s, str): return s s = s.strip() if not s or not (s.startswith("{") or s.startswith("[")): return s try: return json.loads(s) except (json.JSONDecodeError, ValueError): return s def _flush_pending_tool_calls( messages: List[Dict[str, Any]], pending: List[Dict[str, Any]], min_merge_index: int = 0, pending_reasoning: str = "", ) -> str: """Flush accumulated tool calls into messages. If the last message is an assistant message without tool_calls, merge into it (avoids duplicate assistant turns that confuse chat templates). Otherwise create a new assistant message. When ``pending_reasoning`` is set, attach it as ``reasoning_content`` on the synthesized assistant message so reasoning round-trips even when the spec sequence is reasoning → function_call → output (no intervening message item). Returns the passthrough reasoning when no tool calls were flushed, or "" when reasoning was consumed. """ if not pending: return pending_reasoning if ( messages and len(messages) - 1 >= min_merge_index and messages[-1].get("role") == "assistant" and "tool_calls" not in messages[-1] ): messages[-1]["tool_calls"] = list(pending) if pending_reasoning: messages[-1]["reasoning_content"] = pending_reasoning else: msg: Dict[str, Any] = {"role": "assistant", "tool_calls": list(pending)} if pending_reasoning: msg["reasoning_content"] = pending_reasoning messages.append(msg) pending.clear() return "" def _consolidate_system_messages( messages: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: """Move all system messages to the front and merge them into one.""" system_parts: List[str] = [] non_system: List[Dict[str, Any]] = [] for msg in messages: if msg.get("role") == "system": content = msg.get("content", "") if content: system_parts.append(content) else: non_system.append(msg) if not system_parts: return messages return [{"role": "system", "content": "\n\n".join(system_parts)}] + non_system # ============================================================================= # Input Conversion # ============================================================================= def convert_responses_input_to_messages( input_data: Optional[Union[str, List[InputItem]]], instructions: Optional[str] = None, previous_messages: Optional[List[Dict[str, Any]]] = None, consolidate_system_messages: bool = True, ) -> List[Dict[str, Any]]: """Convert Responses API input to internal messages format. Args: input_data: String prompt or list of InputItem objects. instructions: System prompt (prepended as system message). previous_messages: Messages from previous_response_id chain. consolidate_system_messages: If True, merge all system/developer content into one leading system message for strict templates. Server code can set this to False and resolve placement after the target template is known. Returns: List of message dicts compatible with chat template. """ messages: List[Dict[str, Any]] = [] # Collect system/developer content to merge into a single system message # when strict-template compatibility mode is active. In deferred mode, # top-level instructions still form a leading system message, but input # system/developer items keep their original position until template # capability probing decides whether they can be preserved. system_parts: List[str] = [] if instructions: system_parts.append(instructions) # Prepend previous response context if previous_messages: messages.extend(copy.deepcopy(previous_messages)) current_message_start = len(messages) if input_data is None: if system_parts: messages.insert(0, {"role": "system", "content": "\n\n".join(system_parts)}) return ( _consolidate_system_messages(messages) if consolidate_system_messages else messages ) if isinstance(input_data, str): if system_parts: messages.insert(0, {"role": "system", "content": "\n\n".join(system_parts)}) messages.append({"role": "user", "content": input_data}) return ( _consolidate_system_messages(messages) if consolidate_system_messages else messages ) # Process input items # Track pending tool calls for grouping into a single assistant message pending_tool_calls: List[Dict[str, Any]] = [] # Track reasoning content to attach to the next assistant message pending_reasoning: str = "" for item in input_data: # Resolve effective type: EasyInputMessage has no type field item_type = item.type if item_type is None and item.role is not None: item_type = "message" if item_type == "message": # Flush pending tool calls before a new message. Reasoning # passes through when no tool calls were flushed so it lands # on this message instead. pending_reasoning = _flush_pending_tool_calls( messages, pending_tool_calls, min_merge_index=current_message_start, pending_reasoning=pending_reasoning, ) role = item.role or "user" # Map "developer" role to "system" if role == "developer": role = "system" content = item.content if isinstance(content, list): # Convert content parts - preserve images for VLM processing text_parts = [] has_image = False converted_parts: List[Dict[str, Any]] = [] for part in content: if isinstance(part, dict): if part.get("type") in ("input_text", "text", "output_text"): text = part.get("text", "") text_parts.append(text) converted_parts.append({"type": "text", "text": text}) elif part.get("type") == "input_image": # Preserve image data for VLM engines has_image = True image_url = part.get("image_url", part.get("url", "")) detail = part.get("detail", "auto") converted_parts.append( { "type": "input_image", "image_url": image_url, "detail": detail, } ) elif isinstance(part, str): text_parts.append(part) converted_parts.append({"type": "text", "text": part}) if has_image: # Keep as content list so VLM can extract images content = converted_parts else: content = "\n".join(text_parts) if text_parts else "" # Merge system/developer messages into the single system block unless # the server is deferring placement until the template is known. if role == "system": if consolidate_system_messages: system_parts.append(content or "") else: messages.append({"role": "system", "content": content or ""}) else: msg_dict: Dict[str, Any] = {"role": role, "content": content or ""} if role == "assistant" and pending_reasoning: msg_dict["reasoning_content"] = pending_reasoning pending_reasoning = "" messages.append(msg_dict) elif item_type == "reasoning": # Collect reasoning summary text to attach to the next # assistant message as reasoning_content. summary = getattr(item, "summary", None) or ( (item.model_extra or {}).get("summary") if hasattr(item, "model_extra") else None ) if summary: parts = [] for s in summary: if isinstance(s, dict): parts.append(s.get("text", "")) else: parts.append(getattr(s, "text", "")) pending_reasoning = "\n".join(p for p in parts if p) elif item.type == "function_call": # Assistant's tool call — accumulate for grouping call_id = item.call_id or item.id or f"call_{uuid.uuid4().hex[:8]}" pending_tool_calls.append( { "id": call_id, "type": "function", "function": { "name": item.name or "", "arguments": _try_parse_json(item.arguments or "{}"), }, } ) elif item.type == "function_call_output": # Flush pending tool calls first. Any pending reasoning gets # attached to the synthesized assistant tool_calls message. pending_reasoning = _flush_pending_tool_calls( messages, pending_tool_calls, min_merge_index=current_message_start, pending_reasoning=pending_reasoning, ) messages.append( { "role": "tool", "tool_call_id": item.call_id or "", "content": item.output or "", } ) # Flush remaining pending tool calls. If reasoning survived without # a trailing message, attach it to the synthesized tool_calls message. _flush_pending_tool_calls( messages, pending_tool_calls, min_merge_index=current_message_start, pending_reasoning=pending_reasoning, ) # Insert merged system message at position 0 if system_parts: messages.insert(0, {"role": "system", "content": "\n\n".join(system_parts)}) return ( _consolidate_system_messages(messages) if consolidate_system_messages else messages ) # ============================================================================= # Tool Conversion # ============================================================================= def convert_responses_tools( tools: Optional[List[ResponsesTool]], ) -> Optional[List[Dict[str, Any]]]: """Convert Responses API flat tool format to Chat Completions nested format. Responses: {"type": "function", "name": "fn", "parameters": {...}} Chat Completions: {"type": "function", "function": {"name": "fn", "parameters": {...}}} Non-function tool types (local_shell, mcp, web_search, etc.) are skipped since they are not supported by local model chat templates. """ if not tools: return None result = [] for tool in tools: if tool.type == "function" and tool.name: func_def: Dict[str, Any] = {"name": tool.name} if tool.description: func_def["description"] = tool.description if tool.parameters: func_def["parameters"] = tool.parameters if tool.strict is not None: func_def["strict"] = tool.strict result.append({"type": "function", "function": func_def}) # Non-function tools (local_shell, mcp, web_search, etc.) are # silently skipped — local models can't execute them. return result if result else None # ============================================================================= # Response Building # ============================================================================= def build_message_output_item( text: str, item_id: Optional[str] = None, status: str = "completed", ) -> OutputItem: """Build a message-type OutputItem.""" return OutputItem( type="message", id=item_id or generate_id(IDPrefix.MESSAGE), status=status, role="assistant", content=[OutputContent(type="output_text", text=text)], ) def build_function_call_output_item( name: str, arguments: str, call_id: str, item_id: Optional[str] = None, status: str = "completed", ) -> OutputItem: """Build a function_call-type OutputItem.""" return OutputItem( type="function_call", id=item_id or generate_id(IDPrefix.FUNCTION_CALL), status=status, call_id=call_id, name=name, arguments=arguments, ) def build_reasoning_output_item( reasoning_text: str, item_id: Optional[str] = None, status: str = "completed", ) -> OutputItem: """Build a reasoning-type OutputItem with full CoT in summary[0].text.""" from .responses_models import ReasoningSummaryPart summary = [ReasoningSummaryPart(text=reasoning_text)] if reasoning_text else [] return OutputItem( type="reasoning", id=item_id or generate_id(IDPrefix.REASONING), status=status, summary=summary, ) def build_response_usage( input_tokens: int, output_tokens: int, reasoning_tokens: int = 0, cached_tokens: int = 0, ) -> ResponseUsage: """Build ResponseUsage from token counts.""" from .responses_models import OutputTokensDetails return ResponseUsage( input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, input_tokens_details=InputTokensDetails(cached_tokens=cached_tokens), output_tokens_details=OutputTokensDetails(reasoning_tokens=reasoning_tokens), ) # ============================================================================= # SSE Event Formatting # ============================================================================= def format_sse_event(event_type: str, data: Any) -> str: """Format a Responses API SSE event. Returns: "event: {type}\\ndata: {json}\\n\\n" """ if isinstance(data, str): json_str = data elif hasattr(data, "model_dump"): json_str = json.dumps(data.model_dump(exclude_none=True)) elif isinstance(data, dict): json_str = json.dumps(data) else: json_str = json.dumps(data) return f"event: {event_type}\ndata: {json_str}\n\n" # ============================================================================= # Response Store (previous_response_id support) # ============================================================================= MAX_STORED_RESPONSES = 1000 class ResponseStore: """Bounded persisted store for response state and public responses.""" def __init__( self, max_size: int = MAX_STORED_RESPONSES, state_dir: Optional[Union[str, Path]] = None, ): self._store: OrderedDict[str, Dict[str, Any]] = OrderedDict() self._max_size = max_size self._state_dir = Path(state_dir).expanduser().resolve() if state_dir else None if self._state_dir: self._state_dir.mkdir(parents=True, exist_ok=True) self._load_persisted_records() @property def state_dir(self) -> Optional[Path]: """Resolved directory used for persisted response state.""" return self._state_dir def _record_path(self, response_id: str) -> Optional[Path]: if self._state_dir is None: return None return self._state_dir / f"{response_id}.json" def _normalize_record( self, response_id: str, response_data: Dict[str, Any], ) -> Dict[str, Any]: if "public_response" in response_data: record = copy.deepcopy(response_data) record.setdefault("response_id", response_id) record.setdefault( "created_at", record.get("public_response", {}).get("created_at", 0) ) record.setdefault( "previous_response_id", record.get("public_response", {}).get("previous_response_id"), ) record.setdefault("input_messages", []) record.setdefault( "output_messages", normalize_response_output_to_messages( record.get("public_response", {}).get("output", []) ), ) return record public_response = copy.deepcopy(response_data) public_response.setdefault("id", response_id) return { "response_id": response_id, "previous_response_id": public_response.get("previous_response_id"), "input_messages": [], "output_messages": normalize_response_output_to_messages( public_response.get("output", []) ), "public_response": public_response, "created_at": public_response.get("created_at", 0), } def _persist_record(self, record: Dict[str, Any]) -> None: path = self._record_path(record["response_id"]) if path is None: return tmp_path = path.with_suffix(".tmp") with tmp_path.open("w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False) tmp_path.replace(path) def _remove_persisted_record(self, response_id: str) -> None: path = self._record_path(response_id) if path is None or not path.exists(): return path.unlink() def _evict_oldest(self) -> None: while len(self._store) > self._max_size: response_id, _record = self._store.popitem(last=False) self._remove_persisted_record(response_id) def _load_persisted_records(self) -> None: assert self._state_dir is not None loaded: List[Dict[str, Any]] = [] for path in sorted(self._state_dir.glob("*.json")): try: with path.open("r", encoding="utf-8") as f: raw = json.load(f) response_id = raw.get("response_id") or raw.get( "public_response", {} ).get("id") if not response_id: raise ValueError("missing response_id") loaded.append(self._normalize_record(response_id, raw)) except (OSError, ValueError, json.JSONDecodeError) as exc: logger.warning("Skipping corrupt response state file %s: %s", path, exc) loaded.sort( key=lambda record: (record.get("created_at", 0), record["response_id"]) ) for record in loaded: self._store[record["response_id"]] = record self._evict_oldest() def put(self, response_id: str, response_data: Dict[str, Any]) -> None: """Store response state, evicting oldest records if needed.""" record = self._normalize_record(response_id, response_data) if response_id in self._store: self._store.move_to_end(response_id) self._store[response_id] = record self._persist_record(record) self._evict_oldest() def get_record(self, response_id: str) -> Optional[Dict[str, Any]]: """Retrieve a stored response-state record.""" data = self._store.get(response_id) if data is not None: self._store.move_to_end(response_id) return copy.deepcopy(data) return None def get(self, response_id: str) -> Optional[Dict[str, Any]]: """Retrieve the public response object for a stored record.""" data = self.get_record(response_id) if data is None: return None return data.get("public_response") def resolve_chain_messages(self, response_id: str) -> List[Dict[str, Any]]: """Resolve the full previous_response_id chain into message history.""" if response_id not in self._store: raise ResponseStateNotFoundError(f"Response state not found: {response_id}") chain: List[Dict[str, Any]] = [] seen: set[str] = set() current_id: Optional[str] = response_id while current_id: if current_id in seen: raise ResponseStateCorruptError( f"Cycle detected in previous_response_id chain at {current_id}" ) seen.add(current_id) record = self._store.get(current_id) if record is None: raise ResponseStateCorruptError( f"Missing ancestor response state: {current_id}" ) self._store.move_to_end(current_id) chain.append(record) current_id = record.get("previous_response_id") chain.reverse() messages: List[Dict[str, Any]] = [] for record in chain: messages.extend(copy.deepcopy(record.get("input_messages", []))) messages.extend(copy.deepcopy(record.get("output_messages", []))) return _consolidate_system_messages(messages) def delete(self, response_id: str) -> bool: """Delete a stored response. Returns True if found.""" if response_id not in self._store: return False del self._store[response_id] self._remove_persisted_record(response_id) return True def __len__(self) -> int: return len(self._store) # ============================================================================= # Previous Response Conversion # ============================================================================= def convert_stored_response_to_messages( response_data: Dict[str, Any], ) -> List[Dict[str, Any]]: """Convert a stored public response or state record back to messages.""" if "output_messages" in response_data: return copy.deepcopy(response_data.get("output_messages", [])) return normalize_response_output_to_messages(response_data.get("output", [])) def normalize_response_output_to_messages( output_items: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: """Convert response output items to assistant/tool-call history messages.""" messages: List[Dict[str, Any]] = [] pending_tool_calls: List[Dict[str, Any]] = [] pending_reasoning: str = "" for item in output_items: item_type = item.get("type") if item_type == "reasoning": summary = item.get("summary", []) parts = [s.get("text", "") for s in summary if isinstance(s, dict)] pending_reasoning = "\n".join(p for p in parts if p) elif item_type == "message": pending_reasoning = _flush_pending_tool_calls( messages, pending_tool_calls, pending_reasoning=pending_reasoning, ) content_blocks = item.get("content", []) text_parts = [] for block in content_blocks: if block.get("type") == "output_text": text_parts.append(block.get("text", "")) msg_dict: Dict[str, Any] = { "role": item.get("role", "assistant"), "content": "\n".join(text_parts), } if pending_reasoning: msg_dict["reasoning_content"] = pending_reasoning pending_reasoning = "" messages.append(msg_dict) elif item_type == "function_call": call_id = item.get("call_id", f"call_{uuid.uuid4().hex[:8]}") pending_tool_calls.append( { "id": call_id, "type": "function", "function": { "name": item.get("name", ""), "arguments": _try_parse_json(item.get("arguments", "{}")), }, } ) _flush_pending_tool_calls( messages, pending_tool_calls, pending_reasoning=pending_reasoning, ) return _consolidate_system_messages(messages) def build_response_store_record( public_response: Dict[str, Any], input_messages: List[Dict[str, Any]], output_messages: List[Dict[str, Any]], ) -> Dict[str, Any]: """Build a persisted response-state record.""" return { "response_id": public_response.get("id", ""), "previous_response_id": public_response.get("previous_response_id"), "input_messages": copy.deepcopy(input_messages), "output_messages": copy.deepcopy(output_messages), "public_response": copy.deepcopy(public_response), "created_at": public_response.get("created_at", 0), }