[purushah]

Linked issue: #1325

Purpose of change

A MODEL_ROUTER can now route with a registered decision model. Strategies.decision("dm") asks the DECISION_MODEL resource one choice question (options: the candidates; criteria: their describe(...) texts) and routes on the returned choice. The choice's probability becomes the routing score; withMinProbability(p) abstains to the default model below p. Rules, the LLM judge and custom executors are unchanged.

The LLM judge returns a candidate name and nothing else, so a router cannot tell a confident selection from a coin flip or decline a weak one. Decision models return typed answers with probabilities, so the strategy reuses them instead of parsing a verdict out of free text.

Runtime flow

1. ModelRoutingResolver dispatches DECISION_MODEL to DecisionModelRoutingExecutor, which declares internal durable calls and so runs before the route:<router> record, like the judge.
2. The executor resolves the setup, renders the conversation to text with the judge's rendering (anchor candidate's bound prompt, roles, tool calls and results, optional max_context_chars cap, prompt-argument fallback; media dropped and counted) and builds one ChoiceQuestion named model.
3. DecisionModelInvoker.decideWithRetries (new, shared with the built-in decision action) runs setup.decide inside durableExecute under decide:<router> with the configured retry budget, execution reports and model_resource token metrics; retries are accumulated on the routed request like the judge's.
4. The answer maps to a RoutingDecision (table below) with metadata decision_model, decision_probabilities, optional provider_confidence, token counts, decision_context_truncated, decision_media_dropped, and decision_choice on abstain. The resolver persists, normalizes and emits ModelRoutingEvent (source decision_model); the chat path is unchanged.

Key decisions

• The score is the provider's probability for its choice, not the maximum of the map; the framework does not normalize distributions.
• provider_confidence is recorded but not thresholded; not every provider reports it.
• Every candidate must carry a description, and every description must be a String, each enforced by one RoutingCandidateValidator rule shared by the constructor, build() and plan validation; plan validation also requires an explicitly declared setup connection to be registered.
• The call bypasses the built-in decision action (durable calls cannot nest and its flat decide id would collide) but shares its retry loop through DecisionModelInvoker, which the action now uses too.

Behavioral Semantics

Interaction decisions

Answer	min_probability	Result
choice, probability ≥ min or no min	any	route to choice, score = probability, source decision_model
choice, probability < min	set	abstain to default; reason names probability and threshold; decision_choice recorded
choice missing from probabilities	set	abstain to default; reason names the choice and threshold; decision_choice recorded
choice missing from probabilities	not set	route to choice with no score, source decision_model
choice naming a non-candidate	any	abstain to default (defensive)
refusal	any	abstain to default, reason decision model refused[: reason]
predicate or score answer	any	abstain to default, reason decision answer was not a choice
no answer for question model	any	abstain to default, reason decision model returned no answer for question 'model'
request carries no text and no prompt arguments	any	abstain to default without calling the provider
provider error after retries	any	request fails (Decision model '<name>' failed: …); no default, no fallback

Behavioral contracts

1. A choice routes to its candidate with the reported probability as score.
2. A choice below withMinProbability abstains to the default; choice and probability map stay in metadata.
3. A refusal, a non-choice answer or a missing answer abstains to the default; so does a request with no text and no prompt arguments, without a provider call.
4. The call is durable under decide:<router>: on replay the stored outcomes (including a failed attempt) are reused positionally, the provider not called; a restore between the decision call and the route:<router> record routes on the stored choice.
5. Provider failures use the retry budget; exhausting it fails the request, no candidate fallback. Cancellation or interruption raised while the anchor renders the input propagates before the provider is called; it is not treated as an unresolvable anchor (the shared judge helper had this gap; fixed here for both strategies).
6. Fewer than two candidates, an undescribed or blank-described candidate, an unregistered decision model, a declared connection that is blank or unregistered, a declared blank model, or min_probability outside [0, 1]: rejected at declaration or plan construction.
7. Tool rounds, candidate fallback and the other strategies behave as before.

Failure behavior

