Coverage for src / lexigram / ai / relay / stream / claude.py: 90%
157 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Anthropic Claude Messages target stream emitter.
3Maps one canonical :class:`StreamDelta` into valid Claude SSE events.
4The emitter owns the block lifecycle: each text, thinking, and tool-use
5block is started once, closed exactly once, and the terminal
6``message_delta``/``message_stop`` pair is emitted once. Usage rides on
7``message_delta`` so the included usage shape matches the target wire
8format.
9"""
11from __future__ import annotations
13from typing import Any
15from lexigram.ai.relay.errors import stream_state_invalid
16from lexigram.ai.relay.finish_reasons import (
17 finish_reason_to_wire,
18 normalize_finish_reason,
19)
20from lexigram.ai.relay.stream.state import (
21 StreamSnapshot,
22 StreamToolCallRecord,
23 _first_tool_contribution,
24 _pre_text,
25 _pre_thinking,
26 _pre_tool_indices,
27 _started_pre,
28)
29from lexigram.contracts.ai.exceptions import RelayError
30from lexigram.contracts.ai.relay.dto import (
31 ClaudeContent,
32 ClaudeResponse,
33 ClaudeStreamEvent,
34 ClaudeUsage,
35)
36from lexigram.contracts.ai.relay.ir import StreamDelta
37from lexigram.contracts.ai.relay.types import RelayFormat, RelayUsage
38from lexigram.contracts.core.result import Err, Ok, Result
40__all__ = ["claude_emitter"]
43def _claude_stop_reason(reason: str) -> str:
44 """Map a canonical finish reason onto Claude's wire values."""
45 return finish_reason_to_wire(normalize_finish_reason(reason), RelayFormat.CLAUDE)
48def _usage_to_wire(usage: RelayUsage) -> ClaudeUsage:
49 """Serialize canonical usage into the Claude usage shape."""
50 return ClaudeUsage(
51 input_tokens=usage.prompt_tokens,
52 output_tokens=usage.completion_tokens,
53 cache_creation_input_tokens=usage.cache_creation_tokens,
54 cache_read_input_tokens=usage.cache_read_tokens,
55 )
58def _message_start(
59 state: StreamSnapshot, *, role: str = "assistant"
60) -> ClaudeStreamEvent:
61 """Build the ``message_start`` event with a message snapshot."""
62 return ClaudeStreamEvent(
63 type="message_start",
64 message=ClaudeResponse(
65 id=state.stream_id or "",
66 type="message",
67 role=role,
68 model=state.model,
69 content=[],
70 usage=(
71 _usage_to_wire(state.usage)
72 if state.usage is not None
73 else ClaudeUsage()
74 ),
75 ),
76 )
79def _tool_record(state: StreamSnapshot, index: int) -> StreamToolCallRecord | None:
80 return next((r for r in state.tool_calls if r.index == index), None)
83def _thinking_open_pre(state: StreamSnapshot, delta: StreamDelta) -> bool:
84 """Whether a thinking block was already open before this delta."""
85 pre_thinking = _pre_thinking(state, delta)
86 return (
87 bool(pre_thinking)
88 and not _pre_text(state, delta)
89 and not _pre_tool_indices(state, delta)
90 )
93def _text_open_pre(state: StreamSnapshot, delta: StreamDelta) -> bool:
94 """Whether a text block was already open before this delta."""
95 return bool(_pre_text(state, delta)) and not _pre_tool_indices(state, delta)
98def _thinking_block_index(state: StreamSnapshot) -> int:
99 """Deterministic Claude block index for the thinking block."""
100 return int(bool(state.text)) + len(state.tool_calls)
103def _text_block_index(state: StreamSnapshot) -> int:
104 """Deterministic Claude block index for the text block."""
105 return int(bool(state.thinking_text)) + len(state.tool_calls)
108def _thinking_reopen_index(state: StreamSnapshot) -> int:
109 """Block index for a reopened thinking block."""
110 return int(bool(state.text)) + 1 + len(state.tool_calls)
113def _text_reopen_index(state: StreamSnapshot) -> int:
114 """Block index for a reopened text block."""
115 return int(bool(state.thinking_text)) + 1 + len(state.tool_calls)
118def _tool_block_index(state: StreamSnapshot, position: int) -> int:
119 """Deterministic Claude block index for a tool-use block."""
120 return int(bool(state.thinking_text)) + int(bool(state.text)) + position
123def _tool_position(state: StreamSnapshot, index: int) -> int:
124 """Position of a tool call among the stream's tool call records."""
125 for position, record in enumerate(state.tool_calls):
126 if record.index == index:
127 return position
128 return len(state.tool_calls) - 1
131def _content_block_stop(index: int) -> ClaudeStreamEvent:
132 return ClaudeStreamEvent(type="content_block_stop", index=index)
135def _close_open(state: StreamSnapshot, delta: StreamDelta) -> list[ClaudeStreamEvent]:
136 """Close the single currently-open content block, if any.
138 The close index must match the index the block was opened with, so it
139 is reconstructed from the pre-delta state (the block's content types
140 cannot change between its start and its stop).
141 """
142 pre_text = _pre_text(state, delta)
143 pre_thinking = _pre_thinking(state, delta)
144 pre_tools = _pre_tool_indices(state, delta)
145 if bool(pre_thinking) and not pre_text and not pre_tools:
146 return [_content_block_stop(int(bool(pre_text)) + len(pre_tools))]
147 if pre_text and not pre_tools:
148 return [_content_block_stop(int(bool(pre_thinking)) + len(pre_tools))]
149 if pre_tools:
150 ordered = [r.index for r in state.tool_calls if r.index in pre_tools]
151 last = ordered[-1] if ordered else 0
152 position = _tool_position(state, last)
153 return [
154 _content_block_stop(
155 int(bool(pre_thinking)) + int(bool(pre_text)) + position
156 )
157 ]
158 return []
161def _thinking_signature(delta: StreamDelta) -> str | None:
162 signature = delta.passthrough.get("signature")
163 return signature if isinstance(signature, str) and signature else None
166def _text_events(state: StreamSnapshot, delta: StreamDelta) -> list[ClaudeStreamEvent]:
167 events: list[ClaudeStreamEvent] = []
168 if not delta.content:
169 return events
170 first = state.text == delta.content
171 if first:
172 events.extend(_close_open(state, delta))
173 index = _text_block_index(state)
174 events.append(
175 ClaudeStreamEvent(
176 type="content_block_start",
177 index=index,
178 content_block=ClaudeContent(type="text", text=""),
179 )
180 )
181 elif _text_open_pre(state, delta):
182 index = _text_block_index(state)
183 else:
184 events.extend(_close_open(state, delta))
185 index = _text_reopen_index(state)
186 events.append(
187 ClaudeStreamEvent(
188 type="content_block_start",
189 index=index,
190 content_block=ClaudeContent(type="text", text=""),
191 )
192 )
193 events.append(
194 ClaudeStreamEvent(
195 type="content_block_delta",
196 index=index,
197 delta={"type": "text_delta", "text": delta.content},
198 )
199 )
200 return events
203def _thinking_events(
204 state: StreamSnapshot, delta: StreamDelta
205) -> list[ClaudeStreamEvent]:
206 events: list[ClaudeStreamEvent] = []
207 if not delta.thinking_delta:
208 return events
209 first = state.thinking_text == delta.thinking_delta
210 if first:
211 events.extend(_close_open(state, delta))
212 index = _thinking_block_index(state)
213 events.append(
214 ClaudeStreamEvent(
215 type="content_block_start",
216 index=index,
217 content_block=ClaudeContent(type="thinking", thinking=""),
218 )
219 )
220 elif _thinking_open_pre(state, delta):
221 index = _thinking_block_index(state)
222 else:
223 events.extend(_close_open(state, delta))
224 index = _thinking_reopen_index(state)
225 events.append(
226 ClaudeStreamEvent(
227 type="content_block_start",
228 index=index,
229 content_block=ClaudeContent(type="thinking", thinking=""),
230 )
231 )
232 signature = _thinking_signature(delta)
233 thinking_delta_payload: dict[str, Any] = {
234 "type": "thinking_delta",
235 "thinking": delta.thinking_delta,
236 }
237 if signature is not None:
238 thinking_delta_payload["signature"] = signature
239 events.append(
240 ClaudeStreamEvent(
241 type="content_block_delta",
242 index=index,
243 delta=thinking_delta_payload,
244 )
245 )
246 return events
249def _tool_events(state: StreamSnapshot, delta: StreamDelta) -> list[ClaudeStreamEvent]:
250 events: list[ClaudeStreamEvent] = []
251 index = delta.tool_call_index
252 if index is None:
253 return events
254 record = _tool_record(state, index)
255 if record is None:
256 return events
257 first = _first_tool_contribution(state, delta)
258 if first:
259 events.extend(_close_open(state, delta))
260 position = _tool_position(state, index)
261 block_index = _tool_block_index(state, position)
262 if first:
263 events.append(
264 ClaudeStreamEvent(
265 type="content_block_start",
266 index=block_index,
267 content_block=ClaudeContent(
268 type="tool_use",
269 tool_use_id=record.id,
270 name=record.name,
271 input={},
272 ),
273 )
274 )
275 if delta.tool_call_arguments is not None:
276 events.append(
277 ClaudeStreamEvent(
278 type="content_block_delta",
279 index=block_index,
280 delta={
281 "type": "input_json_delta",
282 "partial_json": delta.tool_call_arguments,
283 },
284 )
285 )
286 return events
289def _finish_events(
290 state: StreamSnapshot, delta: StreamDelta
291) -> list[ClaudeStreamEvent]:
292 events: list[ClaudeStreamEvent] = []
293 events.extend(_close_open(state, delta))
294 reason = _claude_stop_reason(delta.finish_reason or "stop")
295 usage = _usage_to_wire(state.usage) if state.usage is not None else None
296 events.append(
297 ClaudeStreamEvent(
298 type="message_delta",
299 delta={"stop_reason": reason},
300 usage=usage,
301 )
302 )
303 events.append(ClaudeStreamEvent(type="message_stop"))
304 return events
307def _usage_events(state: StreamSnapshot, delta: StreamDelta) -> list[ClaudeStreamEvent]:
308 return []
311def claude_emitter(
312 delta: StreamDelta, *, state: StreamSnapshot
313) -> Result[tuple[ClaudeStreamEvent, ...], RelayError]:
314 """Map one canonical delta into Claude SSE events.
316 Args:
317 delta: One canonical stream delta.
318 state: Accumulated session snapshot.
320 Returns:
321 Ok(tuple of events) on success; ``stream_state_invalid`` for an
322 unknown delta kind.
323 """
324 if delta.kind == "role":
325 events: list[ClaudeStreamEvent] = []
326 if not _started_pre(state, delta):
327 events.append(_message_start(state))
328 return Ok(tuple(events))
329 if delta.kind == "content":
330 events = []
331 if not _started_pre(state, delta):
332 events.append(_message_start(state))
333 events.extend(_text_events(state, delta))
334 return Ok(tuple(events))
335 if delta.kind == "thinking":
336 events = []
337 if not _started_pre(state, delta):
338 events.append(_message_start(state))
339 events.extend(_thinking_events(state, delta))
340 return Ok(tuple(events))
341 if delta.kind == "tool_call":
342 events = []
343 if not _started_pre(state, delta):
344 events.append(_message_start(state))
345 events.extend(_tool_events(state, delta))
346 return Ok(tuple(events))
347 if delta.kind == "finish":
348 events = []
349 if not _started_pre(state, delta):
350 events.append(_message_start(state))
351 events.extend(_finish_events(state, delta))
352 return Ok(tuple(events))
353 if delta.kind == "usage":
354 return Ok(tuple(_usage_events(state, delta)))
355 if delta.kind == "status":
356 return Ok(())
357 return Err(stream_state_invalid(f"unknown delta kind {delta.kind!r} for claude"))