[api][plan][java] Add a decision-model routing strategy for MODEL_ROUTER

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

@@ -65,6 +65,9 @@ public class ModelRouter extends Resource {
     /** Descriptor argument naming the candidate the router lands on when the strategy abstains. */
     public static final String DEFAULT_MODEL_KEY = "default_model";
 
+    /** Descriptor argument mapping candidate names to their {@code describe(...)} descriptions. */
+    public static final String CANDIDATE_DESCRIPTIONS_KEY = "candidate_descriptions";
+
     private final List<RoutingCandidate> candidates;
     private final String defaultModel;
     private final boolean fallbackEnabled;
@@ -78,16 +81,6 @@ public ModelRouter(ResourceDescriptor descriptor, ResourceContext resourceContex
         // with the candidate message instead of failing the cast here.
         Object defaultModel = descriptor.getArgument(DEFAULT_MODEL_KEY);
         RoutingCandidateValidator.validate("ModelRouter", names, defaultModel);
-        Map<String, String> descriptions =
-                descriptor.getArgument("candidate_descriptions", Collections.emptyMap());
-        List<RoutingCandidate> parsed = new ArrayList<>();
-        for (String name : names) {
-            parsed.add(new RoutingCandidate(name, descriptions.get(name)));
-        }
-        this.candidates = Collections.unmodifiableList(parsed);
-        this.defaultModel = (String) defaultModel;
-        this.fallbackEnabled =
-                Boolean.TRUE.equals(descriptor.getArgument("fallback", Boolean.FALSE));
         String typeTag = descriptor.getArgument(STRATEGY_TYPE_KEY);
         if (typeTag == null || typeTag.isEmpty()) {
             throw new IllegalArgumentException("ModelRouter requires a routing strategy.");
@@ -100,6 +93,30 @@ public ModelRouter(ResourceDescriptor descriptor, ResourceContext resourceContex
         this.strategy =
                 new RoutingStrategy(
                         RoutingStrategyType.fromTag(typeTag), strategyArgs, executorClass);
+        Object rawDescriptions = descriptor.getArgument(CANDIDATE_DESCRIPTIONS_KEY);
+        if (rawDescriptions != null && !(rawDescriptions instanceof Map)) {
+            // A hand-built or deserialized descriptor may carry any JSON type here; diagnose it
+            // instead of surfacing a raw ClassCastException per routed request.
+            throw new IllegalArgumentException(
+                    String.format(
+                            "ModelRouter declares '%s' as %s; expected a map of candidate name to"
+                                    + " description.",
+                            CANDIDATE_DESCRIPTIONS_KEY,
+                            rawDescriptions.getClass().getSimpleName()));
+        }
+        Map<?, ?> descriptions =
+                rawDescriptions == null ? Collections.emptyMap() : (Map<?, ?>) rawDescriptions;
+        RoutingCandidateValidator.requireStringDescriptions("ModelRouter", names, descriptions);
+        List<RoutingCandidate> parsed = new ArrayList<>();
+        for (String name : names) {
+            parsed.add(new RoutingCandidate(name, (String) descriptions.get(name)));
+        }
+        RoutingCandidateValidator.requireDescriptionsForDecisionStrategy(
+                "ModelRouter", this.strategy, names, descriptions);
+        this.candidates = Collections.unmodifiableList(parsed);
+        this.defaultModel = (String) defaultModel;
+        this.fallbackEnabled =
+                Boolean.TRUE.equals(descriptor.getArgument("fallback", Boolean.FALSE));
     }
 
     /** The configured strategy declaration (type + arguments). */
@@ -234,10 +251,12 @@ public ResourceDescriptor build() {
                     }
                 }
             }
+            RoutingCandidateValidator.requireDescriptionsForDecisionStrategy(
+                    "ModelRouter", strategy, candidates, descriptions);
             Map<String, Object> args = new HashMap<>();
             args.put(CANDIDATES_KEY, new ArrayList<>(candidates));
             if (!descriptions.isEmpty()) {
-                args.put("candidate_descriptions", new HashMap<>(descriptions));
+                args.put(CANDIDATE_DESCRIPTIONS_KEY, new HashMap<>(descriptions));
             }
             if (defaultModel != null) {
                 args.put(DEFAULT_MODEL_KEY, defaultModel);

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

@@ -22,8 +22,10 @@
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -78,4 +80,64 @@ public static void validate(
                             subject, defaultModel, candidates));
         }
     }
