1"""Shared parsing/serialization helpers for the OpenAI Responses mapper."""
2
3from __future__ import annotations
4
5from typing import Any
6
7from lexigram.ai.relay.finish_reasons import responses_incomplete_from_finish
8from lexigram.contracts.ai.relay.dto import ResponsesIncompleteDetails
9from lexigram.contracts.ai.relay.types import RelayFormat
10from lexigram.serialization import dumps_str, loads_str
11
12_TARGET = RelayFormat.OPENAI_RESPONSES
13
14
15def _parse_arguments(arguments: str) -> dict[str, Any] | str:
16 """Parse a wire arguments string into dict form when possible.
17
18 Args:
19 arguments: Raw JSON argument string from the wild.
20
21 Returns:
22 The parsed dict, or the original string when it is empty or not
23 valid JSON.
24 """
25 if not isinstance(arguments, str) or not arguments.strip():
26 return arguments
27 try:
28 value = loads_str(arguments)
29 except (ValueError, TypeError):
30 return arguments
31 if isinstance(value, dict):
32 return value
33 return arguments
34
35
36def _arguments_to_wire(arguments: Any) -> str:
37 """Serialize canonical arguments into a JSON string.
38
39 Args:
40 arguments: Canonical arguments (dict, string, or anything else).
41
42 Returns:
43 A JSON string for the wire, or an empty string when unsupported.
44 """
45 if isinstance(arguments, dict):
46 return dumps_str(arguments)
47 if isinstance(arguments, str):
48 return arguments
49 return ""
50
51
52def _incomplete_for_finish(
53 finish_reason: str | None,
54) -> ResponsesIncompleteDetails | None:
55 """Map a canonical finish reason to an incomplete-details payload."""
56 detail = responses_incomplete_from_finish(finish_reason)
57 if detail is None:
58 return None
59 return ResponsesIncompleteDetails(reason=detail)