Coverage for src / lexigram / ai / relay / mappers / openai_responses.py: 91%
475 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""OpenAI Responses request and response mapper.
3Converts the OpenAI Responses wire DTOs
4(:class:`ResponsesRequest` / :class:`ResponsesResponse`) into the
5canonical relay IR and back. Stream conversion is handled by the shared
6stream lifecycle task and reports ``unsupported_feature`` until then.
7"""
9from __future__ import annotations
11from typing import Any, cast
13from lexigram.ai.relay.context import ConversionContext
14from lexigram.ai.relay.errors import translate, unsupported_feature, unsupported_format
15from lexigram.ai.relay.finish_reasons import (
16 responses_incomplete_from_finish,
17 responses_status_from_finish,
18)
19from lexigram.ai.relay.mappers.base import new_uuid, record_loss
20from lexigram.contracts.ai.agents import ToolDefinition
21from lexigram.contracts.ai.exceptions import RelayError
22from lexigram.contracts.ai.llm import ChatMessage, FunctionCall, ToolCall
23from lexigram.contracts.ai.multimodal import ImageBase64Part, ImageUrlPart, TextPart
24from lexigram.contracts.ai.relay.dto import (
25 ResponsesIncompleteDetails,
26 ResponsesItem,
27 ResponsesRequest,
28 ResponsesResponse,
29 ResponsesUsage,
30)
31from lexigram.contracts.ai.relay.ir import (
32 RelayRequest,
33 RelayResponse,
34 StreamDelta,
35 StreamState,
36)
37from lexigram.contracts.ai.relay.types import RelayFormat, RelayUsage
38from lexigram.contracts.ai.thinking import ThinkingConfig, ThinkingResult
39from lexigram.contracts.core.result import Err, Ok, Result
40from lexigram.serialization import dumps_str, loads_str
42__all__ = ["OpenAIResponsesMapper"]
44_TARGET = RelayFormat.OPENAI_RESPONSES
47def _parse_arguments(arguments: str) -> dict[str, Any] | str:
48 """Parse a wire arguments string into dict form when possible.
50 Args:
51 arguments: Raw JSON argument string from the wild.
53 Returns:
54 The parsed dict, or the original string when it is empty or not
55 valid JSON.
56 """
57 if not isinstance(arguments, str) or not arguments.strip():
58 return arguments
59 try:
60 value = loads_str(arguments)
61 except (ValueError, TypeError):
62 return arguments
63 if isinstance(value, dict):
64 return value
65 return arguments
68def _arguments_to_wire(arguments: Any) -> str:
69 """Serialize canonical arguments into a JSON string.
71 Args:
72 arguments: Canonical arguments (dict, string, or anything else).
74 Returns:
75 A JSON string for the wire, or an empty string when unsupported.
76 """
77 if isinstance(arguments, dict):
78 return dumps_str(arguments)
79 if isinstance(arguments, str):
80 return arguments
81 return ""
84class OpenAIResponsesMapper:
85 """Bidirectional OpenAI Responses converter.
87 Attributes:
88 format: The wire format this mapper handles.
89 """
91 format = _TARGET
93 def request_to_ir(
94 self, payload: Any, *, context: ConversionContext
95 ) -> Result[RelayRequest, RelayError]:
96 """Convert a ``ResponsesRequest`` into canonical ``RelayRequest``.
98 Args:
99 payload: A wire request DTO.
100 context: Per-conversion context with loss sink.
102 Returns:
103 Ok(request) on success, Err(relay_error) on malformed payload.
104 """
105 if not isinstance(payload, ResponsesRequest):
106 return Err(
107 unsupported_format(
108 f"expected ResponsesRequest, got {type(payload).__name__}"
109 )
110 )
111 system_parts: list[str] = []
112 if payload.instructions:
113 system_parts.append(payload.instructions)
114 messages: list[ChatMessage] = []
115 pending_tools: list[ToolCall] = []
116 pending_ids: list[str | None] = []
117 web_search_calls: list[dict[str, Any]] = []
119 def flush_tools() -> None:
120 """Emit accumulated function calls as one assistant turn."""
121 if not pending_tools:
122 return
123 metadata: dict[str, Any] = {}
124 if any(pending_ids):
125 metadata["function_call_item_ids"] = list(pending_ids)
126 messages.append(
127 ChatMessage(
128 role="assistant",
129 content="",
130 tool_calls=list(pending_tools),
131 metadata=metadata or None,
132 )
133 )
134 pending_tools.clear()
135 pending_ids.clear()
137 if isinstance(payload.input, str):
138 messages.append(ChatMessage(role="user", content=payload.input))
139 else:
140 for index, wire_item in enumerate(payload.input):
141 item_type = wire_item.type
142 if item_type == "message":
143 flush_tools()
144 if wire_item.role == "system":
145 system_parts.append(self._message_text_to_ir(wire_item))
146 if index > 0:
147 record_loss(
148 context,
149 field="system_message",
150 target=_TARGET,
151 reason="system_message_reordered",
152 )
153 continue
154 messages.append(self._message_from_item(wire_item, context))
155 elif item_type == "function_call":
156 pending_tools.append(self._tool_from_item(wire_item))
157 pending_ids.append(wire_item.id or wire_item.call_id)
158 elif item_type == "function_call_output":
159 flush_tools()
160 messages.append(self._tool_result_from_item(wire_item))
161 elif item_type == "reasoning":
162 flush_tools()
163 messages.append(self._reasoning_from_item(wire_item))
164 elif item_type == "web_search_call":
165 web_search_calls.append(wire_item.to_dict())
166 record_loss(
167 context,
168 field=f"input[{index}]",
169 target=_TARGET,
170 reason="unsupported_item_preserved",
171 severity="info",
172 )
173 else:
174 record_loss(
175 context,
176 field=f"input[{index}]",
177 target=_TARGET,
178 reason="unknown_item_dropped",
179 )
180 flush_tools()
181 metadata: dict[str, Any] = {}
182 if payload.include is not None:
183 metadata["include"] = list(payload.include)
184 if payload.max_output_tokens is not None:
185 metadata["max_tokens_kind"] = "max_completion_tokens"
186 if web_search_calls:
187 metadata["input_web_search_calls"] = web_search_calls
188 reasoning = payload.reasoning
189 if reasoning is not None:
190 metadata["reasoning"] = reasoning
191 if payload.text is not None:
192 metadata["text"] = payload.text
193 if payload.service_tier is not None:
194 metadata["service_tier"] = payload.service_tier
195 thinking: ThinkingConfig | None = None
196 if isinstance(reasoning, dict):
197 thinking = ThinkingConfig(effort=reasoning.get("effort"))
198 return Ok(
199 RelayRequest(
200 model=context.normalize_model(payload.model),
201 messages=messages,
202 system="\n".join(system_parts) if system_parts else None,
203 tools=self._tools_to_ir(payload.tools, context),
204 temperature=payload.temperature,
205 max_tokens=payload.max_output_tokens,
206 stream=payload.stream,
207 include_usage=bool(payload.include and "usage" in payload.include),
208 parallel_tool_calls=payload.parallel_tool_calls,
209 thinking=thinking,
210 response_format=self._text_to_response_format(payload.text),
211 metadata=metadata,
212 passthrough=dict(payload.passthrough),
213 )
214 )
216 def ir_to_request(
217 self, request: RelayRequest, *, context: ConversionContext
218 ) -> Result[Any, RelayError]:
219 """Convert canonical ``RelayRequest`` into a ``ResponsesRequest``.
221 Args:
222 request: Canonical request IR.
223 context: Per-conversion context with loss sink.
225 Returns:
226 Ok(request) on success, Err(relay_error) on failure.
227 """
228 try:
229 items: list[ResponsesItem] = []
230 instructions = request.system
231 system_parts = [request.system] if request.system else []
232 for message in request.messages:
233 if message.role == "system":
234 system_parts.append(self._system_text(message))
235 record_loss(
236 context,
237 field="system_message",
238 target=_TARGET,
239 reason="system_message_reordered",
240 )
241 continue
242 if message.role == "tool":
243 items.append(self._tool_result_to_item(message))
244 continue
245 if message.tool_calls:
246 items.extend(self._tool_calls_to_items(message, context))
247 continue
248 items.append(self._message_to_item(message, context))
249 items.extend(self._web_search_items(request))
250 if system_parts:
251 instructions = "\n".join(system_parts)
252 handled_metadata = {
253 "include",
254 "reasoning",
255 "text",
256 "service_tier",
257 "input_web_search_calls",
258 "generation_config",
259 "safety_settings",
260 "tool_config",
261 }
262 return Ok(
263 ResponsesRequest(
264 model=context.resolve_model(request.model),
265 input=items,
266 instructions=instructions,
267 tools=(
268 [self._tool_from_ir(tool) for tool in request.tools]
269 if request.tools
270 else None
271 ),
272 temperature=request.temperature,
273 max_output_tokens=request.max_tokens,
274 stream=request.stream,
275 include=self._include_from_ir(request),
276 parallel_tool_calls=request.parallel_tool_calls,
277 reasoning=self._reasoning_from_ir(request, context),
278 text=self._text_from_ir(request),
279 service_tier=request.metadata.get("service_tier"),
280 tool_choice=request.tool_choice,
281 passthrough={
282 **request.passthrough,
283 **{
284 key: value
285 for key, value in request.metadata.items()
286 if key not in handled_metadata
287 },
288 },
289 )
290 )
291 except (RelayError, ValueError, TypeError, KeyError) as exc:
292 return Err(translate(exc, detail="ir_to_request"))
294 def response_to_ir(
295 self, payload: Any, *, context: ConversionContext
296 ) -> Result[RelayResponse, RelayError]:
297 """Convert a ``ResponsesResponse`` into canonical ``RelayResponse``.
299 Args:
300 payload: A wire response DTO.
301 context: Per-conversion context with loss sink.
303 Returns:
304 Ok(response) on success, Err(relay_error) on malformed payload.
305 """
306 if not isinstance(payload, ResponsesResponse):
307 return Err(
308 unsupported_format(
309 f"expected ResponsesResponse, got {type(payload).__name__}"
310 )
311 )
312 try:
313 passthrough = dict(payload.passthrough)
314 if payload.error is not None:
315 passthrough["error"] = payload.error
316 if payload.object != "response":
317 passthrough["object"] = payload.object
318 content_parts: list[str] = []
319 tool_calls: list[ToolCall] = []
320 tool_results: list[ChatMessage] = []
321 reasoning_text: list[str] = []
322 web_search_calls: list[dict[str, Any]] = []
323 for index, output_item in enumerate(payload.output):
324 item_type = output_item.type
325 if item_type == "message":
326 for part in output_item.content or []:
327 if not isinstance(part, dict):
328 content_parts.append(str(part))
329 continue
330 part_type = part.get("type")
331 if part_type == "output_text":
332 content_parts.append(str(part.get("text", "")))
333 else:
334 record_loss(
335 context,
336 field=part_type or "part",
337 target=_TARGET,
338 reason="unknown_part_type",
339 )
340 elif item_type == "reasoning":
341 reasoning_text.extend(self._summary_texts(output_item.summary))
342 elif item_type == "function_call":
343 tool_calls.append(
344 ToolCall(
345 id=output_item.call_id or output_item.id or "",
346 type="function",
347 function=FunctionCall(
348 name=output_item.name or "",
349 arguments=_parse_arguments(output_item.arguments or ""),
350 ),
351 )
352 )
353 elif item_type == "function_call_output":
354 tool_results.append(
355 ChatMessage(
356 role="tool",
357 content=output_item.output or "",
358 tool_call_id=output_item.call_id,
359 )
360 )
361 elif item_type == "web_search_call":
362 web_search_calls.append(output_item.to_dict())
363 record_loss(
364 context,
365 field=f"output[{index}]",
366 target=_TARGET,
367 reason="unsupported_item_preserved",
368 severity="info",
369 )
370 else:
371 record_loss(
372 context,
373 field=f"output[{index}]",
374 target=_TARGET,
375 reason="unknown_item_dropped",
376 )
377 if web_search_calls:
378 passthrough["web_search_calls"] = web_search_calls
379 thinking: ThinkingResult | None = None
380 if reasoning_text:
381 tokens: int | None = None
382 if payload.usage is not None:
383 details = payload.usage.output_tokens_details
384 if isinstance(details, dict) and isinstance(
385 details.get("reasoning_tokens"), int
386 ):
387 tokens = details["reasoning_tokens"]
388 thinking = ThinkingResult(
389 content="".join(reasoning_text), tokens=tokens
390 )
391 return Ok(
392 RelayResponse(
393 model=payload.model,
394 id=payload.id,
395 created=payload.created_at,
396 content="".join(content_parts),
397 thinking=thinking,
398 tool_calls=tool_calls,
399 tool_results=tool_results,
400 finish_reason=self._finish_from_status(
401 payload.status,
402 payload.incomplete_details,
403 bool(tool_calls),
404 ),
405 status=payload.status,
406 incomplete_details=(
407 payload.incomplete_details.to_dict()
408 if payload.incomplete_details is not None
409 else None
410 ),
411 usage=self._usage_from_wire(payload.usage),
412 passthrough=passthrough,
413 )
414 )
415 except (RelayError, ValueError, TypeError, KeyError) as exc:
416 return Err(translate(exc, detail="response_to_ir"))
418 def ir_to_response(
419 self, response: RelayResponse, *, context: ConversionContext
420 ) -> Result[Any, RelayError]:
421 """Convert canonical ``RelayResponse`` into a ``ResponsesResponse``.
423 Args:
424 response: Canonical response IR.
425 context: Per-conversion context with loss sink.
427 Returns:
428 Ok(response) on success, Err(relay_error) on failure.
429 """
430 try:
431 passthrough = dict(response.passthrough)
432 error = passthrough.pop("error", None)
433 object_type = passthrough.pop("object", "response")
434 status, incomplete = self._status_from_finish(response)
435 response_id = response.id or f"chatcmpl-{new_uuid()}"
436 item_status = "incomplete" if status == "incomplete" else "completed"
437 items: list[ResponsesItem] = []
438 content_parts: list[dict[str, Any]] = []
439 if response.content:
440 content_parts.append(
441 {
442 "type": "output_text",
443 "text": response.content,
444 "annotations": [],
445 }
446 )
447 if content_parts:
448 items.append(
449 ResponsesItem(
450 type="message",
451 role="assistant",
452 id=f"{response_id}_msg_0",
453 status=item_status,
454 content=content_parts,
455 quality="",
456 size="",
457 )
458 )
459 if response.thinking is not None and response.thinking.content:
460 items.append(
461 ResponsesItem(
462 type="reasoning",
463 id=f"{response_id}_reasoning_0",
464 status=item_status,
465 role="",
466 content=[
467 {
468 "type": "summary_text",
469 "text": response.thinking.content,
470 "annotations": None,
471 }
472 ],
473 quality="",
474 size="",
475 )
476 )
477 for tool in response.tool_calls:
478 call_id = tool.id or f"call_{new_uuid()}"
479 items.append(
480 ResponsesItem(
481 type="function_call",
482 id=call_id,
483 status=item_status,
484 role="",
485 content=None,
486 quality="",
487 size="",
488 call_id=call_id,
489 name=tool.function.name if tool.function else "",
490 arguments=_arguments_to_wire(
491 tool.function.arguments if tool.function else {}
492 ),
493 )
494 )
495 for index, result in enumerate(response.tool_results):
496 items.append(
497 ResponsesItem(
498 type="function_call_output",
499 id=f"fcoc_{index}",
500 call_id=result.tool_call_id,
501 output=self._result_output(result),
502 )
503 )
504 return Ok(
505 ResponsesResponse(
506 id=response_id,
507 model=context.resolve_model(response.model),
508 output=items,
509 object=object_type,
510 created_at=response.created or 0,
511 status=status,
512 incomplete_details=incomplete,
513 error=error if isinstance(error, dict) else None,
514 usage=self._usage_to_wire(response.usage),
515 passthrough=passthrough,
516 )
517 )
518 except (RelayError, ValueError, TypeError, KeyError) as exc:
519 return Err(translate(exc, detail="ir_to_response"))
521 def stream_to_delta(
522 self, event: Any, *, state: StreamState
523 ) -> Result[tuple[StreamDelta, ...], RelayError]:
524 """Stream conversion is deferred to the shared stream lifecycle task."""
525 return Err(
526 unsupported_feature(
527 "openai_responses stream conversion is not implemented yet"
528 )
529 )
531 def delta_to_stream(
532 self, delta: StreamDelta, *, state: StreamState
533 ) -> Result[tuple[Any, ...], RelayError]:
534 """Stream emission is deferred to the shared stream lifecycle task."""
535 return Err(
536 unsupported_feature(
537 "openai_responses stream conversion is not implemented yet"
538 )
539 )
541 # -- helpers -------------------------------------------------------------
543 @staticmethod
544 def _message_text_to_ir(wire_item: ResponsesItem) -> str:
545 """Extract text from a wire message item."""
546 return "".join(
547 str(part.get("text", ""))
548 for part in wire_item.content or []
549 if isinstance(part, dict) and part.get("type") == "input_text"
550 )
552 @staticmethod
553 def _input_parts_to_ir(
554 content: list[dict[str, Any]] | None, context: ConversionContext
555 ) -> tuple[list[Any], list[dict[str, Any]]]:
556 """Convert wire content parts into canonical parts and files."""
557 converted: list[Any] = []
558 files: list[dict[str, Any]] = []
559 for part in content or []:
560 if not isinstance(part, dict):
561 converted.append(TextPart(text=str(part)))
562 continue
563 part_type = part.get("type")
564 if part_type == "input_text":
565 converted.append(TextPart(text=str(part.get("text", ""))))
566 elif part_type == "input_image":
567 image = part.get("image_url")
568 if isinstance(image, dict):
569 converted.append(
570 ImageUrlPart(
571 url=str(image.get("url", "")),
572 detail=cast("Any", image.get("detail", "auto") or "auto"),
573 )
574 )
575 else:
576 converted.append(
577 ImageUrlPart(
578 url=str(image or ""),
579 detail=cast("Any", part.get("detail", "auto") or "auto"),
580 )
581 )
582 elif part_type == "input_file":
583 files.append(part)
584 record_loss(
585 context,
586 field="content",
587 target=_TARGET,
588 reason="unrepresentable_part_preserved",
589 severity="info",
590 )
591 else:
592 record_loss(
593 context,
594 field=part_type or "part",
595 target=_TARGET,
596 reason="unknown_part_type",
597 )
598 return converted, files
600 def _message_from_item(
601 self, wire_item: ResponsesItem, context: ConversionContext
602 ) -> ChatMessage:
603 """Convert a wire message item into a canonical message."""
604 wire_content = wire_item.content
605 if isinstance(wire_content, str):
606 wire_content = [{"type": "input_text", "text": wire_content}]
607 parts, files = self._input_parts_to_ir(wire_content, context)
608 metadata: dict[str, Any] = {}
609 if wire_item.id:
610 metadata["item_id"] = wire_item.id
611 if files:
612 metadata["input_files"] = files
613 content: str | list[Any]
614 if len(parts) == 1 and isinstance(parts[0], TextPart):
615 content = parts[0].text
616 elif parts:
617 content = parts
618 else:
619 content = ""
620 return ChatMessage(
621 role=wire_item.role or "user",
622 content=content,
623 metadata=metadata or None,
624 )
626 @staticmethod
627 def _tool_from_item(wire_item: ResponsesItem) -> ToolCall:
628 """Convert a wire function_call item into a canonical tool call."""
629 return ToolCall(
630 id=wire_item.call_id or wire_item.id or "",
631 type="function",
632 function=FunctionCall(
633 name=wire_item.name or "",
634 arguments=_parse_arguments(wire_item.arguments or ""),
635 ),
636 )
638 @staticmethod
639 def _tool_result_from_item(wire_item: ResponsesItem) -> ChatMessage:
640 """Convert a wire function_call_output item into a tool message."""
641 metadata: dict[str, Any] = {}
642 if wire_item.id:
643 metadata["item_id"] = wire_item.id
644 return ChatMessage(
645 role="tool",
646 content=wire_item.output or "",
647 tool_call_id=wire_item.call_id,
648 metadata=metadata or None,
649 )
651 @staticmethod
652 def _reasoning_from_item(wire_item: ResponsesItem) -> ChatMessage:
653 """Convert a wire reasoning item into an assistant message."""
654 metadata: dict[str, Any] = {}
655 if wire_item.id:
656 metadata["item_id"] = wire_item.id
657 return ChatMessage(
658 role="assistant",
659 content="",
660 thinking_blocks=list(wire_item.summary or []),
661 metadata=metadata or None,
662 )
664 def _message_to_item(
665 self, message: ChatMessage, context: ConversionContext
666 ) -> ResponsesItem:
667 """Convert a canonical message into a wire message item."""
668 data: dict[str, Any] = {"role": message.role}
669 if message.metadata and message.metadata.get("item_id"):
670 data["id"] = message.metadata["item_id"]
671 files = (message.metadata or {}).get("input_files")
672 has_files = isinstance(files, list) and any(
673 isinstance(item, dict) for item in files
674 )
675 if isinstance(message.content, str):
676 if message.content and not has_files:
677 data["content"] = message.content
678 elif message.content or has_files:
679 parts = self._message_content_parts(message, context)
680 if parts:
681 data["content"] = parts
682 else:
683 parts = self._message_content_parts(message, context)
684 if parts:
685 data["content"] = parts
686 return ResponsesItem(**data)
688 @staticmethod
689 def _message_content_parts(
690 message: ChatMessage, context: ConversionContext
691 ) -> list[dict[str, Any]]:
692 """Serialize canonical content into wire message parts."""
693 parts: list[dict[str, Any]] = []
694 content = message.content
695 if isinstance(content, str):
696 if content:
697 parts.append({"type": "input_text", "text": content})
698 elif isinstance(content, list):
699 for part in content:
700 if isinstance(part, TextPart):
701 parts.append({"type": "input_text", "text": part.text})
702 elif isinstance(part, ImageUrlPart):
703 parts.append(
704 {
705 "type": "input_image",
706 "image_url": part.url,
707 }
708 )
709 elif isinstance(part, ImageBase64Part):
710 parts.append(
711 {
712 "type": "input_image",
713 "image_url": f"data:{part.media_type};base64,{part.data}",
714 }
715 )
716 else:
717 record_loss(
718 context,
719 field="message.content",
720 target=_TARGET,
721 reason="unknown_content_part",
722 )
723 files = (message.metadata or {}).get("input_files")
724 if isinstance(files, list):
725 for file_part in files:
726 if isinstance(file_part, dict):
727 parts.append(file_part)
728 return parts
730 def _tool_calls_to_items(
731 self, message: ChatMessage, context: ConversionContext
732 ) -> list[ResponsesItem]:
733 """Convert a tool-calling assistant turn into wire items."""
734 items: list[ResponsesItem] = []
735 if self._message_content_parts(message, context):
736 items.append(self._message_to_item(message, context))
737 else:
738 items.append(ResponsesItem(role="assistant", content=""))
739 item_ids = (message.metadata or {}).get("function_call_item_ids")
740 for index, tool in enumerate(message.tool_calls or []):
741 data: dict[str, Any] = {
742 "type": "function_call",
743 "call_id": tool.id or f"call_{index + 1}",
744 "name": tool.function.name if tool.function else "",
745 "arguments": _arguments_to_wire(
746 tool.function.arguments if tool.function else {}
747 ),
748 }
749 if isinstance(item_ids, list) and index < len(item_ids) and item_ids[index]:
750 data["id"] = item_ids[index]
751 items.append(ResponsesItem(**data))
752 return items
754 def _thinking_to_items(self, message: ChatMessage) -> list[ResponsesItem]:
755 """Convert canonical thinking blocks into a reasoning item."""
756 summary = list(message.thinking_blocks or [])
757 if not summary:
758 return []
759 data: dict[str, Any] = {"type": "reasoning", "summary": summary}
760 if message.metadata and message.metadata.get("item_id"):
761 data["id"] = message.metadata["item_id"]
762 return [ResponsesItem(**data)]
764 @staticmethod
765 def _tool_result_to_item(message: ChatMessage) -> ResponsesItem:
766 """Convert a canonical tool message into a wire output item."""
767 content = message.content
768 if isinstance(content, list):
769 output = "".join(
770 part.text for part in content if isinstance(part, TextPart)
771 )
772 else:
773 output = content or ""
774 data: dict[str, Any] = {
775 "type": "function_call_output",
776 "output": str(output),
777 "call_id": message.tool_call_id or "call_0",
778 }
779 if message.metadata and message.metadata.get("item_id"):
780 data["id"] = message.metadata["item_id"]
781 return ResponsesItem(**data)
783 @staticmethod
784 def _system_text(message: ChatMessage) -> str:
785 """Extract text from a system-role canonical message."""
786 content = message.content
787 if isinstance(content, str):
788 return content
789 return "".join(part.text for part in content if isinstance(part, TextPart))
791 @staticmethod
792 def _web_search_items(request: RelayRequest) -> list[ResponsesItem]:
793 """Restore preserved web_search_call input items."""
794 raw = request.metadata.get("input_web_search_calls")
795 items: list[ResponsesItem] = []
796 if isinstance(raw, list):
797 for entry in raw:
798 if not isinstance(entry, dict):
799 continue
800 data = dict(entry)
801 items.append(
802 ResponsesItem(
803 type=str(data.pop("type", "web_search_call")),
804 id=data.pop("id", None),
805 passthrough=data,
806 )
807 )
808 return items
810 @staticmethod
811 def _tools_to_ir(
812 tools: list[dict[str, Any]] | None, context: ConversionContext
813 ) -> list[ToolDefinition]:
814 """Convert wire tool dicts into canonical tool definitions."""
815 definitions: list[ToolDefinition] = []
816 if not tools:
817 return definitions
818 for index, tool in enumerate(tools):
819 if not isinstance(tool, dict):
820 record_loss(
821 context,
822 field=f"tools[{index}]",
823 target=_TARGET,
824 reason="non_dict_tool_dropped",
825 )
826 continue
827 if tool.get("type", "function") != "function":
828 record_loss(
829 context,
830 field=f"tools[{index}]",
831 target=_TARGET,
832 reason="non_function_tool_dropped",
833 )
834 continue
835 if isinstance(tool.get("function"), dict):
836 function = tool["function"]
837 else:
838 function = tool
839 parameters = function.get("parameters", {})
840 definitions.append(
841 ToolDefinition(
842 name=str(function.get("name", "")),
843 description=str(function.get("description", "")),
844 parameters=parameters if isinstance(parameters, dict) else {},
845 )
846 )
847 return definitions
849 @staticmethod
850 def _tool_from_ir(tool: ToolDefinition) -> dict[str, Any]:
851 """Serialize a canonical tool definition as a wire tool dict."""
852 return {
853 "type": "function",
854 "name": tool.name,
855 "description": tool.description,
856 "parameters": tool.parameters,
857 }
859 @staticmethod
860 def _text_to_response_format(
861 text: dict[str, Any] | None,
862 ) -> dict[str, Any] | None:
863 """Derive a canonical response format from the wire text config."""
864 if not isinstance(text, dict):
865 return None
866 fmt = text.get("format")
867 if not isinstance(fmt, dict):
868 return None
869 fmt_type = fmt.get("type")
870 if fmt_type == "json_object":
871 return {"type": "json_object"}
872 if fmt_type == "json_schema":
873 result: dict[str, Any] = {"type": "json_schema"}
874 for key in ("schema", "name", "strict"):
875 if key in fmt:
876 result[key] = fmt[key]
877 return result
878 return None
880 @classmethod
881 def _text_from_ir(cls, request: RelayRequest) -> dict[str, Any] | None:
882 """Rebuild the wire text config from canonical response format."""
883 raw = request.metadata.get("text")
884 text: dict[str, Any] = dict(raw) if isinstance(raw, dict) else {}
885 response_format = request.response_format
886 if response_format is None:
887 return text or None
888 fmt_type = response_format.get("type")
889 if fmt_type == "json_object":
890 text["format"] = {"type": "json_object"}
891 elif fmt_type == "json_schema":
892 fmt: dict[str, Any] = {"type": "json_schema"}
893 for key in ("schema", "name", "strict"):
894 if key in response_format:
895 fmt[key] = response_format[key]
896 text["format"] = fmt
897 else:
898 text["format"] = {"type": "text"}
899 return text or None
901 @staticmethod
902 def _include_from_ir(request: RelayRequest) -> list[str] | None:
903 """Rebuild the wire include list from canonical stream settings."""
904 raw = request.metadata.get("include")
905 include = list(raw) if isinstance(raw, list) else []
906 if request.include_usage and "usage" not in include:
907 include.append("usage")
908 return include or None
910 def _reasoning_from_ir(
911 self, request: RelayRequest, context: ConversionContext
912 ) -> dict[str, Any] | None:
913 """Rebuild the wire reasoning config from canonical thinking."""
914 thinking = request.thinking
915 if thinking is not None:
916 if thinking.effort is not None:
917 return {"effort": thinking.effort}
918 record_loss(
919 context,
920 field="thinking",
921 target=_TARGET,
922 reason="effort_only_supported",
923 )
924 raw = request.metadata.get("reasoning")
925 if isinstance(raw, dict):
926 return dict(raw)
927 return None
929 @staticmethod
930 def _summary_texts(
931 summary: list[dict[str, Any]] | None,
932 ) -> list[str]:
933 """Extract text from reasoning summary blocks."""
934 return [
935 str(item.get("text", ""))
936 for item in summary or []
937 if isinstance(item, dict) and item.get("type") == "summary_text"
938 ]
940 @staticmethod
941 def _finish_from_status(
942 status: str | None,
943 incomplete_details: ResponsesIncompleteDetails | None,
944 has_tool_calls: bool,
945 ) -> str | None:
946 """Derive a canonical finish reason from a wire status."""
947 if status == "completed":
948 return "tool_calls" if has_tool_calls else "stop"
949 if status == "incomplete":
950 reason = (
951 incomplete_details.reason if incomplete_details is not None else None
952 )
953 if reason == "max_output_tokens":
954 return "length"
955 if reason == "content_filter":
956 return "content_filter"
957 return "other"
958 if status == "failed":
959 return "other"
960 return None
962 @staticmethod
963 def _status_from_finish(
964 response: RelayResponse,
965 ) -> tuple[str | None, ResponsesIncompleteDetails | None]:
966 """Derive a wire status from canonical finish behavior."""
967 status = response.status
968 incomplete: ResponsesIncompleteDetails | None = None
969 if response.incomplete_details is not None:
970 raw = dict(response.incomplete_details)
971 reason = raw.pop("reason", None)
972 incomplete = ResponsesIncompleteDetails(reason=reason, passthrough=raw)
973 if status is not None:
974 if status == "incomplete" and incomplete is None:
975 derived = _incomplete_for_finish(response.finish_reason)
976 if derived is not None:
977 incomplete = derived
978 return status, incomplete
979 finish = response.finish_reason
980 wire_status, detail = responses_status_from_finish(finish)
981 if detail is None:
982 return wire_status, None
983 return wire_status, ResponsesIncompleteDetails(reason=detail)
985 @staticmethod
986 def _result_output(message: ChatMessage) -> str:
987 """Extract a tool result string from a canonical tool message."""
988 content = message.content
989 if isinstance(content, list):
990 return "".join(part.text for part in content if isinstance(part, TextPart))
991 return str(content or "")
993 @staticmethod
994 def _usage_from_wire(usage: ResponsesUsage | None) -> RelayUsage | None:
995 """Map wire usage into canonical ``RelayUsage``."""
996 if usage is None:
997 return None
998 input_details = usage.input_tokens_details
999 completion_details = usage.completion_tokens_details
1000 if not isinstance(completion_details, dict):
1001 completion_details = usage.output_tokens_details
1002 return RelayUsage(
1003 prompt_tokens=usage.prompt_tokens or usage.input_tokens,
1004 completion_tokens=usage.completion_tokens or usage.output_tokens,
1005 total_tokens_override=usage.total_tokens or None,
1006 cache_read_tokens=(
1007 int(input_details.get("cached_tokens", 0) or 0)
1008 if isinstance(input_details, dict)
1009 else 0
1010 ),
1011 reasoning_tokens=(
1012 int(completion_details.get("reasoning_tokens", 0) or 0)
1013 if isinstance(completion_details, dict)
1014 else 0
1015 ),
1016 input_tokens=usage.input_tokens,
1017 output_tokens=usage.output_tokens,
1018 )
1020 @staticmethod
1021 def _usage_to_wire(usage: RelayUsage | None) -> ResponsesUsage | None:
1022 """Serialize canonical ``RelayUsage`` into wire usage."""
1023 if usage is None:
1024 return None
1025 input_details = (
1026 {"cached_tokens": usage.cache_read_tokens}
1027 if usage.cache_read_tokens
1028 else None
1029 )
1030 return ResponsesUsage(
1031 prompt_tokens=usage.prompt_tokens,
1032 completion_tokens=usage.completion_tokens,
1033 total_tokens=usage.total_tokens,
1034 prompt_tokens_details={"cached_tokens": 0},
1035 completion_tokens_details={"reasoning_tokens": usage.reasoning_tokens},
1036 input_tokens=usage.prompt_tokens,
1037 input_tokens_details=input_details,
1038 output_tokens=usage.completion_tokens,
1039 )
1042def _incomplete_for_finish(
1043 finish_reason: str | None,
1044) -> ResponsesIncompleteDetails | None:
1045 """Map a canonical finish reason to an incomplete-details payload."""
1046 detail = responses_incomplete_from_finish(finish_reason)
1047 if detail is None:
1048 return None
1049 return ResponsesIncompleteDetails(reason=detail)