+
+    /**
+     * A decision-model strategy asks one choice question over the candidates, so it needs at least
+     * two of them, and it routes on their descriptions alone — they are the question's criteria —
+     * so a candidate without one is invisible to the decision model. Applied by the router
+     * constructor, by {@code ModelRouter.Builder#build()} and by plan validation, so every
+     * declaration path fails with the same diagnostic; {@code subject} names the router.
+     */
+    /**
+     * Every description declared for a candidate must be a String. Applied by the router
+     * constructor and by plan validation, so a descriptor that skipped the builder fails with the
+     * same diagnostic at plan construction rather than with a ClassCastException per routed
+     * request; {@code subject} names the router. Entries for names that are not candidates are
+     * ignored, as the constructor ignores them.
+     */
+    public static void requireStringDescriptions(
+            String subject, List<String> candidates, Map<?, ?> descriptions) {
+        for (String name : candidates) {
+            Object description = descriptions.get(name);
+            if (description != null && !(description instanceof String)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "%s candidate '%s' has a %s description; expected a String.",
+                                subject, name, description.getClass().getSimpleName()));
+            }
+        }
+    }
+
+    public static void requireDescriptionsForDecisionStrategy(
+            String subject,
+            RoutingStrategy strategy,
+            List<String> candidates,
+            Map<?, ?> descriptions) {
+        if (strategy.getType() != RoutingStrategyType.DECISION_MODEL) {
+            return;
+        }
+        // A choice question needs something to choose between: decision providers reject a
+        // single option, so a one-candidate router would fail every request on the TaskManager.
+        if (candidates.size() < 2) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "%s uses Strategies.decision, which requires at least two candidates"
+                                    + " (got %s).",
+                            subject, candidates));
+        }
+        List<String> missing = new ArrayList<>();
+        for (String candidate : candidates) {
+            Object description = descriptions == null ? null : descriptions.get(candidate);
+            if (!(description instanceof String) || ((String) description).isBlank()) {
+                missing.add(candidate);
+            }
+        }
+        if (!missing.isEmpty()) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "%s uses Strategies.decision, which requires a description for every"
+                                    + " candidate (describe(name, description)); missing for %s.",
+                            subject, missing));
+        }
+    }
 }

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

@@ -49,10 +49,26 @@ public final class RoutingStrategy implements Serializable {
 
     /**
      * Argument key: optional cap (in characters) on the conversation context included in the judge
-     * input ({@code LLM_JUDGE}). Unset means the judge sees the complete message list.
+     * or decision-model input ({@code LLM_JUDGE}, {@code DECISION_MODEL}). Unset means the model
+     * sees the complete message list.
      */
     public static final String ARG_MAX_CONTEXT_CHARS = "max_context_chars";
 
+    /** Argument key: the decision-model setup name ({@code DECISION_MODEL}). */
+    public static final String ARG_DECISION_MODEL = "decision_model";
+
+    /**
+     * Argument key: optional instructions for the decision model's choice question ({@code
+     * DECISION_MODEL}). Unset means the built-in instruction is used.
+     */
+    public static final String ARG_INSTRUCTIONS = "instructions";
+
+    /**
+     * Argument key: optional minimum probability of the chosen candidate ({@code DECISION_MODEL}).
+     * A choice whose probability is below it abstains to the router's default model.
+     */
+    public static final String ARG_MIN_PROBABILITY = "min_probability";
+
     /** Argument key: the candidate-to-regex rule map ({@code RULE_BASED}). */
     public static final String ARG_RULES = "rules";
 
@@ -101,20 +117,33 @@ private void validate() {
                                     + ARG_PROMPT_TEMPLATE
                                     + "' must be a non-empty String when provided.");
                 }
