core/agent: handle incomplete/failed Responses stream terminals

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

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.
+``POST /responses``. OpenAI Responses and DeepSeek Responses
-
+(``convention = "responses"``) both route through this module; openai-compatible
-This restores the dispatch that lived on the removed ``ResponsesChatClient``
+chat-completions paths never enter it.
-wrapper, now as plain functions called by the kind-based switch in the loop.
 """
 
 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 (
-from plyngent.lmproto.openai.model import ResponsesCreateParam
+    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
+    and usage are emitted once on a terminal event (``response.completed`` /
-    needed to decode function calls + usage). A stream that ends without
+    ``response.incomplete`` — the full Response is needed to decode function
-    ``response.completed`` yields nothing here — the loop treats that as a
+    calls + usage; ``response.incomplete`` maps to a ``length`` finish).
-    missing terminal signal.
+    ``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":
+        if etype in {"error", "response.failed"}:
-            code = event.code if event.code is not UNSET else ""
+            _raise_stream_error(event)
-            message = event.message if event.message is not UNSET else ""
+        if etype in _FINAL_EVENT_TYPES:
-            detail = message.strip() or code
+            final = _decode_final_event(event)
-            label = detail.strip() or etype
+            continue
-            raise RuntimeError(f"openai responses stream error: {label}")  # noqa: TRY003 — transport failure
         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.

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