Unresolvable decision model, a request that cannot be rendered, or an answer that cannot be read: RoutingFailure (failed ChatResponseEvent). Durable-state or recovery failures during the decision call propagate to the runner. Provider exception or null result: retried per MAX_RETRIES / RETRY_WAIT_INTERVAL, then RoutingFailure; cancellation propagates, never persisted. Persistence or recovery errors propagate to the runner, as for the judge.

Tests

Contract	Tests (E = DecisionModelRoutingExecutorTest, C = ChatModelActionRoutingTest)
1	E choiceSelects…AsScore, scoreIsTheChosenCandidateProbabilityNotTheMaximum; C decisionModelChoiceRoutesToNamedCandidate
2	E probabilityBelowMinimumAbstains…, probabilityAtTheMinimumIsAccepted, missingProbabilityForTheChoiceAbstainsWhenAMinimumIsSet, missingProbabilityForTheChoiceRoutesWithoutAScoreWhenNoMinimumIsSet; C decisionBelowMinProbabilityAbstainsToDefault
3	E refusalAbstains…, nonChoiceAnswerAbstains, missingAnswerForTheQuestionAbstains, nonCandidateChoiceAbstains, emptyRequestAbstainsWithoutCallingTheProvider; C decisionRefusalAbstainsToDefault
4	C decisionReplayReusesTheStoredAnswerWithoutCallingTheProvider, asyncDecisionCallRunsAsFlatSiblingBeforeTheRouteRecord; runtime DecisionRoutingRecoveryTest (real journal serializer, sync and async: restore between the two stored decision attempts and the route record; full replay of a completed routed request)
5	C decisionModelUsesTheRetryBudget…, decisionModelFailureFailsTheRequest…, unresolvableDecisionModelFailsTheRequest, durableStateFailureDuringTheDecisionCallPropagatesToTheRunner, cancellationDuringDecisionInputPreparationPropagatesBeforeTheProviderCall, cancellationDuringJudgeInputPreparationPropagatesBeforeTheJudgeCall; E unrenderableRequestIsARoutingFailure
6	RoutingTest decision tests, AgentPlanDecisionModelValidationTest (10 tests)
7	existing routing suites; AgentPlanCrossLanguageTest decision-router snapshot (two routers: the minimum and every optional argument); Python test_python_can_deserialize_java_plan_with_decision_router checks each key's name and JSON type
1, 2, 5 end to end	DecisionModelRoutingE2ETest (e2e integration module): one real Flink job with a scripted Java decision connection and two echo chat connections registered on the agent; confident selection of each candidate, abstention below min_probability to the default model, exhausted retries failing the request without candidate fallback, provider call count including the retry, rendered decision input

mvn -pl api,plan,runtime test   # api 836 pass; runtime 1515 pass; plan 695 run, only AgentPlanDeclareMCPServerTest fails (talks to a port-8000 server on the build host; fails identically on unmodified main)
uv run pytest flink_agents/plan/tests/test_agent_plan_cross_language.py   # 21 passed, 1 skipped
mvn -pl e2e-test/...-integration,e2e-test/...-resource-cross-language test   # 71 pass; cross-language: 4 Python-bridge tests fail with 'No module named flink_agents' / 'ollama' from the system python3, unrelated to routing; DecisionModelCrossLanguageTest passes

End to end on a standalone Flink 2.3.0 cluster with the new examples/openai/OpenAiDecisionRoutingExample (gpt-4o-mini and gpt-4o candidates, the built-in Python OpenAI decision model gpt-6-luna through the cross-language bridge, python.executable pointing at a Python with flink-agents and openai installed): job finished in 7.7 s; output final_model=small … score=0.99 for the greeting and final_model=big … score=0.87 for the SQL request; no TaskManager errors. Runtime notes for the cross-language example: Flink's opt/flink-python jar must be on the cluster classpath, and the embedded interpreter resolves modules through PYTHONPATH (it also sees the base interpreter's site-packages, so a current brotli must be first on the path or the OpenAI client's response decoder fails).

Not verified: Python-side routing (routers remain Java-only), cancellation during the decision call's retry backoff, and a full cluster failover (recovery is covered by the runtime journal test, not a restarted TaskManager).