-                Object cap = arguments.get(ARG_MAX_CONTEXT_CHARS);
-                // Compare as long: intValue() would wrap an out-of-range Long into a tiny (or
-                // negative) cap, and saturate a large Double to Integer.MAX_VALUE ("no cap").
-                if (cap != null
-                        && (!(cap instanceof Number)
-                                || ((Number) cap).longValue() <= 0
-                                || ((Number) cap).longValue() > Integer.MAX_VALUE)) {
+                validateMaxContextChars(arguments.get(ARG_MAX_CONTEXT_CHARS));
+                break;
+            case DECISION_MODEL:
+                Object decisionModel = arguments.get(ARG_DECISION_MODEL);
+                if (!(decisionModel instanceof String) || ((String) decisionModel).isEmpty()) {
+                    throw new IllegalArgumentException(
+                            "Strategies.decision requires a non-empty '"
+                                    + ARG_DECISION_MODEL
+                                    + "'.");
+                }
+                Object instructions = arguments.get(ARG_INSTRUCTIONS);
+                if (instructions != null
+                        && (!(instructions instanceof String)
+                                || ((String) instructions).isBlank())) {
+                    throw new IllegalArgumentException(
+                            "'" + ARG_INSTRUCTIONS + "' must be a non-empty String when provided.");
+                }
+                Object minProbability = arguments.get(ARG_MIN_PROBABILITY);
+                if (minProbability != null
+                        && (!(minProbability instanceof Number)
+                                || !isProbability(((Number) minProbability).doubleValue()))) {
                     throw new IllegalArgumentException(
                             "'"
-                                    + ARG_MAX_CONTEXT_CHARS
-                                    + "' must be a positive integer (at most "
-                                    + Integer.MAX_VALUE
-                                    + ") when provided.");
+                                    + ARG_MIN_PROBABILITY
+                                    + "' must be a number between 0 and 1 when provided.");
                 }
+                validateMaxContextChars(arguments.get(ARG_MAX_CONTEXT_CHARS));
                 break;
             case CUSTOM:
                 if (executorClass == null || executorClass.isEmpty()) {
@@ -141,23 +170,62 @@ private void validate() {
         }
     }
 
+    private static void validateMaxContextChars(@Nullable Object cap) {
+        // Compare as long: intValue() would wrap an out-of-range Long into a tiny (or
+        // negative) cap, and saturate a large Double to Integer.MAX_VALUE ("no cap").
+        if (cap != null
+                && (!(cap instanceof Number)
+                        || ((Number) cap).longValue() <= 0
+                        || ((Number) cap).longValue() > Integer.MAX_VALUE)) {
+            throw new IllegalArgumentException(
+                    "'"
+                            + ARG_MAX_CONTEXT_CHARS
+                            + "' must be a positive integer (at most "
+                            + Integer.MAX_VALUE
+                            + ") when provided.");
+        }
+    }
+
+    private static boolean isProbability(double value) {
+        return !Double.isNaN(value) && value >= 0.0 && value <= 1.0;
+    }
+
     /**
-     * Returns a copy of this {@code LLM_JUDGE} declaration with a cap (in characters) on the
-     * conversation context included in the judge input. Purely opt-in: without it the judge
-     * receives the complete message list. When set, the rendered request and the SYSTEM message are
-     * always kept, remaining messages fill newest-first within the budget, and any truncation is
-     * flagged in the decision metadata ({@code judge_context_truncated}).
+     * Returns a copy of this {@code LLM_JUDGE} or {@code DECISION_MODEL} declaration with a cap (in
+     * characters) on the conversation context included in the judge or decision-model input. Purely
+     * opt-in: without it the model receives the complete message list. When set, the rendered
+     * request and the SYSTEM message are always kept, remaining messages fill newest-first within
+     * the budget, and any truncation is flagged in the decision metadata ({@code
+     * judge_context_truncated} or {@code decision_context_truncated}).
      */
     public RoutingStrategy withMaxContextChars(int maxContextChars) {
-        if (type != RoutingStrategyType.LLM_JUDGE) {
+        if (type != RoutingStrategyType.LLM_JUDGE && type != RoutingStrategyType.DECISION_MODEL) {
             throw new IllegalStateException(
-                    "withMaxContextChars applies to Strategies.llm(...) declarations only.");
+                    "withMaxContextChars applies to Strategies.llm(...) and"
+                            + " Strategies.decision(...) declarations only.");
         }
         Map<String, Object> withCap = new LinkedHashMap<>(arguments);
         withCap.put(ARG_MAX_CONTEXT_CHARS, maxContextChars);
         return new RoutingStrategy(type, withCap, null);
     }
 
