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
commit 762c2c4ae0337314c65fe8f9db9b14d99b8e994f Author: WenjinXie <[email protected]> AuthorDate: Fri Oct 16 13:05:59 2026 +0800 [api][java][python] Declare internal subagents in YAML Resolve agent-local subagent references against all Agents in the current file and Agents already registered in the same execution environment. Load files in order and reject duplicate Agent names before binding. Reuse the existing descriptor builder, plan cycle validation, serialization and runtime invocation path. Keep top-level subagents external and preserve external setup configuration arguments. Cover forward and cross-file references, reused and nested children, invalid declarations, cycles, duplicate names, failure state, shared child actions and schema parity. Verify parent and child application and a serialized-plan invocation on the Java operator. Update both published schema copies and the bundled contract hash. Generated-by: Codex CLI 0.153.4 (GPT-6) Co-authored-by: Codex <[email protected]> --- .../apache/flink/agents/api/yaml/YamlLoader.java | 69 ++++- .../flink/agents/api/yaml/spec/AgentSpec.java | 6 +- .../agents/api/yaml/spec/InternalSubagentSpec.java | 64 +++++ .../flink/agents/api/yaml/spec/SubagentSpec.java | 78 ++++++ .../agents/api/yaml/YamlLoaderLoadYamlTest.java | 78 ++++++ .../flink/agents/api/yaml/YamlSubagentTest.java | 188 +++++++++++++ .../agents/api/yaml/spec/SchemaParityTest.java | 1 + .../flink-agents-dev/assets/yaml-contracts.yaml | 2 +- .../flink-agents-dev/assets/yaml-schema.json | 55 +++- docs/yaml-schema.json | 55 +++- .../flink/agents/plan/YamlSubagentPlanTest.java | 97 +++++++ python/flink_agents/api/yaml/loader.py | 68 ++++- python/flink_agents/api/yaml/specs.py | 27 +- .../flink_agents/api/yaml/tests/test_subagents.py | 301 +++++++++++++++++++++ .../flink_agents/plan/tests/test_yaml_subagents.py | 148 ++++++++++ .../runtime/subagent/InternalSubagentCallTest.java | 56 ++++ 16 files changed, 1263 insertions(+), 30 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java index bbb1242e..3b807bec 100644 --- a/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java @@ -32,14 +32,17 @@ import org.apache.flink.agents.api.resource.ResourceDescriptor; import org.apache.flink.agents.api.resource.ResourceType; import org.apache.flink.agents.api.skills.SkillSourceSpec; import org.apache.flink.agents.api.skills.Skills; +import org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder; import org.apache.flink.agents.api.tools.FunctionTool; import org.apache.flink.agents.api.yaml.spec.ActionSpec; import org.apache.flink.agents.api.yaml.spec.AgentActionRef; import org.apache.flink.agents.api.yaml.spec.AgentSpec; import org.apache.flink.agents.api.yaml.spec.DescriptorSpec; +import org.apache.flink.agents.api.yaml.spec.InternalSubagentSpec; import org.apache.flink.agents.api.yaml.spec.PackageSkillSpec; import org.apache.flink.agents.api.yaml.spec.PromptSpec; import org.apache.flink.agents.api.yaml.spec.SkillsSpec; +import org.apache.flink.agents.api.yaml.spec.SubagentSpec; import org.apache.flink.agents.api.yaml.spec.ToolSpec; import org.apache.flink.agents.api.yaml.spec.UrlSkillSpec; import org.apache.flink.agents.api.yaml.spec.YamlAgentsDocument; @@ -169,6 +172,37 @@ public final class YamlLoader { return Prompt.fromMessages(messages); } + private static ResourceDescriptor buildSubagent(SubagentSpec spec, Map<String, Agent> agents) { + if (spec.getExternal() != null) { + return buildDescriptor(spec.getExternal(), ResourceType.AGENT); + } + InternalSubagentSpec internal = spec.getInternal(); + Agent child = agents.get(internal.getAgent()); + if (child == null) { + throw new IllegalArgumentException( + "Sub-agent '" + + internal.getName() + + "' references unknown agent '" + + internal.getAgent() + + "'; the child must be declared in the same YAML file or already" + + " registered in this environment."); + } + InternalSubagentResourceDescriptorBuilder builder = + new InternalSubagentResourceDescriptorBuilder() + .agent(child) + .description(internal.getDescription()); + if (internal.getInputSchema() != null) { + try { + builder.inputSchema( + new ObjectMapper().writeValueAsString(internal.getInputSchema())); + } catch (IOException e) { + throw new IllegalArgumentException( + "Invalid input schema for sub-agent '" + internal.getName() + "'", e); + } + } + return builder.build(); + } + /** Build a {@link Skills} resource from a parsed {@link SkillsSpec}. */ public static Skills buildSkills(SkillsSpec spec) { List<SkillSourceSpec> sources = new ArrayList<>(); @@ -250,6 +284,10 @@ public final class YamlLoader { /** Parse one YAML file and build the agents it declares. */ public static LoadedFile buildAgents(Path path) { + return buildAgents(path, Collections.emptyMap()); + } + + private static LoadedFile buildAgents(Path path, Map<String, Agent> registeredAgents) { YamlAgentsDocument doc; try { doc = YAML_MAPPER.readValue(Files.readString(path), YamlAgentsDocument.class); @@ -267,10 +305,27 @@ public final class YamlLoader { throw new IllegalArgumentException( "Duplicate agent name '" + spec.getName() + "' in " + path); } + if (registeredAgents.containsKey(spec.getName())) { + throw new IllegalArgumentException( + "Duplicate agent name '" + spec.getName() + "' (loading " + path + ")"); + } agentSpecs.put(spec.getName(), spec); agents.put(spec.getName(), buildAgent(spec)); } + Map<String, Agent> availableAgents = new LinkedHashMap<>(registeredAgents); + availableAgents.putAll(agents); + // Bind after all in-file Agents exist, including references to previously loaded Agents. + for (AgentSpec spec : doc.getAgents()) { + for (SubagentSpec subagent : spec.getSubagents()) { + agents.get(spec.getName()) + .addResource( + subagent.getName(), + ResourceType.AGENT, + buildSubagent(subagent, availableAgents)); + } + } + Map<ResourceType, Map<String, Object>> sharedResources = new EnumMap<>(ResourceType.class); for (ResourceType t : ResourceType.values()) { sharedResources.put(t, new LinkedHashMap<>()); @@ -334,11 +389,12 @@ public final class YamlLoader { /** * Load multiple YAML files. Multiple calls accumulate. Duplicate names — both within a single * file (caught by {@link #buildAgents(Path)}) and across the current environment — raise {@link - * IllegalArgumentException}. + * IllegalArgumentException}. Files are loaded in order; child references may resolve Agents + * registered by earlier files or calls. */ public static void loadYaml(AgentsExecutionEnvironment env, List<Path> paths) { for (Path path : paths) { - LoadedFile loaded = buildAgents(path); + LoadedFile loaded = buildAgents(path, env.getAgents()); for (Map.Entry<String, Agent> entry : loaded.getAgents().entrySet()) { AgentSpec spec = loaded.getAgentSpecs().get(entry.getKey()); @@ -368,14 +424,6 @@ public final class YamlLoader { } } - // Cross-environment agent name uniqueness check. - for (String name : loaded.getAgents().keySet()) { - if (env.getAgents().containsKey(name)) { - throw new IllegalArgumentException( - "Duplicate agent name '" + name + "' (loading " + path + ")"); - } - } - // Commit shared resources — env.addResource enforces dedup with its own message. for (Map.Entry<ResourceType, Map<String, Object>> e : loaded.getSharedResources().entrySet()) { @@ -413,7 +461,6 @@ public final class YamlLoader { 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)); diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java index 1826a61a..6f84d8ad 100644 --- a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/AgentSpec.java @@ -40,7 +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; + private final List<SubagentSpec> subagents; @JsonCreator public AgentSpec( @@ -57,7 +57,7 @@ public final class AgentSpec { @JsonProperty("embedding_model_setups") List<DescriptorSpec> embeddingModelSetups, @JsonProperty("vector_stores") List<DescriptorSpec> vectorStores, @JsonProperty("mcp_servers") List<DescriptorSpec> mcpServers, - @JsonProperty("subagents") List<DescriptorSpec> subagents) { + @JsonProperty("subagents") List<SubagentSpec> subagents) { this.name = name; this.description = description; this.prompts = orEmpty(prompts); @@ -125,7 +125,7 @@ public final class AgentSpec { return mcpServers; } - public List<DescriptorSpec> getSubagents() { + public List<SubagentSpec> getSubagents() { return subagents; } } diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/InternalSubagentSpec.java b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/InternalSubagentSpec.java new file mode 100644 index 00000000..9698bf59 --- /dev/null +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/InternalSubagentSpec.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.agents.api.yaml.spec; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.Map; + +/** A reference to a child Agent in this file or already registered in the environment. */ +public final class InternalSubagentSpec { + private final String name; + private final String agent; + private final String description; + private final Map<String, Object> inputSchema; + + @JsonCreator + public InternalSubagentSpec( + @JsonProperty(value = "name", required = true) String name, + @JsonProperty(value = "agent", required = true) String agent, + @JsonProperty("description") String description, + @JsonProperty("input_schema") Map<String, Object> inputSchema) { + if (agent == null || agent.isBlank()) { + throw new IllegalArgumentException( + "Internal sub-agent 'agent' must be a non-empty name"); + } + this.name = name; + this.agent = agent; + this.description = description == null ? "" : description; + this.inputSchema = inputSchema; + } + + public String getName() { + return name; + } + + public String getAgent() { + return agent; + } + + public String getDescription() { + return description; + } + + public Map<String, Object> getInputSchema() { + return inputSchema; + } +} diff --git a/api/src/main/java/org/apache/flink/agents/api/yaml/spec/SubagentSpec.java b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/SubagentSpec.java new file mode 100644 index 00000000..0516bc94 --- /dev/null +++ b/api/src/main/java/org/apache/flink/agents/api/yaml/spec/SubagentSpec.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.yaml.spec; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.DeserializationContext; +import com.fasterxml.jackson.databind.JsonDeserializer; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; + +import java.io.IOException; + +/** Either an internal Agent reference or an external sub-agent descriptor. */ +@JsonDeserialize(using = SubagentSpec.Deserializer.class) +public final class SubagentSpec { + private final InternalSubagentSpec internal; + private final DescriptorSpec external; + + private SubagentSpec(InternalSubagentSpec internal, DescriptorSpec external) { + this.internal = internal; + this.external = external; + } + + public String getName() { + return internal == null ? external.getName() : internal.getName(); + } + + public InternalSubagentSpec getInternal() { + return internal; + } + + public DescriptorSpec getExternal() { + return external; + } + + static final class Deserializer extends JsonDeserializer<SubagentSpec> { + @Override + public SubagentSpec deserialize(JsonParser parser, DeserializationContext context) + throws IOException { + JsonNode node = parser.getCodec().readTree(parser); + if (!node.has("clazz")) { + for (String field : new String[] {"name", "agent"}) { + if (!node.path(field).isTextual()) { + context.reportInputMismatch( + InternalSubagentSpec.class, + "Internal sub-agent '%s' must be a string", + field); + } + } + if (node.hasNonNull("description") && !node.get("description").isTextual()) { + context.reportInputMismatch( + InternalSubagentSpec.class, + "Internal sub-agent 'description' must be a string or null"); + } + return new SubagentSpec( + parser.getCodec().treeToValue(node, InternalSubagentSpec.class), null); + } + return new SubagentSpec( + null, parser.getCodec().treeToValue(node, DescriptorSpec.class)); + } + } +} diff --git a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java index 13a5155b..00e61694 100644 --- a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlLoaderLoadYamlTest.java @@ -23,6 +23,7 @@ import org.apache.flink.agents.api.AgentsExecutionEnvironment; import org.apache.flink.agents.api.agents.Agent; import org.apache.flink.agents.api.configuration.Configuration; 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.skills.SkillSourceSpec; import org.apache.flink.agents.api.skills.Skills; @@ -31,6 +32,8 @@ import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.table.api.Table; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.nio.file.Files; import java.nio.file.Path; @@ -116,6 +119,81 @@ class YamlLoaderLoadYamlTest { new SkillSourceSpec("local", Map.of("path", "./more"))); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void resolvesPreviouslyLoadedChild(boolean separateCalls, @TempDir Path tmp) throws Exception { + Path childPath = Files.writeString(tmp.resolve("child.yaml"), "agents: [{name: child}]"); + Path parentPath = + Files.writeString( + tmp.resolve("parent.yaml"), + "agents: [{name: parent, subagents: [{name: reviewer, agent: child}]}]"); + AgentsExecutionEnvironment env = new TestEnv(); + if (separateCalls) { + env.loadYaml(childPath); + env.loadYaml(parentPath); + } else { + env.loadYaml(List.of(childPath, parentPath)); + } + ResourceDescriptor descriptor = + (ResourceDescriptor) + env.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("reviewer"); + assertThat(descriptor.getInitialArguments().get("agent")) + .isSameAs(env.getAgents().get("child")); + assertThat(env.getAgents()).containsOnlyKeys("parent", "child"); + assertThat(env.getAgents().get("child").getResources().get(ResourceType.AGENT)).isEmpty(); + assertThatThrownBy(() -> new TestEnv().loadYaml(parentPath)) + .hasMessageContaining("unknown agent 'child'"); + } + + @Test + void childMustBeLoadedFirstAndFailurePreservesState(@TempDir Path tmp) throws Exception { + AgentsExecutionEnvironment env = new TestEnv(); + env.loadYaml( + Files.writeString( + tmp.resolve("existing.yaml"), + "agents: [{name: existing}]\nprompts: [{name: existing_prompt, text: keep}]")); + Agent original = env.getAgents().get("existing"); + Object prompt = env.getResources().get(ResourceType.PROMPT).get("existing_prompt"); + Path childPath = Files.writeString(tmp.resolve("child.yaml"), "agents: [{name: child}]"); + Path parentPath = + Files.writeString( + tmp.resolve("parent.yaml"), + "agents: [{name: parent, subagents: [{name: reviewer, agent: child}]}]\n" + + "prompts: [{name: new_prompt, text: 'not committed'}]"); + assertThatThrownBy(() -> env.loadYaml(List.of(parentPath, childPath))) + .hasMessageContaining("reviewer") + .hasMessageContaining("unknown agent 'child'") + .hasMessageContaining("already registered"); + assertThat(env.getAgents()).containsExactlyEntriesOf(Map.of("existing", original)); + assertThat(env.getResources().get(ResourceType.PROMPT)) + .containsExactlyEntriesOf(Map.of("existing_prompt", prompt)); + env.loadYaml(List.of(childPath, parentPath)); + assertThat(env.getAgents()).containsOnlyKeys("existing", "child", "parent"); + } + + @Test + void currentFileCannotReplaceRegisteredChild(@TempDir Path tmp) throws Exception { + AgentsExecutionEnvironment env = new TestEnv(); + env.loadYaml(Files.writeString(tmp.resolve("child.yaml"), "agents: [{name: child}]")); + Agent original = env.getAgents().get("child"); + Path duplicate = + Files.writeString( + tmp.resolve("duplicate.yaml"), + "agents: [{name: parent, subagents: [{name: reviewer, agent: child}]}, " + + "{name: child, prompts: [{name: replacement, text: bad}]}]\n" + + "prompts: [{name: new_prompt, text: 'not committed'}]"); + assertThatThrownBy(() -> env.loadYaml(duplicate)) + .hasMessageContaining("Duplicate agent name 'child'"); + assertThat(env.getAgents()).containsExactlyEntriesOf(Map.of("child", original)); + assertThat(original.getResources().get(ResourceType.PROMPT)).isEmpty(); + assertThat(env.getResources().values()) + .allSatisfy(resources -> assertThat(resources).isEmpty()); + } + /** Minimal env stub — we can't instantiate LocalExecutionEnvironment from the api module. */ private static final class TestEnv extends AgentsExecutionEnvironment { @Override diff --git a/api/src/test/java/org/apache/flink/agents/api/yaml/YamlSubagentTest.java b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlSubagentTest.java new file mode 100644 index 00000000..bf3c08ef --- /dev/null +++ b/api/src/test/java/org/apache/flink/agents/api/yaml/YamlSubagentTest.java @@ -0,0 +1,188 @@ +/* + * 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.yaml; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; +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.yaml.spec.YamlAgentsDocument; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class YamlSubagentTest { + @TempDir Path directory; + + private YamlLoader.LoadedFile load(String yaml) throws Exception { + Path path = directory.resolve("agents.yaml"); + Files.writeString(path, yaml); + return YamlLoader.buildAgents(path); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void resolvesChildInEitherOrder(boolean childFirst) throws Exception { + String parent = + " - name: parent\n subagents:\n" + + " - name: reviewer\n agent: child\n" + + " description: Review code\n" + + " input_schema: {type: object, properties: {code: {type: string}}}\n"; + String child = + " - name: child\n prompts:\n - {name: child_prompt, text: 'Review {code}'}\n"; + YamlLoader.LoadedFile result = + load("agents:\n" + (childFirst ? child + parent : parent + child)); + ResourceDescriptor descriptor = + (ResourceDescriptor) + result.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("reviewer"); + assertThat(descriptor.getClazz()) + .isEqualTo(InternalSubagentResourceDescriptorBuilder.SETUP_CLASS); + assertThat(descriptor.getInitialArguments().get("agent")) + .isSameAs(result.getAgents().get("child")); + assertThat(descriptor.getInitialArguments()).containsEntry("description", "Review code"); + assertThat( + new ObjectMapper() + .readTree( + (String) + descriptor + .getInitialArguments() + .get("input_schema")) + .at("/properties/code/type") + .asText()) + .isEqualTo("string"); + assertThat(result.getAgents().get("child").getResources().get(ResourceType.PROMPT)) + .containsKey("child_prompt"); + } + + @Test + void externalDeclarationsPreserveProviderArguments() throws Exception { + String external = + "{name: remote, clazz: example.RemoteSetup, type: java, agent: remote-id}"; + YamlLoader.LoadedFile result = + load( + "agents: [{name: parent, subagents: [" + + external + + "]}]\nsubagents: [" + + external + + "]"); + ResourceDescriptor shared = + (ResourceDescriptor) + result.getSharedResources().get(ResourceType.AGENT).get("remote"); + ResourceDescriptor local = + (ResourceDescriptor) + result.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("remote"); + assertThat(shared.getInitialArguments()).containsEntry("agent", "remote-id"); + assertThat(local.getInitialArguments()).containsEntry("agent", "remote-id"); + } + + @ParameterizedTest + @ValueSource(strings = {"", ", description: null", ", description: ''"}) + void optionalMetadata(String metadata) throws Exception { + YamlLoader.LoadedFile result = + load( + "agents: [{name: parent, subagents: [{name: reviewer, agent: child" + + metadata + + "}]}, {name: child}]"); + ResourceDescriptor descriptor = + (ResourceDescriptor) + result.getAgents() + .get("parent") + .getResources() + .get(ResourceType.AGENT) + .get("reviewer"); + assertThat(descriptor.getInitialArguments()) + .containsEntry("description", "") + .doesNotContainKey("input_schema"); + } + + @Test + void sharedInternalReferenceIsRejected() { + assertThatThrownBy( + () -> + load( + "agents: [{name: child}]\nsubagents: [{name: reviewer, agent: child}]")) + .hasMessageContaining("Failed to read YAML") + .hasStackTraceContaining("clazz"); + } + + @ParameterizedTest + @ValueSource( + strings = {"agents: [{name: parent, subagents: [{name: reviewer, agent: missing}]}]"}) + void unknownReferenceNamesTheMissingChild(String yaml) { + assertThatThrownBy(() -> load(yaml)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("reviewer") + .hasMessageContaining("missing") + .hasMessageContaining("same YAML file"); + } + + @ParameterizedTest + @ValueSource( + strings = { + "{name: reviewer, agent: ''}", + "{name: reviewer, agent: ' '}", + "{name: reviewer, agent: null}", + "{name: reviewer, agent: 123}", + "{name: reviewer, agent: true}", + "{name: null, agent: child}", + "{name: 123, agent: child}", + "{name: reviewer, agent: child, description: 123}", + "{name: reviewer, agent: child, type: java}", + "{name: reviewer, agent: child, input_schema: '{}'}", + "{name: reviewer, agent: child, unexpected: true}", + "{name: reviewer}" + }) + void rejectsInvalidDeclarations(String entry) { + assertThatThrownBy( + () -> + new ObjectMapper(new YAMLFactory()) + .readValue( + "agents: [{name: parent, subagents: [" + + entry + + "]}]", + YamlAgentsDocument.class)) + .isInstanceOf(Exception.class); + } + + @ParameterizedTest + @ValueSource( + strings = { + "agents: [{name: child}]\nsubagents: [{name: reviewer, clazz: example.Remote, type: java}, {name: reviewer, clazz: example.Remote, type: java}]", + "agents: [{name: child}, {name: parent, subagents: [{name: reviewer, agent: child}, {name: reviewer, agent: child}]}]" + }) + void duplicateResourceNamesAreRejected(String yaml) { + assertThatThrownBy(() -> load(yaml)).isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java b/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java index 519d9ef2..41abf55f 100644 --- a/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java +++ b/api/src/test/java/org/apache/flink/agents/api/yaml/spec/SchemaParityTest.java @@ -73,6 +73,7 @@ class SchemaParityTest { m.put("ActionSpec", ActionSpec.class); m.put("AgentSpec", AgentSpec.class); m.put("DescriptorSpec", DescriptorSpec.class); + m.put("InternalSubagentSpec", InternalSubagentSpec.class); m.put("PromptMessage", PromptMessage.class); m.put("PromptSpec", PromptSpec.class); m.put("SkillsSpec", SkillsSpec.class); diff --git a/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml b/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml index e1a66669..e18af319 100644 --- a/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml +++ b/dev/agent-skills/flink-agents-dev/assets/yaml-contracts.yaml @@ -13,7 +13,7 @@ contracts: repository: apache/flink-agents ref: main path: docs/yaml-schema.json - blob_sha: ed9800536d5f801e58b40db61b2488c2d5d5daf4 + blob_sha: 1b8270af990c9b80942e5cdf6319b18ff9b678e3 versions_without_yaml_api: - "0.2.1" diff --git a/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json b/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json index ed980053..1b8270af 100644 --- a/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json +++ b/dev/agent-skills/flink-agents-dev/assets/yaml-schema.json @@ -151,7 +151,14 @@ }, "subagents": { "items": { - "$ref": "#/$defs/DescriptorSpec" + "anyOf": [ + { + "$ref": "#/$defs/InternalSubagentSpec" + }, + { + "$ref": "#/$defs/DescriptorSpec" + } + ] }, "title": "Subagents", "type": "array" @@ -236,6 +243,52 @@ "title": "InjectedArg", "type": "object" }, + "InternalSubagentSpec": { + "additionalProperties": false, + "description": "Reference a child Agent in this file or already registered in the environment.", + "properties": { + "agent": { + "pattern": "\\S", + "title": "Agent", + "type": "string" + }, + "description": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Description" + }, + "input_schema": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Input Schema" + }, + "name": { + "title": "Name", + "type": "string" + } + }, + "required": [ + "name", + "agent" + ], + "title": "InternalSubagentSpec", + "type": "object" + }, "MessageRole": { "description": "Role of a message in a chat conversation.", "enum": [ diff --git a/docs/yaml-schema.json b/docs/yaml-schema.json index ed980053..1b8270af 100644 --- a/docs/yaml-schema.json +++ b/docs/yaml-schema.json @@ -151,7 +151,14 @@ }, "subagents": { "items": { - "$ref": "#/$defs/DescriptorSpec" + "anyOf": [ + { + "$ref": "#/$defs/InternalSubagentSpec" + }, + { + "$ref": "#/$defs/DescriptorSpec" + } + ] }, "title": "Subagents", "type": "array" @@ -236,6 +243,52 @@ "title": "InjectedArg", "type": "object" }, + "InternalSubagentSpec": { + "additionalProperties": false, + "description": "Reference a child Agent in this file or already registered in the environment.", + "properties": { + "agent": { + "pattern": "\\S", + "title": "Agent", + "type": "string" + }, + "description": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Description" + }, + "input_schema": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Input Schema" + }, + "name": { + "title": "Name", + "type": "string" + } + }, + "required": [ + "name", + "agent" + ], + "title": "InternalSubagentSpec", + "type": "object" + }, "MessageRole": { "description": "Role of a message in a chat conversation.", "enum": [ diff --git a/plan/src/test/java/org/apache/flink/agents/plan/YamlSubagentPlanTest.java b/plan/src/test/java/org/apache/flink/agents/plan/YamlSubagentPlanTest.java new file mode 100644 index 00000000..b42e349b --- /dev/null +++ b/plan/src/test/java/org/apache/flink/agents/plan/YamlSubagentPlanTest.java @@ -0,0 +1,97 @@ +/* + * 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; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.flink.agents.api.resource.ResourceType; +import org.apache.flink.agents.api.yaml.YamlLoader; +import org.apache.flink.agents.plan.resourceprovider.JavaResourceProvider; +import org.apache.flink.agents.plan.subagent.InternalSubagentPlans; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class YamlSubagentPlanTest { + @TempDir Path directory; + + private AgentPlan compile(String yaml) throws Exception { + Path path = directory.resolve("agents.yaml"); + Files.writeString(path, yaml); + return new AgentPlan(YamlLoader.buildAgents(path).getAgents().get("parent")); + } + + @Test + void nestedAndReusedChildrenSurvivePlanSerialization() throws Exception { + AgentPlan original = + compile( + "agents:\n" + + " - name: parent\n prompts: [{name: parent_only, text: parent}]\n" + + " subagents:\n" + + " - {name: first, agent: child, description: Review, input_schema: {type: object}}\n" + + " - {name: second, agent: child}\n" + + " - name: child\n prompts: [{name: child_only, text: child}]\n" + + " subagents: [{name: leaf, agent: leaf}]\n" + + " - name: leaf\n"); + ObjectMapper mapper = new ObjectMapper(); + AgentPlan plan = mapper.readValue(mapper.writeValueAsString(original), AgentPlan.class); + JavaResourceProvider first = + (JavaResourceProvider) + plan.getResourceProviders().get(ResourceType.AGENT).get("first"); + JavaResourceProvider second = + (JavaResourceProvider) + plan.getResourceProviders().get(ResourceType.AGENT).get("second"); + assertThat(first.getDescriptor().getInitialArguments()) + .containsEntry("scope", "first") + .containsEntry("description", "Review"); + assertThat(first.getDescriptor().getInitialArguments().get("plan")) + .isEqualTo(second.getDescriptor().getInitialArguments().get("plan")); + assertThat( + mapper.readTree( + (String) + first.getDescriptor() + .getInitialArguments() + .get("input_schema")) + .get("type") + .asText()) + .isEqualTo("object"); + AgentPlan child = InternalSubagentPlans.decode(first.getDescriptor()); + assertThat(child.getResourceProviders().get(ResourceType.AGENT)).containsKey("leaf"); + assertThat(child.getResourceProviders().get(ResourceType.PROMPT)) + .containsKey("child_only") + .doesNotContainKey("parent_only"); + } + + @ParameterizedTest + @ValueSource( + strings = { + "agents: [{name: parent, subagents: [{name: self, agent: parent}]}]", + "agents: [{name: parent, subagents: [{name: child, agent: child}]}, {name: child, subagents: [{name: parent, agent: parent}]}]", + "agents: [{name: parent, subagents: [{name: a, agent: a}]}, {name: a, subagents: [{name: b, agent: b}]}, {name: b, subagents: [{name: a, agent: a}]}]" + }) + void recursiveReferencesFailPlanConstruction(String yaml) { + assertThatThrownBy(() -> compile(yaml)).hasMessageContaining("Cyclic sub-agent"); + } +} diff --git a/python/flink_agents/api/yaml/loader.py b/python/flink_agents/api/yaml/loader.py index 77ba5711..16aac3c2 100644 --- a/python/flink_agents/api/yaml/loader.py +++ b/python/flink_agents/api/yaml/loader.py @@ -19,6 +19,7 @@ environment. """ +import json from pathlib import Path from typing import TYPE_CHECKING, Any, Dict, List, Tuple @@ -42,6 +43,7 @@ from flink_agents.api.resource import ( ResourceType, ) from flink_agents.api.skills import Skills, SkillSourceSpec +from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder from flink_agents.api.tools.function_tool import FunctionTool from flink_agents.api.yaml.aliases import ( resolve_clazz, @@ -54,6 +56,7 @@ from flink_agents.api.yaml.specs import ( Language, PromptSpec, SkillsSpec, + SubagentSpec, ToolSpec, YamlAgentsDocument, ) @@ -65,7 +68,6 @@ _DESCRIPTOR_TYPES: Dict[str, ResourceType] = { "embedding_model_setups": ResourceType.EMBEDDING_MODEL, "vector_stores": ResourceType.VECTOR_STORE, "mcp_servers": ResourceType.MCP_SERVER, - "subagents": ResourceType.AGENT, } @@ -152,6 +154,27 @@ def _build_descriptor( return ResourceDescriptor(clazz=python_fqn, **kwargs) +def _build_subagent(spec: SubagentSpec, agents: Dict[str, Agent]) -> ResourceDescriptor: + if isinstance(spec, DescriptorSpec): + return _build_descriptor(spec, ResourceType.AGENT) + child = agents.get(spec.agent) + if child is None: + msg = ( + f"Sub-agent {spec.name!r} references unknown agent {spec.agent!r}; " + "the child must be declared in the same YAML file or already " + "registered in this environment." + ) + raise ValueError(msg) + builder = ( + InternalSubagentResourceDescriptorBuilder() + .agent(child) + .description(spec.description or "") + ) + if spec.input_schema is not None: + builder.input_schema(json.dumps(spec.input_schema)) + return builder.build() + + def _add_descriptors_to_agent( agent: Agent, attr_name: str, descriptors: list[DescriptorSpec] ) -> None: @@ -266,6 +289,7 @@ def _build_agent(agent_spec: AgentSpec) -> Agent: def _build_in_file_state( path: Path | str, + registered_agents: Dict[str, Agent] | None = None, ) -> Tuple[ Dict[str, Agent], Dict[ResourceType, Dict[str, Any]], @@ -273,8 +297,10 @@ def _build_in_file_state( Dict[str, AgentSpec], YamlAgentsDocument, ]: - """Parse one YAML file, perform in-file duplicate detection, and build - the in-memory state without touching any execution environment. + """Parse one YAML file and build its state without modifying the environment. + + Registered Agents are available for child references; their names must not + collide with declarations in this file. Returns: agents: name -> Agent @@ -288,16 +314,38 @@ def _build_in_file_state( exactly one place. """ doc = _load_document(path) + registered_agents = registered_agents or {} agent_specs: Dict[str, AgentSpec] = {} agents: Dict[str, Agent] = {} for spec in doc.agents: if spec.name in agents: msg = f"Duplicate agent name {spec.name!r} in {path}" raise ValueError(msg) + if spec.name in registered_agents: + msg = f"Duplicate agent name {spec.name!r} (loading {path})" + raise ValueError(msg) agent_specs[spec.name] = spec agents[spec.name] = _build_agent(spec) + # Include previously loaded Agents without modifying the environment's registry. + available_agents = {**registered_agents, **agents} + # Create every in-file Agent before binding, allowing forward references. + for spec in doc.agents: + for subagent in spec.subagents: + agents[spec.name].add_resource( + subagent.name, + ResourceType.AGENT, + _build_subagent(subagent, available_agents), + ) + shared_resources: Dict[ResourceType, Dict[str, Any]] = {t: {} for t in ResourceType} + for spec in doc.subagents: + if spec.name in shared_resources[ResourceType.AGENT]: + msg = f"Duplicate shared resource name {spec.name!r} in {path}" + raise ValueError(msg) + shared_resources[ResourceType.AGENT][spec.name] = _build_descriptor( + spec, ResourceType.AGENT + ) for attr, resource_type in _DESCRIPTOR_TYPES.items(): for spec in getattr(doc, attr): if spec.name in shared_resources[resource_type]: @@ -386,8 +434,9 @@ def load_yaml( resources on the environment. Multiple calls accumulate. Duplicate names — both within a single file - and across the current environment — raise ``ValueError``. In-file - duplicate detection is delegated to :func:`_build_in_file_state` so + and across the current environment — raise ``ValueError``. Files are loaded + in order, so child references can use Agents from earlier files or calls. + Agent name checks are delegated to :func:`_build_in_file_state` so that ``load_yaml`` and :func:`build_agents` share the same rules. """ if isinstance(paths, str | Path): @@ -395,15 +444,10 @@ def load_yaml( for path in paths: agents, shared_resources, shared_actions, agent_specs, _ = _build_in_file_state( - path + path, env._agents ) - # Cross-environment duplicate checks. In-file duplicates were - # already caught inside ``_build_in_file_state``. - for name in agents: - if name in env._agents: - msg = f"Duplicate agent name {name!r} (loading {path})" - raise ValueError(msg) + # Agent name conflicts were already caught inside ``_build_in_file_state``. for resource_type, name_to_resource in shared_resources.items(): for name in name_to_resource: if name in env.resources[resource_type]: diff --git a/python/flink_agents/api/yaml/specs.py b/python/flink_agents/api/yaml/specs.py index 063c30f1..6f46fb3e 100644 --- a/python/flink_agents/api/yaml/specs.py +++ b/python/flink_agents/api/yaml/specs.py @@ -29,6 +29,7 @@ from typing import Annotated, Any, Dict, List, Literal from pydantic import ( BaseModel, + BeforeValidator, ConfigDict, Field, StrictBool, @@ -62,6 +63,30 @@ class DescriptorSpec(BaseModel): type: Language | None = None +class InternalSubagentSpec(BaseModel): + """Reference a child Agent in this file or already registered in the environment.""" + + model_config = ConfigDict(extra="forbid") + + name: str + agent: Annotated[str, Field(pattern=r"\S")] + description: str | None = None + input_schema: Dict[str, Any] | None = None + + +def _parse_subagent(value: Any) -> Any: + if isinstance(value, dict): + spec = DescriptorSpec if "clazz" in value else InternalSubagentSpec + return spec.model_validate(value) + return value + + +SubagentSpec = Annotated[ + InternalSubagentSpec | DescriptorSpec, + BeforeValidator(_parse_subagent), +] + + class MessageRole(str, Enum): """Role of a message in a chat conversation.""" @@ -258,7 +283,7 @@ class AgentSpec(BaseModel): embedding_model_setups: List[DescriptorSpec] = Field(default_factory=list) vector_stores: List[DescriptorSpec] = Field(default_factory=list) mcp_servers: List[DescriptorSpec] = Field(default_factory=list) - subagents: List[DescriptorSpec] = Field(default_factory=list) + subagents: List[SubagentSpec] = Field(default_factory=list) class YamlAgentsDocument(BaseModel): diff --git a/python/flink_agents/api/yaml/tests/test_subagents.py b/python/flink_agents/api/yaml/tests/test_subagents.py new file mode 100644 index 00000000..477a18e5 --- /dev/null +++ b/python/flink_agents/api/yaml/tests/test_subagents.py @@ -0,0 +1,301 @@ +################################################################################ +# 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. +################################################################################# +"""YAML declarations of internal and external sub-agents.""" + +import json +from pathlib import Path + +import pytest +import yaml +from pydantic import ValidationError + +from flink_agents.api.resource import ResourceType +from flink_agents.api.subagent import INTERNAL_SUBAGENT_SETUP_CLASS +from flink_agents.api.yaml.loader import build_agents, load_yaml +from flink_agents.api.yaml.specs import YamlAgentsDocument +from flink_agents.api.yaml.tests.test_loader import _registration_env + + +def _write(tmp_path: Path, document: dict, filename: str = "agents.yaml") -> Path: + path = tmp_path / filename + path.write_text(yaml.safe_dump(document)) + return path + + [email protected]("child_first", [False, True]) +def test_internal_reference_preserves_child_and_metadata( + tmp_path: Path, child_first: bool +) -> None: + schema = {"type": "object", "properties": {"code": {"type": "string"}}} + parent = { + "name": "parent", + "subagents": [ + { + "name": "reviewer", + "agent": "child", + "description": "Review code", + "input_schema": schema, + } + ], + } + child = { + "name": "child", + "prompts": [{"name": "child_prompt", "text": "Review {code}"}], + } + agents, _, _ = build_agents( + _write( + tmp_path, {"agents": [child, parent] if child_first else [parent, child]} + ) + ) + descriptor = agents["parent"].resources[ResourceType.AGENT]["reviewer"] + assert descriptor.target_clazz == INTERNAL_SUBAGENT_SETUP_CLASS + assert descriptor.arguments["agent"] is agents["child"] + assert descriptor.arguments["description"] == "Review code" + assert json.loads(descriptor.arguments["input_schema"]) == schema + assert "child_prompt" in agents["child"].resources[ResourceType.PROMPT] + + +def test_external_subagents_preserve_provider_arguments(tmp_path: Path) -> None: + external = {"name": "remote", "clazz": "my_app.RemoteSetup", "agent": "remote-id"} + agents, resources, _ = build_agents( + _write( + tmp_path, + { + "agents": [{"name": "parent", "subagents": [external]}], + "subagents": [external], + }, + ) + ) + assert resources[ResourceType.AGENT]["remote"].arguments == {"agent": "remote-id"} + assert agents["parent"].resources[ResourceType.AGENT]["remote"].arguments == { + "agent": "remote-id" + } + + [email protected]("metadata", [{}, {"description": None}, {"description": ""}]) +def test_optional_internal_metadata(tmp_path: Path, metadata: dict) -> None: + agents, _, _ = build_agents( + _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "child", "agent": "child", **metadata}], + }, + {"name": "child"}, + ] + }, + ) + ) + arguments = agents["parent"].resources[ResourceType.AGENT]["child"].arguments + assert arguments["description"] == "" + assert "input_schema" not in arguments + + +def test_shared_internal_reference_rejected_before_registration(tmp_path: Path) -> None: + env = _registration_env() + with pytest.raises(ValidationError, match="clazz"): + load_yaml( + env, + _write( + tmp_path, + { + "agents": [{"name": "child"}], + "subagents": [{"name": "reviewer", "agent": "child"}], + }, + ), + ) + assert not env._agents + assert all(not entries for entries in env.resources.values()) + + +def test_two_parents_share_child_definition(tmp_path: Path) -> None: + agents, _, _ = build_agents( + _write( + tmp_path, + { + "agents": [ + { + "name": name, + "subagents": [{"name": "reviewer", "agent": "child"}], + } + for name in ["one", "two"] + ] + + [{"name": "child"}] + }, + ) + ) + for name in ["one", "two"]: + assert ( + agents[name].resources[ResourceType.AGENT]["reviewer"].arguments["agent"] + is agents["child"] + ) + + +def test_unknown_child_does_not_register_partial_state(tmp_path: Path) -> None: + document = {"agents": [{"name": "parent"}]} + document["agents"][0]["subagents"] = [{"name": "reviewer", "agent": "missing"}] + env = _registration_env() + with pytest.raises( + ValueError, match="reviewer.*unknown agent.*missing.*same YAML file" + ): + load_yaml(env, _write(tmp_path, document)) + assert not env._agents + assert all(not entries for entries in env.resources.values()) + + [email protected]( + "entry", + [ + {"name": "reviewer", "agent": ""}, + {"name": "reviewer", "agent": " "}, + {"name": "reviewer", "agent": None}, + {"name": "reviewer", "agent": 123}, + {"name": "reviewer", "agent": True}, + {"name": None, "agent": "child"}, + {"name": 123, "agent": "child"}, + {"name": "reviewer", "agent": "child", "description": 123}, + {"name": "reviewer", "agent": "child", "type": "python"}, + {"name": "reviewer", "agent": "child", "input_schema": "{}"}, + {"name": "reviewer", "agent": "child", "unexpected": True}, + {"name": "reviewer"}, + ], +) +def test_rejects_invalid_subagent_declarations(entry: dict) -> None: + with pytest.raises(ValidationError): + YamlAgentsDocument.model_validate( + {"agents": [{"name": "parent", "subagents": [entry]}]} + ) + + [email protected]("shared", [False, True]) +def test_rejects_duplicate_subagent_resource_names( + tmp_path: Path, shared: bool +) -> None: + document = {"agents": [{"name": "parent"}, {"name": "child"}]} + target = document if shared else document["agents"][0] + entry = ( + {"name": "reviewer", "clazz": "my_app.RemoteSetup"} + if shared + else {"name": "reviewer", "agent": "child"} + ) + target["subagents"] = [entry] * 2 + with pytest.raises(ValueError, match="[Dd]uplicate|already"): + build_agents(_write(tmp_path, document)) + + [email protected]("separate_calls", [False, True]) +def test_child_in_previously_loaded_file_is_resolved( + tmp_path: Path, separate_calls: bool +) -> None: + child_path = _write(tmp_path, {"agents": [{"name": "child"}]}, "child.yaml") + parent_path = _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "reviewer", "agent": "child"}], + } + ] + }, + "parent.yaml", + ) + env = _registration_env() + if separate_calls: + load_yaml(env, child_path) + load_yaml(env, parent_path) + else: + load_yaml(env, [child_path, parent_path]) + descriptor = env._agents["parent"].resources[ResourceType.AGENT]["reviewer"] + assert descriptor.arguments["agent"] is env._agents["child"] + assert set(env._agents) == {"parent", "child"} + assert not env._agents["child"].resources[ResourceType.AGENT] + # Registrations belong to one environment, not to a process-global registry. + with pytest.raises(ValueError, match="unknown agent.*child"): + load_yaml(_registration_env(), parent_path) + + +def test_child_must_be_loaded_before_parent_and_failure_preserves_state( + tmp_path: Path, +) -> None: + env = _registration_env() + load_yaml( + env, + _write( + tmp_path, + { + "agents": [{"name": "existing"}], + "prompts": [{"name": "existing_prompt", "text": "keep"}], + }, + ), + ) + original = env._agents["existing"] + prompt = env.resources[ResourceType.PROMPT]["existing_prompt"] + child_path = _write(tmp_path, {"agents": [{"name": "child"}]}, "child.yaml") + parent_path = _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "reviewer", "agent": "child"}], + } + ], + "prompts": [{"name": "new_prompt", "text": "not committed"}], + }, + "parent.yaml", + ) + with pytest.raises( + ValueError, match="reviewer.*unknown agent.*child.*already registered" + ): + load_yaml(env, [parent_path, child_path]) + assert env._agents == {"existing": original} + assert env.resources[ResourceType.PROMPT] == {"existing_prompt": prompt} + load_yaml(env, [child_path, parent_path]) + assert set(env._agents) == {"existing", "child", "parent"} + + +def test_current_file_cannot_replace_registered_child(tmp_path: Path) -> None: + env = _registration_env() + load_yaml(env, _write(tmp_path, {"agents": [{"name": "child"}]})) + original = env._agents["child"] + with pytest.raises(ValueError, match="Duplicate agent name.*child"): + load_yaml( + env, + _write( + tmp_path, + { + "agents": [ + { + "name": "parent", + "subagents": [{"name": "reviewer", "agent": "child"}], + }, + { + "name": "child", + "prompts": [{"name": "replacement", "text": "bad"}], + }, + ], + "prompts": [{"name": "new_prompt", "text": "not committed"}], + }, + ), + ) + assert env._agents == {"child": original} + assert not original.resources[ResourceType.PROMPT] + assert all(not entries for entries in env.resources.values()) diff --git a/python/flink_agents/plan/tests/test_yaml_subagents.py b/python/flink_agents/plan/tests/test_yaml_subagents.py new file mode 100644 index 00000000..8b22d01c --- /dev/null +++ b/python/flink_agents/plan/tests/test_yaml_subagents.py @@ -0,0 +1,148 @@ +################################################################################ +# 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. +################################################################################ +"""Compile YAML parent/child declarations through the production plan path.""" + +import json +from pathlib import Path + +import pytest +import yaml + +from flink_agents.api.resource import ResourceType +from flink_agents.api.yaml.loader import build_agents, load_yaml +from flink_agents.api.yaml.tests.test_loader import _registration_env +from flink_agents.plan.agent_plan import AgentPlan +from flink_agents.plan.configuration import AgentConfiguration +from flink_agents.plan.subagent import decode_plan + + +def _write(tmp_path: Path, agents: list[dict], **shared: object) -> Path: + path = tmp_path / "agents.yaml" + path.write_text(yaml.safe_dump({"agents": agents, **shared})) + return path + + [email protected]("separate_files", [False, True]) +def test_yaml_child_plan_preserves_actions_metadata_and_local_resources( + tmp_path: Path, + separate_files: bool, +) -> None: + path = _write( + tmp_path, + [ + { + "name": "parent", + "prompts": [{"name": "parent_only", "text": "parent"}], + "subagents": [ + { + "name": "reviewer", + "agent": "child", + "description": "Review code", + "input_schema": { + "type": "object", + "properties": {"code": {"type": "string"}}, + }, + } + ], + }, + { + "name": "child", + "actions": ["increment"], + "prompts": [{"name": "child_only", "text": "child"}], + }, + ], + actions=[ + { + "name": "increment", + "function": "flink_agents.api.yaml.tests.fixtures.loader_targets:increment", + "trigger_conditions": ["input"], + } + ], + ) + env = _registration_env() + if separate_files: + document = yaml.safe_load(path.read_text()) + parent = document["agents"].pop(0) + child_path = tmp_path / "child.yaml" + child_path.write_text(yaml.safe_dump(document)) + path.write_text(yaml.safe_dump({"agents": [parent]})) + load_yaml(env, [child_path, path]) + else: + load_yaml(env, path) + original = env._agents["parent"].resources[ResourceType.AGENT]["reviewer"] + plan = AgentPlan.from_agent(env._agents["parent"], AgentConfiguration()) + restored = AgentPlan.model_validate_json( + plan.model_dump_json(serialize_as_any=True) + ) + descriptor = restored.resource_providers[ResourceType.AGENT]["reviewer"].descriptor + assert descriptor.arguments["scope"] == "reviewer" + assert descriptor.arguments["description"] == "Review code" + assert "code" in json.loads(descriptor.arguments["input_schema"])["properties"] + child = decode_plan(descriptor) + assert "increment" in child.actions + assert "child_only" in child.resource_providers[ResourceType.PROMPT] + assert "parent_only" not in child.resource_providers[ResourceType.PROMPT] + assert original.arguments["agent"] is env._agents["child"] + assert "plan" not in original.arguments + + [email protected]( + "edges", + [ + {"parent": "parent"}, + {"parent": "child", "child": "parent"}, + {"parent": "a", "a": "b", "b": "a"}, + ], +) +def test_yaml_cycles_use_existing_plan_validation(tmp_path: Path, edges: dict) -> None: + path = _write( + tmp_path, + [ + {"name": name, "subagents": [{"name": child, "agent": child}]} + for name, child in edges.items() + ], + ) + agents, _, _ = build_agents(path) + with pytest.raises(ValueError, match="Cyclic sub-agent definition"): + AgentPlan.from_agent(agents["parent"], AgentConfiguration()) + + +def test_yaml_nested_children_and_reused_child_compile(tmp_path: Path) -> None: + path = _write( + tmp_path, + [ + { + "name": "parent", + "subagents": [ + {"name": "first", "agent": "child"}, + {"name": "second", "agent": "child"}, + ], + }, + {"name": "child", "subagents": [{"name": "leaf", "agent": "leaf"}]}, + {"name": "leaf"}, + ], + ) + agents, _, _ = build_agents(path) + plan = AgentPlan.from_agent(agents["parent"], AgentConfiguration()) + resources = plan.resource_providers[ResourceType.AGENT] + assert ( + resources["first"].descriptor.arguments["plan"] + == resources["second"].descriptor.arguments["plan"] + ) + child = decode_plan(resources["first"].descriptor) + assert "leaf" in child.resource_providers[ResourceType.AGENT] 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 a932b283..619fc2f4 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 @@ -34,10 +34,12 @@ 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.AgentPlan; +import org.apache.flink.agents.runtime.env.RemoteExecutionEnvironment; import org.apache.flink.agents.runtime.operator.ActionExecutionOperator; import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; import org.junit.jupiter.api.BeforeEach; @@ -45,7 +47,12 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.condition.EnabledForJreRange; import org.junit.jupiter.api.condition.JRE; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -244,6 +251,55 @@ public class InternalSubagentCallTest { // --- tests --- + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @Timeout(60) + void yamlParentInvokesChildAfterPlanSerialization( + boolean separateFiles, @TempDir Path directory) throws Exception { + Path path = directory.resolve("agents.yaml"); + String targets = InternalSubagentCallTest.class.getName(); + Files.writeString( + path, + "agents:\n" + + " - name: parent\n" + + " subagents: [{name: child, agent: echo}]\n" + + " actions:\n" + + " - name: call\n type: java\n trigger_conditions: [input]\n" + + " function: " + + targets + + ":callOnce\n" + + " - name: echo\n actions: [echo]\n" + + "actions:\n" + + " - name: echo\n type: java\n trigger_conditions: [input]\n" + + " function: " + + targets + + "$EchoChildAgent:echo\n"); + StreamExecutionEnvironment flink = StreamExecutionEnvironment.getExecutionEnvironment(); + RemoteExecutionEnvironment env = new RemoteExecutionEnvironment(flink, null); + if (separateFiles) { + String yaml = Files.readString(path); + int childStart = yaml.indexOf(" - name: echo\n"); + Path childPath = + Files.writeString( + directory.resolve("child.yaml"), + "agents:\n" + yaml.substring(childStart)); + Files.writeString(path, yaml.substring(0, childStart)); + env.loadYaml(List.of(childPath, path)); + } else { + env.loadYaml(path); + } + // The child remains independently applicable, including between two parent compilations. + for (String name : List.of("parent", "echo", "parent")) { + env.fromDataStream(flink.fromElements(7L), value -> value).apply(name); + } + AgentPlan original = new AgentPlan(env.getAgents().get("parent")); + ObjectMapper mapper = new ObjectMapper(); + AgentPlan transferred = + mapper.readValue(mapper.writeValueAsString(original), AgentPlan.class); + assertThat(run(transferred, 7L)).containsExactly("echo:7"); + assertThat(CHILD_INVOCATIONS).containsExactly("7"); + } + @Test @Timeout(60) void callerReceivesTheChildsAccumulatedOutput() throws Exception {
