[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/context/RunnerContext.java

@@ -146,6 +146,12 @@ public interface RunnerContext {
      */
     <T> T durableExecuteAsync(DurableCallable<T> callable) throws Exception;
 
+    /** Creates a new session id for a sub-agent call. */
+    String nextSessionId();
+
+    /** Creates a new call id for a sub-agent invocation under the given session. */
+    String nextCallId(String sessionId);
+
     /** Clean up the resource. */
     void close() throws Exception;
 }

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

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

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

@@ -0,0 +1,65 @@
+/*
+ * 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.context.DurableCallable;
+
+/**
+ * Convenience base for the {@link DurableCallable} returned by {@code asAsyncCallable}.
+ *
+ * <p>Keys the durable call by the framework-assigned identity as {@code sessionId#callId} (the
+ * {@link SubagentSetup} contract) and captures exceptions thrown by {@link #callInternal()} into
+ * {@link Result#error(Exception)}, so failures are reported through the result rather than thrown.
+ * Implementations only provide {@link #callInternal()}.
+ */
+public abstract class BaseSubagentCallable implements DurableCallable<Result> {
+
+    private final String sessionId;
+    private final String callId;
+
+    protected BaseSubagentCallable(String sessionId, String callId) {
+        this.sessionId = sessionId;
+        this.callId = callId;
+    }
+
+    @Override
+    public String getId() {
+        return sessionId + "#" + callId;
+    }
+
+    @Override
+    public Class<Result> getResultClass() {
+        return Result.class;
+    }
+
+    @Override
+    public final Result call() {
+        try {
+            return Result.ok(callInternal());
+        } catch (Exception e) {
+            return Result.error(e);
+        }
+    }
+
+    /**
+     * Performs the invocation and returns the JSON-serializable payload. Thrown exceptions are
+     * captured into a failed {@link Result}.
+     */
+    protected abstract Object callInternal() throws Exception;
+}

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

@@ -0,0 +1,98 @@
+/*
+ * 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 java.io.PrintWriter;
+import java.io.Serializable;
+import java.io.StringWriter;
+
+/**
+ * Outcome of a {@link Subagent} call.
+ *
+ * <p>Sub-agent implementations should capture internal failures into a {@code Result} (via {@link
+ * #error}) instead of throwing, so callers can inspect {@link #isSuccess()} without try/catch.
+ *
+ * <p>The failure cause is carried as a serializable {@code errorMessage} — the full stack trace of
+ * the failure — rather than a live exception, so that a {@code Result} can be persisted through
+ * durable execution.
+ */
+public class Result implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final boolean success;
+    private final Object result;
+    private final String errorMessage;
+
+    @JsonCreator
+    public Result(
+            @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 Result ok(Object result) {
+        return new Result(true, result, null);
+    }
+
+    /** Creates a failed result carrying the full stack trace of the given exception. */
+    public static Result error(Exception exception) {
+        return new Result(false, null, exception == null ? null : stackTraceOf(exception));
+    }
+
+    /** Creates a failed result carrying the given message. */
+    public static Result error(String errorMessage) {
+        return new Result(false, null, errorMessage);
+    }
+
+    private static String stackTraceOf(Exception exception) {
+        StringWriter writer = new StringWriter();
+        exception.printStackTrace(new PrintWriter(writer));
+        return writer.toString();
+    }
+
+    public boolean isSuccess() {
+        return success;
+    }
+
+    public Object getResult() {
+        return result;
+    }
+
+    public String getErrorMessage() {
+        return errorMessage;
+    }
+
+    /**
+     * Reconstructs an exception carrying the stored stack trace 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/Subagent.java

@@ -0,0 +1,47 @@
+/*
+ * 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.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+
+/**
+ * Caller-facing interface for all sub-agents (external and internal).
+ *
+ * <p>An invocation is identified by a {@code (sessionId, callId)} pair; the session groups a
+ * conversation across invocations. Both ids are assigned by the framework ({@link
+ * RunnerContext#nextSessionId()} and {@link RunnerContext#nextCallId(String)}): callers may supply
+ * a session id to continue a prior session but never supply a call id.
+ */
+public interface Subagent {
+
+    /** Synchronously invokes the sub-agent, creating a new session. */
+    Result call(RunnerContext ctx, Object prompt) throws Exception;
+
+    /** Synchronously invokes the sub-agent continuing {@code sessionId}. */
+    Result call(RunnerContext ctx, Object prompt, String sessionId) throws Exception;
+
+    /** Produces a deferred, durable callable for this sub-agent call, creating a new session. */
+    DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object prompt);
+
+    /**
+     * Produces a deferred, durable callable for this sub-agent call continuing {@code sessionId}.
+     */
+    DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object prompt, String sessionId);
+}

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

@@ -0,0 +1,81 @@
+/*
+ * 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.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.resource.SerializableResource;
+
+/**
+ * Base setup for an external sub-agent resource. Serialized into the agent plan as an {@code AGENT}
+ * resource.
+ *
+ * <p>Hosts the id-resolution chain behind {@link Subagent}: omitted ids are assigned via the
+ * context before an implementation is ever invoked, and {@link #call} runs the deferred callable
+ * through durable execution. Implementations only provide the terminal {@code asAsyncCallable}.
+ */
+public abstract class SubagentSetup extends SerializableResource implements Subagent {
+
+    @Override
+    @JsonIgnore
+    public ResourceType getResourceType() {
+        return ResourceType.AGENT;
+    }
+
+    @Override
+    public Result call(RunnerContext ctx, Object prompt) throws Exception {
+        return call(ctx, prompt, ctx.nextSessionId());
+    }
+
+    @Override
+    public Result call(RunnerContext ctx, Object prompt, String sessionId) throws Exception {
+        return call(ctx, prompt, sessionId, ctx.nextCallId(sessionId));
+    }
+
+    /**
+     * Synchronously invokes the sub-agent with already-assigned {@code sessionId} and {@code
+     * callId}, running the deferred callable through durable execution. Framework-facing: the ids
+     * are assigned by the shorter variants, never supplied by callers.
+     */
+    public Result call(RunnerContext ctx, Object prompt, String sessionId, String callId)
+            throws Exception {
+        return ctx.durableExecuteAsync(asAsyncCallable(ctx, prompt, sessionId, callId));
+    }
+
+    @Override
+    public DurableCallable<Result> asAsyncCallable(RunnerContext ctx, Object prompt) {
+        return asAsyncCallable(ctx, prompt, ctx.nextSessionId());
+    }
+
+    @Override
+    public DurableCallable<Result> asAsyncCallable(
+            RunnerContext ctx, Object prompt, String sessionId) {
+        return asAsyncCallable(ctx, prompt, sessionId, ctx.nextCallId(sessionId));
+    }
+
+    /**
+     * Produces the deferred, durable callable for one invocation; both ids are already assigned.
+     * The only method implementations must provide. Contract: the returned {@link
+     * DurableCallable#getId()} MUST be derived solely from the {@code (sessionId, callId)} pair.
+     */
+    protected abstract DurableCallable<Result> asAsyncCallable(
+            RunnerContext ctx, Object prompt, String sessionId, String callId);
+}

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

@@ -275,6 +275,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) {
@@ -390,6 +391,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;
+    }
 }