+    /**
+     * Returns a copy of this {@code DECISION_MODEL} declaration with a minimum probability for the
+     * chosen candidate. When the decision model's probability for its choice is below this value,
+     * the strategy abstains and the router's default model answers; the choice and the full
+     * probability map stay in the decision metadata. The probability is the provider's reported
+     * value for the chosen candidate; the framework does not normalize the distribution.
+     */
+    public RoutingStrategy withMinProbability(double minProbability) {
+        if (type != RoutingStrategyType.DECISION_MODEL) {
+            throw new IllegalStateException(
+                    "withMinProbability applies to Strategies.decision(...) declarations only.");
+        }
+        Map<String, Object> withMin = new LinkedHashMap<>(arguments);
+        withMin.put(ARG_MIN_PROBABILITY, minProbability);
+        return new RoutingStrategy(type, withMin, null);
+    }
+
     /**
      * A mis-shaped 'rules' value must fail loudly at declaration time: silently compiling zero
      * rules would disable routing (every request abstains to the default) with no diagnostic. A

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

@@ -34,6 +34,13 @@ public enum RoutingStrategyType {
      */
     LLM_JUDGE,
 
+    /**
+     * Framework-managed decision-model routing: the engine asks a {@code DECISION_MODEL} resource
+     * one choice question over the candidates and routes on the returned choice and its
+     * probability.
+     */
+    DECISION_MODEL,
+
     /** A user-provided {@link CustomRoutingExecutor} referenced by class name. */
     CUSTOM;
 

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

@@ -90,6 +90,44 @@ public static RoutingStrategy llm(String judgeModel, String promptTemplate) {
         return new RoutingStrategy(RoutingStrategyType.LLM_JUDGE, args, null);
     }
 
+    /**
+     * Decision-model routing (framework-managed): a decision model — registered like any other
+     * {@code DECISION_MODEL} resource — answers one choice question whose options are the candidate
+     * names and whose criteria are the candidates' {@code describe(...)} descriptions. The engine
+     * executes the decision call on its durable, metered, observable path. The returned choice
+     * becomes the selected model and its reported probability the routing score. A refusal answer,
+     * a choice whose probability is below {@link RoutingStrategy#withMinProbability(double)}, or a
+     * choice with no reported probability while that minimum is set, abstains to the router's
+     * default model (without a minimum such a choice routes with no score); a decision call that
+     * exhausts its retries fails the request.
+     *
+     * <p>Every candidate must carry a description: the descriptions are all the decision model sees
+     * of the candidates. The decision model receives the conversation text (tool calls and results
+     * included, media and reasoning excluded); use {@link RoutingStrategy#withMaxContextChars(int)}
+     * to cap the context for cost.
+     */
+    public static RoutingStrategy decision(String decisionModel) {
+        Map<String, Object> args = new HashMap<>();
+        args.put(RoutingStrategy.ARG_DECISION_MODEL, decisionModel);
+        return new RoutingStrategy(RoutingStrategyType.DECISION_MODEL, args, null);
+    }
+
+    /**
+     * Like {@link #decision(String)}, with custom instructions for the choice question in place of
+     * the built-in instruction.
+     */
+    public static RoutingStrategy decision(String decisionModel, String instructions) {
+        Map<String, Object> args = new HashMap<>();
+        args.put(RoutingStrategy.ARG_DECISION_MODEL, decisionModel);
+        if (instructions == null) {
+            throw new IllegalArgumentException(
+                    "Strategies.decision instructions must be non-null; use decision(decisionModel)"
+                            + " for the built-in instructions.");
+        }
+        args.put(RoutingStrategy.ARG_INSTRUCTIONS, instructions);
+        return new RoutingStrategy(RoutingStrategyType.DECISION_MODEL, args, null);
+    }
+
     /**
      * A custom executor referenced by class. The class must implement {@link CustomRoutingExecutor}
      * with either a {@code (Map<String,Object>)} constructor or a no-arg constructor. This is the

api/src/main/java/org/apache/flink/agents/api/event/ModelRoutingEvent.java

@@ -51,6 +51,9 @@ public class ModelRoutingEvent extends Event {
     /** The decision came from a framework-managed LLM judge verdict. */
     public static final String SOURCE_LLM_JUDGE = "llm_judge";
 
+    /** The decision came from a framework-managed decision model choice. */
+    public static final String SOURCE_DECISION_MODEL = "decision_model";
+
     /** Attribute/metadata key carrying the decision source (events, payloads, durable records). */
     public static final String DECISION_SOURCE_KEY = "decision_source";