diff --git a/README.md b/README.md index f3bff36458..a54cce2a54 100644 --- a/README.md +++ b/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: diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index 08ce52c32b..1aa0c30f95 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/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) diff --git a/backend/packages/harness/deerflow/models/openai_codex_provider.py b/backend/packages/harness/deerflow/models/openai_codex_provider.py index 856484b428..27f5591ad7 100644 --- a/backend/packages/harness/deerflow/models/openai_codex_provider.py +++ b/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") diff --git a/backend/tests/test_codex_stream_terminal_events.py b/backend/tests/test_codex_stream_terminal_events.py new file mode 100644 index 0000000000..3dd99e3e50 --- /dev/null +++ b/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