[pull] main from apache:main

api/src/main/java/org/apache/flink/agents/api/chat/model/BaseChatModelConnection.java

@@ -52,21 +52,53 @@ public ResourceType getResourceType() {
      * model.
      *
      * <p>Capability is <b>model-dependent</b>, not connection-wide: a single provider connection
-     * commonly serves both models that accept a native schema parameter and models that do not. It
-     * is therefore evaluated against the <i>effective</i> model at request-build time — the model
-     * actually being called, which per-request parameters may override.
+     * commonly serves both models that accept a native schema parameter and models that do not. The
+     * model to ask about is whatever {@link #effectiveModelFor(Map)} returns for the parameters a
+     * request would be built from, and a caller outside the connection asks that hook and nothing
+     * else. Such a caller must not substitute the identifier the request is issued against: on a
+     * deployment-based provider the request targets a deployment name the user chose while
+     * capability belongs to the model backing it, so the two disagree in both directions.
      *
-     * <p>The default {@code false} keeps a connection on the prompt-engineering fallback. An
-     * unrecognized model must report {@code false} so that it degrades to the fallback rather than
-     * failing at the provider.
+     * <p>The default {@code false} keeps a connection on the prompt-engineering fallback. A
+     * connection that classifies by model name must report {@code false} for a name it does not
+     * recognize, so that it degrades to the fallback rather than failing at the provider. A
+     * connection whose capability belongs to the endpoint rather than to the model answers for the
+     * endpoint instead, and may report {@code true} for a name it has never seen.
      *
-     * @param effectiveModel the model the request will be issued against, may be null
+     * @param effectiveModel the model whose capability is being asked about, as returned by {@link
+     *     #effectiveModelFor(Map)}, may be null
      * @return true if a schema can be applied natively for {@code effectiveModel}
      */
     protected boolean supportsNativeStructuredOutput(String effectiveModel) {
         return false;
     }
 
+    /**
+     * The model whose capability {@link #supportsNativeStructuredOutput(String)} should be asked
+     * about, derived from the parameters a request would be built from.
+     *
+     * <p>Overriding this is how a connection whose effective model is not the {@code model}
+     * parameter verbatim keeps the capability answer and the request in agreement. A connection
+     * that falls back to a configured default model when the parameter is absent applies that
+     * fallback here, and a deployment-based provider returns the model backing the deployment
+     * rather than the deployment the request targets.
+     *
+     * <p>Answers for whatever it is given rather than validating: a model that does not resolve
+     * comes back null, and {@link #supportsNativeStructuredOutput(String)} must accept null rather
+     * than throwing. What a null model resolves to is that predicate's own answer; the default, and
+     * every override that classifies by name, reports it not capable.
+     *
+     * <p>An override must read {@code modelParams} without consuming it, so that the same map still
+     * builds the request the answer was about.
+     *
+     * @param modelParams the parameters a request would be built from, may be null
+     * @return the model to ask the capability predicate about, or null if none resolves
+     */
+    @Nullable
+    protected String effectiveModelFor(@Nullable Map<String, Object> modelParams) {
+        return modelParams == null ? null : (String) modelParams.get("model");
+    }
+
     /**
      * Process a chat request and return a chat response.
      *

api/src/test/java/org/apache/flink/agents/api/chat/model/BaseChatModelTest.java

@@ -460,4 +460,32 @@ void testChatWithLongInput() {
         assertNotNull(response);
         assertTrue(response.getContent().length() > 0);
     }
+
+    @Test
+    @DisplayName("Effective model defaults to the model parameter a request would be built from")
+    void testEffectiveModelForReadsTheModelParameter() {
+        RecordingConnection connection = new RecordingConnection();
+        Map<String, Object> modelParams = new HashMap<>();
+        modelParams.put("model", "gpt-4o");
+
+        assertEquals("gpt-4o", connection.effectiveModelFor(modelParams));
+    }
+
+    @Test
+    @DisplayName("Effective model is null when the parameters name no model")
+    void testEffectiveModelForReturnsNullWhenModelParameterAbsent() {
+        RecordingConnection connection = new RecordingConnection();
+
+        // A connection carrying no default of its own has no model to resolve, and the capability
+        // predicate reports a null model not capable rather than throwing.
+        assertNull(connection.effectiveModelFor(Map.of("temperature", 0.5)));
+    }
+
+    @Test
+    @DisplayName("Effective model is null for null parameters")
+    void testEffectiveModelForReturnsNullForNullParameters() {
+        RecordingConnection connection = new RecordingConnection();
+
+        assertNull(connection.effectiveModelFor(null));
+    }
 }

integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java

@@ -182,6 +182,20 @@ protected boolean supportsNativeStructuredOutput(String effectiveModel) {
                                                 || effectiveModel.startsWith(prefix + "-"));
     }
 
+    /**
+     * The {@code model} parameter, falling back to the model configured on the connection when the
+     * call names none, which is how the request itself resolves the model it is issued against.
+     */
+    @Override
+    protected String effectiveModelFor(Map<String, Object> modelParams) {
+        Object modelObj = modelParams != null ? modelParams.get("model") : null;
+        String modelName = modelObj != null ? modelObj.toString() : null;
+        if (modelName == null || modelName.isBlank()) {
+            return this.defaultModel;
+        }
+        return modelName;
+    }
+
     // Models Anthropic documents as rejecting assistant-message prefilling. Source of truth:
     // https://platform.claude.com/docs/en/build-with-claude/working-with-messages#putting-words-in-claudes-mouth
     //

integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java

@@ -396,6 +396,49 @@ void testNativeSchemaNotAppliedWithoutSchema() {
         assertThat(build(CAPABLE_MODEL, null, null).params.outputConfig()).isEmpty();
     }
 
+    @Test
+    @DisplayName("the effective model falls back to the configured default")
+    void testEffectiveModelForFallsBackToTheConfiguredDefault() {
+        // buildRequest resolves the model the same way before feeding the capability predicate,
+        // and connection() is configured with claude-sonnet-4-20250514.
+        assertThat(connection().effectiveModelFor(params(null)))
+                .isEqualTo("claude-sonnet-4-20250514");
+        assertThat(connection().effectiveModelFor(paramsWithModel(CAPABLE_MODEL, null)))
+                .isEqualTo(CAPABLE_MODEL);
+        // A blank model is not a model. The builder substitutes the default for it, so the hook
+        // has to as well or the two disagree on exactly this input.
+        assertThat(connection().effectiveModelFor(paramsWithModel("   ", null)))
+                .isEqualTo("claude-sonnet-4-20250514");
+    }
+
+    @Test
+    @DisplayName("the model the request builder judges is the one the hook names")
+    void testEffectiveModelForNamesTheModelTheBuilderJudges() {
+        // The hook duplicates the builder's resolution rather than centralizing it, so only
+        // capturing what the builder feeds the predicate keeps the two from drifting apart.
+        AtomicReference<String> judged = new AtomicReference<>();
+        AnthropicChatModelConnection connection =
+                new AnthropicChatModelConnection(descriptor("claude-sonnet-4-20250514"), NOOP) {
+                    @Override
+                    protected boolean supportsNativeStructuredOutput(String effectiveModel) {
+                        judged.set(effectiveModel);
+                        return super.supportsNativeStructuredOutput(effectiveModel);
+                    }
+                };
+
+        for (Map<String, Object> modelParams :
+                List.<Map<String, Object>>of(
+                        paramsWithModel(CAPABLE_MODEL, null),
+                        paramsWithModel("   ", null),
+                        params(null))) {
+            String named = connection.effectiveModelFor(modelParams);
+
+            connection.buildRequest(userMessage(), List.of(), modelParams, Answer.class);
+
+            assertThat(judged.get()).isEqualTo(named);
+        }
+    }
+
     @Test
     @DisplayName("a schema that is not a Class keeps the prompt fallback")
     void testNonClassSchemaKeepsFallback() {

integrations/chat-models/bedrock/src/main/java/org/apache/flink/agents/integrations/chatmodels/bedrock/BedrockChatModelConnection.java

@@ -237,6 +237,22 @@ protected boolean supportsNativeStructuredOutput(String effectiveModel) {
         return profile.matches() && NATIVE_STRUCTURED_OUTPUT_MODELS.contains(profile.group(1));
     }
 
+    /**
+     * The {@code model} parameter, falling back to the model configured on the connection when the
+     * call names none, which is how the request itself resolves the model it is issued against.
+     *
+     * <p>Resolving to nothing comes back null rather than raising the way {@code resolveModel}
+     * does, because the capability predicate reports a null model not capable.
+     */
+    @Override
+    protected String effectiveModelFor(Map<String, Object> modelParams) {
+        String model = modelParams != null ? (String) modelParams.get("model") : null;
+        if (model == null || model.isBlank()) {
+            return this.defaultModel;
+        }
+        return model;
+    }
+
     @Override
     public ChatMessage chat(
             List<ChatMessage> messages, List<Tool> tools, Map<String, Object> modelParams) {

integrations/chat-models/bedrock/src/test/java/org/apache/flink/agents/integrations/chatmodels/bedrock/BedrockChatModelConnectionTest.java

@@ -46,6 +46,7 @@
 import software.amazon.awssdk.services.bedrockruntime.model.OutputFormatType;
 
 import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -130,6 +131,61 @@ void testChatThrowsWithoutModel() {
                 .isInstanceOf(RuntimeException.class);
     }
 
+    @Test
+    @DisplayName("the effective model falls back to the configured default")
+    void testEffectiveModelForFallsBackToTheConfiguredDefault() {
+        // buildRequest resolves the model the same way before feeding the capability predicate.
+        assertThat(connection().effectiveModelFor(new HashMap<>()))
+                .isEqualTo("us.anthropic.claude-sonnet-4-20250514-v1:0");
+        assertThat(connection().effectiveModelFor(params("qwen.qwen3-32b-v1:0")))
+                .isEqualTo("qwen.qwen3-32b-v1:0");
+        // A blank model is not a model. resolveModel substitutes the default for it, so the hook
+        // has to as well or the two disagree on exactly this input.
+        assertThat(connection().effectiveModelFor(params("   ")))
+                .isEqualTo("us.anthropic.claude-sonnet-4-20250514-v1:0");
+    }
+
+    @Test
+    @DisplayName("the model the request builder judges is the one the hook names")
+    void testEffectiveModelForNamesTheModelTheBuilderJudges() {
+        // The hook duplicates resolveModel rather than calling it, so only capturing what the
+        // builder feeds the predicate keeps the two from drifting apart.
+        AtomicReference<String> judged = new AtomicReference<>();
+        BedrockChatModelConnection connection =
+                new BedrockChatModelConnection(
+                        descriptor("us-east-1", "us.anthropic.claude-sonnet-4-20250514-v1:0"),
+                        NOOP) {
+                    @Override
+                    protected boolean supportsNativeStructuredOutput(String effectiveModel) {
+                        judged.set(effectiveModel);
+                        return super.supportsNativeStructuredOutput(effectiveModel);
+                    }
+                };
+
+        for (Map<String, Object> modelParams :
+                List.<Map<String, Object>>of(
+                        params("qwen.qwen3-32b-v1:0"), params("   "), new HashMap<>())) {
+            String named = connection.effectiveModelFor(modelParams);
+
+            connection.buildRequest(
+                    List.of(ChatMessage.user("hello")), null, modelParams, Profile.class);
+
+            assertThat(judged.get()).isEqualTo(named);
+        }
+    }
+
+    @Test
+    @DisplayName("the effective model is null rather than throwing when none resolves")
+    void testEffectiveModelForReturnsNullWhenNoModelResolves() {
+        // resolveModel rejects an unresolvable model before a request is built. This query is
+        // part of the connection contract and answers for whatever it is given, so it reports
+        // null, which the capability predicate treats as not capable.
+        BedrockChatModelConnection conn =
+                new BedrockChatModelConnection(descriptor("us-east-1", null), NOOP);
+
+        assertThat(conn.effectiveModelFor(new HashMap<>())).isNull();
+    }
+
     @Test
     @DisplayName("chat accepts a POJO output schema instead of rejecting it as untranslatable")
     void testChatAcceptsPojoOutputSchema() {

integrations/chat-models/gemini/src/main/java/org/apache/flink/agents/integrations/chatmodels/gemini/GeminiChatModelConnection.java

@@ -234,6 +234,23 @@ protected boolean supportsNativeStructuredOutput(String effectiveModel) {
                 && effectiveModel.length() > NATIVE_STRUCTURED_OUTPUT_FAMILY_PREFIX.length();
     }
 
+    /**
+     * The {@code model} parameter, falling back to the model configured on the connection when the
+     * call names none, which is how the request itself resolves the model it is issued against.
+     *
+     * <p>Resolving to nothing comes back null rather than raising the way request building does,
+     * because the capability predicate reports a null model not capable.
+     */
+    @Override
+    protected String effectiveModelFor(Map<String, Object> modelParams) {
+        Object modelObj = modelParams != null ? modelParams.get("model") : null;
+        String modelName = modelObj != null ? modelObj.toString() : null;
+        if (modelName == null || modelName.isBlank()) {
+            return this.defaultModel;
+        }
+        return modelName;
+    }
+
     @Override
     public ChatMessage chat(
             List<ChatMessage> messages,

integrations/chat-models/gemini/src/test/java/org/apache/flink/agents/integrations/chatmodels/gemini/GeminiChatModelConnectionTest.java

@@ -53,6 +53,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -920,4 +921,73 @@ void chatWithoutSchemaDelegatesToTheSchemaCarryingOverload() {
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessageContaining("model name must be provided");
     }
+
+    @Test
+    @DisplayName("The effective model falls back to the configured default")
+    void effectiveModelForFallsBackToTheConfiguredDefault() {
+        // chat resolves the model the same way before feeding the capability predicate, and
+        // connection() is configured with gemini-3-pro-preview.
+        Map<String, Object> modelParams = params();
+
+        assertThat(connection().effectiveModelFor(modelParams)).isEqualTo("gemini-3-pro-preview");
+
+        modelParams.put("model", CAPABLE_MODEL);
+        assertThat(connection().effectiveModelFor(modelParams)).isEqualTo(CAPABLE_MODEL);
+
+        // A blank model is not a model. The request path substitutes the default for it, so the
+        // hook has to as well or the two disagree on exactly this input.
+        modelParams.put("model", "   ");
+        assertThat(connection().effectiveModelFor(modelParams)).isEqualTo("gemini-3-pro-preview");
+    }
+
+    /** Stops the request after the config is assembled, so no call reaches the provider. */
+    private static final class StopBeforeRequest extends RuntimeException {}
+
+    private static Map<String, Object> paramsWithModel(String model) {
+        Map<String, Object> modelParams = params();
+        modelParams.put("model", model);
+        return modelParams;
+    }
+
+    @Test
+    @DisplayName("The model the request builder judges is the one the hook names")
+    void effectiveModelForNamesTheModelTheBuilderJudges() {
+        // chat resolves the model and hands it to buildConfig, which is where the predicate is fed.
+        // Overriding buildConfig to stop once it has run binds the hook to the request without
+        // reaching the provider. Asserting each side against a literal would let the two drift in
+        // step, which is the one failure this has to catch.
+        AtomicReference<String> judged = new AtomicReference<>();
+        GeminiChatModelConnection connection =
+                new GeminiChatModelConnection(
+                        descriptor("test-key", null, "gemini-3-pro-preview"), NOOP) {
+                    @Override
+                    protected boolean supportsNativeStructuredOutput(String effectiveModel) {
+                        judged.set(effectiveModel);
+                        return super.supportsNativeStructuredOutput(effectiveModel);
+                    }
+
+                    @Override
+                    GenerateContentConfig buildConfig(
+                            List<ChatMessage> messages,
+                            List<Tool> tools,
+                            Map<String, Object> arguments,
+                            String modelName,
+                            Object outputSchema) {
+                        super.buildConfig(messages, tools, arguments, modelName, outputSchema);
+                        throw new StopBeforeRequest();
+                    }
+                };
+
+        for (Map<String, Object> modelParams :
+                List.<Map<String, Object>>of(
+                        paramsWithModel(CAPABLE_MODEL), paramsWithModel("   "), params())) {
+            String named = connection.effectiveModelFor(modelParams);
+
+            assertThatThrownBy(
+                            () -> connection.chat(userMessage(), null, modelParams, Report.class))
+                    .hasRootCauseInstanceOf(StopBeforeRequest.class);
+
+            assertThat(judged.get()).isEqualTo(named);
+        }
+    }
 }

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

@@ -223,6 +223,20 @@ protected boolean supportsNativeStructuredOutput(String effectiveModel) {
         return NATIVE_STRUCTURED_OUTPUT_MODELS.contains(effectiveModel);
     }
 
+    /**
+     * The model backing the deployment, read from {@code model_of_azure_deployment}.
+     *
+     * <p>The {@code model} parameter carries the deployment name the request is issued against.
+     * That name is chosen by the user, and although it commonly echoes the model behind it, nothing
+     * keeps the two in step once the deployment is repointed, so it is never the answer here.
+     * Leaving the backing model unset keeps even a capable deployment on the prompt-engineering
+     * fallback rather than classifying a deployment name on its spelling.
+     */
+    @Override
+    protected String effectiveModelFor(Map<String, Object> modelParams) {
+        return modelParams == null ? null : (String) modelParams.get("model_of_azure_deployment");
+    }
+
     /**
      * Whether the configured api-version reaches the structured-output floor.
      *