fix(run_agent): tolerate missing response.created on custom codex streams

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",

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"))

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"