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

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

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Compile-time constants for built-in event types.
+ *
+ * <p>Usage: {@code @Action(EventType.InputEvent)}.
+ *
+ * <p>In CEL trigger conditions, {@code EventType} is exposed as a top-level map variable so {@code
+ * type == EventType.InputEvent} resolves the constant at evaluation time. {@link #allConstants()}
+ * enumerates the constant fields for that activation map and for plan-load validation.
+ */
+public final class EventType {
+
+    public static final String InputEvent = org.apache.flink.agents.api.InputEvent.EVENT_TYPE;
+    public static final String OutputEvent = org.apache.flink.agents.api.OutputEvent.EVENT_TYPE;
+    public static final String ChatRequestEvent =
+            org.apache.flink.agents.api.event.ChatRequestEvent.EVENT_TYPE;
+    public static final String ChatResponseEvent =
+            org.apache.flink.agents.api.event.ChatResponseEvent.EVENT_TYPE;
+    public static final String ToolRequestEvent =
+            org.apache.flink.agents.api.event.ToolRequestEvent.EVENT_TYPE;
+    public static final String ToolResponseEvent =
+            org.apache.flink.agents.api.event.ToolResponseEvent.EVENT_TYPE;
+    public static final String ContextRetrievalRequestEvent =
+            org.apache.flink.agents.api.event.ContextRetrievalRequestEvent.EVENT_TYPE;
+    public static final String ContextRetrievalResponseEvent =
+            org.apache.flink.agents.api.event.ContextRetrievalResponseEvent.EVENT_TYPE;
+
+    /**
+     * Returns all built-in constants as an unmodifiable {@code name → event-type value} map.
+     * Enumerated reflectively from the {@code public static final String} fields of this class so
+     * newly added constants are picked up automatically. Iteration order is unspecified.
+     */
+    public static Map<String, String> allConstants() {
+        Map<String, String> constants = new LinkedHashMap<>();
+        for (Field field : EventType.class.getFields()) {
+            if (Modifier.isStatic(field.getModifiers()) && field.getType() == String.class) {
+                try {
+                    constants.put(field.getName(), (String) field.get(null));
+                } catch (IllegalAccessException e) {
+                    // Unreachable: getFields() only returns public fields.
+                    throw new IllegalStateException("Cannot read EventType." + field.getName(), e);
+                }
+            }
+        }
+        return Collections.unmodifiableMap(constants);
+    }
+
+    private EventType() {}
+}

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

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang3.ClassUtils;
 import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.EventType;
 import org.apache.flink.agents.api.InputEvent;
 import org.apache.flink.agents.api.OutputEvent;
 import org.apache.flink.agents.api.annotation.Action;
@@ -169,7 +170,7 @@ public static void startAction(Event event, RunnerContext ctx) {
         ctx.sendEvent(new ChatRequestEvent(DEFAULT_CHAT_MODEL, inputMessages, outputSchema));
     }
 
-    @Action(listenEventTypes = {ChatResponseEvent.EVENT_TYPE})
+    @Action(EventType.ChatResponseEvent)
     public static void stopAction(Event event, RunnerContext ctx) {
         ChatResponseEvent chatResponse = ChatResponseEvent.fromEvent(event);
         ChatMessage response = chatResponse.getResponse();

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

@@ -24,23 +24,24 @@
 import java.lang.annotation.Target;
 
 /**
- * Annotation for marking a method as an agent action.
+ * Marks a method as an agent action triggered by matching events.
  *
- * <p>This annotation specifies which event types the action should respond to. The annotated method
- * will be triggered when any of the specified event types occur.
+ * <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>Events are specified as type strings via {@link #listenEventTypes()}. Use the {@code
- * EVENT_TYPE} constants on built-in event classes for standard events, or plain strings for custom
- * events.
+ * <pre>{@code
+ * // Built-in event type via the EventType constant
+ * @Action(EventType.InputEvent)
  *
- * <p>Example usage:
+ * // Equivalent via the legacy class constant
+ * @Action(InputEvent.EVENT_TYPE)
  *
- * <pre>{@code
- * @Action(listenEventTypes = {InputEvent.EVENT_TYPE})
- * public void handleInput(Event event, RunnerContext ctx) { ... }
+ * // User-defined event type
+ * @Action("MyCustomEvent")
  *
- * @Action(listenEventTypes = {InputEvent.EVENT_TYPE, "MyCustomEvent"})
- * public void handleMultiple(Event event, RunnerContext ctx) { ... }
+ * // Multiple types (OR semantics)
+ * @Action({EventType.InputEvent, "MyCustomEvent"})
  * }</pre>
  *
  * <p>For a cross-language action, set {@link #target()} to a {@link PythonFunction} with a
@@ -49,7 +50,7 @@
  *
  * <pre>{@code
  * @Action(
- *     listenEventTypes = {InputEvent.EVENT_TYPE},
+ *     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");
@@ -59,12 +60,8 @@
 @Target(ElementType.METHOD)
 @Retention(RetentionPolicy.RUNTIME)
 public @interface Action {
-    /**
-     * List of event type strings that this action should respond to.
-     *
-     * @return Array of event type strings
-     */
-    String[] listenEventTypes();
+    /** Event type name strings; multiple entries have OR semantics. */
+    String[] value();
 
     /**
      * Cross-language target. When {@link PythonFunction#module()} is non-empty, dispatch routes to

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

@@ -33,6 +33,17 @@ public class AgentConfigOptions {
     public static final ConfigOption<LoggerType> EVENT_LOGGER_TYPE =
             new ConfigOption<>("eventLoggerType", LoggerType.class, LoggerType.SLF4J);
 
+    /**
+     * Controls how the CEL condition evaluator handles runtime exceptions and non-Boolean results.
+     * Defaults to {@link CelEvaluationFailurePolicy#WARN_AND_SKIP} for streaming safety; set to
+     * {@link CelEvaluationFailurePolicy#FAIL} on strict-semantics pipelines to trigger failover.
+     */
+    public static final ConfigOption<CelEvaluationFailurePolicy> CEL_EVALUATION_FAILURE_POLICY =
+            new ConfigOption<>(
+                    "celEvaluationFailurePolicy",
+                    CelEvaluationFailurePolicy.class,
+                    CelEvaluationFailurePolicy.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/configuration/CelEvaluationFailurePolicy.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.api.configuration;
+
+/** Behaviour when a CEL trigger condition throws or returns a non-Boolean at evaluation time. */
+public enum CelEvaluationFailurePolicy {
+    /** Log WARN and treat the action as not matching. Streaming-safe default. */
+    WARN_AND_SKIP,
+    /** Rethrow as {@code IllegalStateException}; triggers Flink task failover. */
+    FAIL
+}

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

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.api;
+
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalRequestEvent;
+import org.apache.flink.agents.api.event.ContextRetrievalResponseEvent;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+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;
+
+/** Tests for {@link EventType}. */
+class EventTypeTest {
+
+    @Test
+    void allConstantsEnumeratesEveryBuiltInConstant() {
+        Map<String, String> constants = EventType.allConstants();
+        assertEquals(8, constants.size());
+        assertEquals(InputEvent.EVENT_TYPE, constants.get("InputEvent"));
+        assertEquals(OutputEvent.EVENT_TYPE, constants.get("OutputEvent"));
+        assertEquals(ChatRequestEvent.EVENT_TYPE, constants.get("ChatRequestEvent"));
+        assertEquals(ChatResponseEvent.EVENT_TYPE, constants.get("ChatResponseEvent"));
+        assertEquals(ToolRequestEvent.EVENT_TYPE, constants.get("ToolRequestEvent"));
+        assertEquals(ToolResponseEvent.EVENT_TYPE, constants.get("ToolResponseEvent"));
+        assertEquals(
+                ContextRetrievalRequestEvent.EVENT_TYPE,
+                constants.get("ContextRetrievalRequestEvent"));
+        assertEquals(
+                ContextRetrievalResponseEvent.EVENT_TYPE,
+                constants.get("ContextRetrievalResponseEvent"));
+    }
+}

dist/src/main/resources/META-INF/NOTICE

@@ -14,6 +14,8 @@ This project bundles the following dependencies under the Apache Software Licens
 - com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.18.2
 - com.fasterxml.jackson.module:jackson-module-kotlin:2.18.2
 - com.fasterxml:classmate:1.7.0
+- dev.cel:cel:0.12.0
+- dev.cel:protobuf:0.12.0
 - org.apache.logging.log4j:log4j-api:2.23.1
 - org.apache.logging.log4j:log4j-core:2.23.1
 - org.apache.logging.log4j:log4j-slf4j-impl:2.23.1
@@ -182,6 +184,8 @@ This project bundles the following dependencies under the BSD 3-Clause license.
 See bundled license files for details.
 
 - com.google.protobuf:protobuf-java:3.25.5
+- com.google.re2j:re2j:1.8
+- org.antlr:antlr4-runtime:4.13.2
 - org.ow2.asm:asm:9.3
 
 This project bundles the following dependencies under the EPL2 license.

dist/src/main/resources/META-INF/licenses/LICENSE.antlr4-runtime

@@ -0,0 +1,28 @@
+Copyright (c) 2012-2022 The ANTLR Project. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+
+1. Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright
+notice, this list of conditions and the following disclaimer in the
+documentation and/or other materials provided with the distribution.
+
+3. Neither name of copyright holders nor the names of its contributors
+may be used to endorse or promote products derived from this software
+without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR
+CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

dist/src/main/resources/META-INF/licenses/LICENSE.re2j

@@ -0,0 +1,32 @@
+This is a work derived from Russ Cox's RE2 in Go, whose license
+http://golang.org/LICENSE is as follows:
+
+Copyright (c) 2009 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+     notice, this list of conditions and the following disclaimer.
+
+   * Redistributions in binary form must reproduce the above copyright
+     notice, this list of conditions and the following disclaimer in
+     the documentation and/or other materials provided with the
+     distribution.
+
+   * Neither the name of Google Inc. nor the names of its contributors
+     may be used to endorse or promote products derived from this
+     software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.