fix(models): honor Codex SSE terminal events (#6155)

* fix(models): honor Codex SSE terminal events

* fix(models): preserve malformed Codex error diagnostics

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -284,6 +284,7 @@ It is disabled by default; see the linked guide to enable it.
    ```
 
    - Codex CLI reads `~/.codex/auth.json`
+   - The Codex model provider returns completed responses without waiting for the SSE connection to close. Failed or incomplete responses report the provider's error or reason; partial output is not returned as a successful answer. Non-object error details are reported as text.
    - Claude Code accepts `CLAUDE_CODE_OAUTH_TOKEN`, `ANTHROPIC_AUTH_TOKEN`, `CLAUDE_CODE_CREDENTIALS_PATH`, or `~/.claude/.credentials.json`
    - ACP agent entries are separate from model providers — if you configure `acp_agents.codex`, point it at a Codex ACP adapter such as `npx -y @zed-industries/codex-acp`
    - MiniMax Code speaks ACP directly. Install and authenticate it, then add it as an ACP agent:

backend/packages/harness/deerflow/models/AGENTS.md

@@ -58,6 +58,17 @@ the same policy first so run metadata reports the effective values. Design note:
 `tests/test_reasoning_contract.py`, the contract section of
 `tests/test_model_factory.py`, `tests/test_models_router_reasoning.py`.
 
+### Codex SSE termination (`packages/harness/deerflow/models/openai_codex_provider.py`)
+
+`response.completed` ends stream consumption immediately, before transport EOF;
+retain the output-item recovery path for empty completed output. Terminal
+`response.failed`, `response.incomplete`, and `error` events raise with their
+error code/message or incomplete reason, closing the response and client without
+returning partial output. Non-object error details or response containers are
+reported as text instead of raising `AttributeError`. SSE failures do not enter
+the HTTP-status retry loop.
+Offline HTTP-stream coverage: `tests/test_codex_stream_terminal_events.py`.
+
 ### Claude Code Credentials (`packages/harness/deerflow/models/credential_loader.py`)
 
 - `ClaudeChatModel.model_post_init` calls `load_claude_code_credential()` for every instance, and `create_chat_model` builds fresh instances per run (lead agent, title, summarization, subagents)

backend/packages/harness/deerflow/models/openai_codex_provider.py

@@ -283,6 +283,27 @@ class CodexChatModel(BaseChatModel):
                             streamed_output_items[output_index] = output_item
                     elif event_type == "response.completed":
                         completed_response = data["response"]
+                        # A terminal event completes the request even if the server
+                        # keeps the connection open. Do not let a later read timeout
+                        # replace the completed output with a transport error.
+                        break
+                    elif event_type in ("response.failed", "response.incomplete", "error"):
+                        response = data.get("response") or {}
+                        if event_type == "error":
+                            details = data.get("error") or data
+                        elif not isinstance(response, dict):
+                            details = response
+                        elif event_type == "response.failed":
+                            details = response.get("error") or {}
+                        else:
+                            details = response.get("incomplete_details") or {}
+
+                        if not isinstance(details, dict):
+                            details = {"message": str(details)}
+                        code = details.get("code")
+                        reason = details.get("message") or details.get("reason") or "No details provided"
+                        code_suffix = f" ({code})" if code else ""
+                        raise RuntimeError(f"Codex API {event_type}{code_suffix}: {reason}")
 
         if not completed_response:
             raise RuntimeError("Codex API stream ended without response.completed event")

backend/tests/test_codex_stream_terminal_events.py

@@ -0,0 +1,213 @@
+"""Offline HTTP-stream regressions for Codex terminal response events."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterator
+
+import httpx
+import pytest
+from langchain_core.messages import HumanMessage
+
+from deerflow.models import openai_codex_provider as provider
+from deerflow.models.credential_loader import CodexCliCredential
+
+
+class EventStream(httpx.SyncByteStream):
+    def __init__(self, events: list[dict], tail_error: Exception | None = None):
+        self.events = events
+        self.tail_error = tail_error
+        self.closed = False
+        self.read_past_events = False
+
+    def __iter__(self) -> Iterator[bytes]:
+        for event in self.events:
+            yield f"data: {json.dumps(event)}\n\n".encode()
+        self.read_past_events = True
+        if self.tail_error is not None:
+            raise self.tail_error
+
+    def close(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture
+def model(monkeypatch):
+    monkeypatch.setattr(provider, "load_codex_cli_credential", lambda: CodexCliCredential("fake-token", "fake-account"))
+    return provider.CodexChatModel(retry_max_attempts=3)
+
+
+@pytest.fixture
+def serve_events(monkeypatch):
+    client_class = httpx.Client
+
+    def install(events: list[dict], *, tail_error: Exception | None = None, status_code: int = 200):
+        stream = EventStream(events, tail_error)
+        requests = []
+
+        def handle(request: httpx.Request) -> httpx.Response:
+            requests.append(request)
+            return httpx.Response(status_code, headers={"content-type": "text/event-stream"}, stream=stream)
+
+        transport = httpx.MockTransport(handle)
+        monkeypatch.setattr(provider.httpx, "Client", lambda **kwargs: client_class(transport=transport, **kwargs))
+        return stream, requests
+
+    return install
+
+
+def message_item(text: str) -> dict:
+    return {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]}
+
+
+@pytest.mark.parametrize("recover_streamed_output", [False, True])
+def test_completed_response_does_not_wait_for_transport_eof(model, serve_events, recover_streamed_output):
+    item = message_item("Report ready")
+    response = {
+        "id": "resp-complete",
+        "status": "completed",
+        "output": [] if recover_streamed_output else [item],
+        "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5},
+    }
+    events = [
+        {"type": "response.output_item.done", "output_index": 0, "item": item},
+        {"type": "response.completed", "response": response},
+    ]
+    stream, requests = serve_events(events, tail_error=httpx.ReadTimeout("Connection stayed open after completion"))
+
+    result = model.invoke([HumanMessage(content="Prepare a report")])
+
+    assert result.content == "Report ready"
+    assert result.usage_metadata == {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5}
+    assert len(requests) == 1
+    assert stream.closed
+    assert not stream.read_past_events
+
+
+@pytest.mark.parametrize(
+    ("event", "diagnostics"),
+    [
+        ({"type": "error", "code": "invalid_request", "message": "Unsupported model"}, ["error", "invalid_request", "Unsupported model"]),
+        (
+            {"type": "error", "error": {"code": "flex_unavailable", "message": "Service tier unavailable"}},
+            ["error", "flex_unavailable", "Service tier unavailable"],
+        ),
+        (
+            {"type": "response.failed", "response": {"status": "failed", "error": {"code": "server_error", "message": "Generation failed"}}},
+            ["response.failed", "server_error", "Generation failed"],
+        ),
+        (
+            {"type": "response.incomplete", "response": {"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}}},
+            ["response.incomplete", "max_output_tokens"],
+        ),
+    ],
+)
+def test_terminal_failure_keeps_diagnostics_and_closes_without_retry(model, serve_events, monkeypatch, event, diagnostics):
+    events = [{"type": "response.output_item.done", "output_index": 0, "item": message_item("Partial answer")}, event]
+    stream, requests = serve_events(events, tail_error=httpx.ReadTimeout("Connection stayed open after failure"))
+    monkeypatch.setattr(provider.time, "sleep", lambda _: pytest.fail("SSE terminal errors must not trigger an HTTP retry"))
+
+    with pytest.raises(RuntimeError) as caught:
+        model.invoke([HumanMessage(content="Prepare a report")])
+
+    for diagnostic in diagnostics:
+        assert diagnostic in str(caught.value)
+    assert "Partial answer" not in str(caught.value)
+    assert len(requests) == 1
+    assert stream.closed
+    assert not stream.read_past_events
+
+
+@pytest.mark.parametrize("details", ["rate limited", ["rate limited"], 429])
+@pytest.mark.parametrize(
+    ("event_type", "detail_key"),
+    [("error", "error"), ("response.failed", "error"), ("response.incomplete", "incomplete_details")],
+)
+def test_terminal_failure_preserves_nonobject_details(model, serve_events, monkeypatch, event_type, detail_key, details):
+    event = {"type": event_type}
+    if event_type == "error":
+        event[detail_key] = details
+    else:
+        event["response"] = {detail_key: details}
+    stream, requests = serve_events([event], tail_error=httpx.ReadTimeout("Connection stayed open after failure"))
+    monkeypatch.setattr(provider.time, "sleep", lambda _: pytest.fail("SSE terminal errors must not trigger an HTTP retry"))
+
+    with pytest.raises(RuntimeError) as caught:
+        model.invoke([HumanMessage(content="Hello")])
+
+    assert str(caught.value) == f"Codex API {event_type}: {details}"
+    assert len(requests) == 1
+    assert stream.closed
+    assert not stream.read_past_events
+
+
+@pytest.mark.parametrize("event_type", ["response.failed", "response.incomplete"])
+def test_terminal_failure_preserves_nonobject_response(model, serve_events, event_type):
+    stream, requests = serve_events([{"type": event_type, "response": "rate limited"}])
+
+    with pytest.raises(RuntimeError) as caught:
+        model.invoke([HumanMessage(content="Hello")])
+
+    assert str(caught.value) == f"Codex API {event_type}: rate limited"
+    assert len(requests) == 1
+    assert stream.closed
+    assert not stream.read_past_events
+
+
+@pytest.mark.parametrize(
+    "event",
+    [
+        {"type": "error", "code": None, "message": "Request rejected"},
+        {"type": "response.failed", "response": {"error": None}},
+        {"type": "response.incomplete", "response": {"incomplete_details": None}},
+    ],
+)
+def test_terminal_failure_with_unavailable_details_still_names_event(model, serve_events, event):
+    stream, _ = serve_events([event])
+
+    with pytest.raises(RuntimeError, match=event["type"].replace(".", r"\.")):
+        model.invoke([HumanMessage(content="Hello")])
+
+    assert stream.closed
+    assert not stream.read_past_events
+
+
+def test_transport_eof_without_terminal_event_remains_an_error(model, serve_events):
+    stream, _ = serve_events([{"type": "response.output_item.done", "output_index": 0, "item": message_item("Partial")}])
+
+    with pytest.raises(RuntimeError, match="without response.completed"):
+        model.invoke([HumanMessage(content="Hello")])
+
+    assert stream.closed
+    assert stream.read_past_events
+
+
+def test_http_status_failure_is_not_replaced_by_sse_error(model, serve_events):
+    stream, requests = serve_events([], status_code=400)
+
+    with pytest.raises(httpx.HTTPStatusError) as caught:
+        model.invoke([HumanMessage(content="Hello")])
+
+    assert caught.value.response.status_code == 400
+    assert len(requests) == 1
+    assert stream.closed
+
+
+def test_completed_response_recovers_tool_calls_and_reasoning_before_closing(model, serve_events):
+    tool_call = {"type": "function_call", "name": "save_report", "call_id": "call-report", "arguments": '{"name": "report.md"}'}
+    reasoning = {"type": "reasoning", "summary": [{"type": "summary_text", "text": "Save the completed report"}]}
+    stream, _ = serve_events(
+        [
+            {"type": "response.output_item.done", "output_index": 1, "item": tool_call},
+            {"type": "response.output_item.done", "output_index": 0, "item": reasoning},
+            {"type": "response.completed", "response": {"status": "completed", "output": [], "usage": {}}},
+        ],
+        tail_error=httpx.ReadTimeout("Connection stayed open after tool-call completion"),
+    )
+
+    result = model.invoke([HumanMessage(content="Save the report")])
+
+    assert result.tool_calls == [{"type": "tool_call", "name": "save_report", "id": "call-report", "args": {"name": "report.md"}}]
+    assert result.additional_kwargs["reasoning_content"] == "Save the completed report"
+    assert stream.closed
+    assert not stream.read_past_events