« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Response-direction conversion for the Claude 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 FINISH_REASON_TO_WIRE,
11 finish_reason_to_wire,
12)
13from lexigram.ai.relay.mappers.base import new_uuid, record_loss
14from lexigram.ai.relay.mappers.claude.utils import (
15 _TARGET,
16 _tool_call_from_block,
17 _tool_call_to_block,
18)
19from lexigram.ai.relay.media import resolve_media
20from lexigram.contracts.ai.agents import ToolDefinition
21from lexigram.contracts.ai.exceptions import RelayError
22from lexigram.contracts.ai.llm import ChatMessage, ToolCall
23from lexigram.contracts.ai.multimodal import (
24 ContentPart,
25 ImageBase64Part,
26 ImageUrlPart,
27 TextPart,
28)
29from lexigram.contracts.ai.relay.dto import (
30 ClaudeContent,
31 ClaudeMessage,
32 ClaudeResponse,
33 ClaudeUsage,
34)
35from lexigram.contracts.ai.relay.ir import (
36 RelayRequest,
37 RelayResponse,
38 normalize_finish_reason,
39)
40from lexigram.contracts.ai.relay.types import RelayUsage
41from lexigram.contracts.ai.thinking import ThinkingResult
42from lexigram.contracts.core.result import Err, Ok, Result
44if TYPE_CHECKING:
45 from lexigram.ai.relay.mappers.claude import ClaudeMapper
48class ResponseMixin:
49 """Response conversion: wire ``ClaudeResponse`` to IR and back."""
51 def response_to_ir(
52 self: ClaudeMapper, payload: Any, *, context: ConversionContext
53 ) -> Result[RelayResponse, RelayError]:
54 """Convert a ``ClaudeResponse`` into canonical ``RelayResponse``.
56 Args:
57 payload: A wire response DTO.
58 context: Per-conversion context with loss sink.
60 Returns:
61 Ok(response) on success, Err(relay_error) on malformed payload.
62 """
63 if not isinstance(payload, ClaudeResponse):
64 return Err(
65 unsupported_format(
66 f"expected ClaudeResponse, got {type(payload).__name__}"
67 )
68 )
69 try:
70 text_parts: list[str] = []
71 thinking: ThinkingResult | None = None
72 tool_calls: list[ToolCall] = []
73 tool_results: list[ChatMessage] = []
74 for block in payload.content:
75 if block.type == "text":
76 if block.text is not None:
77 text_parts.append(block.text)
78 elif block.type == "thinking":
79 if thinking is None and block.thinking is not None:
80 thinking = ThinkingResult(
81 content=block.thinking, signature=block.signature
82 )
83 elif block.type == "tool_use":
84 tool_calls.append(_tool_call_from_block(block))
85 elif block.type == "tool_result":
86 result_text = "".join(
87 part.text or ""
88 for part in (block.tool_result_content or [])
89 if part.type == "text"
90 )
91 tool_results.append(
92 ChatMessage(
93 role="tool",
94 content=result_text,
95 tool_call_id=block.tool_use_id,
96 )
97 )
98 else:
99 record_loss(
100 context,
101 field=f"content.{block.type}",
102 target=_TARGET,
103 reason="unknown_block_dropped",
104 )
105 passthrough = dict(payload.passthrough)
106 if payload.stop_sequence is not None:
107 passthrough["stop_sequence"] = payload.stop_sequence
108 return Ok(
109 RelayResponse(
110 model=payload.model,
111 id=payload.id,
112 content="".join(text_parts),
113 thinking=thinking,
114 tool_calls=tool_calls,
115 tool_results=tool_results,
116 finish_reason=normalize_finish_reason(payload.stop_reason),
117 usage=self._usage_from_wire(payload.usage),
118 passthrough=passthrough,
119 )
120 )
121 except (RelayError, ValueError, TypeError, KeyError) as exc:
122 return Err(translate(exc, detail="response_to_ir"))
124 def ir_to_response(
125 self: ClaudeMapper, response: RelayResponse, *, context: ConversionContext
126 ) -> Result[Any, RelayError]:
127 """Convert canonical ``RelayResponse`` into a ``ClaudeResponse``.
129 Args:
130 response: Canonical response IR.
131 context: Per-conversion context with loss sink.
133 Returns:
134 Ok(response) on success, Err(relay_error) on failure.
135 """
136 try:
137 passthrough = dict(response.passthrough)
138 stop_sequence = passthrough.pop("stop_sequence", None)
139 blocks: list[ClaudeContent] = []
140 if response.content:
141 blocks.append(ClaudeContent(type="text", text=response.content))
142 for tool_call in response.tool_calls:
143 blocks.append(_tool_call_to_block(tool_call))
144 stop_reason: str | None = None
145 if response.tool_calls:
146 stop_reason = "tool_use"
147 elif stop_sequence is not None:
148 stop_reason = "stop_sequence"
149 elif response.finish_reason is not None:
150 stop_reason = self._stop_reason_from_ir(response.finish_reason, context)
151 return Ok(
152 ClaudeResponse(
153 id=response.id or f"chatcmpl-{new_uuid()}",
154 model=context.resolve_model(response.model),
155 content=blocks,
156 stop_reason=stop_reason,
157 stop_sequence=stop_sequence
158 if stop_reason == "stop_sequence"
159 else None,
160 usage=self._usage_to_wire(response.usage),
161 passthrough=passthrough,
162 )
163 )
164 except (RelayError, ValueError, TypeError, KeyError) as exc:
165 return Err(translate(exc, detail="ir_to_response"))
167 def _message_from_ir(
168 self: ClaudeMapper, message: ChatMessage, context: ConversionContext
169 ) -> Result[ClaudeMessage, RelayError]:
170 """Convert one canonical message into a Claude message."""
171 if message.role == "tool":
172 return Ok(
173 ClaudeMessage(
174 role="user",
175 content=[
176 ClaudeContent(
177 type="tool_result",
178 tool_use_id=message.tool_call_id,
179 tool_result_content=[
180 ClaudeContent(
181 type="text",
182 text=self._text_from_content(message.content),
183 )
184 ],
185 passthrough=dict(message.metadata or {}),
186 )
187 ],
188 )
189 )
190 if message.role == "assistant":
191 blocks: list[ClaudeContent] = []
192 for block in message.thinking_blocks or []:
193 if isinstance(block, dict) and block.get("type") == "thinking":
194 blocks.append(
195 ClaudeContent(
196 type="thinking",
197 thinking=str(block.get("thinking", "")),
198 signature=(
199 str(block["signature"])
200 if block.get("signature")
201 else None
202 ),
203 )
204 )
205 content_blocks = self._content_to_blocks(message.content, context)
206 if content_blocks.is_err():
207 return Err(content_blocks.unwrap_err())
208 wire_blocks = content_blocks.unwrap()
209 has_text = any(block.type == "text" and block.text for block in wire_blocks)
210 if not has_text:
211 pure_tool_turn = bool(
212 (message.metadata or {}).get("function_call_item_ids")
213 )
214 if (
215 message.tool_calls and not pure_tool_turn
216 ) or not message.tool_calls:
217 wire_blocks = [ClaudeContent(type="text", text="...")]
218 else:
219 wire_blocks = []
220 blocks.extend(wire_blocks)
221 for tool_call in message.tool_calls or []:
222 blocks.append(_tool_call_to_block(tool_call))
223 return Ok(
224 ClaudeMessage(role="assistant", content=self._collapse_content(blocks))
225 )
226 if message.role == "user":
227 if isinstance(message.content, list) and any(
228 isinstance(part, ImageBase64Part) for part in message.content
229 ):
230 return Ok(ClaudeMessage(role="user", content=[]))
231 user_blocks = self._content_to_blocks(message.content, context)
232 if user_blocks.is_err():
233 return Err(user_blocks.unwrap_err())
234 return Ok(
235 ClaudeMessage(
236 role="user", content=self._collapse_content(user_blocks.unwrap())
237 )
238 )
239 record_loss(
240 context,
241 field="messages",
242 target=_TARGET,
243 reason=f"unknown_role_{message.role}_dropped",
244 )
245 return Ok(
246 ClaudeMessage(role="user", content=[ClaudeContent(type="text", text="")])
247 )
249 def _content_to_blocks(
250 self: ClaudeMapper, content: str | list[ContentPart], context: ConversionContext
251 ) -> Result[list[ClaudeContent], RelayError]:
252 """Convert canonical content into a Claude block list."""
253 if isinstance(content, str):
254 return Ok([ClaudeContent(type="text", text=content)])
255 blocks: list[ClaudeContent] = []
256 for part in content:
257 if isinstance(part, TextPart):
258 blocks.append(ClaudeContent(type="text", text=part.text))
259 elif isinstance(part, ImageBase64Part):
260 blocks.append(
261 ClaudeContent(
262 type="image",
263 image_source={
264 "type": "base64",
265 "media_type": part.media_type,
266 "data": part.data,
267 },
268 )
269 )
270 elif isinstance(part, ImageUrlPart):
271 resolved = self._resolve_image(part, context)
272 if resolved.is_err():
273 return Err(resolved.unwrap_err())
274 blocks.append(
275 ClaudeContent(
276 type="image",
277 image_source={
278 "type": "base64",
279 "media_type": resolved.unwrap()[0],
280 "data": resolved.unwrap()[1],
281 },
282 )
283 )
284 else:
285 record_loss(
286 context,
287 field="message.content",
288 target=_TARGET,
289 reason="unknown_content_part",
290 )
291 if not blocks:
292 blocks.append(ClaudeContent(type="text", text=""))
293 return Ok(blocks)
295 @staticmethod
296 def _collapse_content(
297 blocks: list[ClaudeContent],
298 ) -> str | list[ClaudeContent]:
299 """Collapse a single text block into plain string content.
301 The Claude wire protocol accepts either a plain string or a block
302 list for message content; relaykit emits plain strings for
303 single-text messages.
304 """
305 if len(blocks) == 1 and blocks[0].type == "text" and blocks[0].text is not None:
306 return blocks[0].text
307 return blocks
309 @staticmethod
310 def _resolve_image(
311 part: ImageUrlPart, context: ConversionContext
312 ) -> Result[tuple[str, str], RelayError]:
313 """Resolve a URL or data-URI image for Claude.
315 Data URIs decode locally; URLs go through the context resolver.
316 """
317 resolved = resolve_media(
318 part.url,
319 context,
320 field="message.content",
321 target=_TARGET,
322 lossy=False,
323 )
324 if resolved.is_err():
325 return Err(resolved.unwrap_err())
326 image = resolved.unwrap()
327 assert image is not None # lossy=False never drops media # noqa: S101
328 return Ok(image)
330 @staticmethod
331 def _text_from_content(content: str | list[ContentPart]) -> str:
332 """Extract plain text from canonical content."""
333 if isinstance(content, str):
334 return content
335 return "".join(part.text for part in content if isinstance(part, TextPart))
337 @staticmethod
338 def _tool_from_ir(tool: ToolDefinition) -> dict[str, Any]:
339 """Serialize a canonical ``ToolDefinition`` as a Claude wire tool."""
340 return {
341 "name": tool.name,
342 "description": tool.description,
343 "input_schema": tool.parameters,
344 }
346 def _thinking_from_ir(
347 self: ClaudeMapper, request: RelayRequest, context: ConversionContext
348 ) -> dict[str, Any] | None:
349 """Rebuild the Claude ``thinking`` dict from canonical thinking."""
350 thinking = request.thinking
351 if thinking is None:
352 return None
353 if thinking.effort is not None:
354 record_loss(
355 context,
356 field="thinking",
357 target=_TARGET,
358 reason="effort_not_supported",
359 )
360 return None
361 if thinking.suppress:
362 return {"type": "disabled"}
363 return {"type": "enabled", "budget_tokens": thinking.budget_tokens}
365 @staticmethod
366 def _usage_from_wire(usage: ClaudeUsage | None) -> RelayUsage | None:
367 """Map a wire ``ClaudeUsage`` into canonical ``RelayUsage``.
369 Mirrors relaykit's ``buildOpenAIStyleUsageFromClaudeUsage``: the
370 prompt count includes cache reads and cache creations, and the
371 chat ``input_tokens`` is stamped with that total.
372 """
373 if usage is None:
374 return None
375 prompt = (
376 usage.input_tokens
377 + usage.cache_read_input_tokens
378 + usage.cache_creation_input_tokens
379 )
380 return RelayUsage(
381 prompt_tokens=prompt,
382 completion_tokens=usage.output_tokens,
383 cache_read_tokens=usage.cache_read_input_tokens,
384 cache_creation_tokens=usage.cache_creation_input_tokens,
385 input_tokens=prompt,
386 )
388 @staticmethod
389 def _usage_to_wire(usage: RelayUsage | None) -> ClaudeUsage | None:
390 """Serialize canonical ``RelayUsage`` into a ``ClaudeUsage``."""
391 if usage is None:
392 return None
393 return ClaudeUsage(
394 input_tokens=usage.prompt_tokens,
395 output_tokens=usage.completion_tokens,
396 cache_read_input_tokens=usage.cache_read_tokens,
397 cache_creation_input_tokens=usage.cache_creation_tokens,
398 )
400 @staticmethod
401 def _stop_reason_from_ir(
402 finish_reason: str | None, context: ConversionContext
403 ) -> str | None:
404 """Map a canonical finish reason back to a Claude stop reason."""
405 if finish_reason is None:
406 return None
407 if finish_reason in {"function_call", "content_filter"}:
408 record_loss(
409 context,
410 field="finish_reason",
411 target=_TARGET,
412 reason=f"{finish_reason}_adapted",
413 )
414 elif finish_reason not in FINISH_REASON_TO_WIRE:
415 record_loss(
416 context,
417 field="finish_reason",
418 target=_TARGET,
419 reason="finish_reason_adapted",
420 )
421 return finish_reason_to_wire(finish_reason, _TARGET)