diff --git a/src/plyngent/agent/responses_bridge.py b/src/plyngent/agent/responses_bridge.py index 471635a..7c130a3 100644 --- a/src/plyngent/agent/responses_bridge.py +++ b/src/plyngent/agent/responses_bridge.py @@ -1,7 +1,8 @@ """Convert agent chat history/tools to OpenAI Responses API shapes and back. Agent memory and events stay chat-completions-shaped; only the transport uses -Responses. DeepSeek / openai-compatible paths never enter this module. +Responses. OpenAI Responses and DeepSeek Responses (``convention = "responses"``) +both use this bridge; openai-compatible chat-completions paths never enter it. """ from __future__ import annotations diff --git a/src/plyngent/agent/responses_dispatch.py b/src/plyngent/agent/responses_dispatch.py index 32f66ff..2c7e241 100644 --- a/src/plyngent/agent/responses_dispatch.py +++ b/src/plyngent/agent/responses_dispatch.py @@ -1,15 +1,14 @@ """OpenAI Responses dispatch: present ``/responses`` as chat completions to the loop. Agent memory and events stay chat-completions-shaped; only the transport uses -``POST /responses``. DeepSeek / openai-compatible paths never enter this module. - -This restores the dispatch that lived on the removed ``ResponsesChatClient`` -wrapper, now as plain functions called by the kind-based switch in the loop. +``POST /responses``. OpenAI Responses and DeepSeek Responses +(``convention = "responses"``) both route through this module; openai-compatible +chat-completions paths never enter it. """ from __future__ import annotations -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, NoReturn, cast import msgspec from msgspec import UNSET @@ -27,8 +26,13 @@ from plyngent.agent.stream_chunks import ( reasoning_delta_chunk, text_delta_chunk, ) -from plyngent.lmproto.openai.model import Response as ResponseModel -from plyngent.lmproto.openai.model import ResponsesCreateParam +from plyngent.lmproto.openai.model import ( + Response as ResponseModel, +) +from plyngent.lmproto.openai.model import ( + ResponsesCreateParam, + ResponseStreamEvent, +) if TYPE_CHECKING: from collections.abc import AsyncIterator, Sequence @@ -42,6 +46,48 @@ if TYPE_CHECKING: # Responses stream event types that carry reasoning text deltas. _REASONING_DELTA_TYPES = {"response.reasoning_summary_text.delta", "response.reasoning_text.delta"} +# Terminal events that carry the full Response object (status, tool calls, usage). +# ``response.failed`` also carries a Response but is surfaced as an error. +_FINAL_EVENT_TYPES = {"response.completed", "response.incomplete"} + + +def _failed_stream_detail(event: ResponseStreamEvent) -> str: + """Human-readable detail for a ``response.failed`` stream event.""" + if event.response is not UNSET: + raw_error: object = event.response.get("error") + if isinstance(raw_error, dict): + error_map = cast("dict[str, object]", raw_error) + code = error_map.get("code") + message = error_map.get("message") + if isinstance(message, str) and message.strip(): + prefix = f"{code}: " if isinstance(code, str) and code.strip() else "" + return f"{prefix}{message.strip()}" + if isinstance(code, str) and code.strip(): + return code.strip() + if isinstance(event.message, str) and event.message.strip(): + return event.message.strip() + return "unknown failure" + + +def _raise_stream_error(event: ResponseStreamEvent) -> NoReturn: + """Raise for Responses ``error`` / ``response.failed`` stream events.""" + if event.type == "error": + code = event.code if event.code is not UNSET else "" + message = event.message if event.message is not UNSET else "" + detail = message.strip() or code + label = detail.strip() or event.type + raise RuntimeError(f"responses stream error: {label}") # noqa: TRY003 — transport failure + raise RuntimeError(f"responses stream failed: {_failed_stream_detail(event)}") # noqa: TRY003 — transport failure + + +def _decode_final_event(event: ResponseStreamEvent) -> ResponseModel | None: + """Decode the full Response carried by a terminal stream event.""" + if event.response is UNSET: + return None + try: + return msgspec.convert(event.response, ResponseModel) + except TypeError, ValueError, msgspec.ValidationError: + return None async def _stream_as_chat_chunks( @@ -53,32 +99,28 @@ async def _stream_as_chat_chunks( """Yield chat-completions chunks synthesized from a Responses SSE stream. Text / reasoning deltas stream as they arrive; tool calls, finish_reason, - and usage are emitted once on ``response.completed`` (the full Response is - needed to decode function calls + usage). A stream that ends without - ``response.completed`` yields nothing here — the loop treats that as a - missing terminal signal. + and usage are emitted once on a terminal event (``response.completed`` / + ``response.incomplete`` — the full Response is needed to decode function + calls + usage; ``response.incomplete`` maps to a ``length`` finish). + ``response.failed`` raises with the embedded error detail. A stream that + ends without a terminal event yields nothing here — the loop treats that + as a missing terminal signal. """ stream = await client.responses(create, stream=True) final: ResponseModel | None = None async for event in stream: etype = event.type - if etype == "error": - code = event.code if event.code is not UNSET else "" - message = event.message if event.message is not UNSET else "" - detail = message.strip() or code - label = detail.strip() or etype - raise RuntimeError(f"openai responses stream error: {label}") # noqa: TRY003 — transport failure + if etype in {"error", "response.failed"}: + _raise_stream_error(event) + if etype in _FINAL_EVENT_TYPES: + final = _decode_final_event(event) + continue if etype == "response.output_text.delta" and isinstance(event.delta, str) and event.delta: yield text_delta_chunk(model=model, content=event.delta) continue if etype in _REASONING_DELTA_TYPES and isinstance(event.delta, str) and event.delta: yield reasoning_delta_chunk(model=model, content=event.delta) continue - if etype == "response.completed" and event.response is not UNSET: - try: - final = msgspec.convert(event.response, ResponseModel) - except TypeError, ValueError, msgspec.ValidationError: - final = None if final is None: # Stream ended without response.completed — signal missing terminal. diff --git a/tests/test_agent/test_responses_dispatch.py b/tests/test_agent/test_responses_dispatch.py index 65edfae..0cec7d2 100644 --- a/tests/test_agent/test_responses_dispatch.py +++ b/tests/test_agent/test_responses_dispatch.py @@ -328,3 +328,57 @@ async def test_dispatch_stream_raises_on_error_event() -> None: with pytest.raises(RuntimeError, match="boom"): async for _ in cast("Any", stream): pass + + +async def test_dispatch_stream_incomplete_maps_to_length() -> None: + """response.incomplete carries the full response; finish becomes ``length``.""" + import msgspec + + final = _completed_response(text="cut short", status="incomplete") + final_dict = msgspec.to_builtins(final) + events = [ + ResponseStreamEvent(type="response.output_text.delta", delta="cut "), + ResponseStreamEvent(type="response.incomplete", response=final_dict), + ] + client = ScriptedResponsesClient(stream_events=[events]) + param = ChatCompletionsParam(model="gpt-test", messages=[UserChatMessage(content="x")]) + stream = await dispatch_responses(cast("Any", client), param, stream=True) + chunks = [chunk async for chunk in cast("Any", stream)] + assert any(c.choices and c.choices[0].finish_reason == "length" for c in chunks) + + +async def test_dispatch_stream_failed_raises_with_error_detail() -> None: + """response.failed surfaces the embedded error instead of a missing terminal.""" + events = [ + ResponseStreamEvent( + type="response.failed", + response={ + "id": "resp_1", + "object": "response", + "created_at": 1, + "model": "deepseek-v4-flash", + "status": "failed", + "output": [], + "error": {"code": "rate_limit_exceeded", "message": "slow down"}, + }, + ), + ] + client = ScriptedResponsesClient(stream_events=[events]) + param = ChatCompletionsParam(model="gpt-test", messages=[UserChatMessage(content="x")]) + stream = await dispatch_responses(cast("Any", client), param, stream=True) + with pytest.raises(RuntimeError, match="slow down"): + async for _ in cast("Any", stream): + pass + + +async def test_dispatch_stream_failed_falls_back_to_event_message() -> None: + """response.failed without an embedded error uses the event message.""" + events = [ + ResponseStreamEvent(type="response.failed", message="backend error", code="server_error"), + ] + client = ScriptedResponsesClient(stream_events=[events]) + param = ChatCompletionsParam(model="gpt-test", messages=[UserChatMessage(content="x")]) + stream = await dispatch_responses(cast("Any", client), param, stream=True) + with pytest.raises(RuntimeError, match="backend error"): + async for _ in cast("Any", stream): + pass