From f3330461521a982943e7f4bec03e969b78ae0845 Mon Sep 17 00:00:00 2001 From: ZhangYiqun018 <875977494@qq.com> Date: Sun, 12 Apr 2026 11:45:21 +0800 Subject: [PATCH] Handle missing response.created on custom codex streams --- run_agent.py | 75 +++++++++++++++- .../test_run_agent_codex_responses.py | 88 +++++++++++++++++++ tests/run_agent/test_streaming.py | 53 +++++++++++ 3 files changed, 212 insertions(+), 4 deletions(-) diff --git a/run_agent.py b/run_agent.py index b23035454252..201598d90c46 100644 --- a/run_agent.py +++ b/run_agent.py @@ -4224,6 +4224,25 @@ def _create_request_openai_client(self, *, reason: str) -> Any: def _close_request_openai_client(self, client: Any, *, reason: str) -> None: self._close_openai_client(client, reason=reason, shared=False) + def _is_custom_codex_provider(self) -> bool: + provider_name = (self.provider or "").strip().lower() + return provider_name == "custom" or provider_name.startswith("custom:") + + def _should_fallback_custom_codex_missing_created(self, exc: Exception) -> bool: + """Detect custom Responses streams that skip the initial response.created event. + + Some third-party Responses-compatible relays stream usable SSE bytes but + violate the SDK's event ordering contract by starting with + response.in_progress (or another response.* event) before + response.created. curl can display those bytes just fine, but the SDK's + strict state machine raises before Hermes can consume the stream. + """ + if not self._is_custom_codex_provider(): + return False + + err_text = str(exc or "") + return "Expected to have received response.created before response." in err_text + def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta: callable = None): """Execute one streaming Responses API request and return the final response.""" import httpx as _httpx @@ -4326,10 +4345,14 @@ def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta self._client_log_context(), exc, ) - return self._run_codex_create_stream_fallback(api_kwargs, client=active_client) + fallback_kwargs = {"client": active_client} + if on_first_delta is not None: + fallback_kwargs["on_first_delta"] = on_first_delta + return self._run_codex_create_stream_fallback(api_kwargs, **fallback_kwargs) except RuntimeError as exc: err_text = str(exc) missing_completed = "response.completed" in err_text + missing_created = self._should_fallback_custom_codex_missing_created(exc) if missing_completed and attempt < max_stream_retries: logger.debug( "Responses stream closed before completion (attempt %s/%s); retrying. %s", @@ -4343,10 +4366,35 @@ def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta "Responses stream did not emit response.completed; falling back to create(stream=True). %s", self._client_log_context(), ) - return self._run_codex_create_stream_fallback(api_kwargs, client=active_client) + fallback_kwargs = {"client": active_client} + if on_first_delta is not None: + fallback_kwargs["on_first_delta"] = on_first_delta + return self._run_codex_create_stream_fallback(api_kwargs, **fallback_kwargs) + if missing_created and attempt < max_stream_retries: + logger.debug( + "Custom Responses stream skipped response.created (attempt %s/%s); retrying. %s", + attempt + 1, + max_stream_retries + 1, + self._client_log_context(), + ) + continue + if missing_created: + logger.debug( + "Custom Responses stream skipped response.created; falling back to create(stream=True). %s", + self._client_log_context(), + ) + fallback_kwargs = {"client": active_client} + if on_first_delta is not None: + fallback_kwargs["on_first_delta"] = on_first_delta + return self._run_codex_create_stream_fallback(api_kwargs, **fallback_kwargs) raise - def _run_codex_create_stream_fallback(self, api_kwargs: dict, client: Any = None): + def _run_codex_create_stream_fallback( + self, + api_kwargs: dict, + client: Any = None, + on_first_delta: callable = None, + ): """Fallback path for stream completion edge cases on Codex-style Responses backends.""" active_client = client or self._ensure_primary_openai_client(reason="codex_create_stream_fallback") fallback_kwargs = dict(api_kwargs) @@ -4363,6 +4411,8 @@ def _run_codex_create_stream_fallback(self, api_kwargs: dict, client: Any = None terminal_response = None collected_output_items: list = [] collected_text_deltas: list = [] + has_function_calls = False + first_delta_fired = False try: for event in stream_or_response: event_type = getattr(event, "type", None) @@ -4382,6 +4432,23 @@ def _run_codex_create_stream_fallback(self, api_kwargs: dict, client: Any = None delta = event.get("delta", "") if delta: collected_text_deltas.append(delta) + if not has_function_calls: + if not first_delta_fired: + first_delta_fired = True + if on_first_delta: + try: + on_first_delta() + except Exception: + pass + self._fire_stream_delta(delta) + elif "function_call" in str(event_type): + has_function_calls = True + elif "reasoning" in str(event_type) and "delta" in str(event_type): + reasoning_text = getattr(event, "delta", "") + if not reasoning_text and isinstance(event, dict): + reasoning_text = event.get("delta", "") + if reasoning_text: + self._fire_reasoning_delta(reasoning_text) if event_type not in {"response.completed", "response.incomplete", "response.failed"}: continue @@ -4399,7 +4466,7 @@ def _run_codex_create_stream_fallback(self, api_kwargs: dict, client: Any = None "Codex fallback stream: backfilled %d output items", len(collected_output_items), ) - elif collected_text_deltas: + elif collected_text_deltas and not has_function_calls: assembled = "".join(collected_text_deltas) terminal_response.output = [SimpleNamespace( type="message", role="assistant", diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index 533a85ac8351..5ad046263d12 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/tests/run_agent/test_run_agent_codex_responses.py @@ -70,6 +70,27 @@ def _build_copilot_agent(monkeypatch, *, model="gpt-5.4"): return agent +def _build_custom_agent(monkeypatch, *, model="gpt-5.4"): + _patch_agent_bootstrap(monkeypatch) + + agent = run_agent.AIAgent( + model=model, + provider="custom", + api_mode="codex_responses", + base_url="https://relay.example.com/openai", + api_key="relay-token", + quiet_mode=True, + max_iterations=4, + skip_context_files=True, + skip_memory=True, + ) + agent._cleanup_task_resources = lambda task_id: None + agent._persist_session = lambda messages, history=None: None + agent._save_trajectory = lambda messages, user_message, completed: None + agent._save_session_log = lambda messages: None + return agent + + def _codex_message_response(text: str): return SimpleNamespace( output=[ @@ -189,6 +210,12 @@ def _codex_request_kwargs(): } +def _missing_created_error(): + return RuntimeError( + "Expected to have received response.created before response.in_progress" + ) + + def test_api_mode_uses_explicit_provider_when_codex(monkeypatch): _patch_agent_bootstrap(monkeypatch) agent = run_agent.AIAgent( @@ -380,6 +407,67 @@ def _fake_create(**kwargs): assert response.output[0].content[0].text == "streamed create ok" +def test_run_codex_stream_custom_missing_created_retries_then_falls_back(monkeypatch): + agent = _build_custom_agent(monkeypatch) + calls = {"stream": 0, "create": 0} + create_stream = _FakeCreateStream( + [ + SimpleNamespace(type="response.in_progress"), + SimpleNamespace(type="response.output_text.delta", delta="custom "), + SimpleNamespace(type="response.output_text.delta", delta="fallback"), + SimpleNamespace(type="response.completed", response=_codex_message_response("custom fallback")), + ] + ) + + def _fake_stream(**kwargs): + calls["stream"] += 1 + return _FakeResponsesStream(final_error=_missing_created_error()) + + def _fake_create(**kwargs): + calls["create"] += 1 + assert kwargs.get("stream") is True + return create_stream + + agent.client = SimpleNamespace( + responses=SimpleNamespace( + stream=_fake_stream, + create=_fake_create, + ) + ) + + response = agent._run_codex_stream(_codex_request_kwargs()) + assert calls["stream"] == 2 + assert calls["create"] == 1 + assert create_stream.closed is True + assert response.output[0].content[0].text == "custom fallback" + + +def test_run_codex_stream_non_custom_missing_created_raises(monkeypatch): + agent = _build_agent(monkeypatch) + calls = {"stream": 0, "create": 0} + + def _fake_stream(**kwargs): + calls["stream"] += 1 + return _FakeResponsesStream(final_error=_missing_created_error()) + + def _fake_create(**kwargs): + calls["create"] += 1 + return _codex_message_response("should not be used") + + agent.client = SimpleNamespace( + responses=SimpleNamespace( + stream=_fake_stream, + create=_fake_create, + ) + ) + + with pytest.raises(RuntimeError, match="response.created"): + agent._run_codex_stream(_codex_request_kwargs()) + + assert calls["stream"] == 1 + assert calls["create"] == 0 + + def test_run_conversation_codex_plain_text(monkeypatch): agent = _build_agent(monkeypatch) monkeypatch.setattr(agent, "_interruptible_api_call", lambda api_kwargs: _codex_message_response("OK")) diff --git a/tests/run_agent/test_streaming.py b/tests/run_agent/test_streaming.py index 37a61ac37084..65c959412420 100644 --- a/tests/run_agent/test_streaming.py +++ b/tests/run_agent/test_streaming.py @@ -814,3 +814,56 @@ def test_codex_remote_protocol_error_falls_back_to_create_stream(self): assert response is fallback_response mock_fallback.assert_called_once_with({}, client=mock_client) + + def test_codex_create_stream_fallback_emits_deltas_without_response_created(self): + from run_agent import AIAgent + + deltas = [] + + class _CreateStream: + def __init__(self, events): + self._events = list(events) + self.closed = False + + def __iter__(self): + return iter(self._events) + + def close(self): + self.closed = True + + terminal_response = SimpleNamespace(output=[], status="completed") + create_stream = _CreateStream( + [ + SimpleNamespace(type="response.in_progress"), + SimpleNamespace(type="response.output_text.delta", delta="Hello "), + SimpleNamespace(type="response.output_text.delta", delta="fallback"), + SimpleNamespace(type="response.completed", response=terminal_response), + ] + ) + + mock_client = MagicMock() + mock_client.responses.create.return_value = create_stream + + agent = AIAgent( + model="test/model", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + stream_delta_callback=lambda t: deltas.append(t), + ) + agent.api_mode = "codex_responses" + agent.provider = "custom" + agent._interrupt_requested = False + + response = agent._run_codex_create_stream_fallback( + { + "model": "gpt-5.4", + "instructions": "You are Hermes.", + "input": [{"role": "user", "content": "Ping"}], + }, + client=mock_client, + ) + + assert create_stream.closed is True + assert "".join(deltas) == "Hello fallback" + assert response.output[0].content[0].text == "Hello fallback"