[plan][runtime][java] Add CEL Action condition filtering

[rosemarYuan]

Linked issue: #754

Stacked PR — this PR is based on #756 . Please review the [incremental diff], not the full diff against main.

Purpose of change

This is the second PR in the four-PR stack tracked by #754:

1. [api][java][python] Introduce EventType constants and unify Action trigger entry #756 — API changes (under review)
2. This PR — Java CEL runtime
3. Python CEL runtime (planned)
4. Documentation (planned)

It adds the Java-side CEL runtime for @Action trigger conditions, enabling actions to filter events by field-level CEL expressions (e.g. event.tokenCount > 100).

Key components

• ParsedCondition.classify: parses each trigger condition via CEL parser; bare identifier → TypeMatch, anything else → CelExpression.
• CelMacroPolicy: custom has() macro (has(x) → has(attributes.x)); whitelist rejects exists/filter/map etc.
• CelExpressionFacade: Parse → Validate → Check → Compile pipeline with process-wide LRU program cache and AST resource limits.
• CelConditionEvaluator: pre-compiles programs at open time; builds activation map from event; evaluates with configurable failure policy.
• ActionRouter: two-phase routing — actionsByEvent HashMap fast path for TypeMatch, CEL slow path with lazy activation and LinkedHashSet deduplication.
• Operator integration: ActionExecutionOperator / EventRouter delegate to ActionRouter.

Tests

• Plan layer: ParsedConditionTest, ActionParsedConditionsTest, ActionConstructionFailureTest
• Runtime layer: CelExpressionFacadeTest, CelConditionEvaluatorTest, ActionRouterTest, CelResourceLimitsTest
• Serialization: ActionJsonSerializerTest, ActionJsonDeserializerTest (including legacy listen_event_types fallback)
• Cross-language: cel_conformance_cases.yaml, disallowed_macros.yaml + .github/workflows/cel-conformance.yml

API

No new public API beyond what was introduced in #756. This PR adds runtime internals only.

Documentation

• doc-needed
• doc-not-needed (documentation will be covered in PR4)
• doc-included

[weiqingy]

Thanks for taking this on — the serialization discipline stands out: keeping the compiled Programs off the Java-serialization path (router built on the TaskManager) and rebuilding the transient actionsWithCel in both readObject and the constructors are the easy-to-miss details that quietly break on failover. The macro-whitelist-via-CALL-node approach is a neat, well-tested call too. A few questions inline, plus one CI issue worth a look before merge.

1. The dist uber-jar now shades dev.cel and its transitive graph (antlr4-runtime, re2j, protobuf-java). Is the LICENSE/NOTICE update for those tracked somewhere (release cut, or PR4), or expected in this PR? Just so it doesn't fall through the Apache release gate.

[wenjin272]

Thanks for taking this on @rosemarYuan.

I left some comments regarding certain class names and JavaDoc. Additionally, I believe we are missing end-to-end (E2E) tests.

[rosemarYuan]

For Object normalizedAttrs = normalizeValue(event.getAttributes(), 0);, I think here have two question.

The first one is the large-number failure. That part has been fixed: when a number larger than int64 enters condition evaluation, we now convert it to double instead of failing during activation building.

The second one is reducing the amount of data passed into condition evaluation. I agree that we should avoid normalizing every field when the condition only needs a small subset. There are two possible directions here.

1. Shrink the map we build for condition evaluation. At plan/open time, we can analyze each condition and collect the attribute paths it may read. At runtime, we build the input map only from the union of fields needed by the candidate conditions for that event, and normalize only those selected values. If a condition cannot be narrowed safely, we can fall back to the current full-map behavior.
2. Lazy attribute access from the expression engine itself. Java already has the needed capability through CEL Java’s variable resolver path, so values can be normalized only when the expression actually reads them. The problem is Python: the current community implementation does not provide an equivalent lazy access path. The newer Google-backed Python implementation is promising because it wraps the official C++ runtime, but the Python binding does not expose this capability yet, and it currently requires Python 3.11 while we still support Python 3.10.

So my plan is to implement the first approach now: partial map construction with conservative fallback to full normalization. Later, once the Google CEL Python implementation supports Python 3.10 and exposes the required lazy access capability, we can consider migrating to the second approach.

