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

api/src/main/java/org/apache/flink/agents/api/EventType.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.agents.api;
 
+import java.util.Map;
+
 /**
  * Compile-time constants for built-in event types, sourced from each {@code XxxEvent.EVENT_TYPE}.
  *
@@ -40,5 +42,23 @@ public final class EventType {
     public static final String ContextRetrievalResponseEvent =
             org.apache.flink.agents.api.event.ContextRetrievalResponseEvent.EVENT_TYPE;
 
+    private static final Map<String, String> ALL_CONSTANTS =
+            Map.of(
+                    "InputEvent", InputEvent,
+                    "OutputEvent", OutputEvent,
+                    "ChatRequestEvent", ChatRequestEvent,
+                    "ChatResponseEvent", ChatResponseEvent,
+                    "ToolRequestEvent", ToolRequestEvent,
+                    "ToolResponseEvent", ToolResponseEvent,
+                    "ContextRetrievalRequestEvent", ContextRetrievalRequestEvent,
+                    "ContextRetrievalResponseEvent", ContextRetrievalResponseEvent);
+
+    /**
+     * Returns the built-in event type constants as a name-to-value map for condition expressions.
+     */
+    public static Map<String, String> allConstants() {
+        return ALL_CONSTANTS;
+    }
+
     private EventType() {}
 }

api/src/main/java/org/apache/flink/agents/api/agents/Agent.java

@@ -29,6 +29,7 @@
 
 import java.lang.reflect.Method;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 
 /** Base class for defining agent logic. */