Implementation invariants and supporting detail

• decide:<router> is distinct from the built-in decision action's flat decide id; they belong to different actions and cannot collide in one journal.
• DecisionOutcome (result or error text) is the durable record, JSON-serializable like the chat invoker's InvocationOutcome.
• The input helpers both strategies use (effective messages with the anchor's bound prompt, context budget, content projection) live in the package-private RoutingInput, so neither executor depends on the other's internals; the anchor fallback rethrows cancellation and interruption before falling back to the raw messages (a pre-existing gap in the judge path), otherwise behavior unchanged and still pinned by LlmJudgeRoutingExecutorTest.
• ModelRouter.CANDIDATE_DESCRIPTIONS_KEY names the existing candidate_descriptions argument; the serialized plan is unchanged.
• withMaxContextChars validation is shared by the judge and decision declarations.
• Text blocks of one message are joined with a newline in the decision input, without doubling the newlines the tool framing already carries, so a message split around media does not run together and tool rounds do not produce blank lines (toolCallsAndResultsAreRenderedAsText pins the line shape).
• Token counts in the routing metadata are stored in JSON shape (Integer when the value fits an int), so a fresh decision and one replayed from the route record carry the same types; the recovery test asserts them on both paths.
• A setup descriptor that declares no connection or model passes plan validation on purpose: a Python setup subclass may define them as class-level defaults (the cross-language fixture BridgeDecisionSetup does), so only explicitly declared values are checked.
• A media-only, empty or SYSTEM-only conversation contributes no request text, so the prompt arguments become the decision input (mediaOnlyRequestFallsBackToPromptArgs, promptArgsAreTheInputWhenTheConversationCarriesNoRequestText).
• DecisionModelActionTest (32 tests) pins the built-in action's behavior across the move to DecisionModelInvoker.
• Snapshot e2e-test/cross-language-agent-plan-snapshots/java/agent_plan_with_decision_router.json generated with -Dregenerate.snapshots=true.

API

New: RoutingStrategyType.DECISION_MODEL (tag decision_model), Strategies.decision(String) / decision(String, String), RoutingStrategy.withMinProbability(double), ARG_DECISION_MODEL / ARG_INSTRUCTIONS / ARG_MIN_PROBABILITY, ModelRoutingEvent.SOURCE_DECISION_MODEL, ModelRouter.CANDIDATE_DESCRIPTIONS_KEY, RoutingCandidateValidator.requireDescriptionsForDecisionStrategy. withMaxContextChars also accepts decision declarations. No existing signature changes; existing routers, plans and persisted decisions are unaffected. Routers remain Java-only (#1062).

Documentation

• doc-needed
• doc-not-needed
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Claude Code 2.1.309 (Claude Fable 5.1)

[wenjin272]

Thanks for the implementation!

Could we add an E2E test for decision-model routing through a running Flink job? A deterministic mock decision model and mock candidate chat models would be sufficient; it does not need live API credentials or the Python bridge. The test should exercise resource registration and initialization, routing, candidate invocation, and final output, covering selection, low-confidence abstention to the default model, and exhausted decision-model retries.

[purushah]

Added DecisionModelRoutingE2ETest in the e2e integration module. It runs a real Flink job with a scripted Java decision connection and two echo chat models registered on the agent, no credentials and no Python bridge. One job sends four requests and checks a confident selection of each candidate, a choice below the minimum probability abstaining to the default model, and a provider that keeps failing until the retry budget is exhausted, which fails the request without candidate fallback. It also checks the provider call count including the retry and the rendered decision input.

The three changes are separate commits on top of the original one, and the branch is rebased on current main.

[sangkyoonnam]

Please extend the snapshot and the behavior table so a later Python decision-model strategy can follow them. I haven't run anything; I read the diff at c21651514b58. Two gaps are inline.

[wenjin272]

Thanks for addressing the feedback. LGTM.

I re-reviewed the latest changes and ran the focused tests successfully. I also verified the updated example with a real OpenAI API key through the Python bridge: both requests completed successfully and routed to the expected candidates.

[purushah]

Thanks for the review and for verifying the example end to end.