[integrations][openai] Let provider exceptions reach the caller unwrapped

integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java

@@ -289,15 +289,9 @@ private ChatMessage doChat(
             List<Tool> tools,
             Map<String, Object> modelParams,
             Object outputSchema) {
-        try {
-            ChatCompletionCreateParams params =
-                    buildRequest(messages, tools, modelParams, outputSchema);
-            return toResponse(client.chat().completions().create(params), modelParams);
-        } catch (IllegalArgumentException e) {
-            throw e;
-        } catch (Exception e) {
-            throw new RuntimeException("Failed to call Azure OpenAI chat completions API.", e);
-        }
+        ChatCompletionCreateParams params =
+                buildRequest(messages, tools, modelParams, outputSchema);
+        return toResponse(client.chat().completions().create(params), modelParams);
     }
 
     // Package-private so response handling can be asserted against a constructed completion without

integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java

@@ -191,33 +191,29 @@ private ChatMessage doChat(
             List<Tool> tools,
             Map<String, Object> modelParams,
             Object outputSchema) {
-        try {
-            ChatCompletionCreateParams params =
-                    buildRequest(messages, tools, modelParams, outputSchema);
-            ChatCompletion completion = client.chat().completions().create(params);
-            ChatMessage response =
-                    OpenAIChatCompletionsUtils.convertFromOpenAIMessage(
-                            completion.choices().get(0).message());
-
-            // Stash token usage
-            if (completion.usage().isPresent()) {
-                String modelName = modelParams != null ? (String) modelParams.get("model") : null;
-                if (modelName == null || modelName.isBlank()) {
-                    modelName = this.defaultModel;
-                }
-                if (modelName != null && !modelName.isBlank()) {
-                    response.getExtraArgs().put("model_name", modelName);
-                    response.getExtraArgs()
-                            .put("promptTokens", completion.usage().get().promptTokens());
-                    response.getExtraArgs()
-                            .put("completionTokens", completion.usage().get().completionTokens());
-                }
+        ChatCompletionCreateParams params =
+                buildRequest(messages, tools, modelParams, outputSchema);
+        ChatCompletion completion = client.chat().completions().create(params);
+        ChatMessage response =
+                OpenAIChatCompletionsUtils.convertFromOpenAIMessage(
+                        completion.choices().get(0).message());
+
+        // Stash token usage
+        if (completion.usage().isPresent()) {
+            String modelName = modelParams != null ? (String) modelParams.get("model") : null;
+            if (modelName == null || modelName.isBlank()) {
+                modelName = this.defaultModel;
+            }
+            if (modelName != null && !modelName.isBlank()) {
+                response.getExtraArgs().put("model_name", modelName);
+                response.getExtraArgs()
+                        .put("promptTokens", completion.usage().get().promptTokens());
+                response.getExtraArgs()
+                        .put("completionTokens", completion.usage().get().completionTokens());
             }
-
-            return response;
-        } catch (Exception e) {
-            throw new RuntimeException("Failed to call OpenAI chat completions API.", e);
         }
+
+        return response;
     }
 
     // Package-private so the request body (including the native response_format) can be asserted

integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnection.java

@@ -129,28 +129,24 @@ public ChatMessage chat(
             List<ChatMessage> messages,
             List<org.apache.flink.agents.api.tools.Tool> tools,
             Map<String, Object> modelParams) {
-        try {
-            ResponseCreateParams params = buildRequest(messages, tools, modelParams);
-            Response response = client.responses().create(params);
-            ChatMessage result = convertResponse(response);
-
-            if (response.usage().isPresent()) {
-                String modelName = modelParams != null ? (String) modelParams.get("model") : null;
-                if (modelName == null || modelName.isBlank()) {
-                    modelName = this.defaultModel;
-                }
-                if (modelName != null && !modelName.isBlank()) {
-                    result.getExtraArgs().put("model_name", modelName);
-                    result.getExtraArgs().put("promptTokens", response.usage().get().inputTokens());
-                    result.getExtraArgs()
-                            .put("completionTokens", response.usage().get().outputTokens());
-                }
+        ResponseCreateParams params = buildRequest(messages, tools, modelParams);
+        Response response = client.responses().create(params);
+        ChatMessage result = convertResponse(response);
+
+        if (response.usage().isPresent()) {
+            String modelName = modelParams != null ? (String) modelParams.get("model") : null;
+            if (modelName == null || modelName.isBlank()) {
+                modelName = this.defaultModel;
+            }
+            if (modelName != null && !modelName.isBlank()) {
+                result.getExtraArgs().put("model_name", modelName);
+                result.getExtraArgs().put("promptTokens", response.usage().get().inputTokens());
+                result.getExtraArgs()
+                        .put("completionTokens", response.usage().get().outputTokens());
             }
-
-            return result;
-        } catch (Exception e) {
-            throw new RuntimeException("Failed to call OpenAI Responses API.", e);
         }
+
+        return result;
     }
 
     private ResponseCreateParams buildRequest(

integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java

@@ -19,6 +19,7 @@
 package org.apache.flink.agents.integrations.chatmodels.openai;
 
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.openai.errors.BadRequestException;
 import com.openai.models.ChatModel;
 import com.openai.models.ResponseFormatJsonSchema;
 import com.openai.models.chat.completions.ChatCompletion;
@@ -43,6 +44,7 @@
 import org.junit.jupiter.params.provider.NullAndEmptySource;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.io.IOException;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
@@ -96,6 +98,18 @@ private static AzureOpenAIChatModelConnection connection() {
         return connection(CAPABLE_API_VERSION);
     }
 
+    private static AzureOpenAIChatModelConnection connection(
+            String apiVersion, String azureEndpoint, String azureUrlPathMode) {
+        ResourceDescriptor desc =
+                connectionDescriptor()
+                        .addInitialArgument("api_key", "test-key")
+                        .addInitialArgument("api_version", apiVersion)
+                        .addInitialArgument("azure_endpoint", azureEndpoint)
+                        .addInitialArgument("azure_url_path_mode", azureUrlPathMode)
+                        .build();
+        return new AzureOpenAIChatModelConnection(desc, NOOP);
+    }
+
     @Test
     void testConnectionArgumentDefaultsAndZeroTimeout() {
         AzureOpenAIChatModelConnection connection =
@@ -222,6 +236,26 @@ void testChatRejectsReservedKeyInAdditionalKwargs() {
                 .hasMessageContaining("temperature");
     }
 
+    @Test
+    @DisplayName("A provider error reaches the caller as the SDK exception carrying its payload")
+    void testProviderErrorPropagatesUnwrapped() throws IOException {
+        try (FakeOpenAIErrorEndpoint endpoint = FakeOpenAIErrorEndpoint.rejectingWith400()) {
+            // A loopback endpoint is a custom gateway rather than an *.openai.azure.com resource,
+            // so LEGACY is what builds the deployment-scoped Azure request path against it.
+            AzureOpenAIChatModelConnection connection =
+                    connection(CAPABLE_API_VERSION, endpoint.baseUrl(), "LEGACY");
+
+            assertThatThrownBy(() -> connection.chat(userMessage(), List.of(), params(null), null))
+                    .isInstanceOfSatisfying(
+                            BadRequestException.class,
+                            e -> {
+                                assertThat(e.statusCode()).isEqualTo(400);
+                                assertThat(e.code()).contains(FakeOpenAIErrorEndpoint.ERROR_CODE);
+                            })
+                    .hasMessageContaining(FakeOpenAIErrorEndpoint.ERROR_MESSAGE);
+        }
+    }
+
     @Test
     @DisplayName("Native response_format json_schema strict applied for a POJO on a capable model")
     void testNativeAppliedForCapableDeploymentModel() {

integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/FakeOpenAIErrorEndpoint.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.integrations.chatmodels.openai;
+
+import com.sun.net.httpserver.HttpServer;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * A loopback HTTP endpoint that answers every request with the provider error envelope OpenAI
+ * returns for a rejected request, so a connection's error path can be exercised without a live API
+ * call. A 400 is chosen because the SDK does not retry it, which keeps the exchange to a single
+ * request.
+ */
+final class FakeOpenAIErrorEndpoint implements AutoCloseable {
+
+    static final String ERROR_MESSAGE = "The requested model does not exist.";
+    static final String ERROR_CODE = "model_not_found";
+
+    private static final String ERROR_BODY =
+            "{\"error\":{\"message\":\""
+                    + ERROR_MESSAGE
+                    + "\",\"type\":\"invalid_request_error\","
+                    + "\"param\":\"model\",\"code\":\""
+                    + ERROR_CODE
+                    + "\"}}";
+
+    private final HttpServer server;
+
+    private FakeOpenAIErrorEndpoint(HttpServer server) {
+        this.server = server;
+    }
+
+    static FakeOpenAIErrorEndpoint rejectingWith400() throws IOException {
+        HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        byte[] body = ERROR_BODY.getBytes(StandardCharsets.UTF_8);
+        server.createContext(
+                "/",
+                exchange -> {
+                    exchange.getResponseHeaders().add("Content-Type", "application/json");
+                    exchange.sendResponseHeaders(400, body.length);
+                    exchange.getResponseBody().write(body);
+                    exchange.close();
+                });
+        server.setExecutor(null);
+        server.start();
+        return new FakeOpenAIErrorEndpoint(server);
+    }
+
+    String baseUrl() {
+        return "http://127.0.0.1:" + server.getAddress().getPort();
+    }
+
+    @Override
+    public void close() {
+        server.stop(0);
+    }
+}

integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.agents.integrations.chatmodels.openai;
 
+import com.openai.errors.BadRequestException;
 import com.openai.models.ResponseFormatJsonSchema;
 import com.openai.models.chat.completions.ChatCompletionCreateParams;
 import org.apache.flink.agents.api.chat.messages.ChatMessage;
@@ -33,6 +34,7 @@
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
@@ -65,6 +67,16 @@ private static OpenAICompletionsConnection connection() {
         return new OpenAICompletionsConnection(desc, NOOP);
     }
 
+    private static OpenAICompletionsConnection connection(String apiBaseUrl) {
+        ResourceDescriptor desc =
+                ResourceDescriptor.Builder.newBuilder(OpenAICompletionsConnection.class.getName())
+                        .addInitialArgument("api_key", "test-key")
+                        .addInitialArgument("api_base_url", apiBaseUrl)
+                        .addInitialArgument("model", "gpt-4o")
+                        .build();
+        return new OpenAICompletionsConnection(desc, NOOP);
+    }
+
     private static Map<String, Object> params(String model) {
         Map<String, Object> params = new HashMap<>();
         params.put("model", model);
@@ -99,6 +111,42 @@ void testConnectionArgumentValidation() {
         OpenAIClientTestUtils.assertNoTimeoutConfigured(connection);
     }
 
+    @Test
+    @DisplayName("A request-building failure reaches the caller as its own type, not a wrapper")
+    void testRequestBuildingFailurePropagatesUnwrapped() {
+        List<ChatMessage> toolMessageWithoutExternalId =
+                List.of(new ChatMessage(MessageRole.TOOL, "result", Map.of()));
+
+        assertThatThrownBy(
+                        () ->
+                                connection()
+                                        .chat(
+                                                toolMessageWithoutExternalId,
+                                                List.of(),
+                                                params("gpt-4o"),
+                                                null))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("externalId");
+    }
+
+    @Test
+    @DisplayName("A provider error reaches the caller as the SDK exception carrying its payload")
+    void testProviderErrorPropagatesUnwrapped() throws IOException {
+        try (FakeOpenAIErrorEndpoint endpoint = FakeOpenAIErrorEndpoint.rejectingWith400()) {
+            OpenAICompletionsConnection connection = connection(endpoint.baseUrl());
+
+            assertThatThrownBy(
+                            () -> connection.chat(userMessage(), List.of(), params("gpt-4o"), null))
+                    .isInstanceOfSatisfying(
+                            BadRequestException.class,
+                            e -> {
+                                assertThat(e.statusCode()).isEqualTo(400);
+                                assertThat(e.code()).contains(FakeOpenAIErrorEndpoint.ERROR_CODE);
+                            })
+                    .hasMessageContaining(FakeOpenAIErrorEndpoint.ERROR_MESSAGE);
+        }
+    }
+
     @Test
     @DisplayName("Native response_format json_schema strict applied for a POJO on a capable model")
     void testNativeAppliedForPojoCapableModel() {