You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by we...@apache.org on 2026/10/13 12:43:47 UTC
(flink-agents) branch main updated: [api][plan][runtime] Use generic resource descriptors for sub-agents (#1242)
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 5edd3d514 [api][plan][runtime] Use generic resource descriptors for sub-agents (#1242)
5edd3d514 is described below
commit 5edd3d5142f27314a2926caa44cc3585ff203c1b
Author: Wenjin Xie <16...@users.noreply.github.com>
AuthorDate: Tue Oct 13 20:43:42 2026 +0800
[api][plan][runtime] Use generic resource descriptors for sub-agents (#1242)
Generated-by: Codex CLI 0.153.4 (GPT-6)
Co-authored-by: Codex <co...@openai.com>
---
.../org/apache/flink/agents/api/agents/Agent.java | 10 +-
.../InternalSubagentResourceDescriptorBuilder.java | 82 +++++++++++++
.../flink/agents/api/subagent/SubagentSetup.java | 5 +-
...ernalSubagentResourceDescriptorBuilderTest.java | 78 ++++++++++++
.../agents/api/subagent/SubagentRegisterTest.java | 47 ++++++--
docs/content/docs/development/subagents.md | 131 +++++++++++++++++++++
.../java/agent_plan_with_internal_subagent.json | 110 +++++++++--------
.../python/agent_plan_with_internal_subagent.json | 122 +++++++++++++++++++
.../org/apache/flink/agents/plan/AgentPlan.java | 44 ++-----
.../PythonSerializableResourceProvider.java | 34 ------
.../plan/resourceprovider/ResourceProvider.java | 22 +---
.../ResourceProviderJsonDeserializer.java | 15 +--
.../serializer/ResourceProviderJsonSerializer.java | 15 +--
.../plan/subagent/InternalSubagentPlans.java | 93 +++++++++++++++
.../plan/subagent/InternalSubagentProvider.java | 70 -----------
.../agents/plan/AgentPlanCrossLanguageTest.java | 21 +++-
.../agents/plan/AgentPlanSubagentResourceTest.java | 87 ++++++++++----
.../plan/AgentPlanSubagentSerializationTest.java | 15 +--
python/flink_agents/api/agents/agent.py | 11 +-
python/flink_agents/api/subagent.py | 85 ++++++++++++-
python/flink_agents/api/tests/test_subagent.py | 57 ++++++---
.../internal_subagent_test.py | 12 +-
.../subagent_external_integration_agent.py | 18 ++-
.../subagent_integration_agent.py | 10 +-
python/flink_agents/plan/agent_plan.py | 84 +++----------
.../plan/resource/java/java_resource_adapter.py | 9 ++
python/flink_agents/plan/resource_provider.py | 28 ++---
python/flink_agents/plan/subagent.py | 65 +++++++---
.../plan/tests/test_agent_plan_cross_language.py | 61 ++++++----
.../tests/test_agent_plan_subagent_resources.py | 49 +++-----
.../plan/tests/test_resource_provider.py | 51 +++-----
python/flink_agents/runtime/internal_subagent.py | 10 +-
.../flink_agents/runtime/java_resource_adapter.py | 16 +++
python/flink_agents/runtime/resource_cache.py | 8 +-
.../tests/test_internal_subagent_descriptor.py | 108 +++++++++++++++++
.../agents/runtime/context/RunnerContextImpl.java | 2 +-
.../runtime/operator/ActionExecutionOperator.java | 36 +-----
.../runtime/operator/PythonBridgeManager.java | 20 +---
.../runtime/subagent/InternalSubagentSetup.java | 69 +++++------
.../flink/agents/runtime/ResourceCacheTest.java | 27 ++++-
.../runtime/operator/PythonBridgeManagerTest.java | 34 ++++--
.../subagent/BaseAsyncSubagentSetupTest.java | 2 +-
.../runtime/subagent/BaseSubagentSetupTest.java | 15 ++-
.../subagent/DeferredSubagentSubmitTest.java | 2 +-
.../runtime/subagent/InternalSubagentCallTest.java | 56 +++++++++
.../subagent/InternalSubagentDescriptorTest.java | 84 +++++++++++++
.../subagent/InternalSubagentGuardTest.java | 15 ++-
.../subagent/InternalSubagentMatchTest.java | 16 ++-
.../external/ExternalAgentSubagentSetupTest.java | 6 +-
49 files changed, 1461 insertions(+), 606 deletions(-)
diff --git a/api/src/main/java/org/apache/flink/agents/api/agents/Agent.java b/api/src/main/java/org/apache/flink/agents/api/agents/Agent.java
index 26603610a..89819c463 100644
--- a/api/src/main/java/org/apache/flink/agents/api/agents/Agent.java
+++ b/api/src/main/java/org/apache/flink/agents/api/agents/Agent.java
@@ -23,6 +23,7 @@ import org.apache.flink.agents.api.function.JavaFunction;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceType;
import org.apache.flink.agents.api.resource.SerializableResource;
+import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
import org.apache.flink.api.java.tuple.Tuple3;
import javax.annotation.Nullable;
@@ -137,7 +138,8 @@ public class Agent {
* @param name The name indicate the resource.
* @param type The type of the resource.
* @param instance The serializable resource object, the resource descriptor, or — for an {@code
- * AGENT} resource — an {@link Agent} to compile into an internal sub-agent.
+ * AGENT} resource — an {@link Agent} wrapped in a descriptor and compiled into an internal
+ * sub-agent during plan construction.
*/
public Agent addResource(String name, ResourceType type, Object instance) {
if (resources.get(type).containsKey(name)) {
@@ -145,12 +147,14 @@ public class Agent {
}
checkNoChatModelRouterNameClash(name, type, resources);
+ if (type == ResourceType.AGENT && instance instanceof Agent) {
+ instance =
+ new InternalSubagentResourceDescriptorBuilder().agent((Agent) instance).build();
+ }
if (instance instanceof SerializableResource) {
resources.get(type).put(name, instance);
} else if (instance instanceof ResourceDescriptor) {
resources.get(type).put(name, instance);
- } else if (instance instanceof Agent) {
- resources.get(type).put(name, instance);
} else {
throw new IllegalArgumentException(
String.format("Unsupported resource %s", instance.getClass().getName()));
diff --git a/api/src/main/java/org/apache/flink/agents/api/subagent/InternalSubagentResourceDescriptorBuilder.java b/api/src/main/java/org/apache/flink/agents/api/subagent/InternalSubagentResourceDescriptorBuilder.java
new file mode 100644
index 000000000..a192818e7
--- /dev/null
+++ b/api/src/main/java/org/apache/flink/agents/api/subagent/InternalSubagentResourceDescriptorBuilder.java
@@ -0,0 +1,82 @@
+/*
+ * 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.subagent;
+
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+
+import java.util.Objects;
+
+/** Builds an ordinary resource descriptor declaring an internal sub-agent. */
+public final class InternalSubagentResourceDescriptorBuilder {
+ public static final String SETUP_CLASS =
+ "org.apache.flink.agents.runtime.subagent.InternalSubagentSetup";
+ public static final String AGENT = "agent";
+ public static final String PLAN = "plan";
+ public static final String SCOPE = "scope";
+
+ private Agent agent;
+ private String description = "";
+ private String inputSchema;
+ private Class<?> inputType;
+
+ public InternalSubagentResourceDescriptorBuilder() {}
+
+ public InternalSubagentResourceDescriptorBuilder agent(Agent agent) {
+ this.agent = Objects.requireNonNull(agent, "agent");
+ return this;
+ }
+
+ public InternalSubagentResourceDescriptorBuilder description(String description) {
+ this.description = Objects.requireNonNull(description, "description");
+ return this;
+ }
+
+ public InternalSubagentResourceDescriptorBuilder inputSchema(String inputSchema) {
+ this.inputSchema = Objects.requireNonNull(inputSchema, "inputSchema");
+ return this;
+ }
+
+ public InternalSubagentResourceDescriptorBuilder inputType(Class<?> inputType) {
+ this.inputType = Objects.requireNonNull(inputType, "inputType");
+ return this;
+ }
+
+ public ResourceDescriptor build() {
+ Objects.requireNonNull(agent, "An internal sub-agent requires an agent");
+ if (inputType != null && inputSchema != null) {
+ throw new IllegalArgumentException("Declare either inputType or inputSchema, not both");
+ }
+ String schema = inputType == null ? inputSchema : InputSchemas.fromType(inputType);
+ if (inputType != null && schema == null) {
+ throw new IllegalArgumentException("Sub-agent inputType must describe an object");
+ }
+ if (schema != null && schema.isBlank()) {
+ throw new IllegalArgumentException("Sub-agent input schema must not be blank");
+ }
+ ResourceDescriptor.Builder builder =
+ ResourceDescriptor.Builder.newBuilder(SETUP_CLASS)
+ .addInitialArgument(AGENT, agent)
+ .addInitialArgument(SubagentSetup.FIELD_DESCRIPTION, description);
+ if (schema != null) {
+ builder.addInitialArgument(SubagentSetup.FIELD_INPUT_SCHEMA, schema);
+ }
+ return builder.build();
+ }
+}
diff --git a/api/src/main/java/org/apache/flink/agents/api/subagent/SubagentSetup.java b/api/src/main/java/org/apache/flink/agents/api/subagent/SubagentSetup.java
index 130ea0343..1ee314d7e 100644
--- a/api/src/main/java/org/apache/flink/agents/api/subagent/SubagentSetup.java
+++ b/api/src/main/java/org/apache/flink/agents/api/subagent/SubagentSetup.java
@@ -20,10 +20,10 @@ package org.apache.flink.agents.api.subagent;
import com.fasterxml.jackson.annotation.JsonIgnore;
import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.Resource;
import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceType;
-import org.apache.flink.agents.api.resource.SerializableResource;
import javax.annotation.Nullable;
@@ -33,7 +33,7 @@ import java.util.Objects;
* Caller-facing definition of a sub-agent, registered in the agent plan as an {@code AGENT}
* resource.
*/
-public abstract class SubagentSetup extends SerializableResource {
+public abstract class SubagentSetup extends Resource {
/**
* Prefix of the callable name a sub-agent is exposed to a chat model under. Tools are forbidden
@@ -89,6 +89,7 @@ public abstract class SubagentSetup extends SerializableResource {
* passed alongside the descriptor.
*/
protected SubagentSetup(ResourceDescriptor descriptor, ResourceContext resourceContext) {
+ super(descriptor, resourceContext);
this.descriptor =
Objects.requireNonNull(
descriptor,
diff --git a/api/src/test/java/org/apache/flink/agents/api/subagent/InternalSubagentResourceDescriptorBuilderTest.java b/api/src/test/java/org/apache/flink/agents/api/subagent/InternalSubagentResourceDescriptorBuilderTest.java
new file mode 100644
index 000000000..d4fc48db2
--- /dev/null
+++ b/api/src/test/java/org/apache/flink/agents/api/subagent/InternalSubagentResourceDescriptorBuilderTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.subagent;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.*;
+
+class InternalSubagentResourceDescriptorBuilderTest {
+ public static class Request {
+ public String question;
+ }
+
+ @Test
+ void buildsAnOrdinaryDescriptorWithSchemaData() throws Exception {
+ Agent agent = new Agent();
+ ResourceDescriptor descriptor =
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(agent)
+ .description("Answers questions")
+ .inputType(Request.class)
+ .build();
+ assertThat(descriptor).isExactlyInstanceOf(ResourceDescriptor.class);
+ assertThat(descriptor.getInitialArguments())
+ .containsEntry("agent", agent)
+ .containsEntry("description", "Answers questions");
+ String schema = descriptor.getArgument("input_schema");
+ assertThat(new ObjectMapper().readTree(schema).path("properties").has("question")).isTrue();
+ assertThat(descriptor.getInitialArguments()).doesNotContainKey("input_type");
+ }
+
+ @Test
+ void requiresAgentAndUnambiguousInputDeclaration() {
+ assertThatThrownBy(() -> new InternalSubagentResourceDescriptorBuilder().build())
+ .hasMessageContaining("requires an agent");
+ assertThatThrownBy(
+ () ->
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new Agent())
+ .inputType(Request.class)
+ .inputSchema("{}")
+ .build())
+ .hasMessageContaining("not both");
+ assertThatThrownBy(
+ () ->
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new Agent())
+ .inputSchema(" ")
+ .build())
+ .hasMessageContaining("must not be blank");
+ assertThatThrownBy(
+ () ->
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new Agent())
+ .inputType(Object.class)
+ .build())
+ .hasMessageContaining("must describe an object");
+ }
+}
diff --git a/api/src/test/java/org/apache/flink/agents/api/subagent/SubagentRegisterTest.java b/api/src/test/java/org/apache/flink/agents/api/subagent/SubagentRegisterTest.java
index 4b4056f3d..9fa029e06 100644
--- a/api/src/test/java/org/apache/flink/agents/api/subagent/SubagentRegisterTest.java
+++ b/api/src/test/java/org/apache/flink/agents/api/subagent/SubagentRegisterTest.java
@@ -18,7 +18,9 @@
package org.apache.flink.agents.api.subagent;
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceType;
import org.junit.jupiter.api.Test;
@@ -27,38 +29,43 @@ import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.withSettings;
/** Tests registering sub-agents as AGENT resources. */
class SubagentRegisterTest {
@Test
- void registerSubagentSetupAsResource() {
+ void registerSubagentDescriptorAsResource() {
Agent agent = new Agent();
- TestSubagentSetup setup = new TestSubagentSetup(null);
+ ResourceDescriptor setup = new TestSubagentSetup(null).getDescriptor();
agent.addResource("reviewer", ResourceType.AGENT, setup);
Map<String, Object> agentResources = agent.getResources().get(ResourceType.AGENT);
assertEquals(1, agentResources.size());
assertSame(setup, agentResources.get("reviewer"));
- assertEquals(ResourceType.AGENT, setup.getResourceType());
}
@Test
void duplicateNameThrows() {
Agent agent = new Agent();
- agent.addResource("reviewer", ResourceType.AGENT, new TestSubagentSetup(null));
+ agent.addResource(
+ "reviewer", ResourceType.AGENT, new TestSubagentSetup(null).getDescriptor());
assertThrows(
IllegalArgumentException.class,
() ->
agent.addResource(
- "reviewer", ResourceType.AGENT, new TestSubagentSetup(null)));
+ "reviewer",
+ ResourceType.AGENT,
+ new TestSubagentSetup(null).getDescriptor()));
}
@Test
void multipleSubagentsRegistered() {
Agent agent = new Agent();
- TestSubagentSetup reviewer = new TestSubagentSetup(null);
- TestSubagentSetup coder = new TestSubagentSetup(null);
+ ResourceDescriptor reviewer = new TestSubagentSetup(null).getDescriptor();
+ ResourceDescriptor coder = new TestSubagentSetup(null).getDescriptor();
agent.addResource("reviewer", ResourceType.AGENT, reviewer);
agent.addResource("coder", ResourceType.AGENT, coder);
@@ -67,4 +74,30 @@ class SubagentRegisterTest {
assertSame(reviewer, agentResources.get("reviewer"));
assertSame(coder, agentResources.get("coder"));
}
+
+ @Test
+ void rejectsSetupInstances() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new Agent()
+ .addResource(
+ "reviewer",
+ ResourceType.AGENT,
+ new TestSubagentSetup(null)));
+ }
+
+ @Test
+ void sharedRegistrationRejectsSetupInstances() {
+ AgentsExecutionEnvironment env =
+ mock(
+ AgentsExecutionEnvironment.class,
+ withSettings().useConstructor().defaultAnswer(CALLS_REAL_METHODS));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> env.addResource("reviewer", ResourceType.AGENT, new TestSubagentSetup(null)));
+ ResourceDescriptor descriptor =
+ ResourceDescriptor.Builder.newBuilder(TestSubagentSetup.class.getName()).build();
+ assertSame(env, env.addResource("reviewer", ResourceType.AGENT, descriptor));
+ }
}
diff --git a/docs/content/docs/development/subagents.md b/docs/content/docs/development/subagents.md
new file mode 100644
index 000000000..166dadc40
--- /dev/null
+++ b/docs/content/docs/development/subagents.md
@@ -0,0 +1,131 @@
+---
+title: Sub-agents
+weight: 13
+type: docs
+---
+<!--
+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.
+-->
+
+A sub-agent handles a task delegated by a parent agent. An internal sub-agent runs
+an agent workflow in the same job. An external sub-agent calls an agent service
+through a `SubagentSetup` implementation.
+
+## Declare an internal sub-agent
+
+Use `InternalSubagentResourceDescriptorBuilder` to attach a description and input
+schema to a child agent. The builder returns a normal resource descriptor for `addResource`
+or `add_resource`.
+
+```java
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
+
+parent.addResource(
+ "reviewer",
+ ResourceType.AGENT,
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new ReviewAgent())
+ .description("Review code for correctness and potential defects")
+ .inputType(ReviewRequest.class)
+ .build());
+```
+
+```python
+from pydantic import BaseModel
+
+from flink_agents.api.resource import ResourceType
+from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder
+
+class ReviewRequest(BaseModel):
+ code: str
+
+parent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(ReviewAgent())
+ .description("Review code for correctness and potential defects")
+ .input_type(ReviewRequest)
+ .build(),
+)
+```
+
+`ReviewAgent` is your child workflow; the Java `ReviewRequest` is a data class
+with fields describing its input. Alternatively, pass a JSON schema string to
+`inputSchema(...)` in Java or `input_schema(...)` in Python. Choose either an
+input type or an explicit schema. The schema describes the submitted arguments,
+which the child receives as its input event payload; it does not convert that
+payload into an instance of the input type.
+
+When a chat model is configured with `subagents=["reviewer"]`, the description
+helps it choose when to delegate, and the schema tells it which arguments to
+send. A schema is required for model-selected calls. A missing description uses
+a generic delegation description.
+
+For calls made explicitly by an action, the existing shorthand remains available:
+
+```java
+parent.addResource("reviewer", ResourceType.AGENT, new ReviewAgent());
+```
+
+```python
+parent.add_resource("reviewer", ResourceType.AGENT, ReviewAgent())
+```
+
+This shorthand does not declare an input schema. Use the builder when exposing
+the child to a model. Recursive registration, such as a child registering its
+parent, is rejected when the plan is built.
+
+## Declare an external sub-agent
+
+Implement `SubagentSetup` (or the appropriate runtime base class) and register a
+resource descriptor naming that implementation. Do not register a setup instance.
+
+```java
+parent.addResource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteReviewSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/review")
+ .addInitialArgument("description", "Review code using the remote service")
+ .addInitialArgument("input_schema", inputSchema)
+ .build());
+```
+
+The Java setup exposes a public `(ResourceDescriptor, ResourceContext)`
+constructor and implements the sub-agent invocation contract. `endpoint` above
+is configuration read by the example implementation, not a built-in transport.
+
+```python
+parent.add_resource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteReviewSetup",
+ endpoint="https://example.com/review",
+ description="Review code using the remote service",
+ input_schema=input_schema,
+ ),
+)
+```
+
+The Python setup declares these configuration fields and implements the
+invocation contract. Both languages construct the setup at runtime from the
+descriptor. Existing code that passes a setup instance to `addResource` or
+`add_resource` must instead pass a descriptor with the same configuration.
diff --git a/e2e-test/cross-language-agent-plan-snapshots/java/agent_plan_with_internal_subagent.json b/e2e-test/cross-language-agent-plan-snapshots/java/agent_plan_with_internal_subagent.json
index b184fbd30..4da83f942 100644
--- a/e2e-test/cross-language-agent-plan-snapshots/java/agent_plan_with_internal_subagent.json
+++ b/e2e-test/cross-language-agent-plan-snapshots/java/agent_plan_with_internal_subagent.json
@@ -40,61 +40,69 @@
"child" : {
"name" : "child",
"type" : "agent",
- "scope" : "child",
- "childPlan" : {
- "agent_name" : "Agent",
- "actions" : {
- "chat_model_action" : {
- "name" : "chat_model_action",
- "exec" : {
- "func_type" : "JavaFunction",
- "qualname" : "org.apache.flink.agents.plan.actions.ChatModelAction",
- "method_name" : "processChatRequestOrToolResponse",
- "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
+ "descriptor" : {
+ "language" : "java",
+ "target_module" : "",
+ "target_clazz" : "org.apache.flink.agents.runtime.subagent.InternalSubagentSetup",
+ "arguments" : {
+ "plan" : {
+ "agent_name" : "Agent",
+ "actions" : {
+ "chat_model_action" : {
+ "name" : "chat_model_action",
+ "exec" : {
+ "func_type" : "JavaFunction",
+ "qualname" : "org.apache.flink.agents.plan.actions.ChatModelAction",
+ "method_name" : "processChatRequestOrToolResponse",
+ "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
+ },
+ "trigger_conditions" : [ "_chat_request_event", "_tool_response_event" ],
+ "config" : null
+ },
+ "tool_call_action" : {
+ "name" : "tool_call_action",
+ "exec" : {
+ "func_type" : "JavaFunction",
+ "qualname" : "org.apache.flink.agents.plan.actions.ToolCallAction",
+ "method_name" : "processToolRequest",
+ "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
+ },
+ "trigger_conditions" : [ "_tool_request_event" ],
+ "config" : null
+ },
+ "context_retrieval_action" : {
+ "name" : "context_retrieval_action",
+ "exec" : {
+ "func_type" : "JavaFunction",
+ "qualname" : "org.apache.flink.agents.plan.actions.ContextRetrievalAction",
+ "method_name" : "processContextRetrievalRequest",
+ "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
+ },
+ "trigger_conditions" : [ "_context_retrieval_request_event" ],
+ "config" : null
+ },
+ "handle" : {
+ "name" : "handle",
+ "exec" : {
+ "func_type" : "JavaFunction",
+ "qualname" : "org.apache.flink.agents.plan.AgentPlanCrossLanguageTest",
+ "method_name" : "handle",
+ "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
+ },
+ "trigger_conditions" : [ "_input_event" ],
+ "config" : null
+ }
},
- "trigger_conditions" : [ "_chat_request_event", "_tool_response_event" ],
- "config" : null
+ "resource_providers" : { },
+ "config" : {
+ "conf_data" : { }
+ }
},
- "tool_call_action" : {
- "name" : "tool_call_action",
- "exec" : {
- "func_type" : "JavaFunction",
- "qualname" : "org.apache.flink.agents.plan.actions.ToolCallAction",
- "method_name" : "processToolRequest",
- "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
- },
- "trigger_conditions" : [ "_tool_request_event" ],
- "config" : null
- },
- "context_retrieval_action" : {
- "name" : "context_retrieval_action",
- "exec" : {
- "func_type" : "JavaFunction",
- "qualname" : "org.apache.flink.agents.plan.actions.ContextRetrievalAction",
- "method_name" : "processContextRetrievalRequest",
- "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
- },
- "trigger_conditions" : [ "_context_retrieval_request_event" ],
- "config" : null
- },
- "handle" : {
- "name" : "handle",
- "exec" : {
- "func_type" : "JavaFunction",
- "qualname" : "org.apache.flink.agents.plan.AgentPlanCrossLanguageTest",
- "method_name" : "handle",
- "parameter_types" : [ "org.apache.flink.agents.api.Event", "org.apache.flink.agents.api.context.RunnerContext" ]
- },
- "trigger_conditions" : [ "_input_event" ],
- "config" : null
- }
- },
- "resource_providers" : { },
- "config" : {
- "conf_data" : { }
+ "description" : "",
+ "scope" : "child"
}
},
- "__resource_provider_type__" : "InternalSubagentProvider"
+ "__resource_provider_type__" : "JavaResourceProvider"
}
}
},
diff --git a/e2e-test/cross-language-agent-plan-snapshots/python/agent_plan_with_internal_subagent.json b/e2e-test/cross-language-agent-plan-snapshots/python/agent_plan_with_internal_subagent.json
new file mode 100644
index 000000000..bec0031cb
--- /dev/null
+++ b/e2e-test/cross-language-agent-plan-snapshots/python/agent_plan_with_internal_subagent.json
@@ -0,0 +1,122 @@
+{
+ "actions": {
+ "chat_model_action": {
+ "name": "chat_model_action",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.actions.chat_model_action",
+ "qualname": "process_chat_request_or_tool_response"
+ },
+ "trigger_conditions": [
+ "_chat_request_event",
+ "_tool_response_event"
+ ],
+ "config": null
+ },
+ "tool_call_action": {
+ "name": "tool_call_action",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.actions.tool_call_action",
+ "qualname": "process_tool_request"
+ },
+ "trigger_conditions": [
+ "_tool_request_event"
+ ],
+ "config": null
+ },
+ "context_retrieval_action": {
+ "name": "context_retrieval_action",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.actions.context_retrieval_action",
+ "qualname": "process_context_retrieval_request"
+ },
+ "trigger_conditions": [
+ "_context_retrieval_request_event"
+ ],
+ "config": null
+ }
+ },
+ "resource_providers": {
+ "agent": {
+ "child": {
+ "name": "child",
+ "type": "agent",
+ "descriptor": {
+ "language": "java",
+ "target_module": "",
+ "target_clazz": "org.apache.flink.agents.runtime.subagent.InternalSubagentSetup",
+ "arguments": {
+ "description": "Python child",
+ "input_schema": "{\"type\":\"object\",\"properties\":{\"question\":{\"type\":\"string\"}}}",
+ "plan": {
+ "actions": {
+ "handle": {
+ "name": "handle",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.tests.test_agent_plan_cross_language",
+ "qualname": "_dummy_action"
+ },
+ "trigger_conditions": [
+ "_input_event"
+ ],
+ "config": null
+ },
+ "chat_model_action": {
+ "name": "chat_model_action",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.actions.chat_model_action",
+ "qualname": "process_chat_request_or_tool_response"
+ },
+ "trigger_conditions": [
+ "_chat_request_event",
+ "_tool_response_event"
+ ],
+ "config": null
+ },
+ "tool_call_action": {
+ "name": "tool_call_action",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.actions.tool_call_action",
+ "qualname": "process_tool_request"
+ },
+ "trigger_conditions": [
+ "_tool_request_event"
+ ],
+ "config": null
+ },
+ "context_retrieval_action": {
+ "name": "context_retrieval_action",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.actions.context_retrieval_action",
+ "qualname": "process_context_retrieval_request"
+ },
+ "trigger_conditions": [
+ "_context_retrieval_request_event"
+ ],
+ "config": null
+ }
+ },
+ "resource_providers": {},
+ "agent_name": "Agent",
+ "config": {
+ "conf_data": {}
+ }
+ },
+ "scope": "child"
+ }
+ },
+ "__resource_provider_type__": "JavaResourceProvider"
+ }
+ }
+ },
+ "agent_name": "Agent",
+ "config": {
+ "conf_data": {}
+ }
+}
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 806b0a159..32a135dd5 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
@@ -58,7 +58,7 @@ import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
import org.apache.flink.agents.plan.serializer.AgentPlanJsonDeserializer;
import org.apache.flink.agents.plan.serializer.AgentPlanJsonSerializer;
import org.apache.flink.agents.plan.subagent.InternalSubagentCompilationHelper;
-import org.apache.flink.agents.plan.subagent.InternalSubagentProvider;
+import org.apache.flink.agents.plan.subagent.InternalSubagentPlans;
import org.apache.flink.agents.plan.tools.FunctionTool;
import org.apache.flink.agents.plan.tools.bash.BashTool;
import org.slf4j.Logger;
@@ -663,41 +663,13 @@ public class AgentPlan implements Serializable {
} else if (type == ResourceType.AGENT) {
for (Map.Entry<String, Object> kv : entry.getValue().entrySet()) {
String name = kv.getKey();
- Object value = kv.getValue();
- if (value instanceof SubagentSetup) {
- // A live setup registered programmatically travels as its descriptor, so a
- // remote task rebuilds it through the (ResourceDescriptor, ResourceContext)
- // constructor.
- addResourceProvider(
- createDescriptorResourceProvider(
- name,
- ResourceType.AGENT,
- ((SubagentSetup) value).getDescriptor()));
- } else if (value instanceof ResourceDescriptor) {
- // Declared via YAML: the descriptor names a SubagentSetup subclass that is
- // instantiated when the resource is first resolved.
- addResourceProvider(
- createDescriptorResourceProvider(
- name, ResourceType.AGENT, (ResourceDescriptor) value));
- } else if (value instanceof Agent) {
- // Compile a directly-registered child Agent into an internal sub-agent.
- // The child's compiled plan is reused across names and cycles are
- // rejected by InternalSubagentCompilationHelper. The resource name
- // doubles as the sub-agent scope used for runtime resolution.
- Agent child = (Agent) value;
- AgentPlan childPlan =
- InternalSubagentCompilationHelper.getOrCompile(
- child, name, a -> new AgentPlan(a, this.config));
- addResourceProvider(new InternalSubagentProvider(name, childPlan));
- } else {
- throw new IllegalArgumentException(
- "AGENT resource '"
- + name
- + "' must be a SubagentSetup, a ResourceDescriptor, or an"
- + " Agent, but got "
- + value.getClass().getName()
- + ".");
- }
+ ResourceDescriptor descriptor =
+ requireResourceDescriptor(name, type, kv.getValue());
+ descriptor =
+ InternalSubagentPlans.compile(
+ name, descriptor, child -> new AgentPlan(child, this.config));
+ addResourceProvider(
+ createDescriptorResourceProvider(name, ResourceType.AGENT, descriptor));
}
} else {
for (Map.Entry<String, Object> kv : entry.getValue().entrySet()) {
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonSerializableResourceProvider.java b/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonSerializableResourceProvider.java
index 8848f1dae..793ec04d7 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonSerializableResourceProvider.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/PythonSerializableResourceProvider.java
@@ -18,12 +18,10 @@
package org.apache.flink.agents.plan.resourceprovider;
-import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.agents.api.resource.Resource;
import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceType;
import org.apache.flink.agents.api.resource.SerializableResource;
-import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.plan.resource.python.PythonPrompt;
import org.apache.flink.agents.plan.resource.python.PythonTool;
@@ -37,15 +35,6 @@ import java.util.Objects;
* resource for later deserialization.
*/
public class PythonSerializableResourceProvider extends SerializableResourceProvider {
- /**
- * Runtime class materialized for Python-compiled internal sub-agents; referenced by name to
- * keep the {@code runtime -> plan} dependency direction.
- */
- public static final String INTERNAL_SETUP_RUNTIME_CLASS =
- "org.apache.flink.agents.runtime.subagent.InternalSubagentSetup";
-
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
private final Map<String, Object> serialized;
private SerializableResource resource;
@@ -85,29 +74,6 @@ public class PythonSerializableResourceProvider extends SerializableResourceProv
resource = PythonPrompt.fromSerializedMap(serialized);
} else if (this.getType() == ResourceType.TOOL) {
resource = PythonTool.fromSerializedMap(serialized);
- } else if (this.getType() == ResourceType.AGENT) {
- // A Python-compiled internal sub-agent: materialize the Java runtime setup so the
- // operator dispatches the child plan exactly as a Java-compiled one. The
- // serialized map uses snake_case, per the cross-language AgentPlan JSON
- // convention.
- Object childPlanNode = serialized.get("child_plan");
- if (childPlanNode == null) {
- throw new IllegalStateException(
- "A Python-defined external sub-agent setup is Python-owned and must"
- + " be materialized by the Python runtime, provider: "
- + getName());
- }
- AgentPlan childPlan = OBJECT_MAPPER.convertValue(childPlanNode, AgentPlan.class);
- String scope = (String) serialized.get("scope");
- Class<?> clazz =
- Class.forName(
- INTERNAL_SETUP_RUNTIME_CLASS,
- true,
- Thread.currentThread().getContextClassLoader());
- resource =
- (SerializableResource)
- clazz.getConstructor(String.class, AgentPlan.class)
- .newInstance(scope, childPlan);
} else {
throw new UnsupportedOperationException(
"Unsupported resource type: " + this.getType());
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/ResourceProvider.java b/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/ResourceProvider.java
index ad0defc42..e8262cde2 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/ResourceProvider.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/resourceprovider/ResourceProvider.java
@@ -62,26 +62,10 @@ public abstract class ResourceProvider implements java.io.Serializable {
return type;
}
- /**
- * Whether the given provider materializes a resource owned by the Python runtime, so the
- * runtime must ask that runtime to build it instead of resolving it on the Java side.
- *
- * <p>A Python-compiled internal sub-agent carries its child plan in the serialized payload and
- * executes natively on the Java side, so it stays Java-owned; only external Python setups
- * (without a child plan) remain Python-owned.
- */
+ /** Whether this resource is constructed by the Python runtime. */
public static boolean isPythonOwned(ResourceProvider provider) {
- if (provider instanceof PythonResourceProvider) {
- return true;
- }
- if (provider instanceof PythonSerializableResourceProvider) {
- PythonSerializableResourceProvider serializable =
- (PythonSerializableResourceProvider) provider;
- return !(serializable.getType() == ResourceType.AGENT
- && serializable.getSerialized() != null
- && serializable.getSerialized().containsKey("child_plan"));
- }
- return false;
+ return provider instanceof PythonResourceProvider
+ || provider instanceof PythonSerializableResourceProvider;
}
/**
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonDeserializer.java b/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonDeserializer.java
index 710022fcb..1aa71ec14 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonDeserializer.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonDeserializer.java
@@ -26,13 +26,11 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceType;
-import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.JavaSerializableResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.PythonResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.PythonSerializableResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
-import org.apache.flink.agents.plan.subagent.InternalSubagentProvider;
import java.io.IOException;
import java.util.HashMap;
@@ -68,8 +66,7 @@ public class ResourceProviderJsonDeserializer extends StdDeserializer<ResourcePr
return deserializeJavaResourceProvider(node);
} else if (JavaSerializableResourceProvider.class.getSimpleName().equals(providerType)) {
return deserializeJavaSerializableResourceProvider(node);
- } else if (InternalSubagentProvider.class.getSimpleName().equals(providerType)) {
- return deserializeInternalSubagentProvider(node);
+
} else {
throw new IOException("Unsupported resource provider type: " + providerType);
}
@@ -125,14 +122,4 @@ public class ResourceProviderJsonDeserializer extends StdDeserializer<ResourcePr
return new JavaSerializableResourceProvider(
name, ResourceType.fromValue(type), module, clazz, serializedResource);
}
-
- private InternalSubagentProvider deserializeInternalSubagentProvider(JsonNode node) {
- String name = node.get("name").asText();
- try {
- AgentPlan childPlan = mapper.treeToValue(node.get("childPlan"), AgentPlan.class);
- return new InternalSubagentProvider(name, childPlan);
- } catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- }
}
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonSerializer.java b/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonSerializer.java
index 86a470ab3..5e33893f3 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonSerializer.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/serializer/ResourceProviderJsonSerializer.java
@@ -26,7 +26,6 @@ import org.apache.flink.agents.plan.resourceprovider.JavaSerializableResourcePro
import org.apache.flink.agents.plan.resourceprovider.PythonResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.PythonSerializableResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
-import org.apache.flink.agents.plan.subagent.InternalSubagentProvider;
import java.io.IOException;
@@ -58,9 +57,7 @@ public class ResourceProviderJsonSerializer extends StdSerializer<ResourceProvid
} else if (resourceProvider instanceof JavaSerializableResourceProvider) {
serializeJavaSerializableResourceProvider(
jsonGenerator, (JavaSerializableResourceProvider) resourceProvider);
- } else if (resourceProvider instanceof InternalSubagentProvider) {
- serializeInternalSubagentProvider(
- jsonGenerator, (InternalSubagentProvider) resourceProvider);
+
} else {
throw new IllegalArgumentException(
"Unsupported resource provider type: " + resourceProvider.getClass().getName());
@@ -122,14 +119,4 @@ public class ResourceProviderJsonSerializer extends StdSerializer<ResourceProvid
gen.writeStringField(
"__resource_provider_type__", JavaResourceProvider.class.getSimpleName());
}
-
- private void serializeInternalSubagentProvider(
- JsonGenerator gen, InternalSubagentProvider provider) throws IOException {
- gen.writeStringField("name", provider.getName());
- gen.writeStringField("type", provider.getType().getValue());
- gen.writeStringField("scope", provider.getScope());
- gen.writeObjectField("childPlan", provider.getChildPlan());
- gen.writeStringField(
- "__resource_provider_type__", InternalSubagentProvider.class.getSimpleName());
- }
}
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/subagent/InternalSubagentPlans.java b/plan/src/main/java/org/apache/flink/agents/plan/subagent/InternalSubagentPlans.java
new file mode 100644
index 000000000..bc4f514b0
--- /dev/null
+++ b/plan/src/main/java/org/apache/flink/agents/plan/subagent/InternalSubagentPlans.java
@@ -0,0 +1,93 @@
+/*
+ * 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.plan.subagent;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
+import org.apache.flink.agents.plan.AgentPlan;
+import org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider;
+import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/** Compilation and traversal of child plans carried by ordinary resource descriptors. */
+public final class InternalSubagentPlans {
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private InternalSubagentPlans() {}
+
+ public static boolean isInternal(ResourceDescriptor descriptor) {
+ return "java".equals(descriptor.getLanguage())
+ && InternalSubagentResourceDescriptorBuilder.SETUP_CLASS.equals(
+ descriptor.getClazz());
+ }
+
+ public static ResourceDescriptor compile(
+ String name,
+ ResourceDescriptor descriptor,
+ InternalSubagentCompilationHelper.Compiler compiler)
+ throws Exception {
+ if (!isInternal(descriptor)) {
+ return descriptor;
+ }
+ Map<String, Object> args =
+ descriptor.getInitialArguments() == null
+ ? new HashMap<>()
+ : new HashMap<>(descriptor.getInitialArguments());
+ Object agent = args.remove(InternalSubagentResourceDescriptorBuilder.AGENT);
+ if (!(agent instanceof Agent)
+ || args.containsKey(InternalSubagentResourceDescriptorBuilder.PLAN)) {
+ throw new IllegalArgumentException(
+ "Internal sub-agent '"
+ + name
+ + "' requires an Agent declaration, not a compiled plan");
+ }
+ AgentPlan child =
+ InternalSubagentCompilationHelper.getOrCompile((Agent) agent, name, compiler);
+ args.put(
+ InternalSubagentResourceDescriptorBuilder.PLAN,
+ MAPPER.convertValue(child, Map.class));
+ args.put(InternalSubagentResourceDescriptorBuilder.SCOPE, name);
+ return new ResourceDescriptor(InternalSubagentResourceDescriptorBuilder.SETUP_CLASS, args);
+ }
+
+ public static AgentPlan decode(ResourceDescriptor descriptor) {
+ Object data = descriptor.getArgument(InternalSubagentResourceDescriptorBuilder.PLAN);
+ if (!(data instanceof Map)) {
+ throw new IllegalArgumentException(
+ "Internal sub-agent requires a compiled plan object");
+ }
+ return MAPPER.convertValue(data, AgentPlan.class);
+ }
+
+ public static Optional<AgentPlan> plan(ResourceProvider provider) {
+ if (provider.getType() == ResourceType.AGENT && provider instanceof JavaResourceProvider) {
+ ResourceDescriptor descriptor = ((JavaResourceProvider) provider).getDescriptor();
+ if (isInternal(descriptor)) {
+ return Optional.of(decode(descriptor));
+ }
+ }
+ return Optional.empty();
+ }
+}
diff --git a/plan/src/main/java/org/apache/flink/agents/plan/subagent/InternalSubagentProvider.java b/plan/src/main/java/org/apache/flink/agents/plan/subagent/InternalSubagentProvider.java
deleted file mode 100644
index 0a900dab3..000000000
--- a/plan/src/main/java/org/apache/flink/agents/plan/subagent/InternalSubagentProvider.java
+++ /dev/null
@@ -1,70 +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.plan.subagent;
-
-import org.apache.flink.agents.api.resource.Resource;
-import org.apache.flink.agents.api.resource.ResourceContext;
-import org.apache.flink.agents.api.resource.ResourceType;
-import org.apache.flink.agents.plan.AgentPlan;
-import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
-
-/**
- * Plan-side descriptor of an internal sub-agent: the compiled child plan plus its scope name,
- * produced during {@link AgentPlan} compilation when an {@code Agent} is registered as an {@code
- * AGENT} resource. Carries only plan content; all runtime behaviour (call statuses, child caches,
- * bootstrap/await) lives in the runtime class it materializes.
- *
- * <p>Materialization follows route A: the runtime class is referenced by name and instantiated
- * reflectively at {@link #provide}, preserving the {@code runtime -> plan} dependency direction.
- * The eagerly materialized instance implements the runtime's {@code TaskLifecycleListener}, so the
- * operator registers it for lifecycle broadcast through its existing instanceof discovery.
- */
-public class InternalSubagentProvider extends ResourceProvider {
-
- /** Runtime class materialized for this descriptor; referenced by name, not by dependency. */
- public static final String RUNTIME_SETUP_CLASS =
- "org.apache.flink.agents.runtime.subagent.InternalSubagentSetup";
-
- private final String scope;
-
- private final AgentPlan childPlan;
-
- public InternalSubagentProvider(String name, AgentPlan childPlan) {
- super(name, ResourceType.AGENT);
- this.scope = name;
- this.childPlan = childPlan;
- }
-
- public String getScope() {
- return scope;
- }
-
- public AgentPlan getChildPlan() {
- return childPlan;
- }
-
- @Override
- public Resource provide(ResourceContext resourceContext) throws Exception {
- Class<?> clazz =
- Class.forName(
- RUNTIME_SETUP_CLASS, true, Thread.currentThread().getContextClassLoader());
- return (Resource)
- clazz.getConstructor(String.class, AgentPlan.class).newInstance(scope, childPlan);
- }
-}
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanCrossLanguageTest.java b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanCrossLanguageTest.java
index ff6135f7c..f01a99944 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanCrossLanguageTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanCrossLanguageTest.java
@@ -317,7 +317,7 @@ class AgentPlanCrossLanguageTest {
assertThat(actual).isEqualTo(expected);
}
- /** A plan that nests a child agent compiled into an {@code InternalSubagentProvider}. */
+ /** A plan that nests a child agent carried by an ordinary resource descriptor. */
private static AgentPlan compileWithInternalSubagent() throws Exception {
Agent child = new Agent();
child.addAction(
@@ -380,4 +380,23 @@ class AgentPlanCrossLanguageTest {
.get(AgentConfigOptions.CONDITION_EVALUATION_FAILURE_STRATEGY))
.isEqualTo(ConditionEvaluationFailureStrategy.FAIL);
}
+
+ @Test
+ void javaCanReadAndRewritePythonInternalSubagentDescriptor() throws Exception {
+ Path fixture = snapshotDir.resolve("python/agent_plan_with_internal_subagent.json");
+ AgentPlan plan = MAPPER.readValue(Files.readString(fixture), AgentPlan.class);
+ AgentPlan restored = MAPPER.readValue(MAPPER.writeValueAsString(plan), AgentPlan.class);
+ org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider provider =
+ (org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider)
+ restored.getResourceProviders().get(ResourceType.AGENT).get("child");
+ assertThat(provider.getDescriptor().getInitialArguments())
+ .containsEntry("description", "Python child")
+ .containsEntry("scope", "child")
+ .doesNotContainKey("agent");
+ AgentPlan child =
+ org.apache.flink.agents.plan.subagent.InternalSubagentPlans.decode(
+ provider.getDescriptor());
+ assertThat(child.getActions()).containsKey("handle");
+ assertThat(child.getActions().get("handle").getExec()).isInstanceOf(PythonFunction.class);
+ }
}
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentResourceTest.java b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentResourceTest.java
index 90a155d33..d61cdba0f 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentResourceTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentResourceTest.java
@@ -23,10 +23,12 @@ import org.apache.flink.agents.api.agents.Agent;
import org.apache.flink.agents.api.resource.Resource;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
import org.apache.flink.agents.api.subagent.SubagentSetup;
import org.apache.flink.agents.api.subagent.TestSubagentSetup;
+import org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
-import org.apache.flink.agents.plan.subagent.InternalSubagentProvider;
+import org.apache.flink.agents.plan.subagent.InternalSubagentPlans;
import org.apache.flink.agents.plan.tools.bash.BashTool;
import org.junit.jupiter.api.Test;
@@ -42,9 +44,10 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class AgentPlanSubagentResourceTest {
@Test
- void subagentSetupInstanceCompilesIntoAgentProvider() throws Exception {
+ void externalDescriptorCompilesIntoAgentProvider() throws Exception {
Agent agent = new Agent();
- agent.addResource("reviewer", ResourceType.AGENT, new TestSubagentSetup(null));
+ agent.addResource(
+ "reviewer", ResourceType.AGENT, new TestSubagentSetup(null).getDescriptor());
AgentPlan plan = new AgentPlan(agent);
@@ -84,8 +87,8 @@ public class AgentPlanSubagentResourceTest {
agent.getResources().get(ResourceType.AGENT).put("bad", "not-a-subagent");
assertThatThrownBy(() -> new AgentPlan(agent))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("must be a SubagentSetup, a ResourceDescriptor, or an Agent");
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("must be a ResourceDescriptor");
}
@Test
@@ -99,16 +102,16 @@ public class AgentPlanSubagentResourceTest {
Map<String, ResourceProvider> agentProviders =
plan.getResourceProviders().get(ResourceType.AGENT);
assertThat(agentProviders).containsKey("child");
- assertThat(agentProviders.get("child")).isInstanceOf(InternalSubagentProvider.class);
- InternalSubagentProvider provider = (InternalSubagentProvider) agentProviders.get("child");
- assertThat(provider.getScope()).isEqualTo("child");
- assertThat(provider.getChildPlan()).isNotNull();
+ assertThat(agentProviders.get("child")).isInstanceOf(JavaResourceProvider.class);
+ JavaResourceProvider provider = (JavaResourceProvider) agentProviders.get("child");
+ assertThat(provider.getDescriptor().getArgument("scope", "")).isEqualTo("child");
+ assertThat(InternalSubagentPlans.decode(provider.getDescriptor())).isNotNull();
}
/**
* Serializing the provider must close its nested child plan before writing the provider's own
* fields, so the type marker stays at the provider level and the plan deserializes back to an
- * equivalent {@link InternalSubagentProvider}.
+ * equivalent {@link JavaResourceProvider}.
*/
@Test
void internalSubagentProviderSurvivesJsonRoundTrip() throws Exception {
@@ -116,8 +119,8 @@ public class AgentPlanSubagentResourceTest {
Agent child = new Agent();
root.addResource("child", ResourceType.AGENT, child);
AgentPlan plan = new AgentPlan(root);
- InternalSubagentProvider original =
- (InternalSubagentProvider)
+ JavaResourceProvider original =
+ (JavaResourceProvider)
plan.getResourceProviders().get(ResourceType.AGENT).get("child");
ObjectMapper mapper = new ObjectMapper();
@@ -125,12 +128,12 @@ public class AgentPlanSubagentResourceTest {
ResourceProvider provider =
deserialized.getResourceProviders().get(ResourceType.AGENT).get("child");
- assertThat(provider).isInstanceOf(InternalSubagentProvider.class);
- InternalSubagentProvider roundTripped = (InternalSubagentProvider) provider;
- assertThat(roundTripped.getScope()).isEqualTo("child");
- assertThat(roundTripped.getChildPlan()).isNotNull();
- assertThat(roundTripped.getChildPlan().getAgentName())
- .isEqualTo(original.getChildPlan().getAgentName());
+ assertThat(provider).isInstanceOf(JavaResourceProvider.class);
+ JavaResourceProvider roundTripped = (JavaResourceProvider) provider;
+ assertThat(roundTripped.getDescriptor().getArgument("scope", "")).isEqualTo("child");
+ assertThat(InternalSubagentPlans.decode(roundTripped.getDescriptor())).isNotNull();
+ assertThat(InternalSubagentPlans.decode(roundTripped.getDescriptor()).getAgentName())
+ .isEqualTo(InternalSubagentPlans.decode(original.getDescriptor()).getAgentName());
}
@Test
@@ -144,9 +147,12 @@ public class AgentPlanSubagentResourceTest {
Map<String, ResourceProvider> agentProviders =
plan.getResourceProviders().get(ResourceType.AGENT);
- InternalSubagentProvider first = (InternalSubagentProvider) agentProviders.get("first");
- InternalSubagentProvider second = (InternalSubagentProvider) agentProviders.get("second");
- assertThat(first.getChildPlan()).isSameAs(second.getChildPlan());
+ JavaResourceProvider first = (JavaResourceProvider) agentProviders.get("first");
+ JavaResourceProvider second = (JavaResourceProvider) agentProviders.get("second");
+ assertThat(first.getDescriptor().getInitialArguments().get("plan"))
+ .isEqualTo(second.getDescriptor().getInitialArguments().get("plan"));
+ assertThat(first.getDescriptor().getArgument("scope", "")).isEqualTo("first");
+ assertThat(second.getDescriptor().getArgument("scope", "")).isEqualTo("second");
}
@Test
@@ -205,4 +211,43 @@ public class AgentPlanSubagentResourceTest {
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("must not start with the reserved prefix '_subagent_'");
}
+
+ @Test
+ void descriptorMetadataSurvivesCompilationAndJsonWithoutMutatingDeclaration() throws Exception {
+ Agent child = new Agent();
+ ResourceDescriptor declaration =
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(child)
+ .description("Reviews code")
+ .inputSchema("{\"type\":\"object\"}")
+ .build();
+ Agent root = new Agent();
+ root.addResource("reviewer", ResourceType.AGENT, declaration);
+ root.addResource("second", ResourceType.AGENT, declaration);
+ ObjectMapper mapper = new ObjectMapper();
+ AgentPlan restored =
+ mapper.readValue(mapper.writeValueAsString(new AgentPlan(root)), AgentPlan.class);
+ JavaResourceProvider provider =
+ (JavaResourceProvider)
+ restored.getResourceProviders().get(ResourceType.AGENT).get("reviewer");
+ assertThat(provider.getDescriptor().getInitialArguments())
+ .containsEntry("description", "Reviews code")
+ .containsEntry("input_schema", "{\"type\":\"object\"}")
+ .containsEntry("scope", "reviewer")
+ .doesNotContainKey("agent");
+ assertThat(InternalSubagentPlans.plan(provider)).isPresent();
+ assertThat(declaration.getInitialArguments())
+ .containsEntry("agent", child)
+ .doesNotContainKeys("scope", "plan");
+ }
+
+ @Test
+ void descriptorCycleThroughRootIsRejected() {
+ Agent root = new Agent();
+ root.addResource(
+ "self",
+ ResourceType.AGENT,
+ new InternalSubagentResourceDescriptorBuilder().agent(root).build());
+ assertThatThrownBy(() -> new AgentPlan(root)).hasMessageContaining("Cyclic sub-agent");
+ }
}
diff --git a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentSerializationTest.java b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentSerializationTest.java
index 6665a59f3..20d062269 100644
--- a/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentSerializationTest.java
+++ b/plan/src/test/java/org/apache/flink/agents/plan/AgentPlanSubagentSerializationTest.java
@@ -38,22 +38,19 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
-/**
- * An external sub-agent registered programmatically as a live {@link SubagentSetup} instance must
- * survive the plan's serialization round-trip — the transfer that carries a compiled plan from the
- * JobManager to a remote TaskManager — and resolve there to an equivalent setup carrying the same
- * configuration. In-process the plan resolves to the live object, so only the round-trip exercises
- * the descriptor-based rebuild on the far side.
- */
+/** An external descriptor retains its configuration across plan transfer. */
public class AgentPlanSubagentSerializationTest {
@Test
- void liveExternalSubagentIsRebuiltAfterPlanSerializationRoundTrip() throws Exception {
+ void externalDescriptorIsRebuiltAfterPlanSerializationRoundTrip() throws Exception {
Agent agent = new Agent();
agent.addResource(
"reviewer",
ResourceType.AGENT,
- new ExternalReviewSubagent("http://review.internal:8080", 3));
+ ResourceDescriptor.Builder.newBuilder(ExternalReviewSubagent.class.getName())
+ .addInitialArgument("endpoint", "http://review.internal:8080")
+ .addInitialArgument(ExternalReviewSubagent.FIELD_MAX_RETRIES, 3)
+ .build());
AgentPlan plan = new AgentPlan(agent);
diff --git a/python/flink_agents/api/agents/agent.py b/python/flink_agents/api/agents/agent.py
index 4da5a4437..14b11b0f8 100644
--- a/python/flink_agents/api/agents/agent.py
+++ b/python/flink_agents/api/agents/agent.py
@@ -161,7 +161,8 @@ class Agent(ABC):
instance: SerializableResource | ResourceDescriptor | Agent
The serializable resource instance, the descriptor of resource,
or an Agent instance. For the AGENT resource type an ``Agent`` is
- compiled into an internal sub-agent during plan construction.
+ wrapped in a descriptor here and compiled into an internal sub-agent
+ during plan construction.
Returns:
-------
@@ -173,5 +174,13 @@ class Agent(ABC):
msg = f"{resource_type.value} {name} already defined"
raise ValueError(msg)
+ if resource_type == ResourceType.AGENT and isinstance(instance, Agent):
+ from flink_agents.api.subagent import (
+ InternalSubagentResourceDescriptorBuilder,
+ )
+
+ instance = (
+ InternalSubagentResourceDescriptorBuilder().agent(instance).build()
+ )
self._resources[resource_type][name] = instance
return self
diff --git a/python/flink_agents/api/subagent.py b/python/flink_agents/api/subagent.py
index bd2221314..10b62f9e9 100644
--- a/python/flink_agents/api/subagent.py
+++ b/python/flink_agents/api/subagent.py
@@ -23,9 +23,15 @@ from typing import TYPE_CHECKING, Any
from pydantic import BaseModel, field_validator, model_validator
-from flink_agents.api.resource import ResourceType, SerializableResource
+from flink_agents.api.resource import (
+ JavaResourceDescriptor,
+ Resource,
+ ResourceDescriptor,
+ ResourceType,
+)
if TYPE_CHECKING:
+ from flink_agents.api.agents.agent import Agent
from flink_agents.api.runner_context import RunnerContext
_LOG = logging.getLogger(__name__)
@@ -201,7 +207,7 @@ class SubagentFutures(ABC):
"""
-class SubagentSetup(SerializableResource):
+class SubagentSetup(Resource):
"""Caller-facing definition of a sub-agent, registered as an AGENT resource.
Attributes:
@@ -290,3 +296,78 @@ class SubagentSetup(SerializableResource):
The complete ``(session_id, call_id)`` identity is reserved for
implementation use.
"""
+
+
+INTERNAL_SUBAGENT_SETUP_CLASS = (
+ "org.apache.flink.agents.runtime.subagent.InternalSubagentSetup"
+)
+
+
+class InternalSubagentResourceDescriptorBuilder:
+ """Build an ordinary resource descriptor declaring an internal child agent."""
+
+ def __init__(self) -> None:
+ """Start an internal sub-agent descriptor declaration."""
+ self._agent = None
+ self._description = ""
+ self._input_type = None
+ self._input_schema = None
+
+ def agent(self, agent: "Agent") -> "InternalSubagentResourceDescriptorBuilder":
+ """Set the child agent to compile."""
+ from flink_agents.api.agents.agent import Agent
+
+ if not isinstance(agent, Agent):
+ msg = "An internal sub-agent requires an Agent"
+ raise TypeError(msg)
+ self._agent = agent
+ return self
+
+ def description(
+ self, description: str
+ ) -> "InternalSubagentResourceDescriptorBuilder":
+ """Describe when the caller should delegate to this agent."""
+ if not isinstance(description, str):
+ msg = "description must be a string"
+ raise TypeError(msg)
+ self._description = description
+ return self
+
+ def input_type(
+ self, input_type: type
+ ) -> "InternalSubagentResourceDescriptorBuilder":
+ """Derive the input schema from a Pydantic model."""
+ if not isinstance(input_type, type):
+ msg = "input_type must be a type"
+ raise TypeError(msg)
+ self._input_type = input_type
+ return self
+
+ def input_schema(self, schema: str) -> "InternalSubagentResourceDescriptorBuilder":
+ """Declare the input JSON schema explicitly."""
+ if not isinstance(schema, str) or not schema.strip():
+ msg = "Sub-agent input schema must not be blank"
+ raise ValueError(msg)
+ self._input_schema = schema
+ return self
+
+ def build(self) -> ResourceDescriptor:
+ """Return a descriptor whose Agent is compiled by the parent plan."""
+ if self._agent is None:
+ msg = "An internal sub-agent requires an Agent"
+ raise ValueError(msg)
+ if self._input_type is not None and self._input_schema is not None:
+ msg = "Declare either input_type or input_schema, not both"
+ raise ValueError(msg)
+ schema = self._input_schema
+ if self._input_type is not None:
+ schema = _input_schema_from(self._input_type)
+ if schema is None:
+ msg = "Sub-agent input_type must describe an object"
+ raise ValueError(msg)
+ arguments = {"agent": self._agent, "description": self._description}
+ if schema is not None:
+ arguments["input_schema"] = schema
+ return JavaResourceDescriptor(
+ clazz=INTERNAL_SUBAGENT_SETUP_CLASS, arguments=arguments
+ )
diff --git a/python/flink_agents/api/tests/test_subagent.py b/python/flink_agents/api/tests/test_subagent.py
index e1664ad2b..029dcf3e4 100644
--- a/python/flink_agents/api/tests/test_subagent.py
+++ b/python/flink_agents/api/tests/test_subagent.py
@@ -23,7 +23,7 @@ import pytest
from pydantic import ValidationError
from flink_agents.api.agents.agent import Agent
-from flink_agents.api.resource import ResourceType
+from flink_agents.api.resource import ResourceDescriptor, ResourceType
from flink_agents.api.tests.subagent_test_utils import (
NestedTypedTestSubagentSetup,
TestSubagentSetup,
@@ -33,33 +33,36 @@ from flink_agents.api.tests.subagent_test_utils import (
DECLARED_SCHEMA = '{"type":"object","properties":{"path":{"type":"string"}}}'
-def test_register_subagent_setup_as_resource() -> None:
- """A ``SubagentSetup`` registers under the AGENT resource map."""
- agent = Agent()
- setup = TestSubagentSetup()
-
- agent.add_resource("reviewer", ResourceType.AGENT, setup)
-
- agent_resources = agent.resources[ResourceType.AGENT]
- assert len(agent_resources) == 1
- assert agent_resources["reviewer"] is setup
- assert setup.resource_type() == ResourceType.AGENT
-
-
def test_duplicate_name_throws() -> None:
"""Registering a duplicate AGENT name raises."""
agent = Agent()
- agent.add_resource("reviewer", ResourceType.AGENT, TestSubagentSetup())
+ agent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz="flink_agents.api.tests.subagent_test_utils.TestSubagentSetup"
+ ),
+ )
with pytest.raises(ValueError):
- agent.add_resource("reviewer", ResourceType.AGENT, TestSubagentSetup())
+ agent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz="flink_agents.api.tests.subagent_test_utils.TestSubagentSetup"
+ ),
+ )
def test_multiple_subagents_registered() -> None:
"""Multiple distinct AGENT resources coexist."""
agent = Agent()
- reviewer = TestSubagentSetup()
- coder = TestSubagentSetup()
+ reviewer = ResourceDescriptor(
+ clazz="flink_agents.api.tests.subagent_test_utils.TestSubagentSetup"
+ )
+ coder = ResourceDescriptor(
+ clazz="flink_agents.api.tests.subagent_test_utils.TestSubagentSetup"
+ )
agent.add_resource("reviewer", ResourceType.AGENT, reviewer)
agent.add_resource("coder", ResourceType.AGENT, coder)
@@ -180,3 +183,21 @@ def test_the_metadata_serializes_under_the_cross_language_keys() -> None:
assert dumped["description"] == "Reviews a file."
assert dumped["input_schema"] == DECLARED_SCHEMA
assert "inputSchema" not in dumped
+
+
+def test_shared_registration_accepts_descriptor() -> None:
+ from unittest.mock import Mock
+
+ from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+
+ env = Mock(spec=AgentsExecutionEnvironment)
+ env._resources = {kind: {} for kind in ResourceType}
+ descriptor = ResourceDescriptor(
+ clazz="flink_agents.api.tests.subagent_test_utils.TestSubagentSetup"
+ )
+ assert (
+ AgentsExecutionEnvironment.add_resource(
+ env, "reviewer", ResourceType.AGENT, descriptor
+ )
+ is env
+ )
diff --git a/python/flink_agents/e2e_tests/e2e_tests_integration/internal_subagent_test.py b/python/flink_agents/e2e_tests/e2e_tests_integration/internal_subagent_test.py
index 36718d778..f8ede70e1 100644
--- a/python/flink_agents/e2e_tests/e2e_tests_integration/internal_subagent_test.py
+++ b/python/flink_agents/e2e_tests/e2e_tests_integration/internal_subagent_test.py
@@ -17,7 +17,7 @@
################################################################################
"""E2E tests for internal sub-agents on a real Flink job.
-A child agent registered directly as an ``AGENT`` resource exercises the whole
+A child agent registered through a descriptor exercises the whole
operator round-trip: the Python-compiled child plan is materialized into a Java
``InternalSubagentSetup``, the call event is sent on the mailbox thread, the
mailbox is released while the caller awaits, and the child's actions are
@@ -56,6 +56,7 @@ from flink_agents.api.events.event import Event, InputEvent, OutputEvent
from flink_agents.api.execution_environment import AgentsExecutionEnvironment
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder
os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"]
@@ -134,7 +135,14 @@ def _run_and_collect(child: Agent, tmp_path: Path) -> str:
input_stream = env.from_collection(["hello"])
root = RootAgent()
- root.add_resource(CHILD_SCOPE, ResourceType.AGENT, child)
+ root.add_resource(
+ CHILD_SCOPE,
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(child)
+ .description("Delegate the prompt")
+ .build(),
+ )
agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
output_datastream = (
diff --git a/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_external_integration_agent.py b/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_external_integration_agent.py
index c7818574b..b738aa993 100644
--- a/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_external_integration_agent.py
+++ b/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_external_integration_agent.py
@@ -35,7 +35,7 @@ from flink_agents.api.agents.agent import Agent
from flink_agents.api.decorators import action
from flink_agents.api.events.event import Event, InputEvent, OutputEvent
from flink_agents.api.events.event_type import EventType
-from flink_agents.api.resource import ResourceType
+from flink_agents.api.resource import ResourceDescriptor, ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.api.subagent import SubagentResult
from flink_agents.runtime.async_subagent import BaseAsyncSubagentSetup, RunStatus
@@ -137,12 +137,24 @@ class DeferredExternalAgent(Agent):
def build_async_agent() -> Agent:
"""Build the agent registering the async external sub-agent."""
agent = AsyncExternalAgent()
- agent.add_resource("reviewer", ResourceType.AGENT, InMemoryAsyncSubagentSetup())
+ agent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz=f"{InMemoryAsyncSubagentSetup.__module__}.{InMemoryAsyncSubagentSetup.__name__}"
+ ),
+ )
return agent
def build_deferred_agent() -> Agent:
"""Build the agent registering the deferred external sub-agent."""
agent = DeferredExternalAgent()
- agent.add_resource("reviewer", ResourceType.AGENT, InMemoryDeferredSubagentSetup())
+ agent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz=f"{InMemoryDeferredSubagentSetup.__module__}.{InMemoryDeferredSubagentSetup.__name__}"
+ ),
+ )
return agent
diff --git a/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_integration_agent.py b/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_integration_agent.py
index 7eb252c0b..a9cee895f 100644
--- a/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_integration_agent.py
+++ b/python/flink_agents/e2e_tests/e2e_tests_integration/subagent_integration_agent.py
@@ -31,7 +31,7 @@ from flink_agents.api.agents.agent import Agent
from flink_agents.api.decorators import action
from flink_agents.api.events.event import Event, InputEvent, OutputEvent
from flink_agents.api.events.event_type import EventType
-from flink_agents.api.resource import ResourceType
+from flink_agents.api.resource import ResourceDescriptor, ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.api.subagent import SubagentResult
from flink_agents.runtime.deferred_subagent import (
@@ -83,5 +83,11 @@ class SubagentIntegrationAgent(Agent):
def build_agent() -> Agent:
"""Build the agent with the sub-agent registered as an AGENT resource."""
agent = SubagentIntegrationAgent()
- agent.add_resource("reviewer", ResourceType.AGENT, EchoSubagentSetup())
+ agent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz=f"{EchoSubagentSetup.__module__}.{EchoSubagentSetup.__name__}"
+ ),
+ )
return agent
diff --git a/python/flink_agents/plan/agent_plan.py b/python/flink_agents/plan/agent_plan.py
index 211db981f..ae760b0bd 100644
--- a/python/flink_agents/plan/agent_plan.py
+++ b/python/flink_agents/plan/agent_plan.py
@@ -34,7 +34,7 @@ from flink_agents.api.skills import (
LOAD_SKILL_TOOL,
Skills,
)
-from flink_agents.api.subagent import CALLABLE_NAME_PREFIX, SubagentSetup
+from flink_agents.api.subagent import CALLABLE_NAME_PREFIX
from flink_agents.api.tools.function_tool import FunctionTool as ApiFunctionTool
from flink_agents.api.tools.tool import Tool
from flink_agents.plan import subagent as _subagent
@@ -138,31 +138,6 @@ class AgentPlan(BaseModel):
provider
)
)
- elif provider_type == "InternalSubagentProvider":
- # Java compiles a directly-registered child Agent
- # into an InternalSubagentProvider that nests the
- # child plan. Python models the same internal
- # sub-agent as a PythonSerializableResourceProvider
- # pointing at InternalSubagentSetup, so convert to
- # that shape and rebuild the nested child plan
- # recursively.
- child_plan = AgentPlan.model_validate(
- provider["childPlan"]
- )
- self["resource_providers"][type][name] = (
- PythonSerializableResourceProvider(
- name=provider["name"],
- type=ResourceType(provider["type"]),
- module=(
- "flink_agents.runtime.internal_subagent"
- ),
- clazz="InternalSubagentSetup",
- serialized={
- "child_plan": child_plan,
- "scope": provider["scope"],
- },
- )
- )
return self
@staticmethod
@@ -480,53 +455,24 @@ def _get_resource_providers(
all_skills[name] = skills
_add_skills(all_skills, resource_providers)
- for name, value in agent.resources[ResourceType.AGENT].items():
- if isinstance(value, SubagentSetup):
- # A live setup travels as its descriptor, so a remote task rebuilds
- # it from the class name and configuration.
- descriptor = ResourceDescriptor(
- clazz=f"{value.__class__.__module__}.{value.__class__.__name__}",
- arguments=value.model_dump(),
- )
- elif isinstance(value, ResourceDescriptor):
- # Declared via YAML: the descriptor names a SubagentSetup subclass
- # that is instantiated when the resource is first resolved.
- descriptor = value
- elif isinstance(value, Agent):
- # Compile a directly-registered child Agent into an internal
- # sub-agent. The child's compiled plan is reused across names and
- # cycles are rejected by the compilation helper. The resource
- # name doubles as the sub-agent scope used for runtime resolution.
- child_plan = _subagent.get_or_compile(
- value,
- name,
- lambda child: AgentPlan.from_agent(child, config),
- )
- # Reference the runtime setup by name to keep the plan -> runtime
- # dependency direction; the provider materializes
- # flink_agents.runtime.internal_subagent.InternalSubagentSetup at
- # runtime. The live child plan rides in the serialized map and is
- # dumped lazily. This branch carries the child plan itself, so it
- # appends its own provider instead of the shared descriptor one.
- resource_providers.append(
- PythonSerializableResourceProvider(
- name=name,
- type=ResourceType.AGENT,
- module="flink_agents.runtime.internal_subagent",
- clazz="InternalSubagentSetup",
- serialized={"child_plan": child_plan, "scope": name},
- )
- )
- continue
- else:
+ for name, descriptor in agent.resources[ResourceType.AGENT].items():
+ if not isinstance(descriptor, ResourceDescriptor):
msg = (
- f"AGENT resource '{name}' must be a SubagentSetup, a "
- f"ResourceDescriptor, or an Agent, but got "
- f"{type(value).__name__}."
+ f"AGENT resource '{name}' must be a ResourceDescriptor; "
+ "setup instances cannot be registered"
)
raise TypeError(msg)
+ descriptor = _subagent.compile_descriptor(
+ name, descriptor, lambda child: AgentPlan.from_agent(child, config)
+ )
resource_providers.append(
- PythonResourceProvider.get(name=name, descriptor=descriptor)
+ JavaResourceProvider(
+ name=name, type=ResourceType.AGENT, descriptor=descriptor
+ )
+ if descriptor.language == "java"
+ else PythonResourceProvider.get(
+ name=name, descriptor=descriptor, resource_type=ResourceType.AGENT
+ )
)
for resource_type in [
diff --git a/python/flink_agents/plan/resource/java/java_resource_adapter.py b/python/flink_agents/plan/resource/java/java_resource_adapter.py
index dba67df2e..d307f8128 100644
--- a/python/flink_agents/plan/resource/java/java_resource_adapter.py
+++ b/python/flink_agents/plan/resource/java/java_resource_adapter.py
@@ -17,10 +17,19 @@
#################################################################################
from typing import Any, Protocol
+from flink_agents.api.resource import Resource, ResourceDescriptor
+from flink_agents.api.resource_context import ResourceContext
+
class JavaResourceAdapter(Protocol):
"""Services supplied by Runtime to Java-backed Plan resources."""
+ def create_subagent_resource(
+ self, descriptor: ResourceDescriptor, resource_context: ResourceContext
+ ) -> Resource:
+ """Create a Python caller for a Java-orchestrated sub-agent."""
+ ...
+
def getResource(self, name: str, resource_type: str) -> Any:
"""Resolve a resource through its owning runtime."""
...
diff --git a/python/flink_agents/plan/resource_provider.py b/python/flink_agents/plan/resource_provider.py
index c9fbef76d..770cbe19b 100644
--- a/python/flink_agents/plan/resource_provider.py
+++ b/python/flink_agents/plan/resource_provider.py
@@ -204,7 +204,10 @@ class JavaResourceProvider(ResourceProvider):
if descriptor.language != "java":
msg = "JavaResourceProvider requires a Java descriptor"
raise ValueError(msg)
- if resource_type not in JAVA_RESOURCE_MAPPING:
+ if (
+ resource_type not in JAVA_RESOURCE_MAPPING
+ and resource_type != ResourceType.AGENT
+ ):
msg = f"Unsupported Java resource type: {resource_type}"
raise ValueError(msg)
return JavaResourceProvider(
@@ -218,6 +221,10 @@ class JavaResourceProvider(ResourceProvider):
if not self._j_resource_adapter:
err_msg = "java resource adapter is not set"
raise RuntimeError(err_msg)
+ if self.type == ResourceType.AGENT:
+ return self._j_resource_adapter.create_subagent_resource(
+ self.descriptor, resource_context
+ )
j_resource = self._j_resource_adapter.getResource(self.name, self.type.value)
class_path = JAVA_RESOURCE_MAPPING.get(self.type)
@@ -259,18 +266,7 @@ class JavaSerializableResourceProvider(SerializableResourceProvider):
def is_python_owned(provider: ResourceProvider) -> bool:
- """Whether the provider materializes a resource owned by the Python runtime.
-
- The runtime must ask that runtime to build such a resource instead of
- resolving it on the Java side. Mirrors Java ``ResourceProvider.isPythonOwned``.
- """
- if isinstance(provider, PythonResourceProvider):
- return True
- if isinstance(provider, PythonSerializableResourceProvider):
- # A Python-compiled internal sub-agent carries its child plan in the
- # serialized payload and runs natively on the Java side, so it stays
- # Java-owned; only external Python setups (no child plan) are Python-owned.
- return not (
- provider.type == ResourceType.AGENT and "child_plan" in provider.serialized
- )
- return False
+ """Whether this resource is constructed by the Python runtime."""
+ return isinstance(
+ provider, PythonResourceProvider | PythonSerializableResourceProvider
+ )
diff --git a/python/flink_agents/plan/subagent.py b/python/flink_agents/plan/subagent.py
index 50f55e589..4f56d816a 100644
--- a/python/flink_agents/plan/subagent.py
+++ b/python/flink_agents/plan/subagent.py
@@ -15,25 +15,16 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-"""Plan-layer support for internal sub-agents.
-
-An ``Agent`` registered directly as an AGENT resource
-(``add_resource(name, AGENT, ChildAgent())``) is compiled into an internal
-sub-agent: the child's compiled :class:`~flink_agents.plan.agent_plan.AgentPlan`
-plus a scope id (the resource name) are carried by a
-``PythonSerializableResourceProvider`` referencing the runtime setup class
-``flink_agents.runtime.internal_subagent.InternalSubagentSetup`` by name, so
-the plan layer keeps its dependency direction (the setup itself lives in the
-runtime layer, mirroring Java's ``InternalSubagentProvider``).
-
-The ``begin`` / ``end`` / ``get_or_compile`` helpers provide thread-local
-cycle detection and child-plan reuse during compilation, mirroring Java's
-``InternalSubagentCompilationHelper``.
-"""
+"""Compile and traverse child plans carried by ordinary resource descriptors."""
import threading
+from copy import deepcopy
from typing import TYPE_CHECKING, Any, Callable
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.resource import JavaResourceDescriptor, ResourceDescriptor
+from flink_agents.api.subagent import INTERNAL_SUBAGENT_SETUP_CLASS
+
if TYPE_CHECKING:
from flink_agents.plan.agent_plan import AgentPlan
@@ -118,3 +109,47 @@ def get_or_compile(
finally:
del compiling[key]
path.pop()
+
+
+def is_internal(descriptor: ResourceDescriptor) -> bool:
+ """Whether the descriptor constructs the internal sub-agent runtime."""
+ return (
+ descriptor.language == "java"
+ and descriptor.target_clazz == INTERNAL_SUBAGENT_SETUP_CLASS
+ )
+
+
+def compile_descriptor(
+ name: str,
+ descriptor: ResourceDescriptor,
+ compiler: Callable[[Agent], "AgentPlan"],
+) -> ResourceDescriptor:
+ """Compile a child Agent without modifying its declaration."""
+ if not is_internal(descriptor):
+ return descriptor
+ arguments = dict(descriptor.arguments)
+ agent = arguments.pop("agent", None)
+ if not isinstance(agent, Agent) or "plan" in arguments:
+ msg = (
+ f"Internal sub-agent '{name}' requires an Agent declaration, "
+ "not a compiled plan"
+ )
+ raise ValueError(msg)
+ plan = get_or_compile(agent, name, compiler)
+ arguments["plan"] = plan.model_dump(mode="json", serialize_as_any=True)
+ arguments["scope"] = name
+ return JavaResourceDescriptor(
+ clazz=INTERNAL_SUBAGENT_SETUP_CLASS, arguments=arguments
+ )
+
+
+def decode_plan(descriptor: ResourceDescriptor) -> "AgentPlan":
+ """Restore the typed plan from a descriptor's JSON data."""
+ from flink_agents.plan.agent_plan import AgentPlan
+
+ data = descriptor.arguments.get("plan")
+ if not isinstance(data, dict):
+ msg = "Internal sub-agent requires a compiled plan object"
+ raise TypeError(msg)
+ # AgentPlan's validator reconstructs providers in place; preserve the descriptor.
+ return AgentPlan.model_validate(deepcopy(data))
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 a4b09be80..50418b0b5 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
@@ -48,7 +48,6 @@ from flink_agents.plan.function import (
)
from flink_agents.plan.resource_provider import (
JavaResourceProvider,
- PythonSerializableResourceProvider,
)
# python/flink_agents/plan/tests/test_*.py -> repo root is parents[4].
@@ -371,17 +370,7 @@ def test_python_can_deserialize_java_plan_with_python_action() -> None:
def test_python_can_deserialize_java_plan_with_internal_subagent() -> None:
- """Java compiles a directly-registered child Agent into an
- ``InternalSubagentProvider`` that nests the child plan; Python must read it
- back as its own internal sub-agent representation.
-
- The snapshot is emitted by AgentPlanCrossLanguageTest on the Java side
- (regenerate with -Dregenerate.snapshots=true). Python has no
- ``InternalSubagentProvider`` class: it represents the same internal
- sub-agent as a ``PythonSerializableResourceProvider`` referencing
- ``InternalSubagentSetup``, so deserialization must convert to that form and
- recursively rebuild the nested child plan.
- """
+ """Both languages read the same generic descriptor with a nested child plan."""
snapshot = _SNAPSHOT_DIR / "java" / "agent_plan_with_internal_subagent.json"
assert snapshot.exists(), (
f"Java internal sub-agent plan snapshot missing from {snapshot}; "
@@ -391,15 +380,13 @@ def test_python_can_deserialize_java_plan_with_internal_subagent() -> None:
restored = AgentPlan.model_validate_json(snapshot.read_text())
provider = restored.resource_providers[ResourceType.AGENT]["child"]
- assert isinstance(provider, PythonSerializableResourceProvider), (
- f"Expected internal sub-agent to deserialize as "
- f"PythonSerializableResourceProvider, got {type(provider).__name__}"
- )
- assert provider.module == "flink_agents.runtime.internal_subagent"
- assert provider.clazz == "InternalSubagentSetup"
- assert provider.serialized["scope"] == "child"
- child_plan = provider.serialized["child_plan"]
- assert isinstance(child_plan, AgentPlan)
+ from flink_agents.plan.resource_provider import JavaResourceProvider
+ from flink_agents.plan.subagent import decode_plan
+
+ assert isinstance(provider, JavaResourceProvider)
+ assert provider.descriptor.language == "java"
+ assert provider.descriptor.arguments["scope"] == "child"
+ child_plan = decode_plan(provider.descriptor)
assert "handle" in child_plan.actions
@@ -538,3 +525,35 @@ def test_python_can_deserialize_plan_with_java_model_router() -> None:
restored.resource_providers[ResourceType.MODEL_ROUTER]["router"],
JavaResourceProvider,
)
+
+
+def _internal_subagent_plan() -> AgentPlan:
+ from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder
+
+ child = Agent()
+ child.add_action(
+ name="handle",
+ trigger_conditions=[InputEvent.EVENT_TYPE],
+ func=_make_python_function_descriptor(),
+ )
+ root = Agent()
+ root.add_resource(
+ "child",
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(child)
+ .description("Python child")
+ .input_schema('{"type":"object","properties":{"question":{"type":"string"}}}')
+ .build(),
+ )
+ return AgentPlan.from_agent(root, AgentConfiguration())
+
+
+def test_python_internal_subagent_snapshot() -> None:
+ """Java consumes the same generic descriptor emitted by Python compilation."""
+ target = _SNAPSHOT_DIR / "python" / "agent_plan_with_internal_subagent.json"
+ actual = _plan_dump_json(_internal_subagent_plan())
+ if _regenerate_enabled():
+ target.parent.mkdir(parents=True, exist_ok=True)
+ target.write_text(actual + "\n")
+ assert json.loads(target.read_text()) == json.loads(actual)
diff --git a/python/flink_agents/plan/tests/test_agent_plan_subagent_resources.py b/python/flink_agents/plan/tests/test_agent_plan_subagent_resources.py
index f1dd38485..898c79477 100644
--- a/python/flink_agents/plan/tests/test_agent_plan_subagent_resources.py
+++ b/python/flink_agents/plan/tests/test_agent_plan_subagent_resources.py
@@ -21,29 +21,18 @@ import pytest
from flink_agents.api.agents.agent import Agent
from flink_agents.api.resource import ResourceDescriptor, ResourceType
-from flink_agents.api.subagent import SubagentSetup
from flink_agents.api.tests.subagent_test_utils import TestSubagentSetup
from flink_agents.plan.agent_plan import AgentPlan
from flink_agents.plan.configuration import AgentConfiguration
from flink_agents.plan.resource_provider import PythonResourceProvider
-def test_subagent_setup_compiles_into_agent_provider() -> None:
- """A registered SubagentSetup lands in the AGENT provider map and resolves."""
- setup = TestSubagentSetup()
+def test_registered_setup_instance_is_rejected_during_compilation() -> None:
+ """The generic registration API does not make setup instances valid plans."""
agent = Agent()
- agent.add_resource("reviewer", ResourceType.AGENT, setup)
-
- plan = AgentPlan.from_agent(agent, AgentConfiguration())
-
- agents = plan.resource_providers[ResourceType.AGENT]
- assert agents is not None
- assert "reviewer" in agents
- resolved = agents["reviewer"].provide(
- resource_context=None, config=AgentConfiguration()
- )
- assert isinstance(resolved, SubagentSetup)
- assert resolved.resource_type() == ResourceType.AGENT
+ agent.add_resource("reviewer", ResourceType.AGENT, TestSubagentSetup())
+ with pytest.raises(TypeError, match="setup instances cannot be registered"):
+ AgentPlan.from_agent(agent, AgentConfiguration())
def test_agent_descriptor_compiles_into_agent_provider() -> None:
@@ -72,21 +61,21 @@ def test_non_setup_agent_resource_is_rejected() -> None:
with pytest.raises(
TypeError,
- match="must be a SubagentSetup, a ResourceDescriptor, or an Agent",
+ match="must be a ResourceDescriptor",
):
AgentPlan.from_agent(agent, AgentConfiguration())
-def test_live_external_subagent_is_rebuilt_after_plan_json_round_trip() -> None:
- """A live setup crosses the plan JSON boundary as its descriptor and is
- rebuilt with its configuration on the other side.
- """
+def test_external_descriptor_is_rebuilt_after_plan_json_round_trip() -> None:
+ """An external descriptor retains its configuration across the plan boundary."""
agent = Agent()
agent.add_resource(
"reviewer",
ResourceType.AGENT,
- TestSubagentSetup(
- endpoint_url="http://review.internal:8080", fail_on_call=True
+ ResourceDescriptor(
+ clazz="flink_agents.api.tests.subagent_test_utils.TestSubagentSetup",
+ endpoint_url="http://review.internal:8080",
+ fail_on_call=True,
),
)
@@ -119,10 +108,10 @@ def test_child_agent_compiles_into_internal_provider() -> None:
agents = plan.resource_providers[ResourceType.AGENT]
provider = agents["child"]
- assert provider.module == "flink_agents.runtime.internal_subagent"
- assert provider.clazz == "InternalSubagentSetup"
- assert provider.serialized["scope"] == "child"
- assert provider.serialized["child_plan"] is not None
+ assert provider.descriptor.language == "java"
+ assert provider.descriptor.target_clazz.endswith(".InternalSubagentSetup")
+ assert provider.descriptor.arguments["scope"] == "child"
+ assert provider.descriptor.arguments["plan"] is not None
def test_shared_child_agent_compiles_to_single_plan() -> None:
@@ -135,9 +124,9 @@ def test_shared_child_agent_compiles_to_single_plan() -> None:
plan = AgentPlan.from_agent(root, AgentConfiguration())
agents = plan.resource_providers[ResourceType.AGENT]
- first_plan = agents["first"].serialized["child_plan"]
- second_plan = agents["second"].serialized["child_plan"]
- assert first_plan is second_plan
+ first_plan = agents["first"].descriptor.arguments["plan"]
+ second_plan = agents["second"].descriptor.arguments["plan"]
+ assert first_plan == second_plan
def test_cycle_not_through_root_is_rejected_with_cycle_path() -> None:
diff --git a/python/flink_agents/plan/tests/test_resource_provider.py b/python/flink_agents/plan/tests/test_resource_provider.py
index 83408bdee..56e3bdbe1 100644
--- a/python/flink_agents/plan/tests/test_resource_provider.py
+++ b/python/flink_agents/plan/tests/test_resource_provider.py
@@ -24,7 +24,6 @@ from flink_agents.api.resource import Resource, ResourceDescriptor, ResourceType
from flink_agents.plan.resource_provider import (
JavaResourceProvider,
PythonResourceProvider,
- PythonSerializableResourceProvider,
ResourceProvider,
is_python_owned,
)
@@ -105,37 +104,23 @@ def test_python_can_deserialize_java_resource_provider_wire_shape() -> None:
assert provider.descriptor.arguments["max_tokens"] == 1024
-def test_is_python_owned_mirrors_java_child_plan_exclusion(
- resource_provider: ResourceProvider,
+def test_internal_descriptor_is_java_owned(
+ resource_provider: PythonResourceProvider,
) -> None:
- """A Python-compiled internal sub-agent must stay Java-owned on both sides.
-
- ``agent_plan`` builds it as a ``PythonSerializableResourceProvider`` of type
- ``AGENT`` whose serialized payload carries ``child_plan``. ``is_python_owned``
- has to exclude it exactly like Java ``ResourceProvider.isPythonOwned``; otherwise
- Python ``eager_materialize`` returns a resource the Java cache already built
- natively and overwrites the Java ``InternalSubagentSetup``.
- """
- # A plain Python resource descriptor is Python-owned.
- assert is_python_owned(resource_provider) is True
-
- # An internal sub-agent (type AGENT, serialized child_plan) is Java-owned even
- # though it is a PythonSerializableResourceProvider.
- internal_subagent = PythonSerializableResourceProvider(
- name="child",
- type=ResourceType.AGENT,
- module="flink_agents.runtime.internal_subagent",
- clazz="InternalSubagentSetup",
- serialized={"child_plan": {}, "scope": "child"},
- )
- assert is_python_owned(internal_subagent) is False
-
- # An external Python setup serialized without a child plan stays Python-owned.
- external_setup = PythonSerializableResourceProvider(
- name="external",
- type=ResourceType.AGENT,
- module="some.module",
- clazz="ExternalSubagent",
- serialized={"scope": "external"},
+ """Runtime ownership follows the provider language, without payload heuristics."""
+ from flink_agents.api.agents.agent import Agent
+ from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder
+ from flink_agents.plan.agent_plan import AgentPlan
+ from flink_agents.plan.configuration import AgentConfiguration
+
+ assert is_python_owned(resource_provider)
+ root = Agent()
+ root.add_resource(
+ "child",
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder().agent(Agent()).build(),
)
- assert is_python_owned(external_setup) is True
+ provider = AgentPlan.from_agent(root, AgentConfiguration()).resource_providers[
+ ResourceType.AGENT
+ ]["child"]
+ assert not is_python_owned(provider)
diff --git a/python/flink_agents/runtime/internal_subagent.py b/python/flink_agents/runtime/internal_subagent.py
index 0fbc908a3..3e794c44a 100644
--- a/python/flink_agents/runtime/internal_subagent.py
+++ b/python/flink_agents/runtime/internal_subagent.py
@@ -18,10 +18,10 @@
"""Runtime setup for internal sub-agents.
The compiled form of an ``Agent`` registered directly as an AGENT resource:
-the plan layer carries the child
+the compiled descriptor carries the child
:class:`~flink_agents.plan.agent_plan.AgentPlan` and the scope (the resource
-name) in a ``PythonSerializableResourceProvider`` referencing this class by
-name; this runtime class owns the invocation behavior.
+name) in a ``JavaResourceProvider`` referencing the Java runtime setup.
+This Python caller owns the invocation behavior.
Execution mode is the deferred one, inherited from
:class:`~flink_agents.runtime.deferred_subagent.DeferredSubagentSetup`:
@@ -84,7 +84,7 @@ class InternalSubagentCallFactory(Protocol):
class InternalSubagentSetup(DeferredSubagentSetup):
"""Compiled internal sub-agent, produced during ``AgentPlan`` compilation.
- Holds the child agent's compiled ``child_plan`` and its ``scope`` (the
+ Holds the child agent's compiled ``plan`` and its ``scope`` (the
resource name used for runtime resolution). The child work is orchestrated
by the operator: the async callable delegates to
:class:`InternalSubagentCallFactory` on the runtime context. The
@@ -92,7 +92,7 @@ class InternalSubagentSetup(DeferredSubagentSetup):
explicit identities, deferred issue, and the await-only resolve.
"""
- child_plan: AgentPlan
+ plan: AgentPlan
scope: str
def prepare(
diff --git a/python/flink_agents/runtime/java_resource_adapter.py b/python/flink_agents/runtime/java_resource_adapter.py
index 64a8cfa16..93ca57fc7 100644
--- a/python/flink_agents/runtime/java_resource_adapter.py
+++ b/python/flink_agents/runtime/java_resource_adapter.py
@@ -17,6 +17,8 @@
#################################################################################
from typing import Any
+from flink_agents.api.resource import Resource, ResourceDescriptor
+from flink_agents.api.resource_context import ResourceContext
from flink_agents.runtime.flink_metric_group import FlinkMetricGroup
@@ -70,3 +72,17 @@ class JavaResourceAdapterImpl:
msg = "Java resource metric groups must be FlinkMetricGroup or None."
raise TypeError(msg)
resource.setMetricGroup(java_group)
+
+ def create_subagent_resource(
+ self, descriptor: ResourceDescriptor, resource_context: ResourceContext
+ ) -> Resource:
+ """Bind Python calls to an internal sub-agent orchestrated by the operator."""
+ from flink_agents.plan.subagent import decode_plan, is_internal
+ from flink_agents.runtime.internal_subagent import InternalSubagentSetup
+
+ if not is_internal(descriptor):
+ msg = "Calling external Java sub-agents from Python is not supported"
+ raise NotImplementedError(msg)
+ arguments = dict(descriptor.arguments)
+ arguments["plan"] = decode_plan(descriptor)
+ return InternalSubagentSetup(**arguments, resource_context=resource_context)
diff --git a/python/flink_agents/runtime/resource_cache.py b/python/flink_agents/runtime/resource_cache.py
index 1e0db4ecd..4dfc9c6e6 100644
--- a/python/flink_agents/runtime/resource_cache.py
+++ b/python/flink_agents/runtime/resource_cache.py
@@ -90,7 +90,13 @@ class ResourceCache:
raise KeyError(msg)
resource_provider = providers[name]
if isinstance(resource_provider, JavaResourceProvider):
- resource_provider.set_java_resource_adapter(self._j_resource_adapter)
+ from flink_agents.runtime.java_resource_adapter import (
+ JavaResourceAdapterImpl,
+ )
+
+ resource_provider.set_java_resource_adapter(
+ self._j_resource_adapter or JavaResourceAdapterImpl(None)
+ )
resource = resource_provider.provide(
resource_context=self._resource_context, config=self._config
)
diff --git a/python/flink_agents/runtime/tests/test_internal_subagent_descriptor.py b/python/flink_agents/runtime/tests/test_internal_subagent_descriptor.py
new file mode 100644
index 000000000..59b6e2fc7
--- /dev/null
+++ b/python/flink_agents/runtime/tests/test_internal_subagent_descriptor.py
@@ -0,0 +1,108 @@
+################################################################################
+# 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.
+################################################################################
+"""Tests for compiling AGENT resources (SubagentSetup) into the agent plan."""
+
+import json
+from unittest.mock import create_autospec
+
+import pytest
+from pydantic import BaseModel
+
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.resource import JavaResourceDescriptor, ResourceType
+from flink_agents.api.subagent import (
+ InternalSubagentResourceDescriptorBuilder,
+ SubagentResult,
+)
+from flink_agents.plan.agent_plan import AgentPlan
+from flink_agents.plan.configuration import AgentConfiguration
+from flink_agents.plan.resource_provider import JavaResourceProvider
+from flink_agents.runtime.internal_subagent import InternalSubagentCallFactory
+from flink_agents.runtime.resource_cache import ResourceCache
+
+
+class Request(BaseModel):
+ question: str
+
+
+def test_descriptor_round_trip_builds_a_python_caller_with_metadata() -> None:
+ child = Agent()
+ descriptor = (
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(child)
+ .description("Answers questions")
+ .input_type(Request)
+ .build()
+ )
+ assert type(descriptor) is JavaResourceDescriptor
+ root = Agent()
+ root.add_resource("child", ResourceType.AGENT, descriptor)
+ root.add_resource("second", ResourceType.AGENT, descriptor)
+ plan = AgentPlan.from_agent(root, AgentConfiguration())
+ restored = AgentPlan.model_validate_json(
+ plan.model_dump_json(serialize_as_any=True)
+ )
+ provider = restored.resource_providers[ResourceType.AGENT]["child"]
+ assert isinstance(provider, JavaResourceProvider)
+ assert "agent" not in provider.descriptor.arguments
+ assert descriptor.arguments["agent"] is child
+ assert "scope" not in descriptor.arguments
+ before = json.dumps(provider.descriptor.arguments)
+ cache = ResourceCache(restored.resource_providers, AgentConfiguration())
+ setup = cache.get_resource("child", ResourceType.AGENT)
+ assert setup.description == "Answers questions"
+ assert "question" in json.loads(setup.input_schema)["properties"]
+ assert setup.scope == "child"
+ assert cache.get_resource("second", ResourceType.AGENT).scope == "second"
+ assert json.dumps(provider.descriptor.arguments) == before
+ assert cache.get_resource("child", ResourceType.AGENT) is setup
+ # The Python caller retains mailbox bootstrap plus deferred worker await.
+ ctx = create_autospec(InternalSubagentCallFactory, instance=True, spec_set=True)
+ ctx.await_subagent_call.return_value = ["answer"]
+ _, call, _ = setup.prepare(ctx, {"question": "hello"}, "session", "call")
+ ctx.bootstrap_subagent_call.assert_called_once_with(
+ "child", "session", "call", {"question": "hello"}
+ )
+ result = call()
+ assert result == SubagentResult.ok(["answer"])
+ ctx.await_subagent_call.assert_called_once_with("session", "call")
+ cache.close()
+
+
+def test_descriptor_cycle_through_root_is_rejected() -> None:
+ root = Agent()
+ root.add_resource(
+ "self",
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder().agent(root).build(),
+ )
+ with pytest.raises(ValueError, match="Cyclic sub-agent"):
+ AgentPlan.from_agent(root, AgentConfiguration())
+
+
+def test_builder_rejects_missing_agent_and_conflicting_schema() -> None:
+ with pytest.raises(ValueError, match="requires an Agent"):
+ InternalSubagentResourceDescriptorBuilder().build()
+ with pytest.raises(ValueError, match="not both"):
+ InternalSubagentResourceDescriptorBuilder().agent(Agent()).input_type(
+ Request
+ ).input_schema("{}").build()
+ with pytest.raises(ValueError, match="must not be blank"):
+ InternalSubagentResourceDescriptorBuilder().agent(Agent()).input_schema(
+ " "
+ ).build()
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
index 662c0185b..917d835ad 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/context/RunnerContextImpl.java
@@ -771,7 +771,7 @@ public class RunnerContextImpl implements RunnerContext, ExecutionReporter {
if (subagentScope == null) {
return null;
}
- return subagentScope.getCallStatus().getSetup().getChildPlanJson();
+ return subagentScope.getCallStatus().getSetup().getPlanJson();
}
@Override
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
index 4368d87ca..a2b6d8e05 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
@@ -32,8 +32,6 @@ import org.apache.flink.agents.plan.JavaFunction;
import org.apache.flink.agents.plan.PythonFunction;
import org.apache.flink.agents.plan.actions.Action;
import org.apache.flink.agents.plan.resourceprovider.PythonResourceProvider;
-import org.apache.flink.agents.plan.resourceprovider.PythonSerializableResourceProvider;
-import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
import org.apache.flink.agents.plan.utils.CancellationUtils;
import org.apache.flink.agents.runtime.ResourceCache;
import org.apache.flink.agents.runtime.actionstate.ActionState;
@@ -1318,9 +1316,10 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
* notifies them through a single bridge listener.
*
* <p>The bridge listener is registered when the Python runtime has any sub-agent setup to
- * notify: an external Python setup materialized eagerly here, or a Python-compiled internal
- * sub-agent whose Java-owned child plan leaves its caller-facing handle to be built lazily in
- * the Python runtime (see {@link #hasPythonCompiledSubagent()}).
+ * notify: an external Python setup materialized eagerly here, or an internal sub-agent whose
+ * Java-owned child plan leaves its caller-facing handle to be built lazily in the Python
+ * runtime. Any Python action can call an internal sub-agent, regardless of the language in
+ * which its descriptor was declared.
*
* <p>Runs while the operator opens, after the Python bridge is up, because the Python runtime
* materializes the setups it owns.
@@ -1347,35 +1346,12 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
getRuntimeContext().getUserCodeClassLoader(), resourceCache));
}
}
- if (pythonSetupRegistered || hasPythonCompiledSubagent()) {
+ if (pythonSetupRegistered || (pythonBridge.isInitialized() && !internalSetups.isEmpty())) {
addTaskLifecycleListener(
new PythonTaskLifecycleListener(pythonBridge.getPythonActionExecutor()));
}
}
- /**
- * Whether the root plan declares a sub-agent compiled from Python. Such a sub-agent keeps a
- * caller-facing handle in the Python runtime even when its child plan dispatches on the Java
- * side — an internal sub-agent is Java-owned, so it is materialized as a Java {@link
- * InternalSubagentSetup} here and its Python handle is built lazily during the action body. The
- * operator must still bridge the task lifecycle into Python so that handle observes {@code
- * onActionPrepared} and a no-id submit can mint identities. A Python-compiled agent declares
- * its top-level sub-agents in the root plan, so this root-level check also covers nested ones
- * (the bridge listener is operator-wide and fires for child-scope actions too).
- */
- private boolean hasPythonCompiledSubagent() {
- Map<String, ResourceProvider> agentProviders =
- agentPlan
- .getResourceProviders()
- .getOrDefault(ResourceType.AGENT, Collections.emptyMap());
- for (ResourceProvider provider : agentProviders.values()) {
- if (provider instanceof PythonSerializableResourceProvider) {
- return true;
- }
- }
- return false;
- }
-
/**
* Materializes the AGENT resources of a nested child cache and recurses into the child plans of
* internal setups — the plan tree is a DAG because cycles are rejected at compile time, so this
@@ -1438,7 +1414,7 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
setup.getOrCreateChildCache(
getRuntimeContext().getUserCodeClassLoader(), resourceCache);
return new RunnerContextImpl.SubagentScope(
- metricGroup, setup.getChildPlan(), childCache, callStatus);
+ metricGroup, setup.getPlan(), childCache, callStatus);
}
/**
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java
index eeabd2e3b..bf115b06d 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java
@@ -27,7 +27,7 @@ import org.apache.flink.agents.plan.JavaFunction;
import org.apache.flink.agents.plan.PythonFunction;
import org.apache.flink.agents.plan.resourceprovider.JavaSerializableResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
-import org.apache.flink.agents.plan.subagent.InternalSubagentProvider;
+import org.apache.flink.agents.plan.subagent.InternalSubagentPlans;
import org.apache.flink.agents.runtime.PythonMCPResourceDiscovery;
import org.apache.flink.agents.runtime.ResourceCache;
import org.apache.flink.agents.runtime.env.EmbeddedPythonEnvironment;
@@ -55,10 +55,8 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import static org.apache.flink.agents.plan.actions.Utils.requiredVersions;
import static org.apache.flink.agents.plan.actions.Utils.supportAsync;
@@ -266,28 +264,22 @@ class PythonBridgeManager implements AutoCloseable {
* Every plan in the tree rooted at {@code plan}: the plan itself plus each internal sub-agent's
* child plan, recursively. Internal sub-agents are the only nested plans reachable from a
* parent's {@link ResourceType#AGENT} providers, through {@link
- * InternalSubagentProvider#getChildPlan()}. Compilation rejects a cycle, and the identity guard
- * also collapses a child plan shared by more than one parent so it is scanned once.
+ * InternalSubagentPlans#plan(ResourceProvider)}. Compilation rejects cycles. Each descriptor
+ * stores its own child-plan data, including when declarations share an Agent.
*/
static List<AgentPlan> planTree(AgentPlan plan) {
List<AgentPlan> plans = new ArrayList<>();
- collectPlanTree(plan, plans, Collections.newSetFromMap(new IdentityHashMap<>()));
+ collectPlanTree(plan, plans);
return plans;
}
- private static void collectPlanTree(
- AgentPlan plan, List<AgentPlan> acc, Set<AgentPlan> visited) {
- if (!visited.add(plan)) {
- return;
- }
+ private static void collectPlanTree(AgentPlan plan, List<AgentPlan> acc) {
acc.add(plan);
Map<String, ResourceProvider> agents =
plan.getResourceProviders()
.getOrDefault(ResourceType.AGENT, Collections.emptyMap());
for (ResourceProvider provider : agents.values()) {
- if (provider instanceof InternalSubagentProvider) {
- collectPlanTree(((InternalSubagentProvider) provider).getChildPlan(), acc, visited);
- }
+ InternalSubagentPlans.plan(provider).ifPresent(child -> collectPlanTree(child, acc));
}
}
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java b/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java
index 774205210..47e2b203d 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/subagent/InternalSubagentSetup.java
@@ -26,11 +26,14 @@ import org.apache.flink.agents.api.InputEvent;
import org.apache.flink.agents.api.context.DurableCallable;
import org.apache.flink.agents.api.context.RunnerContext;
import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
import org.apache.flink.agents.api.subagent.SubagentResult;
import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.plan.actions.Action;
+import org.apache.flink.agents.plan.subagent.InternalSubagentPlans;
import org.apache.flink.agents.runtime.ResourceCache;
import org.apache.flink.agents.runtime.async.ContinuationActionExecutor;
import org.apache.flink.agents.runtime.async.ContinuationNativeCallable;
@@ -49,7 +52,7 @@ import java.util.Map;
/**
* Runtime setup for an internal sub-agent: a child {@link AgentPlan} compiled from an {@code Agent}
* registered as an {@code AGENT} resource. The plan module serializes the child plan and scope
- * (through {@code InternalSubagentProvider}); this runtime class owns everything the invocation
+ * (through an ordinary resource descriptor); this runtime class owns everything the invocation
* needs at execution time.
*
* <p>Execution mode is the deferred one, inherited from {@link BaseDeferredSubagentSetup}: {@link
@@ -71,61 +74,53 @@ import java.util.Map;
*/
public class InternalSubagentSetup extends BaseDeferredSubagentSetup {
- private static final long serialVersionUID = 1L;
-
/**
* Serializes the child plan with the same configuration {@code RunnerContextImpl} uses for the
* active-scope plan JSON, so the eager materialization the operator runs at open and the lazy
* resolution a child action performs at call time key the Python scope cache by one string.
*/
- private static final ObjectMapper CHILD_PLAN_MAPPER =
+ private static final ObjectMapper PLAN_MAPPER =
new ObjectMapper().registerModule(new JavaTimeModule());
private final String scope;
- private final AgentPlan childPlan;
+ private final AgentPlan plan;
/** Nested map: sessionId → callId → callStatus. */
- private final transient Map<String, Map<String, InternalSubagentCallStatus>> callStatuses =
+ private final Map<String, Map<String, InternalSubagentCallStatus>> callStatuses =
new HashMap<>();
- private final transient Map<String, ResourceCache> childCaches = new HashMap<>();
+ private final Map<String, ResourceCache> childCaches = new HashMap<>();
- private final transient Map<Object, List<String>> keySessionIds = new HashMap<>();
+ private final Map<Object, List<String>> keySessionIds = new HashMap<>();
/**
* The runner contexts this setup registered a session-owner entry on, per session id. Used to
* unregister those entries when the owning record finishes.
*/
- private final transient Map<String, RunnerContextImpl> ownerContexts = new HashMap<>();
+ private final Map<String, RunnerContextImpl> ownerContexts = new HashMap<>();
- /** Lazily built event-to-action matcher for the child plan; not part of the serialized form. */
- private transient ActionMatcher actionMatcher;
+ /** Lazily built event-to-action matcher for the plan. */
+ private ActionMatcher actionMatcher;
- /**
- * Lazily serialized child plan JSON; not part of the serialized form. See {@link
- * #getChildPlanJson()}.
- */
- private transient String childPlanJson;
-
- public InternalSubagentSetup(String scope, AgentPlan childPlan) {
- // An internal sub-agent is built from a compiled child plan rather than a caller
- // descriptor, so it carries a synthetic self-named descriptor to satisfy the base
- // rebuild contract; that contract never reads the resource context, so it stays null.
- super(
- ResourceDescriptor.Builder.newBuilder(InternalSubagentSetup.class.getName())
- .build(),
- null);
- this.scope = scope;
- this.childPlan = childPlan;
+ /** Lazily serialized plan JSON. See {@link #getPlanJson()}. */
+ private String planJson;
+
+ public InternalSubagentSetup(ResourceDescriptor descriptor, ResourceContext context) {
+ super(descriptor, context);
+ this.scope = descriptor.getArgument(InternalSubagentResourceDescriptorBuilder.SCOPE);
+ if (scope == null || scope.isBlank()) {
+ throw new IllegalArgumentException("Internal sub-agent requires a scope");
+ }
+ this.plan = InternalSubagentPlans.decode(descriptor);
}
public String getScope() {
return scope;
}
- public AgentPlan getChildPlan() {
- return childPlan;
+ public AgentPlan getPlan() {
+ return plan;
}
/**
@@ -136,26 +131,26 @@ public class InternalSubagentSetup extends BaseDeferredSubagentSetup {
* resolves its resources against that plan. Both paths then key one Python scope cache, so a
* Python-owned resource of the scope is built exactly once.
*/
- public String getChildPlanJson() {
- if (childPlanJson == null) {
+ public String getPlanJson() {
+ if (planJson == null) {
try {
- childPlanJson = CHILD_PLAN_MAPPER.writeValueAsString(childPlan);
+ planJson = PLAN_MAPPER.writeValueAsString(plan);
} catch (JsonProcessingException e) {
throw new IllegalStateException(
"Failed to serialize the child plan for scope " + scope, e);
}
}
- return childPlanJson;
+ return planJson;
}
/**
* Matches the child plan's actions against a forwarded event, applying the same event-type and
* condition-expression filtering the root router uses. Built lazily because the index depends
- * only on the immutable child plan, and this class is serialized as a resource.
+ * only on the immutable child plan.
*/
public List<Action> matchActions(Event event) {
if (actionMatcher == null) {
- actionMatcher = new ActionMatcher(childPlan);
+ actionMatcher = new ActionMatcher(plan);
}
return actionMatcher.match(event);
}
@@ -346,12 +341,12 @@ public class InternalSubagentSetup extends BaseDeferredSubagentSetup {
*/
public ResourceCache getOrCreateChildCache(
ClassLoader userCodeClassLoader, ResourceCache rootResourceCache) {
- String planJson = getChildPlanJson();
+ String planJson = getPlanJson();
return childCaches.computeIfAbsent(
scope,
k ->
new ResourceCache(
- childPlan.getResourceProviders(),
+ plan.getResourceProviders(),
userCodeClassLoader,
rootResourceCache,
planJson));
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/ResourceCacheTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/ResourceCacheTest.java
index a46c19aa9..475520f1e 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/ResourceCacheTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/ResourceCacheTest.java
@@ -41,7 +41,6 @@ import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.plan.resource.python.PythonChatModelSetup;
import org.apache.flink.agents.plan.resource.python.PythonResourceAdapter;
import org.apache.flink.agents.plan.resource.python.PythonResourceWrapper;
-import org.apache.flink.agents.plan.resourceprovider.JavaSerializableResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.PythonResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
import org.apache.flink.agents.runtime.python.utils.PythonActionExecutor;
@@ -303,13 +302,26 @@ public class ResourceCacheTest {
Map<ResourceType, Map<String, ResourceProvider>> childProviders = new HashMap<>();
childProviders.put(ResourceType.AGENT, new HashMap<>(Map.of("external", pythonAgent)));
InternalSubagentSetup setup =
- new InternalSubagentSetup("child", new AgentPlan(Map.of(), childProviders));
+ new InternalSubagentSetup(
+ org.apache.flink.agents.api.resource.ResourceDescriptor.Builder.newBuilder(
+ org.apache.flink.agents.api.subagent
+ .InternalSubagentResourceDescriptorBuilder
+ .SETUP_CLASS)
+ .addInitialArgument("scope", "child")
+ .addInitialArgument(
+ "plan",
+ new com.fasterxml.jackson.databind.ObjectMapper()
+ .convertValue(
+ new AgentPlan(Map.of(), childProviders),
+ Map.class))
+ .build(),
+ null);
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
ResourceCache rootCache = new ResourceCache(new HashMap<>(), classLoader);
PythonActionExecutor executor = mock(PythonActionExecutor.class);
TestPythonHandle handle = new TestPythonHandle();
- when(executor.eagerMaterialize(ResourceType.AGENT, setup.getChildPlanJson()))
+ when(executor.eagerMaterialize(ResourceType.AGENT, setup.getPlanJson()))
.thenReturn(Collections.singletonMap("external", handle));
rootCache.setPythonActionExecutor(executor);
@@ -321,7 +333,7 @@ public class ResourceCacheTest {
assertThat(materialized).containsExactly(handle);
assertThat(childCache.getResource("external", ResourceType.AGENT)).isSameAs(handle);
// The runtime is asked for the child plan's resources, keyed by the child plan JSON.
- verify(executor).eagerMaterialize(ResourceType.AGENT, setup.getChildPlanJson());
+ verify(executor).eagerMaterialize(ResourceType.AGENT, setup.getPlanJson());
}
@Test
@@ -627,8 +639,11 @@ public class ResourceCacheTest {
Map<String, ResourceProvider> agentProviders = new HashMap<>();
agentProviders.put(
"reviewer",
- JavaSerializableResourceProvider.createResourceProvider(
- "reviewer", ResourceType.AGENT, new TestAgentSetup()));
+ new org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider(
+ "reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(TestAgentSetup.class.getName())
+ .build()));
providers.put(ResourceType.AGENT, agentProviders);
ResourceCache cache = new ResourceCache(providers);
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/PythonBridgeManagerTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/PythonBridgeManagerTest.java
index 96232b8e8..26086b3f6 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/PythonBridgeManagerTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/PythonBridgeManagerTest.java
@@ -25,9 +25,9 @@ import org.apache.flink.agents.api.skills.Skills;
import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.plan.PythonFunction;
import org.apache.flink.agents.plan.actions.Action;
+import org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.PythonResourceProvider;
import org.apache.flink.agents.plan.resourceprovider.ResourceProvider;
-import org.apache.flink.agents.plan.subagent.InternalSubagentProvider;
import org.apache.flink.agents.runtime.env.PythonEnvironmentManager;
import org.apache.flink.agents.runtime.memory.Mem0LongTermMemory;
import org.apache.flink.agents.runtime.python.utils.PythonActionExecutor;
@@ -150,9 +150,10 @@ class PythonBridgeManagerTest {
* {@code open()} decides whether to start the shared Python runtime by inspecting plans, and it
* runs before sub-agent setups exist, so the inspection must walk the whole plan tree. An
* internal sub-agent's child plan is reachable only through {@link
- * InternalSubagentProvider#getChildPlan()}; a Java root with a Python-bearing descendant still
- * has to initialize Python, or the descendant's eager materialization fails its "no Python
- * runtime" guard. Pins the recursion through two nested layers and both detection paths.
+ * org.apache.flink.agents.plan.subagent.InternalSubagentPlans#plan}; a Java root with a
+ * Python-bearing descendant still has to initialize Python, or the descendant's eager
+ * materialization fails its "no Python runtime" guard. Pins the recursion through two nested
+ * layers and both detection paths.
*/
@Test
void planTreeRecursesThroughNestedInternalSubagentChildPlans() throws Exception {
@@ -175,8 +176,7 @@ class PythonBridgeManagerTest {
new AgentPlan(Map.of(pythonAction.getName(), pythonAction), grandchildProviders);
// Middle layer: Java-only, wraps the grandchild as an internal sub-agent.
- InternalSubagentProvider grandchildProvider =
- new InternalSubagentProvider("grandchild", grandchild);
+ JavaResourceProvider grandchildProvider = childProvider("grandchild", grandchild);
Map<ResourceType, Map<String, ResourceProvider>> childProviders = new HashMap<>();
childProviders.put(
ResourceType.AGENT, new HashMap<>(Map.of("grandchild", grandchildProvider)));
@@ -184,13 +184,14 @@ class PythonBridgeManagerTest {
// Root: one Java action, wraps the child as an internal sub-agent.
Action javaAction = TestActions.noopAction();
- InternalSubagentProvider childProvider = new InternalSubagentProvider("child", child);
+ JavaResourceProvider childProvider = childProvider("child", child);
Map<ResourceType, Map<String, ResourceProvider>> rootProviders = new HashMap<>();
rootProviders.put(ResourceType.AGENT, new HashMap<>(Map.of("child", childProvider)));
AgentPlan root = new AgentPlan(Map.of(javaAction.getName(), javaAction), rootProviders);
// Depth-first pre-order: the root, then each descendant through its sub-agent provider.
- assertThat(PythonBridgeManager.planTree(root)).containsExactly(root, child, grandchild);
+ assertThat(PythonBridgeManager.planTree(root)).hasSize(3);
+ assertThat(PythonBridgeManager.planTree(root).get(2).getActions()).containsKey("pyAction");
// The root alone is Java-only; both Python signals live two layers down and must surface.
assertThat(PythonBridgeManager.treeContainsPythonResource(root)).isTrue();
assertThat(PythonBridgeManager.treeContainsPythonAction(root)).isTrue();
@@ -220,7 +221,7 @@ class PythonBridgeManagerTest {
// Root is Java-only and reaches that child solely through an internal sub-agent provider.
Action javaAction = TestActions.noopAction();
- InternalSubagentProvider childProvider = new InternalSubagentProvider("child", child);
+ JavaResourceProvider childProvider = childProvider("child", child);
Map<ResourceType, Map<String, ResourceProvider>> rootProviders = new HashMap<>();
rootProviders.put(ResourceType.AGENT, new HashMap<>(Map.of("child", childProvider)));
AgentPlan root = new AgentPlan(Map.of(javaAction.getName(), javaAction), rootProviders);
@@ -355,4 +356,19 @@ class PythonBridgeManagerTest {
field.setAccessible(true);
field.set(bridge, value);
}
+
+ private static JavaResourceProvider childProvider(String name, AgentPlan plan) {
+ return new JavaResourceProvider(
+ name,
+ ResourceType.AGENT,
+ org.apache.flink.agents.api.resource.ResourceDescriptor.Builder.newBuilder(
+ org.apache.flink.agents.api.subagent
+ .InternalSubagentResourceDescriptorBuilder.SETUP_CLASS)
+ .addInitialArgument("scope", name)
+ .addInitialArgument(
+ "plan",
+ new com.fasterxml.jackson.databind.ObjectMapper()
+ .convertValue(plan, Map.class))
+ .build());
+ }
}
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseAsyncSubagentSetupTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseAsyncSubagentSetupTest.java
index 411f9e88e..eb7db713f 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseAsyncSubagentSetupTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseAsyncSubagentSetupTest.java
@@ -514,7 +514,7 @@ public class BaseAsyncSubagentSetupTest {
private static AgentPlan plan(String actionMethod, MockAsyncSubagentSetup setup)
throws Exception {
Agent agent = new Agent();
- agent.addResource(RESOURCE_NAME, ResourceType.AGENT, setup);
+ agent.addResource(RESOURCE_NAME, ResourceType.AGENT, setup.getDescriptor());
agent.addAction(
new String[] {InputEvent.EVENT_TYPE},
BaseAsyncSubagentSetupTest.class.getMethod(
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseSubagentSetupTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseSubagentSetupTest.java
index aae90b6ff..f49fcd38c 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseSubagentSetupTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/BaseSubagentSetupTest.java
@@ -174,8 +174,10 @@ public class BaseSubagentSetupTest {
@Test
void setupsSharingOneActionAllocateDisjointIds() throws Exception {
Agent agent = new Agent();
- agent.addResource("stream-a", ResourceType.AGENT, new AllocatingCaptureSetup());
- agent.addResource("stream-b", ResourceType.AGENT, new AllocatingCaptureSetup());
+ agent.addResource(
+ "stream-a", ResourceType.AGENT, new AllocatingCaptureSetup().getDescriptor());
+ agent.addResource(
+ "stream-b", ResourceType.AGENT, new AllocatingCaptureSetup().getDescriptor());
agent.addAction(
new String[] {InputEvent.EVENT_TYPE},
BaseSubagentSetupTest.class.getMethod(
@@ -279,7 +281,8 @@ public class BaseSubagentSetupTest {
@Test
void materializationInjectsSubagentNames() throws Exception {
Agent rootAgent = new Agent();
- rootAgent.addResource("root-setup", ResourceType.AGENT, new AllocatingCaptureSetup());
+ rootAgent.addResource(
+ "root-setup", ResourceType.AGENT, new AllocatingCaptureSetup().getDescriptor());
ResourceCache rootCache =
new ResourceCache(new AgentPlan(rootAgent).getResourceProviders());
BaseSubagentSetup rootSetup =
@@ -289,7 +292,8 @@ public class BaseSubagentSetupTest {
// A child cache injects the local resource name as resolved in the child plan: the
// enclosing scope is routing-level context the setup never sees.
Agent childAgent = new Agent();
- childAgent.addResource("nested", ResourceType.AGENT, new AllocatingCaptureSetup());
+ childAgent.addResource(
+ "nested", ResourceType.AGENT, new AllocatingCaptureSetup().getDescriptor());
ResourceCache childCache =
new ResourceCache(
new AgentPlan(childAgent).getResourceProviders(),
@@ -311,7 +315,8 @@ public class BaseSubagentSetupTest {
private static AgentPlan plan() throws Exception {
Agent agent = new Agent();
- agent.addResource(RESOURCE_NAME, ResourceType.AGENT, new AllocatingCaptureSetup());
+ agent.addResource(
+ RESOURCE_NAME, ResourceType.AGENT, new AllocatingCaptureSetup().getDescriptor());
agent.addAction(
new String[] {InputEvent.EVENT_TYPE},
BaseSubagentSetupTest.class.getMethod(
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/DeferredSubagentSubmitTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/DeferredSubagentSubmitTest.java
index b274e90eb..cf26982bb 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/DeferredSubagentSubmitTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/DeferredSubagentSubmitTest.java
@@ -320,7 +320,7 @@ public class DeferredSubagentSubmitTest {
private static AgentPlan plan(String actionMethod, BaseDeferredSubagentSetup setup)
throws Exception {
Agent agent = new Agent();
- agent.addResource(RESOURCE_NAME, ResourceType.AGENT, setup);
+ agent.addResource(RESOURCE_NAME, ResourceType.AGENT, setup.getDescriptor());
agent.addAction(
new String[] {InputEvent.EVENT_TYPE},
DeferredSubagentSubmitTest.class.getMethod(
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java
index 09e0cd3e4..a932b283c 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentCallTest.java
@@ -18,13 +18,18 @@
package org.apache.flink.agents.runtime.subagent;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.agents.api.Event;
import org.apache.flink.agents.api.InputEvent;
import org.apache.flink.agents.api.OutputEvent;
import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.chat.messages.ToolCallBlock;
import org.apache.flink.agents.api.context.MemoryObject;
import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.event.ToolRequestEvent;
+import org.apache.flink.agents.api.event.ToolResponseEvent;
import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
import org.apache.flink.agents.api.subagent.SubagentFuture;
import org.apache.flink.agents.api.subagent.SubagentResult;
import org.apache.flink.agents.api.subagent.SubagentSetup;
@@ -43,6 +48,7 @@ import org.junit.jupiter.api.condition.JRE;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
@@ -299,6 +305,56 @@ public class InternalSubagentCallTest {
assertThat(output).containsExactly("saw:<missing>", "saw:<missing>");
}
+ public static void requestChildTool(Event event, RunnerContext ctx) {
+ ctx.sendEvent(
+ new ToolRequestEvent(
+ "model",
+ List.of(
+ new ToolCallBlock(
+ "call-1",
+ "_subagent_child",
+ Map.of("question", "hello")))));
+ }
+
+ public static void receiveChildTool(Event event, RunnerContext ctx) {
+ ToolResponseEvent response = ToolResponseEvent.fromEvent(event);
+ ctx.sendEvent(new OutputEvent(response.getResponses().get("call-1").getText()));
+ }
+
+ @Test
+ @Timeout(60)
+ void modelToolCallExecutesInternalDescriptorAfterPlanTransfer() throws Exception {
+ Agent root = new Agent();
+ root.addResource(
+ "child",
+ ResourceType.AGENT,
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new EchoChildAgent())
+ .description("Answers questions")
+ .inputSchema(
+ "{\"type\":\"object\",\"properties\":{\"question\":{\"type\":\"string\"}}}")
+ .build());
+ root.addAction(
+ new String[] {InputEvent.EVENT_TYPE},
+ InternalSubagentCallTest.class.getMethod(
+ "requestChildTool", Event.class, RunnerContext.class));
+ root.addAction(
+ new String[] {ToolResponseEvent.EVENT_TYPE},
+ InternalSubagentCallTest.class.getMethod(
+ "receiveChildTool", Event.class, RunnerContext.class));
+ AgentPlan compiled = new AgentPlan(root);
+ // The test supplies the model's tool request and consumes its response itself.
+ Map<String, org.apache.flink.agents.plan.actions.Action> actions =
+ new java.util.LinkedHashMap<>(compiled.getActions());
+ actions.remove("chat_model_action");
+ compiled = new AgentPlan(actions, compiled.getResourceProviders(), compiled.getConfig());
+ ObjectMapper mapper = new ObjectMapper();
+ AgentPlan transferred =
+ mapper.readValue(mapper.writeValueAsString(compiled), AgentPlan.class);
+ assertThat(run(transferred, 1L)).containsExactly("[\"echo:{question=hello}\"]");
+ assertThat(CHILD_INVOCATIONS).containsExactly("{question=hello}");
+ }
+
// --- helpers ---
private static String firstOutput(SubagentResult result) {
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentDescriptorTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentDescriptorTest.java
new file mode 100644
index 000000000..0786c9ee3
--- /dev/null
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentDescriptorTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.runtime.subagent;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
+import org.apache.flink.agents.api.subagent.SubagentSetup;
+import org.apache.flink.agents.plan.AgentPlan;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class InternalSubagentDescriptorTest {
+ private static final String SCHEMA =
+ "{\"type\":\"object\",\"properties\":{\"question\":{\"type\":\"string\"}}}";
+
+ private static final class ChatSetup extends BaseChatModelSetup {
+ ChatSetup(ResourceContext context) {
+ super(new ResourceDescriptor("chat", Map.of("subagents", List.of("child"))), context);
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return Map.of();
+ }
+ }
+
+ @Test
+ void restoredInternalDescriptorDeclaresAModelCallable() throws Exception {
+ Agent root = new Agent();
+ root.addResource(
+ "child",
+ ResourceType.AGENT,
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(new Agent())
+ .description("Answers questions")
+ .inputSchema(SCHEMA)
+ .build());
+ ObjectMapper mapper = new ObjectMapper();
+ AgentPlan restored =
+ mapper.readValue(mapper.writeValueAsString(new AgentPlan(root)), AgentPlan.class);
+ ResourceContext context = mock(ResourceContext.class);
+ SubagentSetup child =
+ (SubagentSetup)
+ restored.getResourceProviders()
+ .get(ResourceType.AGENT)
+ .get("child")
+ .provide(context);
+ assertThat(child).isInstanceOf(InternalSubagentSetup.class);
+ assertThat(child.getResourceContext()).isSameAs(context);
+ assertThat(child.getDescription()).isEqualTo("Answers questions");
+ assertThat(child.getInputSchema()).isEqualTo(SCHEMA);
+ when(context.getResource("child", ResourceType.AGENT)).thenReturn(child);
+ ChatSetup chat = new ChatSetup(context);
+ chat.open();
+ assertThat(chat.getTools()).hasSize(1);
+ assertThat(chat.getTools().get(0).getName()).isEqualTo("_subagent_child");
+ }
+}
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentGuardTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentGuardTest.java
index 710232143..8ab778aec 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentGuardTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentGuardTest.java
@@ -40,10 +40,23 @@ public class InternalSubagentGuardTest {
@EnabledForJreRange(max = JRE.JAVA_20)
void resolvingWithoutContinuationsFailsFast() throws Exception {
InternalSubagentSetup setup =
- new InternalSubagentSetup("scope", new AgentPlan(new Agent()));
+ new InternalSubagentSetup(descriptor(new AgentPlan(new Agent())), null);
assertThatThrownBy(() -> setup.prepare(null, "prompt", "session", "call"))
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining("stackful suspension");
}
+
+ private static org.apache.flink.agents.api.resource.ResourceDescriptor descriptor(
+ AgentPlan plan) {
+ return org.apache.flink.agents.api.resource.ResourceDescriptor.Builder.newBuilder(
+ org.apache.flink.agents.api.subagent
+ .InternalSubagentResourceDescriptorBuilder.SETUP_CLASS)
+ .addInitialArgument("scope", "scope")
+ .addInitialArgument(
+ "plan",
+ new com.fasterxml.jackson.databind.ObjectMapper()
+ .convertValue(plan, java.util.Map.class))
+ .build();
+ }
}
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentMatchTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentMatchTest.java
index 48821d5ef..c3de206f8 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentMatchTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/InternalSubagentMatchTest.java
@@ -46,7 +46,8 @@ class InternalSubagentMatchTest {
void scopeMatchingAppliesTypeAndConditionFilters() throws Exception {
Action typed = action("typed", InputEvent.EVENT_TYPE);
Action conditional = action("conditional", "attributes.route == 'extra'");
- InternalSubagentSetup setup = new InternalSubagentSetup("scope", plan(typed, conditional));
+ InternalSubagentSetup setup =
+ new InternalSubagentSetup(descriptor(plan(typed, conditional)), null);
List<Action> plain = setup.matchActions(new InputEvent("plain"));
assertThat(plain).containsExactly(typed);
@@ -77,4 +78,17 @@ class InternalSubagentMatchTest {
}
return new AgentPlan(byName);
}
+
+ private static org.apache.flink.agents.api.resource.ResourceDescriptor descriptor(
+ AgentPlan plan) {
+ return org.apache.flink.agents.api.resource.ResourceDescriptor.Builder.newBuilder(
+ org.apache.flink.agents.api.subagent
+ .InternalSubagentResourceDescriptorBuilder.SETUP_CLASS)
+ .addInitialArgument("scope", "scope")
+ .addInitialArgument(
+ "plan",
+ new com.fasterxml.jackson.databind.ObjectMapper()
+ .convertValue(plan, java.util.Map.class))
+ .build();
+ }
}
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/external/ExternalAgentSubagentSetupTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/external/ExternalAgentSubagentSetupTest.java
index 4e652384a..6a361280e 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/external/ExternalAgentSubagentSetupTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/external/ExternalAgentSubagentSetupTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.agents.api.context.RunnerContext;
import org.apache.flink.agents.api.resource.ResourceType;
import org.apache.flink.agents.api.subagent.SubagentFuture;
import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.api.subagent.SubagentSetup;
import org.apache.flink.agents.plan.AgentConfiguration;
import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.runtime.operator.ActionExecutionOperator;
@@ -469,7 +470,10 @@ public class ExternalAgentSubagentSetupTest {
private static AgentPlan plan(String actionMethod, Object setup) throws Exception {
Agent agent = new Agent();
- agent.addResource(RESOURCE_NAME, ResourceType.AGENT, setup);
+ agent.addResource(
+ RESOURCE_NAME,
+ ResourceType.AGENT,
+ setup instanceof SubagentSetup ? ((SubagentSetup) setup).getDescriptor() : setup);
agent.addAction(
new String[] {InputEvent.EVENT_TYPE},
ExternalAgentSubagentSetupTest.class.getMethod(