@wenjin272 does this direction sound reasonable?

[wenjin272]

After a thorough review, I believe the current PR has issues in API design, architectural organization, and specific function implementation details. I have provided detailed comments on these issues.

[rosemarYuan]

Thanks for the detailed review, @wenjin272. I’ve reworked the implementation based on your feedback and force-pushed the updated commit stack.

The existing trigger_conditions API is retained as agreed, while classification and expression validation are now separated: Plan owns deterministic classification and fail-fast validation, and Runtime compiles expressions from source during initialization. I also simplified the runtime matching path, renamed it to ActionMatcher, and expanded the E2E and cross-language coverage.

These failures are expected for the current intermediate state of the stack, but they are not unrelated Python-side failures. This PR removes actions_by_event from the shared AgentPlan wire format and derives action routing from trigger_conditions on the Java side. Python still serializes and requires actions_by_event, and does not yet classify or evaluate condition expressions. Therefore, the cross-language snapshot tests are correctly exposing a temporary Java/Python contract mismatch.

This PR should not be considered independently merge-ready until the corresponding Python Plan/runtime changes are integrated, or the existing wire format is retained until both sides can be migrated together.

Could you please take another look when you have time? Thanks!

[wenjin272]

The main blocker for me is that the PR does not define a clear Event -> condition activation contract before implementing it in buildConditionVariables().

The current code treats output, root attributes, and input as three merge sources and introduces output > root > input precedence. This does not match the event model:

InputEvent:  attributes = {"input": payload}
OutputEvent: attributes = {"output": payload}
Other Event: attributes = user-defined fields

InputEvent and OutputEvent should not have sibling user attributes, so this precedence should not exist. Unexpected fields should be rejected or ignored consistently rather than
assigned merge semantics.

The raw input/output instanceof Map checks also make direct Java POJO payloads behave differently from their JSON/cross-language representation. Conditions should use the
Jackson-compatible JSON view consistently. For example, InputEvent(new Result("ok", 42)) should support both input.status and the promoted status, just as the equivalent
base Event with a Map payload does. Payload conversion should remain lazy so type-only conditions do not serialize arbitrary POJOs.

Please also keep the explicit attributes namespace faithful to the real event envelope. For an InputEvent, the expected paths should be:

attributes.input.status  // original event structure
input.status             // promoted root attribute
status                   // promoted payload field

attributes.status should not appear merely because payload fields were flattened.

The tests should be updated around this contract rather than preserving the current merge implementation:

• remove the output > root > input precedence cases that construct invalid built-in event shapes;
• cover Java POJO and JSON-shaped/base-Event parity;
• add a full EventRouter -> ActionMatcher -> ConditionEvaluator -> action test with a POJO input;
• cover multiple trigger entries and single action execution when more than one entry matches;
• rename ConditionTriggerIntegrationAgent/Test to TriggerConditionIntegrationAgent/Test.

Once this activation contract is explicit, the implementation and tests should become substantially simpler and cross-language behavior will be deterministic.

[wenjin272]

Thanks for the updates. Only three comments remain.

[wenjin272]

Some test cases are failing in the Java IT for Flink 1.20.

[rosemarYuan]

Thanks for the follow-up. I’ve addressed the remaining three comments in the latest update.
I also rechecked the latest CI run. All Java unit and integration jobs are now green, including it-java on Flink 1.20. The remaining failures are limited to Python-side compatibility, including the cross-language job: Python AgentPlan still requires actions_by_event, which Java no longer emits, and Python has not yet added CONDITION_EVALUATION_FAILURE_STRATEGY.
I’ll align these in the corresponding Python update.

[rosemarYuan]

Thanks for the follow-up. I’ve addressed the remaining three comments in the latest update.
I also rechecked the latest CI run. All Java unit and integration jobs are now green, including it-java on Flink 1.20. The remaining failures are limited to Python-side compatibility, including the cross-language job: Python AgentPlan still requires actions_by_event, which Java no longer emits, and Python has not yet added CONDITION_EVALUATION_FAILURE_STRATEGY.
I’ll align these in the corresponding Python update.