@@ -42,7 +43,7 @@ public Agent() {
         for (ResourceType type : ResourceType.values()) {
             this.resources.put(type, new HashMap<>());
         }
-        this.actions = new HashMap<>();
+        this.actions = new LinkedHashMap<>();
     }
 
     public Map<String, Tuple3<String[], Function, Map<String, Object>>> getActions() {
@@ -56,8 +57,8 @@ public Map<ResourceType, Map<String, Object>> getResources() {
     /**
      * Add action to agent.
      *
-     * @param triggerConditions Trigger condition strings — each is either an event-type name or a
-     *     future condition-expression form.
+     * @param triggerConditions Raw event-type names or Boolean conditions combined with OR
+     *     semantics. Shape and expression validation occur during {@code AgentPlan} construction.
      * @param method The method of this action, should be static method.
      * @param config The optional config can be used by this action.
      */
@@ -70,8 +71,8 @@ public Agent addAction(
     /**
      * Add action to agent.
      *
-     * @param triggerConditions Trigger condition strings — each is either an event-type name or a
-     *     future condition-expression form.
+     * @param triggerConditions Raw event-type names or Boolean conditions combined with OR
+     *     semantics. Shape and expression validation occur during {@code AgentPlan} construction.
      * @param method The method of this action, should be static method.
      */
     public Agent addAction(String[] triggerConditions, Method method) {
@@ -82,8 +83,8 @@ public Agent addAction(String[] triggerConditions, Method method) {
      * Add action to agent.
      *
      * @param name The action name. Must be unique within this agent.
-     * @param triggerConditions Trigger condition strings — each is either an event-type name or a
-     *     future condition-expression form.
+     * @param triggerConditions Raw event-type names or Boolean conditions combined with OR
+     *     semantics. Shape and expression validation occur during {@code AgentPlan} construction.
      * @param function The api-layer function descriptor; will be promoted to a plan-layer
      *     executable at {@code AgentPlan} construction.
      * @param config Optional config for this action.

api/src/main/java/org/apache/flink/agents/api/annotation/Action.java

@@ -24,46 +24,31 @@
 import java.lang.annotation.Target;
 
 /**
- * Marks a method as an agent action triggered by matching events.
+ * Marks a Java method as an agent action triggered by event types or condition expressions.
  *
- * <p>Each {@link #value()} entry is an event type name string. Use the {@code EVENT_TYPE} constants
- * on built-in event classes, the {@link org.apache.flink.agents.api.EventType} constants, or plain
- * strings for custom events. Multiple entries combine with OR.
+ * <p>Each {@link #value()} entry is either an exact event-type name or a condition expression that
+ * evaluates to boolean. Multiple entries use OR semantics. An entry matching the event-type syntax
+ * is classified as an event type rather than as a condition expression. To combine an event-type
+ * restriction and an attribute predicate using AND, place both in a single expression, for example
+ * {@code type == EventType.InputEvent && score > 5}.
  *
- * <pre>{@code
- * // Built-in event type via the EventType constant
- * @Action(EventType.InputEvent)
+ * <p>The event-type syntax is a bare or dotted name, such as {@code order.created} or {@code
+ * order-created}; each segment may contain hyphens. Boolean attribute conditions must therefore be
+ * explicit, for example {@code ready == true}.
  *
- * // Equivalent via the legacy class constant
- * @Action(InputEvent.EVENT_TYPE)
+ * <p>The API preserves entries as raw strings. Entries are classified and condition expressions are
+ * validated when the agent plan is built.
  *
- * // User-defined event type
- * @Action("MyCustomEvent")
- *
- * // Multiple types (OR semantics)
- * @Action({EventType.InputEvent, "MyCustomEvent"})
- * }</pre>
- *
- * <p>For a cross-language action, set {@link #target()} to a {@link PythonFunction} with a
- * non-empty {@code module}. The annotated Java body is never invoked — throw {@link
- * UnsupportedOperationException} so direct calls outside the framework fail loud:
- *
- * <pre>{@code
- * @Action(
- *     value = EventType.InputEvent,
- *     target = @PythonFunction(module = "my_pkg.handlers", qualname = "handle_input"))
- * public void handleInput(Event event, RunnerContext ctx) {
- *     throw new UnsupportedOperationException("cross-language stub");
- * }
- * }</pre>
+ * <p>Set {@link #target()} to a {@link PythonFunction} configured with a module for a
+ * cross-language action. The annotated Java method is not invoked for such actions.
  */
 @Target(ElementType.METHOD)
 @Retention(RetentionPolicy.RUNTIME)
 public @interface Action {
     /**
-     * Event type name strings; multiple entries have OR semantics. Named {@code value} (not {@code
-     * triggerConditions}) to enable the {@code @Action({...})} shorthand (JLS §9.7.3); corresponds
-     * to Python's {@code *trigger_conditions}.
+     * Raw event-type names or explicit Boolean conditions combined with OR semantics. Named {@code
+     * value} to enable the {@code @Action({...})} shorthand (JLS §9.7.3); corresponds to Python's
+     * {@code *trigger_conditions}.
      */
     String[] value();
 

api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java

@@ -25,6 +25,12 @@
 /** The set of configuration options for agents parameters. */
 public class AgentConfigOptions {
 
+    /** Behaviour when a trigger condition throws or returns a non-Boolean at evaluation time. */
+    public enum ConditionEvaluationFailureStrategy {
+        WARN_AND_SKIP,
+        FAIL
+    }
+
     /**
      * The config parameter specifies which event logger implementation to use. Defaults to {@link
      * LoggerType#SLF4J}, which surfaces events in Flink's Web UI; setting {@link LoggerType#FILE}
@@ -33,6 +39,17 @@ public class AgentConfigOptions {
     public static final ConfigOption<LoggerType> EVENT_LOGGER_TYPE =
             new ConfigOption<>("eventLoggerType", LoggerType.class, LoggerType.SLF4J);
 
+    /**
+     * Specifies how condition evaluation failures are handled. Defaults to {@code WARN_AND_SKIP};
+     * use {@code FAIL} to enforce fail-fast behavior.
+     */
+    public static final ConfigOption<ConditionEvaluationFailureStrategy>
+            CONDITION_EVALUATION_FAILURE_STRATEGY =
+                    new ConfigOption<>(
+                            "action.trigger-condition.evaluate-failure-strategy",
+                            ConditionEvaluationFailureStrategy.class,
+                            ConditionEvaluationFailureStrategy.WARN_AND_SKIP);
+
     /** The config parameter specifies the directory for the FileEvent file. */
     public static final ConfigOption<String> BASE_LOG_DIR =
             new ConfigOption<>("baseLogDir", String.class, null);

api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java

@@ -321,27 +321,31 @@ public static void loadYaml(AgentsExecutionEnvironment env, List<Path> paths) {
         for (Path path : paths) {
             LoadedFile loaded = buildAgents(path);
 
-            // Resolve shared-action string refs first so a bad ref doesn't leave partial state on
-            // the env.
             for (Map.Entry<String, Agent> entry : loaded.getAgents().entrySet()) {
                 AgentSpec spec = loaded.getAgentSpecs().get(entry.getKey());
+                List<ActionSpec> orderedActions = new ArrayList<>();
                 for (AgentActionRef ref : spec.getActions()) {
-                    if (!ref.isReference()) {
-                        continue;
+                    ActionSpec actionSpec = ref.getSpec();
+                    if (ref.isReference()) {
+                        actionSpec = loaded.getSharedActions().get(ref.getReference());
+                        if (actionSpec == null) {
+                            throw new IllegalArgumentException(
+                                    "Agent '"
+                                            + entry.getKey()
+                                            + "' references shared action '"
+                                            + ref.getReference()
+                                            + "' in "
+                                            + path
+                                            + ", but no shared action with that name is defined at"
+                                            + " the file level.");
+                        }
                     }
-                    ActionSpec shared = loaded.getSharedActions().get(ref.getReference());
-                    if (shared == null) {
-                        throw new IllegalArgumentException(
-                                "Agent '"
-                                        + entry.getKey()
-                                        + "' references shared action '"
-                                        + ref.getReference()
-                                        + "' in "
-                                        + path
-                                        + ", but no shared action with that name is defined at"
-                                        + " the file level.");
-                    }
-                    addActionToAgent(entry.getValue(), shared);
+                    orderedActions.add(actionSpec);
+                }
+
+                entry.getValue().getActions().clear();
+                for (ActionSpec actionSpec : orderedActions) {
+                    addActionToAgent(entry.getValue(), actionSpec);
                 }
             }
 
@@ -427,7 +431,10 @@ static Function resolveActionFunction(ActionSpec action) {
         return resolveFunction(action.getName(), action.getFunction(), language, paramTypes);
     }
 
-    /** Register an action on the agent with event aliases resolved. */
+    /**
+     * Registers an action, replacing trigger conditions that exactly match built-in event aliases
+     * while leaving expression conditions unchanged.
+     */
     static void addActionToAgent(Agent agent, ActionSpec action) {
         Function fn = resolveActionFunction(action);
         String[] triggerConditions =

api/src/main/java/org/apache/flink/agents/api/yaml/spec/ActionSpec.java

@@ -21,16 +21,18 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.agents.api.yaml.Language;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
 /**
  * Action referencing a user function plus its trigger conditions.
  *
- * <p>Each entry in {@code trigger_conditions} is either an event-type name (bare identifier) or a
- * future condition-expression form — the runtime classifies the string when it loads the plan.
+ * <p>Each entry in {@code trigger_conditions} is either an event-type name or a condition
+ * expression. Entries are classified and validated during {@code AgentPlan} construction.
  */
 @JsonIgnoreProperties(ignoreUnknown = false)
 public final class ActionSpec {
@@ -41,15 +43,34 @@ public final class ActionSpec {
     private final Language type;
 
     @JsonCreator
-    public ActionSpec(
+    private ActionSpec(
             @JsonProperty(value = "name", required = true) String name,
             @JsonProperty("function") String function,
             @JsonProperty(value = "trigger_conditions", required = true)
-                    List<String> triggerConditions,
+                    JsonNode triggerConditionsNode,
             @JsonProperty("config") Map<String, Object> config,
             @JsonProperty("type") Language type) {
+        this(name, function, parseTriggerConditions(name, triggerConditionsNode), config, type);
+    }
+
+    public ActionSpec(
+            String name,
+            String function,
+            List<String> triggerConditions,
+            Map<String, Object> config,
+            Language type) {
         if (triggerConditions == null || triggerConditions.isEmpty()) {
-            throw new IllegalArgumentException("trigger_conditions must not be empty");
+            throw new IllegalArgumentException(
+                    "'trigger_conditions' is required and must contain at least one entry");
+        }
+        for (int index = 0; index < triggerConditions.size(); index++) {
+            String triggerCondition = triggerConditions.get(index);
+            if (triggerCondition == null || triggerCondition.isBlank()) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "'trigger_conditions' entry #%d for action '%s' must be a non-blank string",
+                                index + 1, name));
+            }
         }
         this.name = name;
         this.function = function;
@@ -58,6 +79,32 @@ public ActionSpec(
         this.type = type;
     }
 
+    private static List<String> parseTriggerConditions(
+            String actionName, JsonNode triggerConditionsNode) {
+        if (triggerConditionsNode == null || triggerConditionsNode.isNull()) {
+            return null;
+        }
+        if (!triggerConditionsNode.isArray()) {
+            throw new IllegalArgumentException("'trigger_conditions' must be an array");
+        }
+
+        List<String> triggerConditions = new ArrayList<>(triggerConditionsNode.size());
+        for (int index = 0; index < triggerConditionsNode.size(); index++) {
+            JsonNode entry = triggerConditionsNode.get(index);
+            if (entry == null || entry.isNull()) {
+                triggerConditions.add(null);
+            } else if (entry.isTextual()) {
+                triggerConditions.add(entry.textValue());
+            } else {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "'trigger_conditions' entry #%d for action '%s' must be a string, but found %s",
+                                index + 1, actionName, entry.getNodeType()));
+            }
+        }
+        return triggerConditions;
+    }
+
     public String getName() {
         return name;
     }

api/src/test/java/org/apache/flink/agents/api/EventTypeTest.java

@@ -26,22 +26,29 @@
 import org.apache.flink.agents.api.event.ToolResponseEvent;
 import org.junit.jupiter.api.Test;
 
+import java.util.Map;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /** Tests for {@link EventType}. */
 class EventTypeTest {
 
     @Test
-    void builtInConstantsMatchEventClassConstants() {
-        assertEquals(InputEvent.EVENT_TYPE, EventType.InputEvent);
-        assertEquals(OutputEvent.EVENT_TYPE, EventType.OutputEvent);
-        assertEquals(ChatRequestEvent.EVENT_TYPE, EventType.ChatRequestEvent);
-        assertEquals(ChatResponseEvent.EVENT_TYPE, EventType.ChatResponseEvent);
-        assertEquals(ToolRequestEvent.EVENT_TYPE, EventType.ToolRequestEvent);
-        assertEquals(ToolResponseEvent.EVENT_TYPE, EventType.ToolResponseEvent);
-        assertEquals(
-                ContextRetrievalRequestEvent.EVENT_TYPE, EventType.ContextRetrievalRequestEvent);
+    void allConstantsProvidesAnUnmodifiableNameToValueMap() {
         assertEquals(
-                ContextRetrievalResponseEvent.EVENT_TYPE, EventType.ContextRetrievalResponseEvent);
+                Map.of(
+                        "InputEvent", InputEvent.EVENT_TYPE,
+                        "OutputEvent", OutputEvent.EVENT_TYPE,
+                        "ChatRequestEvent", ChatRequestEvent.EVENT_TYPE,
+                        "ChatResponseEvent", ChatResponseEvent.EVENT_TYPE,
+                        "ToolRequestEvent", ToolRequestEvent.EVENT_TYPE,
+                        "ToolResponseEvent", ToolResponseEvent.EVENT_TYPE,
+                        "ContextRetrievalRequestEvent", ContextRetrievalRequestEvent.EVENT_TYPE,
+                        "ContextRetrievalResponseEvent", ContextRetrievalResponseEvent.EVENT_TYPE),
+                EventType.allConstants());
+        assertThrows(
+                UnsupportedOperationException.class,
+                () -> EventType.allConstants().put("custom", "custom"));
     }
 }