[api][plan][runtime] Use generic resource descriptors for sub-agents

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

@@ -23,6 +23,7 @@
 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;
@@ -114,20 +115,23 @@ public void addResourcesIfAbsent(Map<ResourceType, Map<String, Object>> resource
      * @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)) {
             throw new IllegalArgumentException(String.format("%s %s already defined.", type, name));
         }
         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()));

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();
+    }
+}

api/src/main/java/org/apache/flink/agents/api/subagent/SubagentSetup.java

@@ -20,10 +20,10 @@
 
 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 @@
  * 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,

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");
+    }
+}

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 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 @@ void multipleSubagentsRegistered() {
         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));
+    }
 }