[api][plan][runtime] Introduce AGENT resource type and sub-agent invocation API

api/pom.xml

@@ -68,4 +68,21 @@ under the License.
         </dependency>
     </dependencies>
 
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>3.4.2</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
 </project>

api/src/main/java/org/apache/flink/agents/api/resource/ResourceType.java

@@ -33,7 +33,8 @@ public enum ResourceType {
     TOOL("tool"),
     MCP_SERVER("mcp_server"),
     SKILLS("skills"),
-    MODEL_ROUTER("model_router");
+    MODEL_ROUTER("model_router"),
+    AGENT("agent");
 
     private final String value;
 

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

@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+/**
+ * Handle for one sub-agent invocation, identified by the {@code (sessionId, callId)} pair that keys
+ * the invocation.
+ */
+public abstract class SubagentFuture {
+
+    private final String sessionId;
+    private final String callId;
+
+    protected SubagentFuture(String sessionId, String callId) {
+        this.sessionId = sessionId;
+        this.callId = callId;
+    }
+
+    public String getSessionId() {
+        return sessionId;
+    }
+
+    public String getCallId() {
+        return callId;
+    }
+
+    /** Whether the invocation has reached a terminal state. */
+    public abstract boolean isDone();
+
+    /**
+     * Resolves the invocation, waiting until it reaches a terminal state. Failures converge into a
+     * failed {@link SubagentResult} rather than a separately reported exceptional completion.
+     */
+    public abstract SubagentResult await() throws Exception;
+
+    /** Requests cancellation of the invocation. */
+    public void cancel() {}
+
+    /** Groups this handle with others to be resolved together through {@link SubagentFutures}. */
+    public abstract SubagentFutures combine(SubagentFuture... others);
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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 java.util.List;
+
+/**
+ * A group of sub-agent handles to be resolved together. A group is not itself an invocation and
+ * carries no {@code (sessionId, callId)} identity.
+ */
+public abstract class SubagentFutures {
+
+    /** Whether every handle in the group has reached a terminal state. */
+    public abstract boolean isDone();
+
+    /**
+     * Waits for every handle in the group and returns their outcomes in the order the handles were
+     * added. Like {@link SubagentFuture#await()}, failures surface through failed {@link
+     * SubagentResult}s.
+     */
+    public abstract List<SubagentResult> awaitAll() throws Exception;
+
+    /** Requests cancellation of every handle in the group. */
+    public void cancel() {}
+
+    /** Adds more handles to the group. */
+    public abstract SubagentFutures combine(SubagentFuture... others);
+}

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

@@ -0,0 +1,119 @@
+/*
+ * 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.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+
+/**
+ * Outcome of a sub-agent call issued through {@link SubagentSetup}. A successful outcome carries a
+ * JSON-serializable payload, and a failed one carries a serializable error message.
+ *
+ * <p>Implementations capture their internal failures into a result through {@link #error} instead
+ * of throwing, so callers inspect {@link #isSuccess()} rather than catching. Because the failure is
+ * carried as a message rather than a live exception, the whole result can be persisted through
+ * durable execution and survive a failover.
+ */
+public class SubagentResult implements Serializable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(SubagentResult.class);
+
+    private static final long serialVersionUID = 1L;
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private final boolean success;
+    private final Object result;
+    private final String errorMessage;
+
+    @JsonCreator
+    public SubagentResult(
+            @JsonProperty("success") boolean success,
+            @JsonProperty("result") Object result,
+            @JsonProperty("errorMessage") String errorMessage) {
+        this.success = success;
+        this.result = result;
+        this.errorMessage = errorMessage;
+    }
+
+    /** Creates a successful result carrying the given value. */
+    public static SubagentResult ok(Object result) {
+        return new SubagentResult(true, result, null);
+    }
+
+    /**
+     * Creates a failed result carrying the exception's type and message. The full stack trace is
+     * logged here rather than persisted, keeping the durable payload bounded.
+     */
+    public static SubagentResult error(Exception exception) {
+        if (exception == null) {
+            return new SubagentResult(false, null, null);
+        }
+        LOG.warn("Sub-agent call failed; persisting the exception summary.", exception);
+        return new SubagentResult(false, null, summaryOf(exception));
+    }
+
+    /** Creates a failed result carrying the given message. */
+    public static SubagentResult error(String errorMessage) {
+        return new SubagentResult(false, null, errorMessage);
+    }
+
+    private static String summaryOf(Exception exception) {
+        return exception.getClass().getName() + ": " + exception.getMessage();
+    }
+
+    public boolean isSuccess() {
+        return success;
+    }
+
+    public Object getResult() {
+        return result;
+    }
+
+    /**
+     * Returns the payload converted to {@code resultClass}.
+     *
+     * <p>Durable recovery re-binds the persisted payload through a plain {@link ObjectMapper}
+     * without polymorphic typing, so after a failover replay {@link #getResult()} hands back a
+     * {@code LinkedHashMap} instead of the caller's type. This accessor converts the payload to the
+     * expected class uniformly on both the first execution and a replay.
+     */
+    public <T> T getResult(Class<T> resultClass) {
+        return OBJECT_MAPPER.convertValue(result, resultClass);
+    }
+
+    public String getErrorMessage() {
+        return errorMessage;
+    }
+
+    /**
+     * Reconstructs an exception carrying the stored summary as its message, or null if this result
+     * is successful.
+     */
+    @JsonIgnore
+    public Exception getException() {
+        return success ? null : new RuntimeException(errorMessage);
+    }
+}

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

@@ -0,0 +1,58 @@
+/*
+ * 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.annotation.JsonIgnore;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.resource.SerializableResource;
+
+/**
+ * Caller-facing definition of a sub-agent, registered in the agent plan as an {@code AGENT}
+ * resource.
+ */
+public abstract class SubagentSetup extends SerializableResource {
+
+    @Override
+    @JsonIgnore
+    public ResourceType getResourceType() {
+        return ResourceType.AGENT;
+    }
+
+    /**
+     * Issues a new invocation with an implementation-assigned identity. This is the preferred form.
+     */
+    public abstract SubagentFuture submit(RunnerContext ctx, Object prompt) throws Exception;
+
+    /**
+     * Issues an invocation that continues the conversation of an earlier invocation. Pass the
+     * {@code sessionId} of the earlier invocation to continue it. The session id is available on
+     * the handle returned by that invocation. Whether a conversation can be continued across
+     * actions is up to the concrete implementation.
+     */
+    public abstract SubagentFuture submit(RunnerContext ctx, Object prompt, String sessionId)
+            throws Exception;
+
+    /**
+     * Issues an invocation under the given {@code (sessionId, callId)} identity. This form is
+     * reserved for implementation use.
+     */
+    public abstract SubagentFuture submit(
+            RunnerContext ctx, Object prompt, String sessionId, String callId) throws Exception;
+}

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

@@ -292,6 +292,7 @@ public static LoadedFile buildAgents(Path path) {
         addSharedDescriptors(
                 sharedResources, ResourceType.VECTOR_STORE, doc.getVectorStores(), path);
         addSharedDescriptors(sharedResources, ResourceType.MCP_SERVER, doc.getMcpServers(), path);
+        addSharedDescriptors(sharedResources, ResourceType.AGENT, doc.getSubagents(), path);
 
         for (ToolSpec t : doc.getTools()) {
             if (sharedResources.get(ResourceType.TOOL).put(t.getName(), buildTool(t)) != null) {
@@ -411,6 +412,7 @@ private static Agent buildAgent(AgentSpec spec) {
         addAgentDescriptors(agent, ResourceType.EMBEDDING_MODEL, spec.getEmbeddingModelSetups());
         addAgentDescriptors(agent, ResourceType.VECTOR_STORE, spec.getVectorStores());
         addAgentDescriptors(agent, ResourceType.MCP_SERVER, spec.getMcpServers());
+        addAgentDescriptors(agent, ResourceType.AGENT, spec.getSubagents());
 
         for (ToolSpec t : spec.getTools()) {
             agent.addResource(t.getName(), ResourceType.TOOL, buildTool(t));

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

@@ -40,6 +40,7 @@ public final class AgentSpec {
     private final List<DescriptorSpec> embeddingModelSetups;
     private final List<DescriptorSpec> vectorStores;
     private final List<DescriptorSpec> mcpServers;
+    private final List<DescriptorSpec> subagents;
 
     @JsonCreator
     public AgentSpec(
@@ -55,7 +56,8 @@ public AgentSpec(
                     List<DescriptorSpec> embeddingModelConnections,
             @JsonProperty("embedding_model_setups") List<DescriptorSpec> embeddingModelSetups,
             @JsonProperty("vector_stores") List<DescriptorSpec> vectorStores,
-            @JsonProperty("mcp_servers") List<DescriptorSpec> mcpServers) {
+            @JsonProperty("mcp_servers") List<DescriptorSpec> mcpServers,
+            @JsonProperty("subagents") List<DescriptorSpec> subagents) {
         this.name = name;
         this.description = description;
         this.prompts = orEmpty(prompts);
@@ -68,6 +70,7 @@ public AgentSpec(
         this.embeddingModelSetups = orEmpty(embeddingModelSetups);
         this.vectorStores = orEmpty(vectorStores);
         this.mcpServers = orEmpty(mcpServers);
+        this.subagents = orEmpty(subagents);
     }
 
     private static <T> List<T> orEmpty(List<T> list) {
@@ -121,4 +124,8 @@ public List<DescriptorSpec> getVectorStores() {
     public List<DescriptorSpec> getMcpServers() {
         return mcpServers;
     }
+
+    public List<DescriptorSpec> getSubagents() {
+        return subagents;
+    }
 }