« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Response-direction conversion for the OpenAI Responses mapper."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.ai.relay.context import ConversionContext
8from lexigram.ai.relay.errors import translate, unsupported_format
9from lexigram.ai.relay.finish_reasons import (
10 responses_status_from_finish,
11)
12from lexigram.ai.relay.mappers.base import new_uuid, record_loss
13from lexigram.ai.relay.mappers.openai_responses.utils import (
14 _TARGET,
15 _arguments_to_wire,
16 _incomplete_for_finish,
17 _parse_arguments,
18)
19from lexigram.contracts.ai.exceptions import RelayError
20from lexigram.contracts.ai.llm import ChatMessage, FunctionCall, ToolCall
21from lexigram.contracts.ai.multimodal import TextPart
22from lexigram.contracts.ai.relay.dto import (
23 ResponsesIncompleteDetails,
24 ResponsesItem,
25 ResponsesResponse,
26 ResponsesUsage,
27)
28from lexigram.contracts.ai.relay.ir import RelayResponse
29from lexigram.contracts.ai.relay.types import RelayUsage
30from lexigram.contracts.ai.thinking import ThinkingResult
31from lexigram.contracts.core.result import Err, Ok, Result
33if TYPE_CHECKING:
34 from lexigram.ai.relay.mappers.openai_responses import OpenAIResponsesMapper
37class ResponseMixin:
38 """Response conversion: wire ``ResponsesResponse`` to IR and back."""
40 def response_to_ir(
41 self: OpenAIResponsesMapper,
42 payload: Any,
43 *,
44 context: ConversionContext,
45 ) -> Result[RelayResponse, RelayError]:
46 """Convert a ``ResponsesResponse`` into canonical ``RelayResponse``.
48 Args:
49 payload: A wire response DTO.
50 context: Per-conversion context with loss sink.
52 Returns:
53 Ok(response) on success, Err(relay_error) on malformed payload.
54 """
55 if not isinstance(payload, ResponsesResponse):
56 return Err(
57 unsupported_format(
58 f"expected ResponsesResponse, got {type(payload).__name__}"
59 )
60 )
61 try:
62 passthrough = dict(payload.passthrough)
63 if payload.error is not None:
64 passthrough["error"] = payload.error
65 if payload.object != "response":
66 passthrough["object"] = payload.object
67 content_parts: list[str] = []
68 tool_calls: list[ToolCall] = []
69 tool_results: list[ChatMessage] = []
70 reasoning_text: list[str] = []
71 web_search_calls: list[dict[str, Any]] = []
72 for index, output_item in enumerate(payload.output):
73 item_type = output_item.type
74 if item_type == "message":
75 for part in output_item.content or []:
76 if not isinstance(part, dict):
77 content_parts.append(str(part))
78 continue
79 part_type = part.get("type")
80 if part_type == "output_text":
81 content_parts.append(str(part.get("text", "")))
82 else:
83 record_loss(
84 context,
85 field=part_type or "part",
86 target=_TARGET,
87 reason="unknown_part_type",
88 )
89 elif item_type == "reasoning":
90 reasoning_text.extend(self._summary_texts(output_item.summary))
91 elif item_type == "function_call":
92 tool_calls.append(
93 ToolCall(
94 id=output_item.call_id or output_item.id or "",
95 type="function",
96 function=FunctionCall(
97 name=output_item.name or "",
98 arguments=_parse_arguments(output_item.arguments or ""),
99 ),
100 )
101 )
102 elif item_type == "function_call_output":
103 tool_results.append(
104 ChatMessage(
105 role="tool",
106 content=output_item.output or "",
107 tool_call_id=output_item.call_id,
108 )
109 )
110 elif item_type == "web_search_call":
111 web_search_calls.append(output_item.to_dict())
112 record_loss(
113 context,
114 field=f"output[{index}]",
115 target=_TARGET,
116 reason="unsupported_item_preserved",
117 severity="info",
118 )
119 else:
120 record_loss(
121 context,
122 field=f"output[{index}]",
123 target=_TARGET,
124 reason="unknown_item_dropped",
125 )
126 if web_search_calls:
127 passthrough["web_search_calls"] = web_search_calls
128 thinking: ThinkingResult | None = None
129 if reasoning_text:
130 tokens: int | None = None
131 if payload.usage is not None:
132 details = payload.usage.output_tokens_details
133 if isinstance(details, dict) and isinstance(
134 details.get("reasoning_tokens"), int
135 ):
136 tokens = details["reasoning_tokens"]
137 thinking = ThinkingResult(
138 content="".join(reasoning_text), tokens=tokens
139 )
140 return Ok(
141 RelayResponse(
142 model=payload.model,
143 id=payload.id,
144 created=payload.created_at,
145 content="".join(content_parts),
146 thinking=thinking,
147 tool_calls=tool_calls,
148 tool_results=tool_results,
149 finish_reason=self._finish_from_status(
150 payload.status,
151 payload.incomplete_details,
152 bool(tool_calls),
153 ),
154 status=payload.status,
155 incomplete_details=(
156 payload.incomplete_details.to_dict()
157 if payload.incomplete_details is not None
158 else None
159 ),
160 usage=self._usage_from_wire(payload.usage),
161 passthrough=passthrough,
162 )
163 )
164 except (RelayError, ValueError, TypeError, KeyError) as exc:
165 return Err(translate(exc, detail="response_to_ir"))
167 def ir_to_response(
168 self: OpenAIResponsesMapper,
169 response: RelayResponse,
170 *,
171 context: ConversionContext,
172 ) -> Result[Any, RelayError]:
173 """Convert canonical ``RelayResponse`` into a ``ResponsesResponse``.
175 Args:
176 response: Canonical response IR.
177 context: Per-conversion context with loss sink.
179 Returns:
180 Ok(response) on success, Err(relay_error) on failure.
181 """
182 try:
183 passthrough = dict(response.passthrough)
184 error = passthrough.pop("error", None)
185 object_type = passthrough.pop("object", "response")
186 status, incomplete = self._status_from_finish(response)
187 response_id = response.id or f"chatcmpl-{new_uuid()}"
188 item_status = "incomplete" if status == "incomplete" else "completed"
189 items: list[ResponsesItem] = []
190 content_parts: list[dict[str, Any]] = []
191 if response.content:
192 content_parts.append(
193 {
194 "type": "output_text",
195 "text": response.content,
196 "annotations": [],
197 }
198 )
199 if content_parts:
200 items.append(
201 ResponsesItem(
202 type="message",
203 role="assistant",
204 id=f"{response_id}_msg_0",
205 status=item_status,
206 content=content_parts,
207 quality="",
208 size="",
209 )
210 )
211 if response.thinking is not None and response.thinking.content:
212 items.append(
213 ResponsesItem(
214 type="reasoning",
215 id=f"{response_id}_reasoning_0",
216 status=item_status,
217 role="",
218 content=[
219 {
220 "type": "summary_text",
221 "text": response.thinking.content,
222 "annotations": None,
223 }
224 ],
225 quality="",
226 size="",
227 )
228 )
229 for tool in response.tool_calls:
230 call_id = tool.id or f"call_{new_uuid()}"
231 items.append(
232 ResponsesItem(
233 type="function_call",
234 id=call_id,
235 status=item_status,
236 role="",
237 content=None,
238 quality="",
239 size="",
240 call_id=call_id,
241 name=tool.function.name if tool.function else "",
242 arguments=_arguments_to_wire(
243 tool.function.arguments if tool.function else {}
244 ),
245 )
246 )
247 for index, result in enumerate(response.tool_results):
248 items.append(
249 ResponsesItem(
250 type="function_call_output",
251 id=f"fcoc_{index}",
252 call_id=result.tool_call_id,
253 output=self._result_output(result),
254 )
255 )
256 return Ok(
257 ResponsesResponse(
258 id=response_id,
259 model=context.resolve_model(response.model),
260 output=items,
261 object=object_type,
262 created_at=response.created or 0,
263 status=status,
264 incomplete_details=incomplete,
265 error=error if isinstance(error, dict) else None,
266 usage=self._usage_to_wire(response.usage),
267 passthrough=passthrough,
268 )
269 )
270 except (RelayError, ValueError, TypeError, KeyError) as exc:
271 return Err(translate(exc, detail="ir_to_response"))
273 @staticmethod
274 def _summary_texts(
275 summary: list[dict[str, Any]] | None,
276 ) -> list[str]:
277 """Extract text from reasoning summary blocks."""
278 return [
279 str(item.get("text", ""))
280 for item in summary or []
281 if isinstance(item, dict) and item.get("type") == "summary_text"
282 ]
284 @staticmethod
285 def _finish_from_status(
286 status: str | None,
287 incomplete_details: ResponsesIncompleteDetails | None,
288 has_tool_calls: bool,
289 ) -> str | None:
290 """Derive a canonical finish reason from a wire status."""
291 if status == "completed":
292 return "tool_calls" if has_tool_calls else "stop"
293 if status == "incomplete":
294 reason = (
295 incomplete_details.reason if incomplete_details is not None else None
296 )
297 if reason == "max_output_tokens":
298 return "length"
299 if reason == "content_filter":
300 return "content_filter"
301 return "other"
302 if status == "failed":
303 return "other"
304 return None
306 @staticmethod
307 def _status_from_finish(
308 response: RelayResponse,
309 ) -> tuple[str | None, ResponsesIncompleteDetails | None]:
310 """Derive a wire status from canonical finish behavior."""
311 status = response.status
312 incomplete: ResponsesIncompleteDetails | None = None
313 if response.incomplete_details is not None:
314 raw = dict(response.incomplete_details)
315 reason = raw.pop("reason", None)
316 incomplete = ResponsesIncompleteDetails(reason=reason, passthrough=raw)
317 if status is not None:
318 if status == "incomplete" and incomplete is None:
319 derived = _incomplete_for_finish(response.finish_reason)
320 if derived is not None:
321 incomplete = derived
322 return status, incomplete
323 finish = response.finish_reason
324 wire_status, detail = responses_status_from_finish(finish)
325 if detail is None:
326 return wire_status, None
327 return wire_status, ResponsesIncompleteDetails(reason=detail)
329 @staticmethod
330 def _result_output(message: ChatMessage) -> str:
331 """Extract a tool result string from a canonical tool message."""
332 content = message.content
333 if isinstance(content, list):
334 return "".join(part.text for part in content if isinstance(part, TextPart))
335 return str(content or "")
337 @staticmethod
338 def _usage_from_wire(usage: ResponsesUsage | None) -> RelayUsage | None:
339 """Map wire usage into canonical ``RelayUsage``."""
340 if usage is None:
341 return None
342 input_details = usage.input_tokens_details
343 completion_details = usage.completion_tokens_details
344 if not isinstance(completion_details, dict):
345 completion_details = usage.output_tokens_details
346 return RelayUsage(
347 prompt_tokens=usage.prompt_tokens or usage.input_tokens,
348 completion_tokens=usage.completion_tokens or usage.output_tokens,
349 total_tokens_override=usage.total_tokens or None,
350 cache_read_tokens=(
351 int(input_details.get("cached_tokens", 0) or 0)
352 if isinstance(input_details, dict)
353 else 0
354 ),
355 reasoning_tokens=(
356 int(completion_details.get("reasoning_tokens", 0) or 0)
357 if isinstance(completion_details, dict)
358 else 0
359 ),
360 input_tokens=usage.input_tokens,
361 output_tokens=usage.output_tokens,
362 )
364 @staticmethod
365 def _usage_to_wire(usage: RelayUsage | None) -> ResponsesUsage | None:
366 """Serialize canonical ``RelayUsage`` into wire usage."""
367 if usage is None:
368 return None
369 input_details = (
370 {"cached_tokens": usage.cache_read_tokens}
371 if usage.cache_read_tokens
372 else None
373 )
374 return ResponsesUsage(
375 prompt_tokens=usage.prompt_tokens,
376 completion_tokens=usage.completion_tokens,
377 total_tokens=usage.total_tokens,
378 prompt_tokens_details={"cached_tokens": 0},
379 completion_tokens_details={"reasoning_tokens": usage.reasoning_tokens},
380 input_tokens=usage.prompt_tokens,
381 input_tokens_details=input_details,
382 output_tokens=usage.completion_tokens,
383 )