This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 1bf3d1c1a [api][plan][java][python] Support direct declaration of
cross-language actions (#1159)
1bf3d1c1a is described below
commit 1bf3d1c1a11b8ab8d64a8f03fc2aa2b4dcb7ced5
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Thu Oct 1 14:52:59 2026 +0800
[api][plan][java][python] Support direct declaration of cross-language
actions (#1159)
Generated-by: Qoder
---
.../apache/flink/agents/api/annotation/Action.java | 25 ++--
.../agents/api/annotation/PythonFunction.java | 31 -----
.../flink/agents/api/function/JavaFunction.java | 17 ++-
.../flink/agents/api/function/PythonFunction.java | 25 +++-
.../agents/api/function/JavaFunctionTest.java | 15 +++
.../agents/api/function/PythonFunctionTest.java | 21 +++
docs/content/docs/development/workflow_agent.md | 30 ++---
.../test/JavaAgentWithPythonActionAgent.java | 20 +--
.../org/apache/flink/agents/plan/AgentPlan.java | 145 ++++++++++++++-------
.../apache/flink/agents/plan/AgentPlanTest.java | 142 +++++++++++++++-----
python/flink_agents/api/decorators.py | 106 ++++++++++-----
python/flink_agents/api/function.py | 20 ++-
python/flink_agents/api/tests/test_decorators.py | 97 +++++++-------
python/flink_agents/api/tests/test_function.py | 64 +++++++++
.../python_agent_with_java_action.py | 17 +--
python/flink_agents/plan/agent_plan.py | 45 ++++---
python/flink_agents/plan/tests/test_agent_plan.py | 68 ++++++++--
.../plan/tests/test_agent_plan_cross_language.py | 39 ++++++
18 files changed, 651 insertions(+), 276 deletions(-)
diff --git
a/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java
b/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java
index 256879bf5..57c0c0b55 100644
--- a/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java
+++ b/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java
@@ -24,7 +24,18 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
- * Marks a Java method as an agent action triggered by event types or
condition expressions.
+ * Marks a Java member as an agent action triggered by event types or
condition expressions.
+ *
+ * <p>The annotated member is always the action's real definition:
+ *
+ * <ul>
+ * <li>On a {@code static} method with the {@code (Event, RunnerContext)}
signature, the method is
+ * the native Java implementation.
+ * <li>On a {@code static final} field whose value is an api-layer {@link
+ * org.apache.flink.agents.api.function.Function} descriptor (for
example {@link
+ * org.apache.flink.agents.api.function.PythonFunction}), the field is
the cross-language
+ * target and no placeholder method body is required.
+ * </ul>
*
* <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
@@ -38,11 +49,8 @@ import java.lang.annotation.Target;
*
* <p>The API preserves entries as raw strings. Entries are classified and
condition expressions are
* validated when the agent plan is built.
- *
- * <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)
+@Target({ElementType.METHOD, ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
public @interface Action {
/**
@@ -53,9 +61,8 @@ public @interface Action {
String[] value();
/**
- * Cross-language target. When {@link PythonFunction#module()} is
non-empty, dispatch routes to
- * the Python target and the annotated Java body is unused. Default (empty
{@code module}) keeps
- * the action native Java.
+ * Optional action name override. When empty (the default), the annotated
method or field name
+ * is used as the action name.
*/
- PythonFunction target() default @PythonFunction;
+ String name() default "";
}
diff --git
a/api/src/main/java/org/apache/flink/agents/api/annotation/PythonFunction.java
b/api/src/main/java/org/apache/flink/agents/api/annotation/PythonFunction.java
deleted file mode 100644
index 424b1077f..000000000
---
a/api/src/main/java/org/apache/flink/agents/api/annotation/PythonFunction.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.annotation;
-
-/**
- * Python target descriptor used inside {@link Action#target()}. Empty {@link
#module()} = no
- * cross-language target (action stays native Java). When non-empty, the Java
method body is never
- * invoked — throw {@link UnsupportedOperationException} from the stub so
direct calls outside the
- * framework fail loud.
- */
-public @interface PythonFunction {
- String module() default "";
-
- String qualname() default "";
-}
diff --git
a/api/src/main/java/org/apache/flink/agents/api/function/JavaFunction.java
b/api/src/main/java/org/apache/flink/agents/api/function/JavaFunction.java
index 2e438d9ce..a5f348c24 100644
--- a/api/src/main/java/org/apache/flink/agents/api/function/JavaFunction.java
+++ b/api/src/main/java/org/apache/flink/agents/api/function/JavaFunction.java
@@ -56,14 +56,27 @@ public final class JavaFunction implements Function,
Serializable {
@JsonProperty(FIELD_QUAL_NAME) String qualName,
@JsonProperty(FIELD_METHOD_NAME) String methodName,
@JsonProperty(FIELD_PARAMETER_TYPES) List<String> parameterTypes) {
- this.qualName = Objects.requireNonNull(qualName, "qualName");
- this.methodName = Objects.requireNonNull(methodName, "methodName");
+ this.qualName = requireNonEmpty(qualName, FIELD_QUAL_NAME);
+ this.methodName = requireNonEmpty(methodName, FIELD_METHOD_NAME);
this.parameterTypes =
parameterTypes == null
? Collections.emptyList()
: Collections.unmodifiableList(new
ArrayList<>(parameterTypes));
}
+ /**
+ * Rejects a null or empty identifier. A descriptor names a cross-language
target, so an empty
+ * identifier can never resolve; failing here makes every construction
path ({@link
+ * #fromMethod}, {@code new}, and Jackson deserialization) reject a
partially specified target
+ * up front rather than deferring the failure to dispatch time.
+ */
+ private static String requireNonEmpty(String value, String name) {
+ if (Objects.requireNonNull(value, name).isEmpty()) {
+ throw new IllegalArgumentException(name + " must not be empty");
+ }
+ return value;
+ }
+
/**
* Build a descriptor from a reflected {@link Method}. Each parameter type
is captured via
* {@link Class#getName()} — the same form {@link Class#forName(String)}
accepts when the api
diff --git
a/api/src/main/java/org/apache/flink/agents/api/function/PythonFunction.java
b/api/src/main/java/org/apache/flink/agents/api/function/PythonFunction.java
index e249a693f..eebb9e88d 100644
--- a/api/src/main/java/org/apache/flink/agents/api/function/PythonFunction.java
+++ b/api/src/main/java/org/apache/flink/agents/api/function/PythonFunction.java
@@ -44,8 +44,29 @@ public final class PythonFunction implements Function,
Serializable {
public PythonFunction(
@JsonProperty(FIELD_MODULE) String module,
@JsonProperty(FIELD_QUAL_NAME) String qualName) {
- this.module = Objects.requireNonNull(module, "module");
- this.qualName = Objects.requireNonNull(qualName, "qualName");
+ this.module = requireNonEmpty(module, FIELD_MODULE);
+ this.qualName = requireNonEmpty(qualName, FIELD_QUAL_NAME);
+ }
+
+ /**
+ * Rejects a null or empty identifier. A descriptor names a cross-language
target, so an empty
+ * identifier can never resolve; failing here makes every construction
path ({@link #of}, {@code
+ * new}, and Jackson deserialization) reject a partially specified target
up front rather than
+ * deferring the failure to dispatch time.
+ */
+ private static String requireNonEmpty(String value, String name) {
+ if (Objects.requireNonNull(value, name).isEmpty()) {
+ throw new IllegalArgumentException(name + " must not be empty");
+ }
+ return value;
+ }
+
+ /**
+ * Convenience factory for a Python cross-language target. Reads naturally
in the
+ * annotated-field form: {@code PythonFunction.of("my_pkg.handlers",
"handle_input")}.
+ */
+ public static PythonFunction of(String module, String qualName) {
+ return new PythonFunction(module, qualName);
}
public String getModule() {
diff --git
a/api/src/test/java/org/apache/flink/agents/api/function/JavaFunctionTest.java
b/api/src/test/java/org/apache/flink/agents/api/function/JavaFunctionTest.java
index b79eaa792..c3c73cd57 100644
---
a/api/src/test/java/org/apache/flink/agents/api/function/JavaFunctionTest.java
+++
b/api/src/test/java/org/apache/flink/agents/api/function/JavaFunctionTest.java
@@ -24,6 +24,7 @@ import java.lang.reflect.Method;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
class JavaFunctionTest {
@@ -75,4 +76,18 @@ class JavaFunctionTest {
JavaFunction back = m.readValue(json, JavaFunction.class);
assertThat(back).isEqualTo(fn);
}
+
+ @Test
+ void rejectsEmptyQualName() {
+ assertThatThrownBy(() -> new JavaFunction("", "add", List.of("int")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("qualName");
+ }
+
+ @Test
+ void rejectsEmptyMethodName() {
+ assertThatThrownBy(() -> new JavaFunction("com.example.X", "",
List.of("int")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("methodName");
+ }
}
diff --git
a/api/src/test/java/org/apache/flink/agents/api/function/PythonFunctionTest.java
b/api/src/test/java/org/apache/flink/agents/api/function/PythonFunctionTest.java
index d2ae089ca..b21746bef 100644
---
a/api/src/test/java/org/apache/flink/agents/api/function/PythonFunctionTest.java
+++
b/api/src/test/java/org/apache/flink/agents/api/function/PythonFunctionTest.java
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
class PythonFunctionTest {
@@ -48,4 +49,24 @@ class PythonFunctionTest {
PythonFunction back = m.readValue(json, PythonFunction.class);
assertThat(back).isEqualTo(fn);
}
+
+ @Test
+ void rejectsEmptyModule() {
+ assertThatThrownBy(() -> new PythonFunction("", "MyClass.method"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("module");
+ }
+
+ @Test
+ void rejectsEmptyQualName() {
+ assertThatThrownBy(() -> new PythonFunction("pkg.mod", ""))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("qualName");
+ }
+
+ @Test
+ void factoryRejectsEmptyIdentifiers() {
+ assertThatThrownBy(() -> PythonFunction.of("", ""))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
}
diff --git a/docs/content/docs/development/workflow_agent.md
b/docs/content/docs/development/workflow_agent.md
index 85c3a94b2..565a35aca 100644
--- a/docs/content/docs/development/workflow_agent.md
+++ b/docs/content/docs/development/workflow_agent.md
@@ -238,7 +238,7 @@ public class ReviewAnalysisAgent extends Agent {
An action is a piece of code that can be executed. It declares one or more
trigger conditions and is
triggered when an event matches one of them.
-Use `@action(*trigger_conditions, target=None)` to decorate a Python function
or `@Action({...})` to
+Use `@action(*trigger_conditions, name=None)` to decorate a Python function or
`@Action({...})` to
annotate a Java method. The function or method accepts the triggering `Event`
and a `RunnerContext`.
It sends events through the context rather than returning a result, so declare
its return type as
`None` in Python or `void` in Java. A native Java action must be `public
static`. Python actions can
@@ -675,7 +675,7 @@ when an outer call occupies an async or interpreter worker
while waiting for ano
### Cross-language Actions
-An action declared in one language can dispatch its body to the other language
by setting a `target` on the decorator/annotation. The decorated function or
annotated method then acts as a stub — it should raise so direct calls outside
the framework fail loud.
+An action declared in one language can dispatch its body to the other
language. The declared member *is* the executable target, so no placeholder
body is needed: apply `action(...)` to a `Function` descriptor in Python, or
annotate a `static final` `Function` field in Java. The member name becomes the
action name, and can be overridden with `name=` (Python) or `name` (Java).
{{< tabs "Cross-language Actions" >}}
@@ -684,29 +684,21 @@ An action declared in one language can dispatch its body
to the other language b
from flink_agents.api.function import JavaFunction
class MyAgent(Agent):
- @action(
- InputEvent.EVENT_TYPE,
- # Action signatures are fixed (Event, RunnerContext), so for_action
- # fills the Java parameter types for you — only the class and method.
- target=JavaFunction.for_action("com.example.MyHandlers",
"handleInput"),
- )
- @staticmethod
- def handle_input(event: Event, ctx: RunnerContext) -> None:
- raise NotImplementedError("cross-language stub")
+ # Action signatures are fixed (Event, RunnerContext), so for_action
+ # fills the Java parameter types for you — only the class and method.
+ handle_input = action(InputEvent.EVENT_TYPE)(
+ JavaFunction.for_action("com.example.MyHandlers", "handleInput"))
```
{{< /tab >}}
{{< tab "Java" >}}
```java
+import org.apache.flink.agents.api.function.PythonFunction;
+
public class MyAgent extends Agent {
- @Action(
- value = EventType.InputEvent,
- target = @PythonFunction(
- module = "my_pkg.handlers",
- qualname = "handle_input"))
- public static void handleInput(Event event, RunnerContext ctx) {
- throw new UnsupportedOperationException("cross-language stub");
- }
+ @Action(EventType.InputEvent)
+ private static final PythonFunction handleInput =
+ PythonFunction.of("my_pkg.handlers", "handle_input");
}
```
{{< /tab >}}
diff --git
a/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/JavaAgentWithPythonActionAgent.java
b/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/JavaAgentWithPythonActionAgent.java
index b74ba3f60..fc1cbe4a8 100644
---
a/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/JavaAgentWithPythonActionAgent.java
+++
b/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/JavaAgentWithPythonActionAgent.java
@@ -18,6 +18,7 @@
package org.apache.flink.agents.resource.test;
import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.annotation.Action;
import org.apache.flink.agents.api.function.PythonFunction;
import org.apache.flink.api.java.functions.KeySelector;
@@ -27,16 +28,15 @@ public class JavaAgentWithPythonActionAgent extends Agent {
"flink_agents.e2e_tests.e2e_tests_resource_cross_language.python_action_handler";
public static final String PYTHON_QUALNAME = "multiply_by_two";
- public JavaAgentWithPythonActionAgent() {
- addAction(
- "multiply_by_two",
- new String[] {
- "type == EventType.InputEvent && input > 1 && input < 7",
- "type == EventType.InputEvent && input > 3 && input < 9"
- },
- new PythonFunction(PYTHON_MODULE, PYTHON_QUALNAME),
- null);
- }
+ /** Cross-language action declared on the field that is its Python target.
*/
+ @Action(
+ value = {
+ "type == EventType.InputEvent && input > 1 && input < 7",
+ "type == EventType.InputEvent && input > 3 && input < 9"
+ },
+ name = "multiply_by_two")
+ private static final PythonFunction multiplyByTwo =
+ PythonFunction.of(PYTHON_MODULE, PYTHON_QUALNAME);
public static class SingleKeySelector implements KeySelector<Long, Long> {
@Override
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
b/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
index 74fbc5274..4d99e3d32 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java
@@ -66,6 +66,7 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
+import java.lang.reflect.AnnotatedElement;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
@@ -230,6 +231,9 @@ public class AgentPlan implements Serializable {
}
private void registerAction(Action action) {
+ if (actions.containsKey(action.getName())) {
+ throw new IllegalStateException("Duplicate action name '" +
action.getName() + "'.");
+ }
actions.put(action.getName(), action);
}
@@ -239,58 +243,43 @@ public class AgentPlan implements Serializable {
addBuiltAction(ToolCallAction.getToolCallAction());
addBuiltAction(ContextRetrievalAction.getContextRetrievalAction());
- // Scan the agent class for methods annotated with @Action
Class<?> agentClass = agent.getClass();
- // getDeclaredMethods() skips inherited @Action methods; reject loudly.
- for (Class<?> parent = agentClass.getSuperclass();
- parent != null && parent != Agent.class;
- parent = parent.getSuperclass()) {
- for (Method inherited : parent.getDeclaredMethods()) {
- if (inherited.isAnnotationPresent(
- org.apache.flink.agents.api.annotation.Action.class)) {
- throw new IllegalStateException(
- "Inherited @Action '"
- + parent.getName()
- + "#"
- + inherited.getName()
- + "' is not supported; declare on the
concrete agent.");
- }
- }
- }
+ // getDeclaredMethods()/getDeclaredFields() skip inherited @Action
members; reject loudly.
+ rejectInheritedActions(agentClass);
+
+ // @Action on a method: the method body is the native Java
implementation.
for (Method method : agentClass.getDeclaredMethods()) {
- if
(!method.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class))
{
+ org.apache.flink.agents.api.annotation.Action actionAnnotation =
+
method.getAnnotation(org.apache.flink.agents.api.annotation.Action.class);
+ if (actionAnnotation == null) {
continue;
}
+ org.apache.flink.agents.plan.Function execFunction =
+ new org.apache.flink.agents.plan.JavaFunction(
+ method.getDeclaringClass(),
+ method.getName(),
+ method.getParameterTypes());
+ extractActions(
+ resolveActionName(actionAnnotation.name(),
method.getName()),
+ actionAnnotation.value(),
+ execFunction,
+ null);
+ }
+
+ // @Action on a field: the field value is the cross-language target
descriptor.
+ for (Field field : agentClass.getDeclaredFields()) {
org.apache.flink.agents.api.annotation.Action actionAnnotation =
- Objects.requireNonNull(
- method.getAnnotation(
-
org.apache.flink.agents.api.annotation.Action.class));
- String[] triggerConditions = actionAnnotation.value();
- org.apache.flink.agents.api.annotation.PythonFunction target =
- actionAnnotation.target();
- String targetModule = target.module();
- String targetQualname = target.qualname();
- boolean moduleSet = !targetModule.isEmpty();
- boolean qualnameSet = !targetQualname.isEmpty();
-
- org.apache.flink.agents.plan.Function execFunction;
- if (!moduleSet && !qualnameSet) {
- execFunction =
- new org.apache.flink.agents.plan.JavaFunction(
- method.getDeclaringClass(),
- method.getName(),
- method.getParameterTypes());
- } else if (moduleSet && qualnameSet) {
- execFunction =
- new org.apache.flink.agents.plan.PythonFunction(
- targetModule, targetQualname);
- } else {
- throw new IllegalStateException(
- "PythonFunction target on '"
- + method.getName()
- + "' must set both module and qualname");
+
field.getAnnotation(org.apache.flink.agents.api.annotation.Action.class);
+ if (actionAnnotation == null) {
+ continue;
}
- extractActions(method.getName(), triggerConditions, execFunction,
null);
+ org.apache.flink.agents.api.function.Function descriptor =
+ readActionDescriptorField(field);
+ extractActions(
+ resolveActionName(actionAnnotation.name(),
field.getName()),
+ actionAnnotation.value(),
+ toPlanFunction(descriptor),
+ null);
}
for (var action : agent.getActions().entrySet()) {
@@ -300,6 +289,70 @@ public class AgentPlan implements Serializable {
}
}
+ private static String resolveActionName(String override, String
memberName) {
+ return override == null || override.isEmpty() ? memberName : override;
+ }
+
+ /** Reject any {@code @Action} declared on a superclass; it must sit on
the concrete agent. */
+ private static void rejectInheritedActions(Class<?> agentClass) {
+ for (Class<?> parent = agentClass.getSuperclass();
+ parent != null && parent != Agent.class;
+ parent = parent.getSuperclass()) {
+ for (Method inherited : parent.getDeclaredMethods()) {
+ rejectInheritedAction(parent, inherited.getName(), inherited);
+ }
+ for (Field inherited : parent.getDeclaredFields()) {
+ rejectInheritedAction(parent, inherited.getName(), inherited);
+ }
+ }
+ }
+
+ private static void rejectInheritedAction(
+ Class<?> parent, String memberName, AnnotatedElement element) {
+ if
(element.isAnnotationPresent(org.apache.flink.agents.api.annotation.Action.class))
{
+ throw new IllegalStateException(
+ "Inherited @Action '"
+ + parent.getName()
+ + "#"
+ + memberName
+ + "' is not supported; declare on the concrete
agent.");
+ }
+ }
+
+ /**
+ * Read an {@code @Action}-annotated field as an api-layer {@link
+ * org.apache.flink.agents.api.function.Function} descriptor. The field
must be {@code static
+ * final} and hold a non-null descriptor; anything else fails with an
actionable error at plan
+ * construction. Cross-language target resolution stays deferred to the
runtime boundary that
+ * owns the real user-code classloader or Python interpreter.
+ */
+ private static org.apache.flink.agents.api.function.Function
readActionDescriptorField(
+ Field field) throws IllegalAccessException {
+ int modifiers = field.getModifiers();
+ if (!Modifier.isStatic(modifiers) || !Modifier.isFinal(modifiers)) {
+ throw new IllegalStateException(
+ "@Action field '"
+ + field.getName()
+ + "' must be declared 'static final' and hold a
Function descriptor.");
+ }
+ field.setAccessible(true);
+ Object value = field.get(null);
+ if (value == null) {
+ throw new IllegalStateException(
+ "@Action field '" + field.getName() + "' must hold a
non-null descriptor.");
+ }
+ if (!(value instanceof org.apache.flink.agents.api.function.Function))
{
+ throw new IllegalStateException(
+ "@Action field '"
+ + field.getName()
+ + "' must hold an api-layer Function descriptor
(for example "
+ + "PythonFunction or JavaFunction), but got "
+ + value.getClass().getName()
+ + ".");
+ }
+ return (org.apache.flink.agents.api.function.Function) value;
+ }
+
private static ResourceDescriptor requireResourceDescriptor(
String name, ResourceType type, Object value) {
if (!(value instanceof ResourceDescriptor)) {
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
index 3a0a2a3fa..f4e16e074 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanTest.java
@@ -294,27 +294,21 @@ public class AgentPlanTest {
}
}
- /** Cross-language action via {@code @Action(target =
@PythonFunction(...))}. */
+ /** Cross-language action via an {@code @Action}-annotated descriptor
field. */
public static class AgentWithCrossLanguageAction extends Agent {
- @org.apache.flink.agents.api.annotation.Action(
- value = EventType.InputEvent,
- target =
- @org.apache.flink.agents.api.annotation.PythonFunction(
- module = "my_pkg.handlers",
- qualname = "handle_input"))
- public static void handle(Event event, RunnerContext ctx) {
- throw new UnsupportedOperationException("cross-language stub");
- }
+ @org.apache.flink.agents.api.annotation.Action(EventType.InputEvent)
+ private static final PythonFunction handle =
+ PythonFunction.of("my_pkg.handlers", "handle_input");
}
@Test
- public void testActionWithPythonTargetCompilesToPythonFunctionExec()
throws Exception {
+ public void testFieldDescriptorCompilesToPythonFunctionExec() throws
Exception {
AgentPlan plan = new AgentPlan(new AgentWithCrossLanguageAction());
Action action = plan.getActions().get("handle");
assertThat(action).isNotNull();
assertThat(action.getExec())
- .as("non-empty target.module() must compile to a plan
PythonFunction exec")
+ .as("a PythonFunction descriptor field must compile to a plan
PythonFunction exec")
.isInstanceOf(org.apache.flink.agents.plan.PythonFunction.class);
org.apache.flink.agents.plan.PythonFunction exec =
@@ -324,7 +318,7 @@ public class AgentPlanTest {
assertThat(action.getTriggerConditions()).containsExactly(InputEvent.EVENT_TYPE);
}
- /** Plain {@code @Action} (no {@code target}) compiles to a native Java
exec. */
+ /** Plain {@code @Action} on a method compiles to a native Java exec. */
public static class AgentWithNativeJavaAction extends Agent {
@org.apache.flink.agents.api.annotation.Action(EventType.InputEvent)
public static void handle(Event event, RunnerContext ctx) {
@@ -333,52 +327,89 @@ public class AgentPlanTest {
}
@Test
- public void testActionWithEmptyTargetCompilesToJavaFunctionExec() throws
Exception {
+ public void testNativeMethodActionCompilesToJavaFunctionExec() throws
Exception {
AgentPlan plan = new AgentPlan(new AgentWithNativeJavaAction());
Action action = plan.getActions().get("handle");
assertThat(action).isNotNull();
assertThat(action.getExec())
- .as("empty target.module() must compile to a plan JavaFunction
exec")
+ .as("an @Action method must compile to a plan JavaFunction
exec")
.isInstanceOf(JavaFunction.class);
}
- /** Partially-set target (module without qualname) — must be rejected at
compile. */
- public static class AgentWithHalfSetPythonTargetMissingQualname extends
Agent {
+ /** {@code @Action(name = ...)} overrides the default member name. */
+ public static class AgentWithRenamedFieldAction extends Agent {
@org.apache.flink.agents.api.annotation.Action(
value = EventType.InputEvent,
- target =
@org.apache.flink.agents.api.annotation.PythonFunction(module = "pkg"))
- public static void handle(Event event, RunnerContext ctx) {
- throw new UnsupportedOperationException("cross-language stub");
- }
+ name = "renamed")
+ private static final PythonFunction handle =
+ PythonFunction.of("my_pkg.handlers", "handle_input");
}
@Test
- public void testActionWithPythonTargetMissingQualnameIsRejected() {
- assertThatThrownBy(() -> new AgentPlan(new
AgentWithHalfSetPythonTargetMissingQualname()))
- .isInstanceOf(IllegalStateException.class)
- .hasMessageContaining("handle")
- .hasMessageContaining("qualname");
+ public void testNameOverrideReplacesMemberName() throws Exception {
+ AgentPlan plan = new AgentPlan(new AgentWithRenamedFieldAction());
+ assertThat(plan.getActions()).containsKey("renamed");
+ assertThat(plan.getActions()).doesNotContainKey("handle");
}
- /** Partially-set target (qualname without module) — must be rejected at
compile. */
- public static class AgentWithHalfSetPythonTargetMissingModule extends
Agent {
+ /** {@code @Action(name = ...)} on a method overrides the default method
name. */
+ public static class AgentWithRenamedMethodAction extends Agent {
@org.apache.flink.agents.api.annotation.Action(
value = EventType.InputEvent,
- target =
- @org.apache.flink.agents.api.annotation.PythonFunction(
- qualname = "handle_input"))
+ name = "renamedMethod")
public static void handle(Event event, RunnerContext ctx) {
- throw new UnsupportedOperationException("cross-language stub");
+ // intentionally empty
}
}
@Test
- public void testActionWithPythonTargetMissingModuleIsRejected() {
- assertThatThrownBy(() -> new AgentPlan(new
AgentWithHalfSetPythonTargetMissingModule()))
+ public void testNameOverrideReplacesMethodName() throws Exception {
+ AgentPlan plan = new AgentPlan(new AgentWithRenamedMethodAction());
+ assertThat(plan.getActions()).containsKey("renamedMethod");
+ assertThat(plan.getActions()).doesNotContainKey("handle");
+ }
+
+ /** {@code @Action} on a non-{@code static final} field is rejected at
compile. */
+ public static class AgentWithNonFinalActionField extends Agent {
+ @org.apache.flink.agents.api.annotation.Action(EventType.InputEvent)
+ private PythonFunction handle = PythonFunction.of("my_pkg.handlers",
"handle_input");
+ }
+
+ @Test
+ public void testNonStaticFinalActionFieldIsRejected() {
+ assertThatThrownBy(() -> new AgentPlan(new
AgentWithNonFinalActionField()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("handle")
+ .hasMessageContaining("static final");
+ }
+
+ /** {@code @Action} on a field holding a non-descriptor value is rejected
at compile. */
+ public static class AgentWithNonDescriptorActionField extends Agent {
+ @org.apache.flink.agents.api.annotation.Action(EventType.InputEvent)
+ private static final String handle = "not a descriptor";
+ }
+
+ @Test
+ public void testNonDescriptorActionFieldIsRejected() {
+ assertThatThrownBy(() -> new AgentPlan(new
AgentWithNonDescriptorActionField()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("handle")
+ .hasMessageContaining("Function descriptor");
+ }
+
+ /** {@code @Action} on a field holding a null descriptor is rejected at
compile. */
+ public static class AgentWithNullActionField extends Agent {
+ @org.apache.flink.agents.api.annotation.Action(EventType.InputEvent)
+ private static final PythonFunction handle = null;
+ }
+
+ @Test
+ public void testNullActionFieldIsRejected() {
+ assertThatThrownBy(() -> new AgentPlan(new AgentWithNullActionField()))
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining("handle")
- .hasMessageContaining("module");
+ .hasMessageContaining("non-null");
}
/**
@@ -402,6 +433,47 @@ public class AgentPlanTest {
.hasMessageContaining("Inherited @Action");
}
+ /** An {@code @Action} descriptor field on a parent class is rejected too.
*/
+ public abstract static class BaseAgentWithInheritedActionField extends
Agent {
+ @org.apache.flink.agents.api.annotation.Action(EventType.InputEvent)
+ protected static final PythonFunction sharedAction =
+ PythonFunction.of("my_pkg.handlers", "handle_input");
+ }
+
+ public static class ConcreteAgentInheritingActionField
+ extends BaseAgentWithInheritedActionField {}
+
+ @Test
+ public void testActionFieldInheritedFromParentAgentClassIsRejected() {
+ assertThatThrownBy(() -> new AgentPlan(new
ConcreteAgentInheritingActionField()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("sharedAction")
+ .hasMessageContaining("BaseAgentWithInheritedActionField")
+ .hasMessageContaining("Inherited @Action");
+ }
+
+ /** Two members resolving to the same action name are rejected at compile.
*/
+ public static class AgentWithDuplicateActionName extends Agent {
+ @org.apache.flink.agents.api.annotation.Action(EventType.InputEvent)
+ public static void handle(Event event, RunnerContext ctx) {
+ // intentionally empty
+ }
+
+ @org.apache.flink.agents.api.annotation.Action(
+ value = EventType.OutputEvent,
+ name = "handle")
+ private static final PythonFunction other =
+ PythonFunction.of("my_pkg.handlers", "handle_input");
+ }
+
+ @Test
+ public void testDuplicateActionNameIsRejected() {
+ assertThatThrownBy(() -> new AgentPlan(new
AgentWithDuplicateActionName()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Duplicate action name")
+ .hasMessageContaining("handle");
+ }
+
@Test
public void testAgentPlanResourceProviders() throws Exception {
// Test that AgentPlan initializes resource providers correctly
diff --git a/python/flink_agents/api/decorators.py
b/python/flink_agents/api/decorators.py
index 14e2227fb..aaf914e5c 100644
--- a/python/flink_agents/api/decorators.py
+++ b/python/flink_agents/api/decorators.py
@@ -15,9 +15,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+from dataclasses import dataclass
from typing import Callable, Type
-from flink_agents.api.function import Function, JavaFunction, PythonFunction
+from flink_agents.api.function import Function
from flink_agents.api.tools.tool_parameter_injection import (
InjectedArg,
normalize_injected_args,
@@ -25,25 +26,49 @@ from flink_agents.api.tools.tool_parameter_injection import
(
)
-def _validate_target(target: Function, owner: str) -> None:
- """Reject targets with empty required identifiers, attributed to
``owner``."""
- if isinstance(target, PythonFunction):
- if not target.module or not target.qualname:
- msg = (
- f"PythonFunction target on '{owner}' must set both module and
qualname"
- )
- raise ValueError(msg)
- elif isinstance(target, JavaFunction):
- if not target.qualname or not target.method_name:
- msg = f"JavaFunction target on '{owner}' must set both qualname
and method_name"
- raise ValueError(msg)
+@dataclass(frozen=True)
+class ActionDeclaration:
+ """Immutable action declaration produced by applying ``action`` to a
descriptor.
+
+ Applying ``action`` to a cross-language :class:`Function` descriptor
returns
+ this wrapper instead of mutating the descriptor, so a descriptor shared by
+ multiple declarations keeps each declaration's trigger conditions and name
+ separate.
+
+ Attributes:
+ ----------
+ trigger_conditions : tuple[str, ...]
+ Raw event-type names or Boolean condition expressions.
+ func : Function
+ The api-layer executable descriptor dispatched for this action.
+ name : str | None
+ Optional action name override; ``None`` means use the attribute name.
+ """
+
+ trigger_conditions: tuple
+ func: Function
+ name: str | None = None
def action(
*trigger_conditions: str,
- target: Function | None = None,
+ name: str | None = None,
) -> Callable:
- """Decorator for marking a function as an agent action.
+ """Mark a function or a cross-language descriptor as an agent action.
+
+ Two declaration forms, mirroring Java's ``@Action`` on a method or a field:
+
+ * Native — decorate a callable; the decorated function is the action body::
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def handle(event: Event, ctx: RunnerContext) -> None: ...
+
+ * Cross-language — apply to a :class:`Function` descriptor; the decorated
+ attribute is the executable target, so no placeholder body is needed::
+
+ handle = action(EventType.InputEvent)(
+ JavaFunction.for_action("com.example.Handlers", "handle"))
Each trigger condition is an event-type name or Boolean condition
expression. Multiple conditions combine with OR semantics. To combine an
@@ -55,40 +80,51 @@ def action(
----------
trigger_conditions : str
Raw event-type names or Boolean condition expressions.
- target : Function, optional
- Cross-language function descriptor dispatched instead of the
- decorated body. The body becomes a stub — raise
- ``NotImplementedError`` so direct calls fail loud.
+ name : str, optional
+ Action name override. An empty string or ``None`` falls back to the
+ decorated member's attribute name (mirroring Java's ``@Action.name``).
Returns:
-------
Callable
- Decorator function that marks the target function with trigger
conditions.
+ Decorator that, for a callable target, returns the tagged function, and
+ for a :class:`Function` descriptor target, returns an immutable
+ :class:`ActionDeclaration`.
Raises:
------
TypeError
- If an entry is not a string, or ``target`` is not a
- :class:`Function` descriptor.
+ If a trigger condition is not a string, or the decorated object is
+ neither a callable nor an api-layer :class:`Function` descriptor.
"""
for entry in trigger_conditions:
if not isinstance(entry, str):
msg = f"action trigger condition must be a string, got {entry!r}"
raise TypeError(msg)
- if target is not None and not isinstance(target, Function):
- msg = (
- f"action(target=...) must be an api-layer Function descriptor, "
- f"got {type(target).__name__}"
- )
- raise TypeError(msg)
-
- def decorator(func: Callable) -> Callable:
- if target is not None:
- _validate_target(target, func.__qualname__)
- func._target = target
- func._trigger_conditions = trigger_conditions
- return func
+ # Mirror Java's resolveActionName: an empty-string override means "no
+ # override" (a Java annotation cannot distinguish its "" default from an
+ # explicit ""), so the action falls back to its attribute name.
+ if name == "":
+ name = None
+
+ def decorator(target: Callable | Function) -> Callable | ActionDeclaration:
+ if isinstance(target, Function):
+ return ActionDeclaration(
+ trigger_conditions=trigger_conditions,
+ func=target,
+ name=name,
+ )
+ if not callable(target):
+ msg = (
+ f"action() must decorate a callable or an api-layer Function "
+ f"descriptor, got {type(target).__name__}"
+ )
+ raise TypeError(msg)
+ target._trigger_conditions = trigger_conditions
+ if name is not None:
+ target._action_name = name
+ return target
return decorator
diff --git a/python/flink_agents/api/function.py
b/python/flink_agents/api/function.py
index d95375a40..0263d8482 100644
--- a/python/flink_agents/api/function.py
+++ b/python/flink_agents/api/function.py
@@ -27,7 +27,7 @@ import inspect
from abc import ABC
from typing import Any, Callable, List, Tuple
-from pydantic import BaseModel, model_serializer
+from pydantic import BaseModel, ValidationInfo, field_validator,
model_serializer
#: Java parameter types of an action method. Action signatures are fixed
#: ``(Event, RunnerContext)``, so callers never have to spell them out. A tuple
@@ -59,6 +59,15 @@ class PythonFunction(Function):
module: str
qualname: str
+ @field_validator("module", "qualname")
+ @classmethod
+ def _identifiers_must_not_be_empty(cls, value: str, info: ValidationInfo)
-> str:
+ """Reject an empty identifier at construction so every path fails
fast."""
+ if not value:
+ msg = f"{info.field_name} must not be empty"
+ raise ValueError(msg)
+ return value
+
@model_serializer
def __serialize(self) -> dict[str, Any]:
return {
@@ -107,6 +116,15 @@ class JavaFunction(Function):
method_name: str
parameter_types: List[str]
+ @field_validator("qualname", "method_name")
+ @classmethod
+ def _identifiers_must_not_be_empty(cls, value: str, info: ValidationInfo)
-> str:
+ """Reject an empty identifier at construction so every path fails
fast."""
+ if not value:
+ msg = f"{info.field_name} must not be empty"
+ raise ValueError(msg)
+ return value
+
@classmethod
def for_action(cls, qualname: str, method_name: str) -> "JavaFunction":
"""Build a descriptor for a Java action, filling the fixed signature.
diff --git a/python/flink_agents/api/tests/test_decorators.py
b/python/flink_agents/api/tests/test_decorators.py
index 882f5ea2b..07ff823ec 100644
--- a/python/flink_agents/api/tests/test_decorators.py
+++ b/python/flink_agents/api/tests/test_decorators.py
@@ -17,10 +17,10 @@
#################################################################################
import pytest
-from flink_agents.api.decorators import action, tool
+from flink_agents.api.decorators import ActionDeclaration, action, tool
from flink_agents.api.events.event import Event, InputEvent, OutputEvent
from flink_agents.api.events.event_type import EventType
-from flink_agents.api.function import JavaFunction, PythonFunction
+from flink_agents.api.function import JavaFunction
from flink_agents.api.runner_context import RunnerContext
from flink_agents.api.tools import InjectedArg
@@ -104,77 +104,74 @@ def _java_target() -> JavaFunction:
return JavaFunction.for_action("com.example.Handlers", "handle")
-def test_action_decorator_with_cross_language_target() -> None:
+def test_action_applied_to_descriptor_returns_immutable_declaration() -> None:
target = _java_target()
- @action(EventType.InputEvent, target=target)
- def stub(event: Event, ctx: RunnerContext) -> None:
- msg = "cross-language stub"
- raise NotImplementedError(msg)
+ declaration = action(EventType.InputEvent)(target)
- assert stub._trigger_conditions == (InputEvent.EVENT_TYPE,)
- assert stub._target is target
+ assert isinstance(declaration, ActionDeclaration)
+ assert declaration.trigger_conditions == (InputEvent.EVENT_TYPE,)
+ assert declaration.func is target
+ assert declaration.name is None
-def test_action_decorator_rejects_non_function_target() -> None:
- with pytest.raises(TypeError, match="api-layer Function descriptor"):
-
- @action(EventType.InputEvent, target="not a function") # type:
ignore[arg-type]
- def stub(event: Event, ctx: RunnerContext) -> None:
- pass
+def test_action_declaration_does_not_mutate_descriptor() -> None:
+ target = _java_target()
+ action(EventType.InputEvent)(target)
-def test_action_decorator_without_target_does_not_set_attribute() -> None:
- @action(EventType.InputEvent)
- def regular(event: Event, ctx: RunnerContext) -> None:
- pass
+ # The descriptor is pure data and must stay free of declaration metadata,
+ # so the same descriptor can back multiple independent declarations.
+ assert not hasattr(target, "_trigger_conditions")
+ assert not hasattr(target, "_action_name")
- assert not hasattr(regular, "_target")
+def test_action_declaration_keeps_shared_descriptor_declarations_distinct() ->
None:
+ shared = _java_target()
-def test_action_decorator_rejects_java_target_with_empty_qualname() -> None:
- bad = JavaFunction(qualname="", method_name="handle", parameter_types=[])
- with pytest.raises(ValueError, match="qualname"):
+ first = action(EventType.InputEvent)(shared)
+ second = action(EventType.OutputEvent, name="renamed")(shared)
- @action(EventType.InputEvent, target=bad)
- def stub(event: Event, ctx: RunnerContext) -> None:
- pass
+ assert first.trigger_conditions == (InputEvent.EVENT_TYPE,)
+ assert second.trigger_conditions == (OutputEvent.EVENT_TYPE,)
+ assert second.name == "renamed"
+ assert first.func is second.func is shared
-def test_action_decorator_rejects_java_target_with_empty_method_name() -> None:
- bad = JavaFunction(qualname="com.example.X", method_name="",
parameter_types=[])
- with pytest.raises(ValueError, match="method_name"):
+def test_action_rejects_non_callable_non_descriptor() -> None:
+ with pytest.raises(TypeError, match="callable or an api-layer Function"):
+ action(EventType.InputEvent)("not a function")
- @action(EventType.InputEvent, target=bad)
- def stub(event: Event, ctx: RunnerContext) -> None:
- pass
+def test_native_action_does_not_produce_declaration() -> None:
+ @action(EventType.InputEvent)
+ def regular(event: Event, ctx: RunnerContext) -> None:
+ pass
-def test_action_decorator_rejects_python_target_with_empty_module() -> None:
- bad = PythonFunction(module="", qualname="handle")
- with pytest.raises(ValueError, match="module"):
+ assert not isinstance(regular, ActionDeclaration)
+ assert regular._trigger_conditions == (InputEvent.EVENT_TYPE,)
+ assert not hasattr(regular, "_action_name")
- @action(EventType.InputEvent, target=bad)
- def stub(event: Event, ctx: RunnerContext) -> None:
- pass
+def test_native_action_records_name_override() -> None:
+ @action(EventType.InputEvent, name="renamed")
+ def regular(event: Event, ctx: RunnerContext) -> None:
+ pass
-def test_action_decorator_rejects_python_target_with_empty_qualname() -> None:
- bad = PythonFunction(module="pkg.mod", qualname="")
- with pytest.raises(ValueError, match="qualname"):
+ assert regular._action_name == "renamed"
- @action(EventType.InputEvent, target=bad)
- def stub(event: Event, ctx: RunnerContext) -> None:
- pass
+def test_empty_name_override_is_normalized_to_none() -> None:
+ # An empty-string override means "no override" (mirrors Java
resolveActionName):
+ # it must not be recorded on a native action nor stored on a declaration.
+ @action(EventType.InputEvent, name="")
+ def regular(event: Event, ctx: RunnerContext) -> None:
+ pass
-def test_action_decorator_target_error_names_decorated_function() -> None:
- bad = PythonFunction(module="pkg.mod", qualname="")
- with pytest.raises(ValueError, match="my_named_stub"):
+ assert not hasattr(regular, "_action_name")
- @action(EventType.InputEvent, target=bad)
- def my_named_stub(event: Event, ctx: RunnerContext) -> None:
- pass
+ declaration = action(EventType.InputEvent, name="")(_java_target())
+ assert declaration.name is None
def test_tool_decorator_supports_injected_args() -> None:
diff --git a/python/flink_agents/api/tests/test_function.py
b/python/flink_agents/api/tests/test_function.py
new file mode 100644
index 000000000..3e3ba6b05
--- /dev/null
+++ b/python/flink_agents/api/tests/test_function.py
@@ -0,0 +1,64 @@
+################################################################################
+# 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.
+#################################################################################
+"""Construction-time validation for the api-layer Function descriptors.
+
+A descriptor identifies a cross-language target by its required identifiers, so
+an empty identifier can never resolve. Rejecting it at construction makes every
+declaration path fail fast -- ``action()``, ``add_action``, direct construction
+and deserialization alike -- instead of deferring the failure to runtime.
+"""
+import pytest
+
+from flink_agents.api.function import JavaFunction, PythonFunction
+
+
+def test_python_function_accepts_valid_identifiers() -> None:
+ fn = PythonFunction(module="pkg.mod", qualname="MyClass.method")
+ assert fn.module == "pkg.mod"
+ assert fn.qualname == "MyClass.method"
+
+
+def test_python_function_rejects_empty_module() -> None:
+ with pytest.raises(ValueError, match="module"):
+ PythonFunction(module="", qualname="handle")
+
+
+def test_python_function_rejects_empty_qualname() -> None:
+ with pytest.raises(ValueError, match="qualname"):
+ PythonFunction(module="pkg.mod", qualname="")
+
+
+def test_java_function_accepts_valid_identifiers() -> None:
+ fn = JavaFunction.for_action("com.example.Handlers", "handle")
+ assert fn.qualname == "com.example.Handlers"
+ assert fn.method_name == "handle"
+
+
+def test_java_function_rejects_empty_qualname() -> None:
+ with pytest.raises(ValueError, match="qualname"):
+ JavaFunction(qualname="", method_name="handle", parameter_types=[])
+
+
+def test_java_function_rejects_empty_method_name() -> None:
+ with pytest.raises(ValueError, match="method_name"):
+ JavaFunction(qualname="com.example.X", method_name="",
parameter_types=[])
+
+
+def test_java_function_for_action_rejects_empty_qualname() -> None:
+ with pytest.raises(ValueError, match="qualname"):
+ JavaFunction.for_action("", "handle")
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_resource_cross_language/python_agent_with_java_action.py
b/python/flink_agents/e2e_tests/e2e_tests_resource_cross_language/python_agent_with_java_action.py
index 981beb487..3a9f2125c 100644
---
a/python/flink_agents/e2e_tests/e2e_tests_resource_cross_language/python_agent_with_java_action.py
+++
b/python/flink_agents/e2e_tests/e2e_tests_resource_cross_language/python_agent_with_java_action.py
@@ -32,17 +32,12 @@ JAVA_HANDLER_METHOD = "multiplyByTwo"
class PythonAgentWithJavaActionAgent(Agent):
"""Python agent whose overlapping expressions dispatch one Java action."""
- def __init__(self) -> None:
- """Create a PythonAgentWithJavaActionAgent."""
- super().__init__()
- self.add_action(
- name="multiply_by_two",
- trigger_conditions=[
- "type == EventType.InputEvent && input > 1 && input < 7",
- "type == EventType.InputEvent && input > 3 && input < 9",
- ],
- func=JavaFunction.for_action(JAVA_HANDLER_QUALNAME,
JAVA_HANDLER_METHOD),
- )
+ # Cross-language action declared on the attribute that is its Java target;
+ # the attribute name is the action name.
+ multiply_by_two = action(
+ "type == EventType.InputEvent && input > 1 && input < 7",
+ "type == EventType.InputEvent && input > 3 && input < 9",
+ )(JavaFunction.for_action(JAVA_HANDLER_QUALNAME, JAVA_HANDLER_METHOD))
class InvalidTriggerConditionAgent(Agent):
diff --git a/python/flink_agents/plan/agent_plan.py
b/python/flink_agents/plan/agent_plan.py
index a8034b168..79a2e8e89 100644
--- a/python/flink_agents/plan/agent_plan.py
+++ b/python/flink_agents/plan/agent_plan.py
@@ -20,6 +20,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, cast
from pydantic import BaseModel, field_serializer, model_validator
from flink_agents.api.agents.agent import Agent
+from flink_agents.api.decorators import ActionDeclaration
from flink_agents.api.function import Function as ApiFunction
from flink_agents.api.function import JavaFunction as ApiJavaFunction
from flink_agents.api.function import PythonFunction as ApiPythonFunction
@@ -143,7 +144,9 @@ class AgentPlan(BaseModel):
"""Build a AgentPlan from user defined agent."""
actions = {}
for action in _get_actions(agent) + BUILT_IN_ACTIONS:
- assert action.name not in actions, f"Duplicate action name:
{action.name}"
+ if action.name in actions:
+ msg = f"Duplicate action name: {action.name}"
+ raise RuntimeError(msg)
actions[action.name] = action
resource_providers = {}
@@ -196,9 +199,10 @@ class AgentPlan(BaseModel):
return self.actions[action_name].config.get(key, None)
-def _action_marker(value: Any) -> tuple | None:
- """Return ``(inner_callable, trigger_conditions, target)`` if ``value`` is
@action.
+def _native_action_marker(value: Any) -> tuple | None:
+ """Return the native ``@action`` marker as a 3-tuple, or ``None``.
+ The tuple is ``(inner_callable, trigger_conditions, name_override)``.
``@action`` may set ``_trigger_conditions`` on the outer wrapper (when
``@action``
is the outer decorator) or on ``__func__`` (when ``@staticmethod`` is outer
and ``@action`` inner). Accept either by checking both candidates.
@@ -215,7 +219,12 @@ def _action_marker(value: Any) -> tuple | None:
)
if marker is None:
return None
- return inner, marker._trigger_conditions, getattr(marker, "_target", None)
+ return inner, marker._trigger_conditions, getattr(marker, "_action_name",
None)
+
+
+def _is_action_attr(value: Any) -> bool:
+ """True if ``value`` is an @action member: a declaration or a tagged
callable."""
+ return isinstance(value, ActionDeclaration) or
_native_action_marker(value) is not None
def _get_actions(agent: Agent) -> List[Action]:
@@ -231,13 +240,13 @@ def _get_actions(agent: Agent) -> List[Action]:
List[Action]
List of Action defined in the agent.
"""
- # __dict__ skips inherited @action methods; reject loudly.
+ # __dict__ skips inherited @action members; reject loudly.
agent_class = agent.__class__
for parent in agent_class.__mro__[1:]:
if parent is Agent or parent is object:
break
for parent_name, parent_value in parent.__dict__.items():
- if _action_marker(parent_value) is not None:
+ if _is_action_attr(parent_value):
msg = (
f"Inherited @action '{parent.__qualname__}.{parent_name}'
is "
f"not supported; declare on the concrete agent."
@@ -246,19 +255,25 @@ def _get_actions(agent: Agent) -> List[Action]:
actions = []
for name, value in agent_class.__dict__.items():
- marker = _action_marker(value)
+ if isinstance(value, ActionDeclaration):
+ # Cross-language: the attribute holds an immutable declaration that
+ # wraps the executable descriptor.
+ actions.append(
+ Action(
+ name=value.name if value.name is not None else name,
+ exec=_to_plan_function(value.func),
+ trigger_conditions=list(value.trigger_conditions),
+ )
+ )
+ continue
+ marker = _native_action_marker(value)
if marker is None:
continue
- inner, trigger_conditions, target = marker
- exec_ = (
- _to_plan_function(target)
- if target is not None
- else PythonFunction.from_callable(inner)
- )
+ inner, trigger_conditions, name_override = marker
actions.append(
Action(
- name=name,
- exec=exec_,
+ name=name_override if name_override is not None else name,
+ exec=PythonFunction.from_callable(inner),
trigger_conditions=list(trigger_conditions),
)
)
diff --git a/python/flink_agents/plan/tests/test_agent_plan.py
b/python/flink_agents/plan/tests/test_agent_plan.py
index b040acdaf..ae8d1f857 100644
--- a/python/flink_agents/plan/tests/test_agent_plan.py
+++ b/python/flink_agents/plan/tests/test_agent_plan.py
@@ -183,20 +183,15 @@ _JAVA_HANDLER_QUALNAME = (
)
-class AgentWithCrossLanguageDecoratedAction(Agent):
- @action(
- EventType.InputEvent,
- target=JavaFunction.for_action(_JAVA_HANDLER_QUALNAME, "handleInput"),
+class AgentWithCrossLanguageDescriptorAction(Agent):
+ handle = action(EventType.InputEvent)(
+ JavaFunction.for_action(_JAVA_HANDLER_QUALNAME, "handleInput")
)
- @staticmethod
- def handle(event: Event, ctx: RunnerContext) -> None:
- msg = "cross-language stub"
- raise NotImplementedError(msg)
-def test_decorated_action_with_target_compiles_to_plan_java_function() -> None:
+def test_descriptor_action_compiles_to_plan_java_function() -> None:
plan = AgentPlan.from_agent(
- AgentWithCrossLanguageDecoratedAction(), AgentConfiguration()
+ AgentWithCrossLanguageDescriptorAction(), AgentConfiguration()
)
action = plan.actions["handle"]
assert action.exec.qualname == _JAVA_HANDLER_QUALNAME
@@ -204,6 +199,59 @@ def
test_decorated_action_with_target_compiles_to_plan_java_function() -> None:
assert action.trigger_conditions == [InputEvent.EVENT_TYPE]
+class _BaseAgentWithInheritedDescriptorAction(Agent):
+ """Base with a cross-language descriptor @action — verifies the inheritance
+ guard also covers ``ActionDeclaration`` members, not just native callables.
+ """
+
+ shared = action(EventType.InputEvent)(
+ JavaFunction.for_action(_JAVA_HANDLER_QUALNAME, "handleInput")
+ )
+
+
+class
_ConcreteAgentInheritingDescriptorAction(_BaseAgentWithInheritedDescriptorAction):
+ """Concrete agent inheriting the descriptor ``shared`` action."""
+
+
+def test_descriptor_action_inherited_from_parent_agent_class_is_rejected() ->
None:
+ with pytest.raises(RuntimeError, match="Inherited @action") as exc:
+ AgentPlan.from_agent(
+ _ConcreteAgentInheritingDescriptorAction(), AgentConfiguration()
+ )
+ assert "shared" in str(exc.value)
+ assert "_BaseAgentWithInheritedDescriptorAction" in str(exc.value)
+
+
+class AgentWithDuplicateActionName(Agent):
+ """Two members whose ``name`` overrides collide — must fail fast."""
+
+ first = action(EventType.InputEvent, name="dup")(
+ JavaFunction.for_action(_JAVA_HANDLER_QUALNAME, "handleInput")
+ )
+ second = action(EventType.InputEvent, name="dup")(
+ JavaFunction.for_action(_JAVA_HANDLER_QUALNAME, "handleInput")
+ )
+
+
+def test_duplicate_action_name_is_rejected() -> None:
+ with pytest.raises(RuntimeError, match="Duplicate action name"):
+ AgentPlan.from_agent(AgentWithDuplicateActionName(),
AgentConfiguration())
+
+
+class AgentWithEmptyNameOverride(Agent):
+ """An empty ``name`` override falls back to the attribute name (Java
parity)."""
+
+ handle = action(EventType.InputEvent, name="")(
+ JavaFunction.for_action(_JAVA_HANDLER_QUALNAME, "handleInput")
+ )
+
+
+def test_empty_name_override_falls_back_to_attribute_name() -> None:
+ plan = AgentPlan.from_agent(AgentWithEmptyNameOverride(),
AgentConfiguration())
+ assert "handle" in plan.actions
+ assert "" not in plan.actions
+
+
class MyEvent(Event):
"""Event for testing purposes."""
diff --git a/python/flink_agents/plan/tests/test_agent_plan_cross_language.py
b/python/flink_agents/plan/tests/test_agent_plan_cross_language.py
index 93e652811..4e27f0cfb 100644
--- a/python/flink_agents/plan/tests/test_agent_plan_cross_language.py
+++ b/python/flink_agents/plan/tests/test_agent_plan_cross_language.py
@@ -28,6 +28,7 @@ from flink_agents.api.core_options import (
AgentConfigOptions,
ConditionEvaluationFailureStrategy,
)
+from flink_agents.api.decorators import action
from flink_agents.api.events.event import Event, InputEvent
from flink_agents.api.function import (
JavaFunction as ApiJavaFunction,
@@ -119,6 +120,44 @@ def test_compile_agent_with_java_function_descriptor() ->
None:
assert action.trigger_conditions== [InputEvent.EVENT_TYPE]
+def test_compile_agent_with_descriptor_action_class_attribute() -> None:
+ """``action(...)(descriptor)`` as a class attribute becomes a plan action.
+
+ The descriptor-first declaration form mirrors Java's ``@Action`` field: the
+ attribute name is the action name and the wrapped descriptor is the
+ executable target, with no placeholder body required.
+ """
+
+ class DescriptorAgent(Agent):
+ handle = action(InputEvent.EVENT_TYPE)(
+ ApiJavaFunction.for_action("com.example.Handlers", "handle")
+ )
+
+ plan = AgentPlan.from_agent(DescriptorAgent(), AgentConfiguration())
+ compiled = plan.actions["handle"]
+
+ assert isinstance(compiled.exec, PlanJavaFunction), (
+ f"Expected plan JavaFunction, got {type(compiled.exec).__name__}"
+ )
+ assert compiled.exec.qualname == "com.example.Handlers"
+ assert compiled.exec.method_name == "handle"
+ assert compiled.trigger_conditions == [InputEvent.EVENT_TYPE]
+
+
+def test_compile_agent_with_named_descriptor_action_class_attribute() -> None:
+ """``name=`` on the descriptor form overrides the attribute name."""
+
+ class NamedDescriptorAgent(Agent):
+ handle = action(InputEvent.EVENT_TYPE, name="renamed")(
+ ApiJavaFunction.for_action("com.example.Handlers", "handle")
+ )
+
+ plan = AgentPlan.from_agent(NamedDescriptorAgent(), AgentConfiguration())
+
+ assert "renamed" in plan.actions
+ assert "handle" not in plan.actions
+
+
def test_python_plan_compile_does_not_validate_java_class_exists() -> None:
"""Python plan compile must not require the Java class to exist locally.
