You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "wenjin272 (via GitHub)" <gi...@apache.org> on 2026/05/11 08:42:47 UTC
[PR] Skill repo extending [flink-agents]
wenjin272 opened a new pull request, #655:
URL: https://github.com/apache/flink-agents/pull/655
<!--
* Thank you very much for contributing to Flink Agents.
* Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc.
-->
<!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. -->
Linked issue: #692, #593, #594
### Purpose of change
Extend the way Flink-Agents loads external agent skills:
* load agent skills from remote urls (only http/https point to a zip)
* load agent skills from python package
* load agent skills from java classpath
### Tests
ut
### API
Yes, add `from_xxx` to `Skills`.
### Documentation
<!-- Do not remove this section. Check the proper box only. -->
- [ ] `doc-needed` <!-- Your PR changes impact docs -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-included` <!-- Your PR already contains the necessary documentation updates -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3240567851
##########
api/src/main/java/org/apache/flink/agents/api/skills/Skills.java:
##########
@@ -51,33 +57,76 @@ public class Skills extends SerializableResource {
/** Reserved name of the built-in bash tool used to execute skill scripts. */
public static final String BASH_TOOL = "bash";
- private List<String> paths;
+ private final List<String> paths;
Review Comment:
In our design, skills with the same name are not supported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3263199741
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/URLSkillRepository.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import org.apache.flink.agents.runtime.skill.AgentSkill;
+import org.apache.flink.agents.runtime.skill.SkillRepository;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Skill repository backed by an http(s) URL pointing to a zip.
+ *
+ * <p>The zip is downloaded to a temp file and extracted into a process-local temp directory. The
+ * downloaded zip itself is removed once extraction completes; the extracted directory is released
+ * via {@link #close()} (cascaded through {@code SkillManager} → {@code ResourceContextImpl} →
+ * {@code ResourceCache} on operator close). A JVM shutdown hook acts as fallback cleanup if {@code
+ * close()} is never called.
+ *
+ * <p>Composes a {@link SkillDirectoryReader} for the on-disk parsing; the materialization happens
+ * upfront in the constructor (no abuse of {@code super(...)}).
+ */
+public final class URLSkillRepository implements SkillRepository {
+
+ private static final int REQUEST_TIMEOUT_MS = 90_000;
+
+ private final String url;
Review Comment:
**Field name `owned` is an adjective without a subject.**
`Materialized` is already a noun. Rename the field to `materialization` (matches the class name), or rename the inner class to `MaterializedDir` so the field reads `materializedDir`. Current call sites `this.owned.close()` / `if (owned == null)` don't read well.
Same applies to `ClasspathSkillRepository.java:73` and `FileSystemSkillRepository.java:51`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -18,33 +18,63 @@
package org.apache.flink.agents.runtime.skill;
+import org.apache.flink.agents.api.skills.SkillSourceSpec;
import org.apache.flink.agents.api.skills.Skills;
-import org.apache.flink.agents.runtime.skill.repository.FileSystemSkillRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Loads and indexes all skills referenced by a {@link Skills} configuration.
*
* <p>Mirrors the Python {@code flink_agents.runtime.skill.skill_manager.SkillManager}.
+ *
+ * <p>Owned by {@code ResourceContextImpl}; closed via {@code ResourceCache.close()} on operator
+ * close, which cascades to each repository's temp directory. Avoid constructing a {@code
+ * SkillManager} outside that flow without {@code try-with-resources}.
*/
-public class SkillManager {
+public class SkillManager implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SkillManager.class);
private final Skills config;
+ private final ClassLoader classLoader;
private final Map<String, AgentSkill> skills = new LinkedHashMap<>();
private final Map<String, SkillRepository> repos = new HashMap<>();
- public SkillManager(Skills config) {
+ /**
+ * Construct a {@code SkillManager} that resolves {@code classpath:} sources against {@code
+ * classLoader}. Production code passes the Flink user-code class loader (threaded through
+ * {@code ResourceCache} from {@code ActionExecutionOperator}); tests / standalone use may use
+ * {@link #SkillManager(Skills)}.
+ */
+ public SkillManager(Skills config, ClassLoader classLoader) {
Review Comment:
**Constructor-failure leak: partial `loadAll()` leaves registered repos with no closeable handle.**
If the 2nd of 3 sources succeeds (repo + hook registered) but the 3rd throws, the constructor exits with `IllegalStateException` → caller never gets a `SkillManager` reference → cannot call `close()` → 2nd repo's hook + temp dir leak until JVM exit.
Same monotonic-leak class as review #10, just triggered by partial-construction instead of hot-reload. 5-line fix: wrap `loadAll`'s loop in try/catch, call `close()` before rethrowing.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -111,38 +148,94 @@ public List<String> getSkillDirs(List<String> names) {
@Nullable
public Path getSkillDir(String skillName) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- return ((FileSystemSkillRepository) repo).getBaseDir().resolve(skillName);
- }
- return null;
+ return repo == null ? null : repo.getSkillDir(skillName);
}
/** Resolve a skill resource's relative path to an absolute path, or {@code null} if missing. */
@Nullable
public Path resolveResourcePath(String skillName, String resourcePath) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- Path resolved =
- ((FileSystemSkillRepository) repo)
- .getBaseDir()
- .resolve(skillName)
- .resolve(resourcePath);
- if (Files.isRegularFile(resolved)) {
- return resolved;
+ if (repo == null) {
+ return null;
+ }
+ Path dir = repo.getSkillDir(skillName);
+ if (dir == null) {
+ return null;
+ }
+ Path resolved = dir.resolve(resourcePath);
+ return Files.isRegularFile(resolved) ? resolved : null;
+ }
+
+ private void loadAll() {
+ for (SkillSourceSpec spec : config.getSources()) {
+ try {
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from " + spec.getScheme() + ":" + spec.getParams(),
+ e);
+ }
+ }
+ }
+
+ /** Build a {@link SkillOrigin} from a spec for diagnostics (WARN on duplicates, etc.). */
Review Comment:
**`originOf` re-introduces the parallel scheme ladder the registry was meant to eliminate.**
```java
switch (spec.getScheme()) {
case "local": ...; case "url": ...;
case "classpath": ...; case "package": ...;
}
```
Python's `_origin_of` mirrors this. Adding scheme `s3` now requires 3 file edits (registry ×2 + originOf ×2), not 1. Java's `case "package"` is dead code — no Java `package` handler exists.
Fix: add `default String describeLocation(params)` to `SkillSourceHandler`. Each handler describes its own location at registration.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/AgentSkill.java:
##########
@@ -44,6 +44,7 @@ public final class AgentSkill {
@Nullable private volatile Map<String, String> resources;
@Nullable private Supplier<Map<String, String>> resourceLoader;
private volatile boolean activated;
+ @Nullable private volatile SkillOrigin origin;
Review Comment:
**Java/Python serialization surface for `origin` is asymmetric.**
Python's `AgentSkill` extends `BaseModel`, so `origin` is in default `model_dump()` output. Java's `AgentSkill` is a POJO with no Jackson annotation — any future debug/checkpoint/Pemja serialization path silently drops it on Java while Python emits it. Same class of cross-language drop A3 fixed at the `Skills` envelope.
Add `@JsonProperty("origin")` + `@JsonInclude(NON_NULL)`. Near-zero cost now; cost grows once `AgentSkill` travels on the wire.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillRepository.java:
##########
@@ -20,14 +20,19 @@
import javax.annotation.Nullable;
+import java.nio.file.Path;
import java.util.List;
import java.util.Map;
/**
* Source of skills. Mirrors the Python {@code
* flink_agents.runtime.skill.skill_repository.SkillRepository}.
+ *
+ * <p>Implementations that own a materialized temp directory (URL / classpath / zip) should override
Review Comment:
**Three sibling repos duplicate the same 5 delegating methods + state** (`reader`, `materialization`, plus `getSkill`/`getSkills`/`getResources`/`getSkillDir`/`close`) — ~45 lines of boilerplate, +15 per future source.
Extract `AbstractMaterializedSkillRepository` holding the shared state and implementing the delegators once; concrete repos reduce to source-specific I/O + an abstract `getOrigin()`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -111,38 +148,94 @@ public List<String> getSkillDirs(List<String> names) {
@Nullable
public Path getSkillDir(String skillName) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- return ((FileSystemSkillRepository) repo).getBaseDir().resolve(skillName);
- }
- return null;
+ return repo == null ? null : repo.getSkillDir(skillName);
}
/** Resolve a skill resource's relative path to an absolute path, or {@code null} if missing. */
@Nullable
public Path resolveResourcePath(String skillName, String resourcePath) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- Path resolved =
- ((FileSystemSkillRepository) repo)
- .getBaseDir()
- .resolve(skillName)
- .resolve(resourcePath);
- if (Files.isRegularFile(resolved)) {
- return resolved;
+ if (repo == null) {
+ return null;
+ }
+ Path dir = repo.getSkillDir(skillName);
+ if (dir == null) {
+ return null;
+ }
+ Path resolved = dir.resolve(resourcePath);
+ return Files.isRegularFile(resolved) ? resolved : null;
+ }
+
+ private void loadAll() {
+ for (SkillSourceSpec spec : config.getSources()) {
+ try {
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from " + spec.getScheme() + ":" + spec.getParams(),
+ e);
+ }
+ }
+ }
+
+ /** Build a {@link SkillOrigin} from a spec for diagnostics (WARN on duplicates, etc.). */
+ private static SkillOrigin originOf(SkillSourceSpec spec) {
+ Map<String, String> p = spec.getParams();
+ String location;
+ switch (spec.getScheme()) {
+ case "local":
+ location = p.getOrDefault("path", "");
+ break;
+ case "url":
+ location = p.getOrDefault("url", "");
+ break;
+ case "classpath":
+ location = p.getOrDefault("resource", "");
+ break;
+ case "package":
+ location = p.getOrDefault("package", "") + "/" + p.getOrDefault("resource", "");
+ break;
+ default:
+ location = p.toString();
+ }
+ return new SkillOrigin(spec.getScheme(), location);
+ }
+
+ /**
+ * Close every owned {@link SkillRepository}, releasing any temp directories materialized for
+ * URL / classpath-zip / classpath-jar sources. Idempotent.
+ */
+ @Override
+ public void close() {
+ // repos may map multiple skill names to the same repo instance; dedup by identity.
+ Set<SkillRepository> unique = Collections.newSetFromMap(new IdentityHashMap<>());
+ unique.addAll(repos.values());
Review Comment:
**`SkillManager.close()` swallows exceptions silently; inconsistent with `ResourceCache.close()`.**
```java
for (SkillRepository repo : unique) {
try { repo.close(); } catch (Exception ignored) {}
}
```
`ResourceCache.close()` (line 135-160) collects first-exception + suppressed and rethrows. Same close cascade, two styles. Real failures (locked file, permission denied, disk full) give callers no signal. `ResourceContextImpl.ensureSkillManager()` line 116 has the same silent catch on config swap.
Minimum: `LOG.warn(..., e)`. Preferable: align with `ResourceCache.close()`'s rethrow pattern.
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -532,11 +533,11 @@ private void addSkills(Map<String, Skills> skillsObjects) throws Exception {
ResourceType.TOOL,
new ResourceDescriptor(BashTool.class.getName(), new HashMap<>())));
- LinkedHashSet<String> paths = new LinkedHashSet<>();
+ LinkedHashSet<SkillSourceSpec> sources = new LinkedHashSet<>();
for (Skills s : skillsObjects.values()) {
Review Comment:
**Multi-`@Skills` merge order is non-deterministic on Java; WARN reports collisions but the "winner" is unspecified.**
`AgentPlan.addSkills` iterates `skillsObjects.values()` in `LinkedHashMap` order, but insertion order comes from `Class.getMethods()` — JLS doesn't guarantee declaration order. Test self-discloses at `AgentPlanDeclareSkillsTest:163`: "Class.getMethods() does not preserve declaration order".
Two `@Skills` methods with the same skill name pick different winners across deployments / JDK versions; WARN fires but is non-reproducible. Python `dict` preserves insertion order — Python is deterministic, Java is the bad case.
Min fix: sort method names lexicographically before merging + javadoc note. Stronger: `@Skills(priority=N)` or fail-fast on collision.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #655:
URL: https://github.com/apache/flink-agents/pull/655#issuecomment-4485909320
Hi, @wzhero1. Ty for the careful review.
I have updated and addressed the second round comments:
- **`owned` rename + 5× delegating boilerplate**: the three repos collapsed into `AbstractMaterializedSkillRepository` (Java) / `MaterializedSkillRepository` (Python); the
field becomes `materialization` on the base. Python's `PackageSkillRepository` simplified to own its tempdir (no more `as_file` ctx + atexit two-step). — 7152f9e2
- **`originOf` scheme ladder**: moved onto `SkillSourceHandler.describeLocation`; built-ins register both functions together. Dead `case "package"` is gone. — e20359eb
- **Partial-load repo leak + `close()` swallow**: constructor closes already-registered repos before propagating; close cascade aligned with `ResourceCache.close()`
(first-exception rethrow + suppressed). — 1c16c86a
- **`@Skills` merge non-determinism**: `addSkills` sorts keys lexically before merging; mixed-scheme test re-asserts as `List` to lock the order. — 8c2fe46c
- **`AgentSkill` JSON asymmetry**: not taking — see inline reply.
Six of seven taken; cross-language symmetry preserved on every fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Skill repo extending [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #655:
URL: https://github.com/apache/flink-agents/pull/655#issuecomment-4418927529
This pr based on #650
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3240589796
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/SkillMaterializer.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.Comparator;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
+
+/**
+ * Internal helpers for materializing skill sources (zip files, URL downloads, classpath JAR
+ * entries) into a local temp directory. Cleanup is registered as a JVM shutdown hook so the temp
+ * directories are removed when the process exits.
+ */
+public final class SkillMaterializer {
+
+ private static final String TEMP_DIR_PREFIX = "flink-agents-skills-";
+
+ private SkillMaterializer() {}
+
+ /**
+ * Register a JVM shutdown hook that removes {@code path} recursively. Failures during deletion
+ * are silently ignored (best-effort cleanup).
+ */
+ public static void registerCleanup(Path path) {
+ Thread hook =
+ new Thread(() -> deleteRecursively(path), "skill-cleanup-" + path.getFileName());
+ Runtime.getRuntime().addShutdownHook(hook);
+ }
+
+ /**
+ * Extract a zip into a fresh temp directory and return that directory. Validates every entry
+ * against zip-slip (paths must resolve inside the extraction directory). Registers a JVM
+ * shutdown hook to remove the extraction directory at process exit.
+ *
+ * @throws IOException if any zip entry resolves outside the extraction directory.
+ */
+ public static Path extractZipSafely(Path zipPath) throws IOException {
Review Comment:
Similar to the question above, I believe we can iterate on and implement this improvement in future versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3266018155
##########
python/flink_agents/runtime/skill/repository/skill_directory_reader.py:
##########
@@ -0,0 +1,137 @@
+################################################################################
+# 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.
+################################################################################
+"""Read-only accessor for an on-disk directory of skills.
+
+Composed by :class:`SkillRepository` implementations to handle the
+"parse SKILL.md under base_dir" half of their work, leaving each repo
+free to manage its own materialization and ``close()`` story.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, Dict, List
+
+from flink_agents.runtime.skill.skill_parser import SkillParser
+
+if TYPE_CHECKING:
+ from flink_agents.runtime.skill.agent_skill import AgentSkill
+
+logger = logging.getLogger(__name__)
+
+
+class SkillDirectoryReader:
+ """Reads skills from an already-materialized directory. No lifecycle."""
+
+ SKILL_MD_FILE = "SKILL.md"
+
+ def __init__(self, base_dir: Path) -> None:
+ """Wrap ``base_dir`` (must be an existing directory)."""
+ if base_dir is None:
+ msg = "Base directory cannot be None"
+ raise ValueError(msg)
+ resolved = Path(base_dir).resolve()
+ if not resolved.exists():
+ msg = f"Path does not exist: {resolved}"
+ raise ValueError(msg)
+ if not resolved.is_dir():
+ msg = f"Path must be a directory: {resolved}"
+ raise ValueError(msg)
+ self._base_dir = resolved
+
+ @property
+ def base_dir(self) -> Path:
+ """Absolute base directory."""
+ return self._base_dir
+
+ def get_skill_dir(self, name: str) -> Path:
+ """``base_dir / name``; existence not checked."""
+ return self._base_dir / name
+
+ def get_skill(self, name: str) -> AgentSkill | None:
+ """Parse ``base_dir/name/SKILL.md``; ``None`` if absent."""
+ skill_dir = self._base_dir / name
+ skill_md_path = skill_dir / self.SKILL_MD_FILE
+ if not skill_md_path.exists():
+ return None
+ return self._load_skill(skill_dir)
+
+ def get_skills(self) -> List[AgentSkill]:
+ """All skills under ``base_dir``, sorted by name."""
+ skills: List[AgentSkill] = []
+ for skill_name in self._list_skill_names():
+ skill = self.get_skill(skill_name)
+ if skill is not None:
+ skills.append(skill)
+ return skills
+
+ def get_resources(self, name: str) -> Dict[str, str]:
+ """All non-``SKILL.md`` files under the named skill, keyed by relative path."""
+ skill_dir = self._base_dir / name
+ if not skill_dir.is_dir():
+ return {}
+ return self._load_resources(skill_dir)
+
+ def _list_skill_names(self) -> List[str]:
+ return sorted(
+ [
+ entry.name
+ for entry in self._base_dir.iterdir()
+ if entry.is_dir() and (entry / self.SKILL_MD_FILE).exists()
+ ]
+ )
+
+ def _load_skill(self, skill_dir: Path) -> AgentSkill | None:
+ skill_md_path = skill_dir / self.SKILL_MD_FILE
+ if not skill_md_path.exists():
+ return None
+ try:
+ content = skill_md_path.read_text()
+ skill = SkillParser.parse_skill(content)
+ if skill.name != skill_dir.name:
+ logger.warning(
+ "The skill name %s is different from the base directory %s.",
+ skill.name,
+ skill_dir.name,
+ )
+ except Exception as e:
+ err_msg = f"Failed to load skill from {skill_dir}"
+ raise ValueError(err_msg) from e
+ else:
+ return skill
+
+ def _load_resources(self, skill_dir: Path) -> Dict[str, str]:
+ resources: Dict[str, str] = {}
+ for root, _dirs, files in os.walk(skill_dir):
+ root_path = Path(root)
+ for file_name in files:
+ if file_name == self.SKILL_MD_FILE:
+ continue
+ file_path = root_path / file_name
+ relative_path = str(file_path.relative_to(skill_dir))
+ try:
+ resources[relative_path] = file_path.read_text()
+ except UnicodeDecodeError:
+ resources[relative_path] = f"base64: {file_path.read_bytes()}"
Review Comment:
**This branch is not base64-encoding — binary resources are silently corrupted.**
```python
resources[relative_path] = f"base64: {file_path.read_bytes()}"
```
The f-string returns the `bytes` repr (literal `"base64: b'\\x89PNG...'"`), not encoded data. `base64.b64decode(value.removeprefix("base64:"))` will raise on any real payload. Java sibling uses `Base64.getEncoder().encodeToString(...)`, so the cross-language contract diverges silently and every binary resource read via `AgentSkill.get_resource()` on Python is unusable.
Fix: `"base64:" + base64.b64encode(file_path.read_bytes()).decode("ascii")`. Worth adding a PNG round-trip test through both languages — none exists today.
##########
python/flink_agents/runtime/skill/skill_manager.py:
##########
@@ -93,42 +108,91 @@ def get_skill_dirs(self, *names: str) -> List[str]:
"""Return absolute directory paths for the given skill names.
If no names are provided, returns directories for all filesystem-backed
- skills. Unknown names and skills not backed by a filesystem repo are
- silently skipped.
+ skills. Skills not backed by a filesystem repo are silently skipped.
+
+ Raises:
+ ValueError: If any provided name is not a registered skill.
"""
selected = names if names else tuple(self._repos.keys())
dirs: List[str] = []
for skill_name in selected:
repo = self._repos.get(skill_name)
- if isinstance(repo, FileSystemSkillRepository):
- dirs.append(str(repo.base_dir / skill_name))
+ if repo is None:
+ msg = (
+ f"Skill {skill_name} not found, "
+ f"available skill names are: {list(self._repos.keys())}"
+ )
+ raise ValueError(msg)
+ dir_path = repo.get_skill_dir(skill_name)
+ if dir_path is not None:
+ dirs.append(str(dir_path))
return dirs
def get_skill_dir(self, skill_name: str) -> Path | None:
"""Return absolute directory path for a single skill, if filesystem-backed."""
repo = self._repos.get(skill_name)
- if isinstance(repo, FileSystemSkillRepository):
- return repo.base_dir / skill_name
- return None
+ return None if repo is None else repo.get_skill_dir(skill_name)
def resolve_resource_path(self, skill_name: str, resource_path: str) -> Path | None:
"""Resolve a skill resource's relative path to an absolute filesystem path.
Returns None if the skill's repository doesn't support path resolution.
"""
repo = self._repos.get(skill_name)
- if isinstance(repo, FileSystemSkillRepository):
- resolved = repo.base_dir / skill_name / resource_path
- if resolved.exists() and resolved.is_file():
- return resolved
- return None
-
- def _load_skills_from_paths(self) -> None:
- for path in self._config.paths:
- repo = FileSystemSkillRepository(path)
- for skill in repo.get_skills():
- skill.set_resource_loader(
- lambda name=skill.name, r=repo: r.get_resources(name)
+ if repo is None:
+ return None
+ dir_path = repo.get_skill_dir(skill_name)
+ if dir_path is None:
+ return None
+ resolved = dir_path / resource_path
+ return resolved if resolved.is_file() else None
+
+ def _load_skills(self) -> None:
+ for spec in self._config.sources:
+ try:
+ handler = skill_source_registry.get(spec.scheme)
+ repo = handler.open(spec.params)
+ except (OSError, ValueError) as e:
+ msg = (
+ f"Failed to load skills from {spec.scheme}:{spec.params}"
)
- self._skills[skill.name] = skill
- self._repos[skill.name] = repo
+ raise RuntimeError(msg) from e
+ self._register_repo(repo, _origin_of(spec))
Review Comment:
**Partial-load leak: a failed source leaves earlier-opened repos uncleaned.**
Java's `loadAll()` now wraps each `open()` in try/catch and closes already-opened repos on failure (round-2 follow-up). Python here still just re-raises — if source #3 of 5 fails, sources #1 and #2 keep their extracted temp dirs and `atexit` handlers for the interpreter's lifetime. In a long-lived TaskManager Python worker every failed reload leaks.
Same try/finally + first-exception pattern as the Java side: collect opened repos in a list, close them on the failure path before re-raising.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -111,38 +148,113 @@ public List<String> getSkillDirs(List<String> names) {
@Nullable
public Path getSkillDir(String skillName) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- return ((FileSystemSkillRepository) repo).getBaseDir().resolve(skillName);
- }
- return null;
+ return repo == null ? null : repo.getSkillDir(skillName);
}
/** Resolve a skill resource's relative path to an absolute path, or {@code null} if missing. */
@Nullable
public Path resolveResourcePath(String skillName, String resourcePath) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- Path resolved =
- ((FileSystemSkillRepository) repo)
- .getBaseDir()
- .resolve(skillName)
- .resolve(resourcePath);
- if (Files.isRegularFile(resolved)) {
- return resolved;
+ if (repo == null) {
+ return null;
+ }
+ Path dir = repo.getSkillDir(skillName);
+ if (dir == null) {
+ return null;
+ }
+ Path resolved = dir.resolve(resourcePath);
+ return Files.isRegularFile(resolved) ? resolved : null;
+ }
+
+ private void loadAll() {
+ for (SkillSourceSpec spec : config.getSources()) {
+ try {
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ IllegalStateException toThrow =
+ new IllegalStateException(
+ "Failed to load skills from "
+ + spec.getScheme()
+ + ":"
+ + spec.getParams(),
+ e);
+ // Release repos registered before this point. The caller never receives a
+ // SkillManager reference (we're throwing from the constructor path), so
+ // without this cleanup their shutdown hooks + temp dirs would leak until
+ // JVM exit.
+ try {
+ closeRepos();
+ } catch (Exception cleanupError) {
+ toThrow.addSuppressed(cleanupError);
+ }
+ throw toThrow;
+ }
+ }
+ }
+
+ /**
+ * Build a {@link SkillOrigin} from a spec for diagnostics (WARN on duplicates, etc.). The
+ * location description is delegated to the handler registered for the scheme — see {@link
+ * SkillSourceHandler#describeLocation(Map)} — so adding a new scheme is a single registry call.
+ */
+ private static SkillOrigin originOf(SkillSourceSpec spec) {
+ SkillSourceHandler handler = SkillSourceRegistry.get(spec.getScheme());
+ return new SkillOrigin(spec.getScheme(), handler.describeLocation(spec.getParams()));
+ }
+
+ /**
+ * Close every owned {@link SkillRepository}, releasing any temp directories materialized for
+ * URL / classpath-zip / classpath-jar sources. Idempotent (delegated repos use {@link
+ * java.util.concurrent.atomic.AtomicBoolean} guards).
+ *
+ * <p>Mirrors {@code ResourceCache.close()}: the first failure is rethrown after every repo has
+ * been attempted, with subsequent failures attached as suppressed exceptions. This surfaces
+ * real shutdown bugs (locked files, permission denied, disk full) instead of silently
+ * swallowing them.
+ */
+ @Override
+ public void close() throws Exception {
+ closeRepos();
+ }
+
+ private void closeRepos() throws Exception {
+ // repos may map multiple skill names to the same repo instance; dedup by identity.
+ Set<SkillRepository> unique = Collections.newSetFromMap(new IdentityHashMap<>());
+ unique.addAll(repos.values());
+ Exception firstException = null;
+ for (SkillRepository repo : unique) {
+ try {
+ repo.close();
+ } catch (Exception e) {
+ if (firstException == null) {
+ firstException = e;
+ } else {
+ firstException.addSuppressed(e);
+ }
}
}
- return null;
+ if (firstException != null) {
+ throw firstException;
+ }
}
- private void loadFromPaths() {
- for (String path : config.getPaths()) {
- FileSystemSkillRepository repo = new FileSystemSkillRepository(path);
- for (AgentSkill skill : repo.getSkills()) {
- final String skillName = skill.getName();
- skill.setResourceLoader(() -> repo.getResources(skillName));
- skills.put(skillName, skill);
- repos.put(skillName, repo);
+ private void registerRepo(SkillRepository repo, SkillOrigin origin) {
+ for (AgentSkill skill : repo.getSkills()) {
+ final String skillName = skill.getName();
+ skill.setResourceLoader(() -> repo.getResources(skillName));
+ skill.setOrigin(origin);
+ AgentSkill previous = skills.put(skillName, skill);
+ if (previous != null) {
+ LOG.warn(
+ "Skill '{}' from {} overrides earlier registration from {}",
+ skillName,
+ origin,
+ previous.getOrigin() == null ? "<unknown>" : previous.getOrigin());
}
+ repos.put(skillName, repo);
Review Comment:
**Duplicate skill name overwrites the repo-index entry; the displaced repo is orphaned from `closeRepos()`. Bug exists in both languages.**
```java
repos.put(skillName, repo); // overwrites previous repo ref
```
`closeRepos()` iterates `repos.values()`, so when two sources contribute skill `"foo"` the displaced repo is no longer reachable — its temp dir survives until JVM exit, only the shutdown hook ever fires. The recently-added close-cascade plumbing can't help because the orphan is never visited. Python mirrors the same shape at `python/.../skill_manager.py:177`.
Fix: track opened repos in a separate `List<SkillRepository>` and iterate *that* in `closeRepos()` (dedup with `IdentityHashMap`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #655:
URL: https://github.com/apache/flink-agents/pull/655
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3240583037
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/URLSkillRepository.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * Skill repository backed by an http(s) URL pointing to a zip.
+ *
+ * <p>The zip is downloaded to a temp file and extracted into a process-local temp directory
+ * (cleaned up at JVM exit). The downloaded zip itself is removed once extraction completes.
+ */
+public class URLSkillRepository extends FileSystemSkillRepository {
+
+ private static final int REQUEST_TIMEOUT_MS = 90_000;
+
+ public URLSkillRepository(String url) throws IOException {
+ super(materialize(url));
+ }
+
+ private static Path materialize(String url) throws IOException {
Review Comment:
By design, we do not intend to support URL download validation and security protections in this version. However, I believe this could indeed become an issue in production environments, so we should consider adding it in the next release.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on PR #655:
URL: https://github.com/apache/flink-agents/pull/655#issuecomment-4441023842
Thanks for the PR. Left some comments as below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3240567851
##########
api/src/main/java/org/apache/flink/agents/api/skills/Skills.java:
##########
@@ -51,33 +57,76 @@ public class Skills extends SerializableResource {
/** Reserved name of the built-in bash tool used to execute skill scripts. */
public static final String BASH_TOOL = "bash";
- private List<String> paths;
+ private final List<String> paths;
Review Comment:
In current design, skills with the same name are not supported. If there are future user requirements, we can expand further.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3241672079
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/ClasspathSkillRepository.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+
+/**
+ * Skill repository backed by a classpath resource — either a directory under {@code
+ * src/main/resources} (typical dev / Maven layout) or a path inside a JAR on the classpath (typical
+ * deployment).
+ *
+ * <p>For directory resources the underlying {@link FileSystemSkillRepository} uses the real on-disk
+ * path. For JAR-entry resources, the matching entries are extracted into a process-local temp
+ * directory (cleaned up at JVM exit).
+ *
+ * <p>When a {@link URLClassLoader} is used and the JAR does not contain an explicit directory entry
+ * for the resource prefix, this class falls back to scanning the classloader's URLs directly to
+ * synthesize the correct {@code jar:} URL.
+ */
+public class ClasspathSkillRepository extends FileSystemSkillRepository {
+
+ public ClasspathSkillRepository(String resource) throws IOException {
+ this(resource, Thread.currentThread().getContextClassLoader());
+ }
+
+ /**
+ * Constructor that accepts an explicit classloader. Useful in tests to inject a {@code
+ * URLClassLoader} pointing at a freshly-built jar.
+ */
+ public ClasspathSkillRepository(String resource, ClassLoader classLoader) throws IOException {
+ super(materialize(resource, classLoader));
+ }
+
+ private static Path materialize(String resource, ClassLoader classLoader) throws IOException {
+ URL url = classLoader.getResource(resource);
+ if (url == null) {
+ // Fallback for URLClassLoader: JARs without explicit directory entries will not return
+ // a URL for a prefix via getResource(). Scan the classloader's URLs directly to find
+ // any JAR entry that starts with the resource prefix.
+ url = findInUrlClassLoader(resource, classLoader);
+ }
+ if (url == null) {
+ throw new IllegalArgumentException("Classpath resource not found: " + resource);
+ }
+ String protocol = url.getProtocol();
+ if ("file".equals(protocol)) {
+ Path p;
+ try {
+ p = Paths.get(url.toURI());
+ } catch (URISyntaxException e) {
+ throw new IOException("Bad classpath URL: " + url, e);
+ }
+ if (Files.isDirectory(p)) {
+ return p;
+ }
+ if (Files.isRegularFile(p) && p.toString().toLowerCase().endsWith(".zip")) {
+ return SkillMaterializer.extractZipSafely(p);
+ }
+ throw new IllegalArgumentException(
+ "Classpath resource must be a directory or a .zip: " + p);
+ }
+ if ("jar".equals(protocol)) {
+ return SkillMaterializer.extractClasspathFromJar(url, resource);
+ }
+ throw new IllegalArgumentException(
Review Comment:
What is the significance of highlighting `URLClassLoader` and `ChildFirstClassLoader` here? I understand that ChildFirstClassLoader is also a subclass of URLClassLoader.
Since we set the contextClassLoader to Flink's UserCodeClassloader before executing a JavaAction, it is used in most scenarios. However, there was an issue with the initialization of the skill-managed path in Python Actions. I have modified the code to explicitly pass the UserCodeClassloader.
Typically, I understand that the Flink user JAR does not use the jar-in-jar approach.
`getResourceAsStream` requires a file path as input, but the `resource` here is typically a directory name, so getResourceAsStream cannot be used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3234514268
##########
api/src/main/java/org/apache/flink/agents/api/skills/Skills.java:
##########
@@ -51,33 +57,76 @@ public class Skills extends SerializableResource {
/** Reserved name of the built-in bash tool used to execute skill scripts. */
public static final String BASH_TOOL = "bash";
- private List<String> paths;
+ private final List<String> paths;
+ private final List<String> urls;
+ private final List<String> classpathResources;
Review Comment:
**Adding a new source today forces 5 parallel changes** (`Skills` field + factory + `XxxSkillRepository` + `SkillManager` for-loop + `AgentPlan` merge branch).
Suggest a discriminated union — single `sources: List<SkillSourceSpec(scheme, location)>` field + a `scheme → handler` registry. New source = one registration, no other file touched. Bonus: cross-language ser/de becomes uniform.
Wire-format-affecting; much cheaper to fix before 0.3.0 ships.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/ClasspathSkillRepository.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+
+/**
+ * Skill repository backed by a classpath resource — either a directory under {@code
+ * src/main/resources} (typical dev / Maven layout) or a path inside a JAR on the classpath (typical
+ * deployment).
+ *
+ * <p>For directory resources the underlying {@link FileSystemSkillRepository} uses the real on-disk
+ * path. For JAR-entry resources, the matching entries are extracted into a process-local temp
+ * directory (cleaned up at JVM exit).
+ *
+ * <p>When a {@link URLClassLoader} is used and the JAR does not contain an explicit directory entry
+ * for the resource prefix, this class falls back to scanning the classloader's URLs directly to
+ * synthesize the correct {@code jar:} URL.
+ */
+public class ClasspathSkillRepository extends FileSystemSkillRepository {
+
+ public ClasspathSkillRepository(String resource) throws IOException {
+ this(resource, Thread.currentThread().getContextClassLoader());
+ }
+
+ /**
+ * Constructor that accepts an explicit classloader. Useful in tests to inject a {@code
+ * URLClassLoader} pointing at a freshly-built jar.
+ */
+ public ClasspathSkillRepository(String resource, ClassLoader classLoader) throws IOException {
+ super(materialize(resource, classLoader));
+ }
+
+ private static Path materialize(String resource, ClassLoader classLoader) throws IOException {
+ URL url = classLoader.getResource(resource);
+ if (url == null) {
+ // Fallback for URLClassLoader: JARs without explicit directory entries will not return
+ // a URL for a prefix via getResource(). Scan the classloader's URLs directly to find
+ // any JAR entry that starts with the resource prefix.
+ url = findInUrlClassLoader(resource, classLoader);
+ }
+ if (url == null) {
+ throw new IllegalArgumentException("Classpath resource not found: " + resource);
+ }
+ String protocol = url.getProtocol();
+ if ("file".equals(protocol)) {
+ Path p;
+ try {
+ p = Paths.get(url.toURI());
+ } catch (URISyntaxException e) {
+ throw new IOException("Bad classpath URL: " + url, e);
+ }
+ if (Files.isDirectory(p)) {
+ return p;
+ }
+ if (Files.isRegularFile(p) && p.toString().toLowerCase().endsWith(".zip")) {
+ return SkillMaterializer.extractZipSafely(p);
+ }
+ throw new IllegalArgumentException(
+ "Classpath resource must be a directory or a .zip: " + p);
+ }
+ if ("jar".equals(protocol)) {
+ return SkillMaterializer.extractClasspathFromJar(url, resource);
+ }
+ throw new IllegalArgumentException(
Review Comment:
**⚠️ Flink fat-jar / nested-jar deployment is not covered.**
Flink user JARs typically use Maven Shade or jar-in-jar; `getResource("skills")` returns protocols like `jar:nested:` that throw right here. `findInUrlClassLoader` also assumes `URLClassLoader`, not Flink's `ChildFirstClassLoader`.
Issue #593 explicitly targets this scenario. Before merge: (1) an end-to-end test on a Flink mini-cluster, (2) use `ClassLoader.getResources()` + `getResourceAsStream` instead of dispatching on URL protocol.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/URLSkillRepository.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * Skill repository backed by an http(s) URL pointing to a zip.
+ *
+ * <p>The zip is downloaded to a temp file and extracted into a process-local temp directory
+ * (cleaned up at JVM exit). The downloaded zip itself is removed once extraction completes.
+ */
+public class URLSkillRepository extends FileSystemSkillRepository {
Review Comment:
**Inheritance is being used as composition.**
A URL repo is not logically *a* FileSystem repo — it fetches over the network, then delegates. Side effects: the constructor performs network I/O before `super(...)`, the original `url` is lost for diagnostics, there's no `close()` lifecycle, and the fetch step can't be mocked alone.
Suggest a `SkillSource { Path materialize() }` interface composed into one `MaterializingSkillRepository`. Same applies to `ClasspathSkillRepository` and `PackageSkillRepository`.
##########
api/src/main/java/org/apache/flink/agents/api/skills/Skills.java:
##########
@@ -51,33 +57,76 @@ public class Skills extends SerializableResource {
/** Reserved name of the built-in bash tool used to execute skill scripts. */
public static final String BASH_TOOL = "bash";
- private List<String> paths;
+ private final List<String> paths;
Review Comment:
**Skill names are flat — no namespace, no version.** With this PR opening multiple loading sources, two teams' `github` skills will collide (last writer silently wins).
Options: (1) `namespace:` in SKILL.md frontmatter → `team-a/github`, (2) per-source namespace `fromUrl(url, namespace="team-a")`, (3) source identity *is* the namespace. Wire-format-affecting — easier now than after 0.3.0.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -134,15 +137,33 @@ public Path resolveResourcePath(String skillName, String resourcePath) {
return null;
}
- private void loadFromPaths() {
+ private void loadAll() {
for (String path : config.getPaths()) {
- FileSystemSkillRepository repo = new FileSystemSkillRepository(path);
- for (AgentSkill skill : repo.getSkills()) {
- final String skillName = skill.getName();
- skill.setResourceLoader(() -> repo.getResources(skillName));
- skills.put(skillName, skill);
- repos.put(skillName, repo);
+ registerRepo(new FileSystemSkillRepository(path));
+ }
+ for (String url : config.getUrls()) {
+ try {
+ registerRepo(new URLSkillRepository(url));
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to load skills from URL: " + url, e);
+ }
+ }
+ for (String resource : config.getClasspathResources()) {
+ try {
+ registerRepo(new ClasspathSkillRepository(resource));
+ } catch (IOException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from classpath: " + resource, e);
}
}
}
Review Comment:
**`AgentSkill` doesn't know where it came from.** Origin is captured only inside a lambda; `SkillManager` keeps a parallel `repos` map to compensate. Logs and errors can't name the source, the duplicate-name WARN can't either, and serialization loses origin entirely.
Suggest a declarative `record SkillOrigin(String scheme, String location, Instant loadedAt)` field on `AgentSkill`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/SkillMaterializer.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.Comparator;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
+
+/**
+ * Internal helpers for materializing skill sources (zip files, URL downloads, classpath JAR
+ * entries) into a local temp directory. Cleanup is registered as a JVM shutdown hook so the temp
+ * directories are removed when the process exits.
+ */
+public final class SkillMaterializer {
+
+ private static final String TEMP_DIR_PREFIX = "flink-agents-skills-";
+
+ private SkillMaterializer() {}
+
+ /**
+ * Register a JVM shutdown hook that removes {@code path} recursively. Failures during deletion
+ * are silently ignored (best-effort cleanup).
+ */
+ public static void registerCleanup(Path path) {
+ Thread hook =
+ new Thread(() -> deleteRecursively(path), "skill-cleanup-" + path.getFileName());
+ Runtime.getRuntime().addShutdownHook(hook);
+ }
+
+ /**
+ * Extract a zip into a fresh temp directory and return that directory. Validates every entry
+ * against zip-slip (paths must resolve inside the extraction directory). Registers a JVM
+ * shutdown hook to remove the extraction directory at process exit.
+ *
+ * @throws IOException if any zip entry resolves outside the extraction directory.
+ */
+ public static Path extractZipSafely(Path zipPath) throws IOException {
+ Path extractDir = Files.createTempDirectory(TEMP_DIR_PREFIX);
+ // Register cleanup before validation so the empty tempdir is always reclaimed,
+ // even if validation raises.
+ registerCleanup(extractDir);
+ try (ZipFile zf = new ZipFile(zipPath.toFile())) {
+ // First pass: zip-slip validation.
+ Enumeration<? extends ZipEntry> entries = zf.entries();
+ while (entries.hasMoreElements()) {
Review Comment:
**Two-pass scan can be merged into one** — validate-then-extract per entry in a single pass. Not a perf concern, just shorter.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -134,15 +137,33 @@ public Path resolveResourcePath(String skillName, String resourcePath) {
return null;
}
- private void loadFromPaths() {
+ private void loadAll() {
for (String path : config.getPaths()) {
- FileSystemSkillRepository repo = new FileSystemSkillRepository(path);
- for (AgentSkill skill : repo.getSkills()) {
- final String skillName = skill.getName();
- skill.setResourceLoader(() -> repo.getResources(skillName));
- skills.put(skillName, skill);
- repos.put(skillName, repo);
+ registerRepo(new FileSystemSkillRepository(path));
+ }
+ for (String url : config.getUrls()) {
+ try {
+ registerRepo(new URLSkillRepository(url));
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to load skills from URL: " + url, e);
+ }
+ }
+ for (String resource : config.getClasspathResources()) {
+ try {
+ registerRepo(new ClasspathSkillRepository(resource));
+ } catch (IOException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from classpath: " + resource, e);
}
}
}
+
+ private void registerRepo(SkillRepository repo) {
Review Comment:
**Duplicate skill names overwrite silently with no log.** Multi-source name collisions go through `skills.put()` and the later registration wins — no WARN, no error. Users debugging "why doesn't my `SKILL.md` take effect" get no diagnostic.
Minimum: WARN with both origins. Better: document the `paths → urls → classpath` precedence. (Naming the source concretely needs the `SkillOrigin` change on `AgentSkill`.)
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/SkillMaterializer.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.Comparator;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
+
+/**
+ * Internal helpers for materializing skill sources (zip files, URL downloads, classpath JAR
+ * entries) into a local temp directory. Cleanup is registered as a JVM shutdown hook so the temp
+ * directories are removed when the process exits.
+ */
+public final class SkillMaterializer {
+
+ private static final String TEMP_DIR_PREFIX = "flink-agents-skills-";
+
+ private SkillMaterializer() {}
+
+ /**
+ * Register a JVM shutdown hook that removes {@code path} recursively. Failures during deletion
+ * are silently ignored (best-effort cleanup).
+ */
+ public static void registerCleanup(Path path) {
+ Thread hook =
+ new Thread(() -> deleteRecursively(path), "skill-cleanup-" + path.getFileName());
+ Runtime.getRuntime().addShutdownHook(hook);
+ }
+
+ /**
+ * Extract a zip into a fresh temp directory and return that directory. Validates every entry
+ * against zip-slip (paths must resolve inside the extraction directory). Registers a JVM
+ * shutdown hook to remove the extraction directory at process exit.
+ *
+ * @throws IOException if any zip entry resolves outside the extraction directory.
+ */
+ public static Path extractZipSafely(Path zipPath) throws IOException {
+ Path extractDir = Files.createTempDirectory(TEMP_DIR_PREFIX);
+ // Register cleanup before validation so the empty tempdir is always reclaimed,
+ // even if validation raises.
+ registerCleanup(extractDir);
+ try (ZipFile zf = new ZipFile(zipPath.toFile())) {
+ // First pass: zip-slip validation.
+ Enumeration<? extends ZipEntry> entries = zf.entries();
+ while (entries.hasMoreElements()) {
+ ZipEntry entry = entries.nextElement();
+ Path resolved = extractDir.resolve(entry.getName()).normalize();
+ if (!resolved.startsWith(extractDir)) {
+ throw new IOException("Unsafe zip entry: " + entry.getName());
+ }
+ }
+ // Second pass: extract.
+ entries = zf.entries();
+ while (entries.hasMoreElements()) {
+ ZipEntry entry = entries.nextElement();
+ Path target = extractDir.resolve(entry.getName()).normalize();
+ if (entry.isDirectory()) {
+ Files.createDirectories(target);
+ } else {
+ Files.createDirectories(target.getParent());
+ try (InputStream in = zf.getInputStream(entry)) {
+ Files.copy(in, target);
+ }
+ }
+ }
+ }
+ return extractDir;
+ }
+
+ /**
+ * Extract every JAR entry whose name starts with {@code resourcePrefix + "/"} into a fresh temp
+ * directory. The prefix itself is stripped (so entries under {@code skills/skill-a/...} extract
+ * as {@code skill-a/...}).
+ *
+ * <p>Registers a JVM shutdown hook for cleanup. Rejects entries that would resolve outside the
+ * extraction directory (zip-slip).
+ */
+ public static Path extractClasspathFromJar(URL jarUrl, String resourcePrefix)
+ throws IOException {
+ // Parse the JAR file URL from the jar: URL. The format is jar:<jar-file-url>!/[entry].
+ // We extract just the inner jar-file URL so we can open the whole JarFile and enumerate
+ // all entries — JarURLConnection.getJarFile() would fail when the entry specifier names a
+ // prefix that has no corresponding stored directory entry.
+ String spec = jarUrl.toString();
+ int sep = spec.indexOf("!/");
+ String innerSpec = sep >= 0 ? spec.substring(4, sep) : spec.substring(4);
Review Comment:
**Magic number.** `4` is `"jar:".length()`. Extract `private static final int JAR_URL_PREFIX_LEN = "jar:".length();`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/ClasspathSkillRepository.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+
+/**
+ * Skill repository backed by a classpath resource — either a directory under {@code
+ * src/main/resources} (typical dev / Maven layout) or a path inside a JAR on the classpath (typical
+ * deployment).
+ *
+ * <p>For directory resources the underlying {@link FileSystemSkillRepository} uses the real on-disk
+ * path. For JAR-entry resources, the matching entries are extracted into a process-local temp
+ * directory (cleaned up at JVM exit).
+ *
+ * <p>When a {@link URLClassLoader} is used and the JAR does not contain an explicit directory entry
+ * for the resource prefix, this class falls back to scanning the classloader's URLs directly to
+ * synthesize the correct {@code jar:} URL.
+ */
+public class ClasspathSkillRepository extends FileSystemSkillRepository {
+
+ public ClasspathSkillRepository(String resource) throws IOException {
+ this(resource, Thread.currentThread().getContextClassLoader());
+ }
+
+ /**
+ * Constructor that accepts an explicit classloader. Useful in tests to inject a {@code
+ * URLClassLoader} pointing at a freshly-built jar.
+ */
+ public ClasspathSkillRepository(String resource, ClassLoader classLoader) throws IOException {
+ super(materialize(resource, classLoader));
+ }
+
+ private static Path materialize(String resource, ClassLoader classLoader) throws IOException {
+ URL url = classLoader.getResource(resource);
+ if (url == null) {
+ // Fallback for URLClassLoader: JARs without explicit directory entries will not return
+ // a URL for a prefix via getResource(). Scan the classloader's URLs directly to find
+ // any JAR entry that starts with the resource prefix.
+ url = findInUrlClassLoader(resource, classLoader);
Review Comment:
**`findInUrlClassLoader` silently uses only the first matching jar.** Multiple plugin jars with the same prefix → every jar after the first is dropped, no log. Minimum: log the choice. Ideally merge across all matches, or fail loudly when more than one matches.
Also assumes `URLClassLoader`, not Flink's `ChildFirstClassLoader` — see the nested-jar thread on `materialize`.
##########
python/flink_agents/api/skills.py:
##########
@@ -43,23 +59,56 @@ def my_skills() -> Skills:
class Skills(SerializableResource):
"""A resource describing where to load agent skills from.
- Use :meth:`from_local_dir` to construct — direct field construction is
- reserved for internal serialization and not part of the public API.
+ Use one of the ``from_*`` factory methods to construct — direct field
+ construction is reserved for internal serialization and not part of the
+ public API.
"""
- # Internal location list populated through the factory methods; kept
- # as a public pydantic field so it can be serialized/deserialized.
+ # Filesystem paths. Each entry may be a directory whose immediate
+ # subdirectories each contain a ``SKILL.md`` file, or a ``.zip`` file
+ # whose top-level entries are the skill subdirectories.
paths: List[str] = Field(default_factory=list)
+ # http(s) URLs. Each URL must point to a ``.zip`` whose top level is
+ # the baseDir.
+ urls: List[str] = Field(default_factory=list)
+
+ # (package_name, resource_path) pairs. Each entry references an installed
+ # Python package and a path inside it; the resource may be a directory or
+ # a ``.zip`` file.
+ packages: List[Tuple[str, str]] = Field(default_factory=list)
+
@classmethod
def from_local_dir(cls, *paths: str) -> Skills:
- """Create a Skills resource from one or more local filesystem directories.
+ """Create a Skills resource from one or more local paths.
- Each path points to a directory whose immediate subdirectories each
- contain a ``SKILL.md`` file.
+ Each path may be a directory or a ``.zip`` file. For a directory, its
+ immediate subdirectories must each contain a ``SKILL.md`` file. For
+ a zip, its top-level entries are the skill subdirectories.
"""
return cls(paths=list(paths))
+ @classmethod
+ def from_url(cls, *urls: str) -> Skills:
+ """Create a Skills resource from one or more http(s) URLs.
+
+ Each URL must point to a ``.zip`` whose top level is the baseDir
+ (i.e. skill subdirectories sit at the top of the zip).
+ """
+ return cls(urls=list(urls))
+
+ @classmethod
+ def from_package(cls, package: str, resource: str) -> Skills:
Review Comment:
**API asymmetry.** Java `fromClasspath(String...)` is varargs; Python `from_package(package, resource)` is single-pair, so two packages require two `@skills` functions.
Make it symmetric: `def from_package(cls, *pairs: Tuple[str, str])`. See also the cross-language drop thread on the `packages` field.
##########
plan/src/test/java/org/apache/flink/agents/plan/AgentPlanDeclareSkillsTest.java:
##########
@@ -126,6 +127,6 @@ void programmaticSkillsAddResourceParticipates() throws Exception {
.provide(
org.apache.flink.agents.api.resource.ResourceContext
.fromGetResource((n, t) -> null));
- assertEquals(java.util.List.of("/tmp/skill-d"), merged.getPaths());
+ assertEquals(List.of("/tmp/skill-d"), merged.getPaths());
Review Comment:
**Merge-coverage gap.** `AgentPlan.addSkills` now merges three `LinkedHashSet`s but the test only asserts on `getPaths()`. Add cases for overlapping `urls`, overlapping `classpathResources`, and a mixed three-source case asserting all three end up in the merged result.
##########
python/flink_agents/api/skills.py:
##########
@@ -43,23 +59,56 @@ def my_skills() -> Skills:
class Skills(SerializableResource):
"""A resource describing where to load agent skills from.
- Use :meth:`from_local_dir` to construct — direct field construction is
- reserved for internal serialization and not part of the public API.
+ Use one of the ``from_*`` factory methods to construct — direct field
+ construction is reserved for internal serialization and not part of the
+ public API.
"""
- # Internal location list populated through the factory methods; kept
- # as a public pydantic field so it can be serialized/deserialized.
+ # Filesystem paths. Each entry may be a directory whose immediate
+ # subdirectories each contain a ``SKILL.md`` file, or a ``.zip`` file
+ # whose top-level entries are the skill subdirectories.
paths: List[str] = Field(default_factory=list)
+ # http(s) URLs. Each URL must point to a ``.zip`` whose top level is
+ # the baseDir.
+ urls: List[str] = Field(default_factory=list)
Review Comment:
**Cross-language deserialization silently drops fields.** Python writes `packages` → Java drops it → 0 skills loaded, no warning. Mirror for Java `classpathResources` → Python.
Short term: emit a WARN on unrecognized source fields during deserialization. Long term: unify under one cross-language scheme (see the `sources` discriminator suggested on `Skills.java`).
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/URLSkillRepository.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * Skill repository backed by an http(s) URL pointing to a zip.
+ *
+ * <p>The zip is downloaded to a temp file and extracted into a process-local temp directory
+ * (cleaned up at JVM exit). The downloaded zip itself is removed once extraction completes.
+ */
+public class URLSkillRepository extends FileSystemSkillRepository {
+
+ private static final int REQUEST_TIMEOUT_MS = 90_000;
+
+ public URLSkillRepository(String url) throws IOException {
+ super(materialize(url));
+ }
+
+ private static Path materialize(String url) throws IOException {
Review Comment:
**URL zip is loaded with no integrity verification** — vulnerable to truncated downloads, stale CDN cache, MITM with a private CA, and proxy rewrites.
`Skills.fromUrl(url, "sha256:…")` with verification before extraction; mandatory in production. Same place to add an optional host allowlist for SSRF mitigation.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/SkillMaterializer.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.Comparator;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
+
+/**
+ * Internal helpers for materializing skill sources (zip files, URL downloads, classpath JAR
+ * entries) into a local temp directory. Cleanup is registered as a JVM shutdown hook so the temp
+ * directories are removed when the process exits.
+ */
+public final class SkillMaterializer {
+
+ private static final String TEMP_DIR_PREFIX = "flink-agents-skills-";
+
+ private SkillMaterializer() {}
+
+ /**
+ * Register a JVM shutdown hook that removes {@code path} recursively. Failures during deletion
+ * are silently ignored (best-effort cleanup).
+ */
+ public static void registerCleanup(Path path) {
+ Thread hook =
+ new Thread(() -> deleteRecursively(path), "skill-cleanup-" + path.getFileName());
+ Runtime.getRuntime().addShutdownHook(hook);
+ }
+
+ /**
+ * Extract a zip into a fresh temp directory and return that directory. Validates every entry
+ * against zip-slip (paths must resolve inside the extraction directory). Registers a JVM
+ * shutdown hook to remove the extraction directory at process exit.
+ *
+ * @throws IOException if any zip entry resolves outside the extraction directory.
+ */
+ public static Path extractZipSafely(Path zipPath) throws IOException {
Review Comment:
**Missing zip-bomb protection.** No cap on entry count, per-entry size, total size, or compression ratio — a small DEFLATE bomb can exhaust TaskManager disk/memory.
Add limits (e.g. ≤ 50 MB per entry, ≤ 200 MB total, ≤ 10 000 entries, reject ratio > 100:1) and tests asserting the rejection.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -44,7 +47,7 @@ public class SkillManager {
public SkillManager(Skills config) {
Review Comment:
**Constructor is synchronous, blocking, and not recoverable.** `loadAll()` fetches sources serially; one slow URL stalls startup, one failure kills the whole `SkillManager`, and `repos` is `final` so there's no hot reload.
Follow-up: parallel materialization (`CompletableFuture`), fail-soft mode with a failure list, lazy load on first `getSkill()`, optional `reloadSource()` API.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/SkillMaterializer.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.Comparator;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
+
+/**
+ * Internal helpers for materializing skill sources (zip files, URL downloads, classpath JAR
+ * entries) into a local temp directory. Cleanup is registered as a JVM shutdown hook so the temp
+ * directories are removed when the process exits.
+ */
+public final class SkillMaterializer {
+
+ private static final String TEMP_DIR_PREFIX = "flink-agents-skills-";
+
+ private SkillMaterializer() {}
+
+ /**
+ * Register a JVM shutdown hook that removes {@code path} recursively. Failures during deletion
+ * are silently ignored (best-effort cleanup).
+ */
+ public static void registerCleanup(Path path) {
Review Comment:
**Shutdown hooks accumulate unboundedly.** Each zip / URL / classpath repo registers one hook; long-running processes that rebuild `SkillManager` (retries, hot reload) leak hooks monotonically.
Fix: cache materializations by source digest (best — eliminates the leak entirely), or expose `close()` that calls `removeShutdownHook` + deletes immediately. At minimum javadoc that `SkillManager` must not be reconstructed in a loop.
##########
api/src/main/java/org/apache/flink/agents/api/skills/Skills.java:
##########
@@ -51,33 +57,76 @@ public class Skills extends SerializableResource {
/** Reserved name of the built-in bash tool used to execute skill scripts. */
public static final String BASH_TOOL = "bash";
- private List<String> paths;
+ private final List<String> paths;
+ private final List<String> urls;
+ private final List<String> classpathResources;
/** Required by Jackson. */
public Skills() {
this.paths = Collections.emptyList();
+ this.urls = Collections.emptyList();
+ this.classpathResources = Collections.emptyList();
}
@JsonCreator
- public Skills(@JsonProperty("paths") List<String> paths) {
+ public Skills(
Review Comment:
**Source-incompatible constructor change.** Wire compat is fine; but `new Skills(List.of("/data/skills"))` no longer compiles. Either keep a `@Deprecated` 1-arg constructor that forwards, or call it out in the 0.3.0 release notes.
If `Skills` later migrates to a single `sources` field (see the discriminator thread), this signature changes again — better to break once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #655:
URL: https://github.com/apache/flink-agents/pull/655#issuecomment-4452438090
Hi, @wzhero1, Ty for your thorough review.
I fixed some design issues and bugs, including
- Skills rewritten as a single discriminated sources: List<SkillSourceSpec(scheme, params)> + SkillSourceRegistry. Uniform Java/Python wire format.
- URL / Classpath / Package / FileSystem repos are now sibling SkillRepository impls composing a shared SkillDirectoryReader. No more inheritance-as-composition.
- SkillOrigin(scheme, location) attached to every AgentSkill at registration.
- SkillRepository extends AutoCloseable; SkillManager.close() cascades through ResourceContextImpl on operator close (failover-safe).
- Duplicate skill names now LOG.warn with both origins; last-write-wins kept.
- All matching classpath JARs merged via extractClasspathFromJars; same-path collisions WARN.
- Single-pass zip extract; JAR_URL_PREFIX_LEN named constant.
The remaining issues related to robustness and security will be iterated on in subsequent phases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3241705957
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/ClasspathSkillRepository.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+
+/**
+ * Skill repository backed by a classpath resource — either a directory under {@code
+ * src/main/resources} (typical dev / Maven layout) or a path inside a JAR on the classpath (typical
+ * deployment).
+ *
+ * <p>For directory resources the underlying {@link FileSystemSkillRepository} uses the real on-disk
+ * path. For JAR-entry resources, the matching entries are extracted into a process-local temp
+ * directory (cleaned up at JVM exit).
+ *
+ * <p>When a {@link URLClassLoader} is used and the JAR does not contain an explicit directory entry
+ * for the resource prefix, this class falls back to scanning the classloader's URLs directly to
+ * synthesize the correct {@code jar:} URL.
+ */
+public class ClasspathSkillRepository extends FileSystemSkillRepository {
+
+ public ClasspathSkillRepository(String resource) throws IOException {
+ this(resource, Thread.currentThread().getContextClassLoader());
+ }
+
+ /**
+ * Constructor that accepts an explicit classloader. Useful in tests to inject a {@code
+ * URLClassLoader} pointing at a freshly-built jar.
+ */
+ public ClasspathSkillRepository(String resource, ClassLoader classLoader) throws IOException {
+ super(materialize(resource, classLoader));
+ }
+
+ private static Path materialize(String resource, ClassLoader classLoader) throws IOException {
+ URL url = classLoader.getResource(resource);
+ if (url == null) {
+ // Fallback for URLClassLoader: JARs without explicit directory entries will not return
+ // a URL for a prefix via getResource(). Scan the classloader's URLs directly to find
+ // any JAR entry that starts with the resource prefix.
+ url = findInUrlClassLoader(resource, classLoader);
Review Comment:
I think merge across all matches makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3263199741
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/URLSkillRepository.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import org.apache.flink.agents.runtime.skill.AgentSkill;
+import org.apache.flink.agents.runtime.skill.SkillRepository;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Skill repository backed by an http(s) URL pointing to a zip.
+ *
+ * <p>The zip is downloaded to a temp file and extracted into a process-local temp directory. The
+ * downloaded zip itself is removed once extraction completes; the extracted directory is released
+ * via {@link #close()} (cascaded through {@code SkillManager} → {@code ResourceContextImpl} →
+ * {@code ResourceCache} on operator close). A JVM shutdown hook acts as fallback cleanup if {@code
+ * close()} is never called.
+ *
+ * <p>Composes a {@link SkillDirectoryReader} for the on-disk parsing; the materialization happens
+ * upfront in the constructor (no abuse of {@code super(...)}).
+ */
+public final class URLSkillRepository implements SkillRepository {
+
+ private static final int REQUEST_TIMEOUT_MS = 90_000;
+
+ private final String url;
Review Comment:
**Field name `owned` is an adjective without a subject.** maybe `ownerMaterialization`
Same applies to `ClasspathSkillRepository.java:73` and `FileSystemSkillRepository.java:51`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on PR #655:
URL: https://github.com/apache/flink-agents/pull/655#issuecomment-4495619767
approved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3264374788
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/AgentSkill.java:
##########
@@ -44,6 +44,7 @@ public final class AgentSkill {
@Nullable private volatile Map<String, String> resources;
@Nullable private Supplier<Map<String, String>> resourceLoader;
private volatile boolean activated;
+ @Nullable private volatile SkillOrigin origin;
Review Comment:
Took a closer look and decided to skip this for now. Grep'd every `AgentSkill` reference in the repo (Java + Python) — no serialization path uses it today, and Jackson's
default bean introspection would emit `getOrigin()` as `"origin"` automatically even without the annotation, so the "silently drops" failure mode doesn't actually fire under
the project's default `ObjectMapper`.
Also worth noting: I ran Python's `model_dump()` and it emits `"origin": null` by default (pydantic v2 doesn't drop None fields), so `@JsonInclude(NON_NULL)` would have made
the two sides byte-asymmetric. I'd rather design the schema (NON_NULL vs null, `resources` handling, `@JsonCreator`) when there's an actual wire path with concrete
requirements. Happy to revisit if you have a specific path in mind.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3240913481
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -44,7 +47,7 @@ public class SkillManager {
public SkillManager(Skills config) {
Review Comment:
This behavior will be optimized in subsequent iterations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #655:
URL: https://github.com/apache/flink-agents/pull/655#issuecomment-4488644387
Hi, @wzhero1, thanks for round three. All three taken:
- **Python base64 silently corrupting binary resources**: spot on — f-string was inlining the bytes repr, not real base64. Switched to
`base64.b64encode(...).decode("ascii")`. The single-line `startsWith("base64: ")` assertion is exactly how this hid for a cycle; both Java and Python tests now b64-decode the
value and assert the round-trip against a PNG signature + non-UTF-8 byte sequence. — e7956487
- **Python partial-load leak**: Python `_load_skills` now closes already-opened repos before re-raising, mirroring the Java fix from round two. — aa7c9785
- **Duplicate-name repo orphan in `close()`**: confirmed on both sides — the skill-name → repo map drops the displaced repo's reference, so iterating it on close orphaned the
earlier repo. Added an `openedRepos`/`_opened_repos` list (identity / `id()` dedup) and changed close to iterate that. New test on each side: two sources contributing the
same skill name, both repos must be closed. — 8b10d3ab
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3263199104
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/URLSkillRepository.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import org.apache.flink.agents.runtime.skill.AgentSkill;
+import org.apache.flink.agents.runtime.skill.SkillRepository;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Skill repository backed by an http(s) URL pointing to a zip.
+ *
+ * <p>The zip is downloaded to a temp file and extracted into a process-local temp directory. The
+ * downloaded zip itself is removed once extraction completes; the extracted directory is released
+ * via {@link #close()} (cascaded through {@code SkillManager} → {@code ResourceContextImpl} →
+ * {@code ResourceCache} on operator close). A JVM shutdown hook acts as fallback cleanup if {@code
+ * close()} is never called.
+ *
+ * <p>Composes a {@link SkillDirectoryReader} for the on-disk parsing; the materialization happens
+ * upfront in the constructor (no abuse of {@code super(...)}).
+ */
+public final class URLSkillRepository implements SkillRepository {
+
+ private static final int REQUEST_TIMEOUT_MS = 90_000;
+
+ private final String url;
Review Comment:
**Field name `owned` is an adjective without a subject.**
`Materialized` is already a noun. Rename the field to `materialization` (matches the class name), or rename the inner class to `MaterializedDir` so the field reads `materializedDir`. Current call sites `this.owned.close()` / `if (owned == null)` don't read well.
Same applies to `ClasspathSkillRepository.java:73` and `FileSystemSkillRepository.java:51`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillRepository.java:
##########
@@ -20,14 +20,19 @@
import javax.annotation.Nullable;
+import java.nio.file.Path;
import java.util.List;
import java.util.Map;
/**
* Source of skills. Mirrors the Python {@code
* flink_agents.runtime.skill.skill_repository.SkillRepository}.
+ *
+ * <p>Implementations that own a materialized temp directory (URL / classpath / zip) should override
Review Comment:
**Three sibling repos duplicate the same 5 delegating methods + state** (`reader`, `materialization`, plus `getSkill`/`getSkills`/`getResources`/`getSkillDir`/`close`) — ~45 lines of boilerplate, +15 per future source.
Extract `AbstractMaterializedSkillRepository` holding the shared state and implementing the delegators once; concrete repos reduce to source-specific I/O + an abstract `getOrigin()`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -111,38 +148,94 @@ public List<String> getSkillDirs(List<String> names) {
@Nullable
public Path getSkillDir(String skillName) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- return ((FileSystemSkillRepository) repo).getBaseDir().resolve(skillName);
- }
- return null;
+ return repo == null ? null : repo.getSkillDir(skillName);
}
/** Resolve a skill resource's relative path to an absolute path, or {@code null} if missing. */
@Nullable
public Path resolveResourcePath(String skillName, String resourcePath) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- Path resolved =
- ((FileSystemSkillRepository) repo)
- .getBaseDir()
- .resolve(skillName)
- .resolve(resourcePath);
- if (Files.isRegularFile(resolved)) {
- return resolved;
+ if (repo == null) {
+ return null;
+ }
+ Path dir = repo.getSkillDir(skillName);
+ if (dir == null) {
+ return null;
+ }
+ Path resolved = dir.resolve(resourcePath);
+ return Files.isRegularFile(resolved) ? resolved : null;
+ }
+
+ private void loadAll() {
+ for (SkillSourceSpec spec : config.getSources()) {
+ try {
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from " + spec.getScheme() + ":" + spec.getParams(),
+ e);
+ }
+ }
+ }
+
+ /** Build a {@link SkillOrigin} from a spec for diagnostics (WARN on duplicates, etc.). */
Review Comment:
**`originOf` re-introduces the parallel scheme ladder the registry was meant to eliminate.**
```java
switch (spec.getScheme()) {
case "local": ...; case "url": ...;
case "classpath": ...; case "package": ...;
}
```
Python's `_origin_of` mirrors this. Adding scheme `s3` now requires 3 file edits (registry ×2 + originOf ×2), not 1. Java's `case "package"` is dead code — no Java `package` handler exists.
Fix: add `default String describeLocation(params)` to `SkillSourceHandler`. Each handler describes its own location at registration.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/AgentSkill.java:
##########
@@ -44,6 +44,7 @@ public final class AgentSkill {
@Nullable private volatile Map<String, String> resources;
@Nullable private Supplier<Map<String, String>> resourceLoader;
private volatile boolean activated;
+ @Nullable private volatile SkillOrigin origin;
Review Comment:
**Java/Python serialization surface for `origin` is asymmetric.**
Python's `AgentSkill` extends `BaseModel`, so `origin` is in default `model_dump()` output. Java's `AgentSkill` is a POJO with no Jackson annotation — any future debug/checkpoint/Pemja serialization path silently drops it on Java while Python emits it. Same class of cross-language drop A3 fixed at the `Skills` envelope.
Add `@JsonProperty("origin")` + `@JsonInclude(NON_NULL)`. Near-zero cost now; cost grows once `AgentSkill` travels on the wire.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -18,33 +18,63 @@
package org.apache.flink.agents.runtime.skill;
+import org.apache.flink.agents.api.skills.SkillSourceSpec;
import org.apache.flink.agents.api.skills.Skills;
-import org.apache.flink.agents.runtime.skill.repository.FileSystemSkillRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Loads and indexes all skills referenced by a {@link Skills} configuration.
*
* <p>Mirrors the Python {@code flink_agents.runtime.skill.skill_manager.SkillManager}.
+ *
+ * <p>Owned by {@code ResourceContextImpl}; closed via {@code ResourceCache.close()} on operator
+ * close, which cascades to each repository's temp directory. Avoid constructing a {@code
+ * SkillManager} outside that flow without {@code try-with-resources}.
*/
-public class SkillManager {
+public class SkillManager implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SkillManager.class);
private final Skills config;
+ private final ClassLoader classLoader;
private final Map<String, AgentSkill> skills = new LinkedHashMap<>();
private final Map<String, SkillRepository> repos = new HashMap<>();
- public SkillManager(Skills config) {
+ /**
+ * Construct a {@code SkillManager} that resolves {@code classpath:} sources against {@code
+ * classLoader}. Production code passes the Flink user-code class loader (threaded through
+ * {@code ResourceCache} from {@code ActionExecutionOperator}); tests / standalone use may use
+ * {@link #SkillManager(Skills)}.
+ */
+ public SkillManager(Skills config, ClassLoader classLoader) {
Review Comment:
**Constructor-failure leak: partial `loadAll()` leaves registered repos with no closeable handle.**
If the 2nd of 3 sources succeeds (repo + hook registered) but the 3rd throws, the constructor exits with `IllegalStateException` → caller never gets a `SkillManager` reference → cannot call `close()` → 2nd repo's hook + temp dir leak until JVM exit.
Same monotonic-leak class as review #10, just triggered by partial-construction instead of hot-reload. 5-line fix: wrap `loadAll`'s loop in try/catch, call `close()` before rethrowing.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -111,38 +148,94 @@ public List<String> getSkillDirs(List<String> names) {
@Nullable
public Path getSkillDir(String skillName) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- return ((FileSystemSkillRepository) repo).getBaseDir().resolve(skillName);
- }
- return null;
+ return repo == null ? null : repo.getSkillDir(skillName);
}
/** Resolve a skill resource's relative path to an absolute path, or {@code null} if missing. */
@Nullable
public Path resolveResourcePath(String skillName, String resourcePath) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- Path resolved =
- ((FileSystemSkillRepository) repo)
- .getBaseDir()
- .resolve(skillName)
- .resolve(resourcePath);
- if (Files.isRegularFile(resolved)) {
- return resolved;
+ if (repo == null) {
+ return null;
+ }
+ Path dir = repo.getSkillDir(skillName);
+ if (dir == null) {
+ return null;
+ }
+ Path resolved = dir.resolve(resourcePath);
+ return Files.isRegularFile(resolved) ? resolved : null;
+ }
+
+ private void loadAll() {
+ for (SkillSourceSpec spec : config.getSources()) {
+ try {
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from " + spec.getScheme() + ":" + spec.getParams(),
+ e);
+ }
+ }
+ }
+
+ /** Build a {@link SkillOrigin} from a spec for diagnostics (WARN on duplicates, etc.). */
+ private static SkillOrigin originOf(SkillSourceSpec spec) {
+ Map<String, String> p = spec.getParams();
+ String location;
+ switch (spec.getScheme()) {
+ case "local":
+ location = p.getOrDefault("path", "");
+ break;
+ case "url":
+ location = p.getOrDefault("url", "");
+ break;
+ case "classpath":
+ location = p.getOrDefault("resource", "");
+ break;
+ case "package":
+ location = p.getOrDefault("package", "") + "/" + p.getOrDefault("resource", "");
+ break;
+ default:
+ location = p.toString();
+ }
+ return new SkillOrigin(spec.getScheme(), location);
+ }
+
+ /**
+ * Close every owned {@link SkillRepository}, releasing any temp directories materialized for
+ * URL / classpath-zip / classpath-jar sources. Idempotent.
+ */
+ @Override
+ public void close() {
+ // repos may map multiple skill names to the same repo instance; dedup by identity.
+ Set<SkillRepository> unique = Collections.newSetFromMap(new IdentityHashMap<>());
+ unique.addAll(repos.values());
Review Comment:
**`SkillManager.close()` swallows exceptions silently; inconsistent with `ResourceCache.close()`.**
```java
for (SkillRepository repo : unique) {
try { repo.close(); } catch (Exception ignored) {}
}
```
`ResourceCache.close()` (line 135-160) collects first-exception + suppressed and rethrows. Same close cascade, two styles. Real failures (locked file, permission denied, disk full) give callers no signal. `ResourceContextImpl.ensureSkillManager()` line 116 has the same silent catch on config swap.
Minimum: `LOG.warn(..., e)`. Preferable: align with `ResourceCache.close()`'s rethrow pattern.
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -532,11 +533,11 @@ private void addSkills(Map<String, Skills> skillsObjects) throws Exception {
ResourceType.TOOL,
new ResourceDescriptor(BashTool.class.getName(), new HashMap<>())));
- LinkedHashSet<String> paths = new LinkedHashSet<>();
+ LinkedHashSet<SkillSourceSpec> sources = new LinkedHashSet<>();
for (Skills s : skillsObjects.values()) {
Review Comment:
**Multi-`@Skills` merge order is non-deterministic on Java; WARN reports collisions but the "winner" is unspecified.**
`AgentPlan.addSkills` iterates `skillsObjects.values()` in `LinkedHashMap` order, but insertion order comes from `Class.getMethods()` — JLS doesn't guarantee declaration order. Test self-discloses at `AgentPlanDeclareSkillsTest:163`: "Class.getMethods() does not preserve declaration order".
Two `@Skills` methods with the same skill name pick different winners across deployments / JDK versions; WARN fires but is non-reproducible. Python `dict` preserves insertion order — Python is deterministic, Java is the bad case.
Min fix: sort method names lexicographically before merging + javadoc note. Stronger: `@Skills(priority=N)` or fail-fast on collision.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][runtime] Extend agent skill loading repository. [flink-agents]
Posted by "wzhero1 (via GitHub)" <gi...@apache.org>.
wzhero1 commented on code in PR #655:
URL: https://github.com/apache/flink-agents/pull/655#discussion_r3263199104
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/repository/URLSkillRepository.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.runtime.skill.repository;
+
+import org.apache.flink.agents.runtime.skill.AgentSkill;
+import org.apache.flink.agents.runtime.skill.SkillRepository;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Skill repository backed by an http(s) URL pointing to a zip.
+ *
+ * <p>The zip is downloaded to a temp file and extracted into a process-local temp directory. The
+ * downloaded zip itself is removed once extraction completes; the extracted directory is released
+ * via {@link #close()} (cascaded through {@code SkillManager} → {@code ResourceContextImpl} →
+ * {@code ResourceCache} on operator close). A JVM shutdown hook acts as fallback cleanup if {@code
+ * close()} is never called.
+ *
+ * <p>Composes a {@link SkillDirectoryReader} for the on-disk parsing; the materialization happens
+ * upfront in the constructor (no abuse of {@code super(...)}).
+ */
+public final class URLSkillRepository implements SkillRepository {
+
+ private static final int REQUEST_TIMEOUT_MS = 90_000;
+
+ private final String url;
Review Comment:
**Field name `owned` is an adjective without a subject.**
`Materialized` is already a noun. Rename the field to `materialization` (matches the class name), or rename the inner class to `MaterializedDir` so the field reads `materializedDir`. Current call sites `this.owned.close()` / `if (owned == null)` don't read well.
Same applies to `ClasspathSkillRepository.java:73` and `FileSystemSkillRepository.java:51`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillRepository.java:
##########
@@ -20,14 +20,19 @@
import javax.annotation.Nullable;
+import java.nio.file.Path;
import java.util.List;
import java.util.Map;
/**
* Source of skills. Mirrors the Python {@code
* flink_agents.runtime.skill.skill_repository.SkillRepository}.
+ *
+ * <p>Implementations that own a materialized temp directory (URL / classpath / zip) should override
Review Comment:
**Three sibling repos duplicate the same 5 delegating methods + state** (`reader`, `materialization`, plus `getSkill`/`getSkills`/`getResources`/`getSkillDir`/`close`) — ~45 lines of boilerplate, +15 per future source.
Extract `AbstractMaterializedSkillRepository` holding the shared state and implementing the delegators once; concrete repos reduce to source-specific I/O + an abstract `getOrigin()`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -111,38 +148,94 @@ public List<String> getSkillDirs(List<String> names) {
@Nullable
public Path getSkillDir(String skillName) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- return ((FileSystemSkillRepository) repo).getBaseDir().resolve(skillName);
- }
- return null;
+ return repo == null ? null : repo.getSkillDir(skillName);
}
/** Resolve a skill resource's relative path to an absolute path, or {@code null} if missing. */
@Nullable
public Path resolveResourcePath(String skillName, String resourcePath) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- Path resolved =
- ((FileSystemSkillRepository) repo)
- .getBaseDir()
- .resolve(skillName)
- .resolve(resourcePath);
- if (Files.isRegularFile(resolved)) {
- return resolved;
+ if (repo == null) {
+ return null;
+ }
+ Path dir = repo.getSkillDir(skillName);
+ if (dir == null) {
+ return null;
+ }
+ Path resolved = dir.resolve(resourcePath);
+ return Files.isRegularFile(resolved) ? resolved : null;
+ }
+
+ private void loadAll() {
+ for (SkillSourceSpec spec : config.getSources()) {
+ try {
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from " + spec.getScheme() + ":" + spec.getParams(),
+ e);
+ }
+ }
+ }
+
+ /** Build a {@link SkillOrigin} from a spec for diagnostics (WARN on duplicates, etc.). */
+ private static SkillOrigin originOf(SkillSourceSpec spec) {
+ Map<String, String> p = spec.getParams();
+ String location;
+ switch (spec.getScheme()) {
+ case "local":
+ location = p.getOrDefault("path", "");
+ break;
+ case "url":
+ location = p.getOrDefault("url", "");
+ break;
+ case "classpath":
+ location = p.getOrDefault("resource", "");
+ break;
+ case "package":
+ location = p.getOrDefault("package", "") + "/" + p.getOrDefault("resource", "");
+ break;
+ default:
+ location = p.toString();
+ }
+ return new SkillOrigin(spec.getScheme(), location);
+ }
+
+ /**
+ * Close every owned {@link SkillRepository}, releasing any temp directories materialized for
+ * URL / classpath-zip / classpath-jar sources. Idempotent.
+ */
+ @Override
+ public void close() {
+ // repos may map multiple skill names to the same repo instance; dedup by identity.
+ Set<SkillRepository> unique = Collections.newSetFromMap(new IdentityHashMap<>());
+ unique.addAll(repos.values());
Review Comment:
**`SkillManager.close()` swallows exceptions silently; inconsistent with `ResourceCache.close()`.**
```java
for (SkillRepository repo : unique) {
try { repo.close(); } catch (Exception ignored) {}
}
```
`ResourceCache.close()` (line 135-160) collects first-exception + suppressed and rethrows. Same close cascade, two styles. Real failures (locked file, permission denied, disk full) give callers no signal. `ResourceContextImpl.ensureSkillManager()` line 116 has the same silent catch on config swap.
Minimum: `LOG.warn(..., e)`. Preferable: align with `ResourceCache.close()`'s rethrow pattern.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/AgentSkill.java:
##########
@@ -44,6 +44,7 @@ public final class AgentSkill {
@Nullable private volatile Map<String, String> resources;
@Nullable private Supplier<Map<String, String>> resourceLoader;
private volatile boolean activated;
+ @Nullable private volatile SkillOrigin origin;
Review Comment:
**Java/Python serialization surface for `origin` is asymmetric.**
Python's `AgentSkill` extends `BaseModel`, so `origin` is in default `model_dump()` output. Java's `AgentSkill` is a POJO with no Jackson annotation — any future debug/checkpoint/Pemja serialization path silently drops it on Java while Python emits it. Same class of cross-language drop A3 fixed at the `Skills` envelope.
Add `@JsonProperty("origin")` + `@JsonInclude(NON_NULL)`. Near-zero cost now; cost grows once `AgentSkill` travels on the wire.
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -532,11 +533,11 @@ private void addSkills(Map<String, Skills> skillsObjects) throws Exception {
ResourceType.TOOL,
new ResourceDescriptor(BashTool.class.getName(), new HashMap<>())));
- LinkedHashSet<String> paths = new LinkedHashSet<>();
+ LinkedHashSet<SkillSourceSpec> sources = new LinkedHashSet<>();
for (Skills s : skillsObjects.values()) {
Review Comment:
**Multi-`@Skills` merge order is non-deterministic on Java; WARN reports collisions but the "winner" is unspecified.**
`AgentPlan.addSkills` iterates `skillsObjects.values()` in `LinkedHashMap` order, but insertion order comes from `Class.getMethods()` — JLS doesn't guarantee declaration order. Test self-discloses at `AgentPlanDeclareSkillsTest:163`: "Class.getMethods() does not preserve declaration order".
Two `@Skills` methods with the same skill name pick different winners across deployments / JDK versions; WARN fires but is non-reproducible. Python `dict` preserves insertion order — Python is deterministic, Java is the bad case.
Min fix: sort method names lexicographically before merging + javadoc note. Stronger: `@Skills(priority=N)` or fail-fast on collision.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -18,33 +18,63 @@
package org.apache.flink.agents.runtime.skill;
+import org.apache.flink.agents.api.skills.SkillSourceSpec;
import org.apache.flink.agents.api.skills.Skills;
-import org.apache.flink.agents.runtime.skill.repository.FileSystemSkillRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Loads and indexes all skills referenced by a {@link Skills} configuration.
*
* <p>Mirrors the Python {@code flink_agents.runtime.skill.skill_manager.SkillManager}.
+ *
+ * <p>Owned by {@code ResourceContextImpl}; closed via {@code ResourceCache.close()} on operator
+ * close, which cascades to each repository's temp directory. Avoid constructing a {@code
+ * SkillManager} outside that flow without {@code try-with-resources}.
*/
-public class SkillManager {
+public class SkillManager implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SkillManager.class);
private final Skills config;
+ private final ClassLoader classLoader;
private final Map<String, AgentSkill> skills = new LinkedHashMap<>();
private final Map<String, SkillRepository> repos = new HashMap<>();
- public SkillManager(Skills config) {
+ /**
+ * Construct a {@code SkillManager} that resolves {@code classpath:} sources against {@code
+ * classLoader}. Production code passes the Flink user-code class loader (threaded through
+ * {@code ResourceCache} from {@code ActionExecutionOperator}); tests / standalone use may use
+ * {@link #SkillManager(Skills)}.
+ */
+ public SkillManager(Skills config, ClassLoader classLoader) {
Review Comment:
**Constructor-failure leak: partial `loadAll()` leaves registered repos with no closeable handle.**
If the 2nd of 3 sources succeeds (repo + hook registered) but the 3rd throws, the constructor exits with `IllegalStateException` → caller never gets a `SkillManager` reference → cannot call `close()` → 2nd repo's hook + temp dir leak until JVM exit.
Same monotonic-leak class as review #10, just triggered by partial-construction instead of hot-reload. 5-line fix: wrap `loadAll`'s loop in try/catch, call `close()` before rethrowing.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java:
##########
@@ -111,38 +148,94 @@ public List<String> getSkillDirs(List<String> names) {
@Nullable
public Path getSkillDir(String skillName) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- return ((FileSystemSkillRepository) repo).getBaseDir().resolve(skillName);
- }
- return null;
+ return repo == null ? null : repo.getSkillDir(skillName);
}
/** Resolve a skill resource's relative path to an absolute path, or {@code null} if missing. */
@Nullable
public Path resolveResourcePath(String skillName, String resourcePath) {
SkillRepository repo = repos.get(skillName);
- if (repo instanceof FileSystemSkillRepository) {
- Path resolved =
- ((FileSystemSkillRepository) repo)
- .getBaseDir()
- .resolve(skillName)
- .resolve(resourcePath);
- if (Files.isRegularFile(resolved)) {
- return resolved;
+ if (repo == null) {
+ return null;
+ }
+ Path dir = repo.getSkillDir(skillName);
+ if (dir == null) {
+ return null;
+ }
+ Path resolved = dir.resolve(resourcePath);
+ return Files.isRegularFile(resolved) ? resolved : null;
+ }
+
+ private void loadAll() {
+ for (SkillSourceSpec spec : config.getSources()) {
+ try {
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from " + spec.getScheme() + ":" + spec.getParams(),
+ e);
+ }
+ }
+ }
+
+ /** Build a {@link SkillOrigin} from a spec for diagnostics (WARN on duplicates, etc.). */
Review Comment:
**`originOf` re-introduces the parallel scheme ladder the registry was meant to eliminate.**
```java
switch (spec.getScheme()) {
case "local": ...; case "url": ...;
case "classpath": ...; case "package": ...;
}
```
Python's `_origin_of` mirrors this. Adding scheme `s3` now requires 3 file edits (registry ×2 + originOf ×2), not 1. Java's `case "package"` is dead code — no Java `package` handler exists.
Fix: add `default String describeLocation(params)` to `SkillSourceHandler`. Each handler describes its own location at registration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org