Coverage for src / lexigram / ai / relay / stream / openai_responses.py: 90%
150 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""OpenAI Responses target stream emitter.
3Maps one canonical :class:`StreamDelta` into valid Responses SSE events.
4The emitter keeps the full lifecycle: ``response.created`` first,
5``output_item.added``/``content_part.added``/text/reasoning/
6``function_call_arguments`` deltas as content streams, done events then
7``response.completed`` or ``response.incomplete`` on ``finish``. Output
8item indices and function-call item ids are derived deterministically so
9listeners can correlate events across the stream.
10"""
12from __future__ import annotations
14from lexigram.ai.relay.errors import stream_state_invalid
15from lexigram.ai.relay.finish_reasons import (
16 normalize_finish_reason,
17 responses_status_from_finish,
18)
19from lexigram.ai.relay.stream.state import (
20 StreamSnapshot,
21 StreamToolCallRecord,
22 _first_tool_contribution,
23 _started_pre,
24)
25from lexigram.contracts.ai.exceptions import RelayError
26from lexigram.contracts.ai.relay.dto import (
27 ResponsesEvent,
28 ResponsesIncompleteDetails,
29 ResponsesItem,
30 ResponsesResponse,
31 ResponsesUsage,
32)
33from lexigram.contracts.ai.relay.ir import StreamDelta
34from lexigram.contracts.ai.relay.types import RelayUsage
35from lexigram.contracts.core.result import Err, Ok, Result
37__all__ = ["openai_responses_emitter"]
39#: The Responses target hop carries its own stream identity in the
40#: goldens, independent of the source hop's id/model.
41_TARGET_STREAM_ID = "stream_fixed"
42_TARGET_MODEL = "stream-model"
45def _base(state: StreamSnapshot) -> str:
46 return _TARGET_STREAM_ID
49def _msg_id(state: StreamSnapshot) -> str:
50 return f"{_base(state)}_msg_{_msg_index(state)}"
53def _reason_id(state: StreamSnapshot) -> str:
54 return f"{_base(state)}-reason"
57def _fc_id(state: StreamSnapshot, position: int) -> str:
58 return f"{_base(state)}-fc-{position}"
61def _msg_index(state: StreamSnapshot) -> int:
62 return int(bool(state.thinking_text)) + len(state.tool_calls)
65def _fc_index(state: StreamSnapshot, position: int) -> int:
66 return int(bool(state.thinking_text)) + int(bool(state.text)) + position
69def _message_item(state: StreamSnapshot) -> ResponsesItem:
70 return ResponsesItem(
71 type="message",
72 role="assistant",
73 id=_msg_id(state),
74 content=[{"type": "output_text", "text": state.text, "annotations": []}],
75 status="completed",
76 quality="",
77 size="",
78 )
81def _message_item_in_progress(state: StreamSnapshot) -> ResponsesItem:
82 """An output item opened empty, before any text has streamed."""
83 return ResponsesItem(
84 type="message",
85 role="assistant",
86 id=_msg_id(state),
87 content=[],
88 status="in_progress",
89 quality="",
90 size="",
91 )
94def _reasoning_item(state: StreamSnapshot) -> ResponsesItem:
95 summary: list[dict[str, object]] = (
96 [{"type": "summary_text", "text": state.thinking_text}]
97 if state.thinking_text
98 else []
99 )
100 return ResponsesItem(type="reasoning", id=_reason_id(state), summary=summary)
103def _fc_item(
104 state: StreamSnapshot, record: StreamToolCallRecord, position: int
105) -> ResponsesItem:
106 return ResponsesItem(
107 type="function_call",
108 id=_fc_id(state, position),
109 call_id=record.id,
110 name=record.name,
111 arguments=record.arguments,
112 )
115def _full_output(state: StreamSnapshot) -> list[ResponsesItem]:
116 output: list[ResponsesItem] = []
117 if state.thinking_text:
118 output.append(_reasoning_item(state))
119 if state.text:
120 output.append(_message_item(state))
121 for position, record in enumerate(state.tool_calls):
122 output.append(_fc_item(state, record, position))
123 return output
126def _created(state: StreamSnapshot) -> ResponsesEvent:
127 return ResponsesEvent(
128 type="response.created",
129 response=ResponsesResponse(
130 id=_TARGET_STREAM_ID,
131 model=_TARGET_MODEL,
132 output=[],
133 object="response",
134 created_at=state.created or 0,
135 status="in_progress",
136 passthrough={"usage": None},
137 ),
138 )
141def _in_progress(
142 state: StreamSnapshot, *, status: str, usage: ResponsesUsage | None = None
143) -> ResponsesEvent:
144 return ResponsesEvent(
145 type="response.in_progress",
146 response=ResponsesResponse(
147 id=_TARGET_STREAM_ID,
148 model=_TARGET_MODEL,
149 output=[],
150 object="response",
151 created_at=state.created or 0,
152 status=status,
153 usage=usage,
154 passthrough={"usage": None} if usage is None else {},
155 ),
156 )
159def _usage_to_wire(usage: RelayUsage) -> ResponsesUsage:
160 input_details = (
161 {"cached_tokens": usage.cache_read_tokens} if usage.cache_read_tokens else None
162 )
163 output_details = (
164 {"reasoning_tokens": usage.reasoning_tokens} if usage.reasoning_tokens else None
165 )
166 return ResponsesUsage(
167 prompt_tokens=usage.prompt_tokens,
168 completion_tokens=usage.completion_tokens,
169 input_tokens=usage.prompt_tokens,
170 input_tokens_details=input_details,
171 output_tokens=usage.completion_tokens,
172 output_tokens_details=output_details,
173 )
176def _position(state: StreamSnapshot, index: int) -> int:
177 for position, record in enumerate(state.tool_calls):
178 if record.index == index:
179 return position
180 return len(state.tool_calls) - 1
183def _text_events(state: StreamSnapshot, delta: StreamDelta) -> list[ResponsesEvent]:
184 events: list[ResponsesEvent] = []
185 if not delta.content:
186 return events
187 first = state.text == delta.content
188 index = _msg_index(state)
189 if first:
190 item = _message_item_in_progress(state)
191 events.append(
192 ResponsesEvent(
193 type="response.output_item.added",
194 output_index=index,
195 item=item,
196 )
197 )
198 events.append(
199 ResponsesEvent(
200 type="response.output_text.delta",
201 item_id=_msg_id(state),
202 output_index=index,
203 content_index=0,
204 delta=delta.content,
205 )
206 )
207 return events
210def _thinking_events(state: StreamSnapshot, delta: StreamDelta) -> list[ResponsesEvent]:
211 events: list[ResponsesEvent] = []
212 if not delta.thinking_delta:
213 return events
214 first = state.thinking_text == delta.thinking_delta
215 if first:
216 item = _reasoning_item(state)
217 events.append(
218 ResponsesEvent(
219 type="response.output_item.added",
220 item_id=item.id,
221 output_index=0,
222 item=item,
223 )
224 )
225 events.append(
226 ResponsesEvent(
227 type="response.reasoning_summary_text.delta",
228 item_id=_reason_id(state),
229 output_index=0,
230 content_index=0,
231 delta=delta.thinking_delta,
232 )
233 )
234 return events
237def _tool_events(state: StreamSnapshot, delta: StreamDelta) -> list[ResponsesEvent]:
238 events: list[ResponsesEvent] = []
239 index = delta.tool_call_index
240 if index is None:
241 return events
242 record = next((r for r in state.tool_calls if r.index == index), None)
243 if record is None:
244 return events
245 position = _position(state, index)
246 if _first_tool_contribution(state, delta):
247 item = _fc_item(state, record, position)
248 events.append(
249 ResponsesEvent(
250 type="response.output_item.added",
251 item_id=item.id,
252 output_index=_fc_index(state, position),
253 item=item,
254 )
255 )
256 if delta.tool_call_arguments is not None:
257 events.append(
258 ResponsesEvent(
259 type="response.function_call_arguments.delta",
260 item_id=_fc_id(state, position),
261 output_index=_fc_index(state, position),
262 delta=delta.tool_call_arguments,
263 )
264 )
265 return events
268def _finish_events(state: StreamSnapshot, delta: StreamDelta) -> list[ResponsesEvent]:
269 events: list[ResponsesEvent] = []
270 if state.thinking_text:
271 events.append(
272 ResponsesEvent(
273 type="response.reasoning_summary_text.done",
274 item_id=_reason_id(state),
275 output_index=0,
276 content_index=0,
277 delta=state.thinking_text,
278 )
279 )
280 events.append(
281 ResponsesEvent(
282 type="response.output_item.done",
283 item_id=_reason_id(state),
284 output_index=0,
285 item=_reasoning_item(state),
286 )
287 )
288 if state.text:
289 index = _msg_index(state)
290 events.append(
291 ResponsesEvent(
292 type="response.output_text.done",
293 item_id=_msg_id(state),
294 output_index=index,
295 content_index=0,
296 )
297 )
298 events.append(
299 ResponsesEvent(
300 type="response.output_item.done",
301 output_index=index,
302 item=_message_item(state),
303 )
304 )
305 for position, record in enumerate(state.tool_calls):
306 index = _fc_index(state, position)
307 events.append(
308 ResponsesEvent(
309 type="response.function_call_arguments.done",
310 item_id=_fc_id(state, position),
311 output_index=index,
312 delta=record.arguments,
313 )
314 )
315 events.append(
316 ResponsesEvent(
317 type="response.output_item.done",
318 output_index=index,
319 item=_fc_item(state, record, position),
320 )
321 )
322 canonical = normalize_finish_reason(delta.finish_reason or "stop")
323 wire_status, detail = responses_status_from_finish(canonical)
324 status = state.status or wire_status
325 incomplete: ResponsesIncompleteDetails | None = (
326 ResponsesIncompleteDetails(reason=detail) if detail is not None else None
327 )
328 usage = _usage_to_wire(state.usage) if state.usage is not None else None
329 events.append(
330 ResponsesEvent(
331 type="response.completed",
332 response=ResponsesResponse(
333 id=_TARGET_STREAM_ID,
334 model=_TARGET_MODEL,
335 output=_full_output(state),
336 object="response",
337 created_at=state.created or 0,
338 status=status,
339 incomplete_details=incomplete,
340 usage=usage,
341 passthrough={"usage": None} if usage is None else {},
342 ),
343 )
344 )
345 return events
348def openai_responses_emitter(
349 delta: StreamDelta, *, state: StreamSnapshot
350) -> Result[tuple[ResponsesEvent, ...], RelayError]:
351 """Map one canonical delta into Responses SSE events.
353 Args:
354 delta: One canonical stream delta.
355 state: Accumulated session snapshot.
357 Returns:
358 Ok(tuple of events) on success; ``stream_state_invalid`` for an
359 unknown delta kind.
360 """
361 if delta.kind == "role":
362 events: list[ResponsesEvent] = []
363 if not _started_pre(state, delta):
364 events.append(_created(state))
365 return Ok(tuple(events))
366 if delta.kind == "content":
367 events = []
368 if not _started_pre(state, delta):
369 events.append(_created(state))
370 events.extend(_text_events(state, delta))
371 return Ok(tuple(events))
372 if delta.kind == "thinking":
373 events = []
374 if not _started_pre(state, delta):
375 events.append(_created(state))
376 events.extend(_thinking_events(state, delta))
377 return Ok(tuple(events))
378 if delta.kind == "tool_call":
379 events = []
380 if not _started_pre(state, delta):
381 events.append(_created(state))
382 events.extend(_tool_events(state, delta))
383 return Ok(tuple(events))
384 if delta.kind == "finish":
385 events = []
386 if not _started_pre(state, delta):
387 events.append(_created(state))
388 events.extend(_finish_events(state, delta))
389 return Ok(tuple(events))
390 if delta.kind == "usage":
391 return Ok(())
392 if delta.kind == "status":
393 events = []
394 if not _started_pre(state, delta):
395 events.append(_created(state))
396 events.append(_in_progress(state, status=delta.status or "in_progress"))
397 return Ok(tuple(events))
398 return Err(
399 stream_state_invalid(f"unknown delta kind {delta.kind!r} for openai_responses")
400 )