This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 546d1a36 [hotfix][java][python] Release skill repositories on any load
failure and remove dead code (#951)
546d1a36 is described below
commit 546d1a36b7ceb546f22ffb42a793547b279a1306
Author: Weiqing Yang <[email protected]>
AuthorDate: Tue Aug 4 02:29:57 2026 -0700
[hotfix][java][python] Release skill repositories on any load failure and
remove dead code (#951)
---
python/flink_agents/runtime/skill/skill_manager.py | 47 ++++-----
.../flink_agents/runtime/skill/skill_repository.py | 20 ----
.../runtime/skill/tests/test_manager.py | 104 ++++++++++++++++++++
.../flink/agents/runtime/skill/SkillManager.java | 67 ++++++-------
.../agents/runtime/skill/SkillManagerTest.java | 107 ++++++++++++++++++---
5 files changed, 241 insertions(+), 104 deletions(-)
diff --git a/python/flink_agents/runtime/skill/skill_manager.py
b/python/flink_agents/runtime/skill/skill_manager.py
index c5ccf592..99f6e87a 100644
--- a/python/flink_agents/runtime/skill/skill_manager.py
+++ b/python/flink_agents/runtime/skill/skill_manager.py
@@ -137,37 +137,24 @@ class SkillManager:
repo = self._repos.get(skill_name)
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 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)
- self._opened_repos.append(repo)
- except (OSError, ValueError) as e:
- # Release repos opened by earlier iterations — the caller never
- # receives a SkillManager reference to clean them up via
close()
- # itself, so without this their temp dirs / atexit handlers
leak
- # until interpreter exit.
- self.close()
- msg = (
- f"Failed to load skills from {spec.scheme}:{spec.params}"
- )
- raise RuntimeError(msg) from e
- self._register_repo(repo, _origin_of(spec))
+ try:
+ for spec in self._config.sources:
+ try:
+ handler = skill_source_registry.get(spec.scheme)
+ repo = handler.open(spec.params)
+ self._opened_repos.append(repo)
+ except (OSError, ValueError) as e:
+ msg = f"Failed to load skills from
{spec.scheme}:{spec.params}"
+ raise RuntimeError(msg) from e
+ self._register_repo(repo, _origin_of(spec))
+ except BaseException:
+ # Release every repo opened so far — the caller never receives a
+ # SkillManager reference to clean them up via close() itself, so
+ # without this their temp dirs / atexit handlers leak until
+ # interpreter exit.
+ self.close()
+ raise
def _register_repo(self, repo: "SkillRepository", origin: SkillOrigin) ->
None:
for skill in repo.get_skills():
diff --git a/python/flink_agents/runtime/skill/skill_repository.py
b/python/flink_agents/runtime/skill/skill_repository.py
index 5cca1f8b..1b76afe6 100644
--- a/python/flink_agents/runtime/skill/skill_repository.py
+++ b/python/flink_agents/runtime/skill/skill_repository.py
@@ -16,32 +16,12 @@
# limitations under the License.
#################################################################################
from abc import ABC, abstractmethod
-from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List
from flink_agents.runtime.skill.agent_skill import AgentSkill
-@dataclass
-class SkillRepositoryInfo:
- """Information about a skill repository.
-
- Attributes:
- ----------
- repo_type : str
- The type of repository (e.g., "filesystem", "classpath", "url").
- location : str
- The location of the repository (e.g., path, URL).
- writeable : bool
- Whether the repository supports write operations.
- """
-
- repo_type: str
- location: str
- writeable: bool
-
-
class SkillRepository(ABC):
"""Source of skills, loaded from filesystem / classpath / URL / package.
diff --git a/python/flink_agents/runtime/skill/tests/test_manager.py
b/python/flink_agents/runtime/skill/tests/test_manager.py
index 4944f846..5d4c062c 100644
--- a/python/flink_agents/runtime/skill/tests/test_manager.py
+++ b/python/flink_agents/runtime/skill/tests/test_manager.py
@@ -357,3 +357,107 @@ class TestSkillManagerMixedSources:
assert closed == ["skill-1"], (
"the repo opened before the partial-load failure must be closed"
)
+
+ def test_load_failure_outside_wrapped_types_closes_repos_and_propagates(
+ self,
+ ) -> None:
+ # Contract: a source failure that is neither OSError nor ValueError
still
+ # closes the repos opened before it, and reaches the caller unwrapped.
+ from typing import Dict, List
+
+ from flink_agents.runtime.skill import skill_source_registry
+ from flink_agents.runtime.skill.agent_skill import AgentSkill
+ from flink_agents.runtime.skill.skill_repository import SkillRepository
+
+ closed: List[str] = []
+
+ class FakeRepo(SkillRepository):
+ def get_skill(self, name: str) -> AgentSkill | None:
+ return self.get_skills()[0] if name == "skill-1" else None
+
+ def get_skills(self) -> List[AgentSkill]:
+ return [AgentSkill(name="skill-1", description="dummy",
content="body")]
+
+ def get_resources(self, name: str) -> Dict[str, str]:
+ return {}
+
+ def close(self) -> None:
+ closed.append("skill-1")
+
+ counter = {"n": 0}
+
+ def opener(params) -> SkillRepository:
+ counter["n"] += 1
+ if counter["n"] == 2:
+ msg = "corrupt archive"
+ raise zipfile.BadZipFile(msg)
+ return FakeRepo()
+
+ skill_source_registry.register("test-badzip-close", opener)
+
+ config = Skills(
+ sources=[
+ SkillSourceSpec(scheme="test-badzip-close", params={}),
+ SkillSourceSpec(scheme="test-badzip-close", params={}),
+ ]
+ )
+
+ with pytest.raises(zipfile.BadZipFile):
+ SkillManager(config)
+ assert closed == ["skill-1"], (
+ "a load failure outside (OSError, ValueError) must still close the
"
+ "repos opened before it"
+ )
+
+ def test_registration_failure_closes_earlier_repo_and_propagates(self) ->
None:
+ # Contract: a failure raised while registering a repo — after its
open()
+ # already succeeded — still closes the repo from an earlier source.
+ from typing import Dict, List
+
+ from flink_agents.runtime.skill import skill_source_registry
+ from flink_agents.runtime.skill.agent_skill import AgentSkill
+ from flink_agents.runtime.skill.skill_repository import SkillRepository
+
+ closed: List[str] = []
+
+ class FakeRepo(SkillRepository):
+ def __init__(self, tag: str, *, boom: bool) -> None:
+ self._tag = tag
+ self._boom = boom
+
+ def get_skill(self, name: str) -> AgentSkill | None:
+ return None
+
+ def get_skills(self) -> List[AgentSkill]:
+ if self._boom:
+ msg = "exploding during registration"
+ raise KeyError(msg)
+ return [AgentSkill(name=self._tag, description="d",
content="b")]
+
+ def get_resources(self, name: str) -> Dict[str, str]:
+ return {}
+
+ def close(self) -> None:
+ closed.append(self._tag)
+
+ counter = {"n": 0}
+
+ def opener(params) -> SkillRepository:
+ counter["n"] += 1
+ return FakeRepo(f"skill-{counter['n']}", boom=counter["n"] == 2)
+
+ skill_source_registry.register("test-register-boom", opener)
+
+ config = Skills(
+ sources=[
+ SkillSourceSpec(scheme="test-register-boom", params={}),
+ SkillSourceSpec(scheme="test-register-boom", params={}),
+ ]
+ )
+
+ with pytest.raises(KeyError):
+ SkillManager(config)
+ assert closed == ["skill-1", "skill-2"], (
+ "a registration failure must close every repo opened so far — the "
+ "earlier source's repo and the one whose registration failed"
+ )
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java
index 4c0eae6c..9111edf6 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java
@@ -26,7 +26,6 @@ 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;
@@ -159,48 +158,38 @@ public class SkillManager implements AutoCloseable {
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 == 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);
- openedRepos.add(repo);
- 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 {
+ for (SkillSourceSpec spec : config.getSources()) {
try {
- closeRepos();
- } catch (Exception cleanupError) {
- toThrow.addSuppressed(cleanupError);
+ SkillRepository repo =
+ SkillSourceRegistry.get(spec.getScheme())
+ .open(spec.getParams(), classLoader);
+ openedRepos.add(repo);
+ registerRepo(repo, originOf(spec));
+ } catch (IOException | IllegalArgumentException e) {
+ throw new IllegalStateException(
+ "Failed to load skills from "
+ + spec.getScheme()
+ + ":"
+ + spec.getParams(),
+ e);
}
- throw toThrow;
}
+ } catch (Throwable t) {
+ // Release every repo opened so far, on any failure path. 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. The original failure propagates unchanged; a
cleanup
+ // failure rides along as suppressed so neither is lost —
including an Error,
+ // which closeRepos() does not catch per-repo and which would
otherwise
+ // replace the original failure outright.
+ try {
+ closeRepos();
+ } catch (Throwable cleanupError) {
+ t.addSuppressed(cleanupError);
+ }
+ throw t;
}
}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/skill/SkillManagerTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/skill/SkillManagerTest.java
index dfc05f35..daf8660f 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/skill/SkillManagerTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/skill/SkillManagerTest.java
@@ -106,14 +106,6 @@ class SkillManagerTest {
assertTrue(ex.getMessage().contains("github"));
}
- @Test
- void resolveResourcePathLocatesBundledFile() {
- SkillManager manager = new SkillManager(configFromResources());
- Path resolved = manager.resolveResourcePath("nano-banana-pro",
"scripts/generate_image.py");
- assertNotNull(resolved);
- assertTrue(Files.isRegularFile(resolved));
- }
-
private static void zipDir(Path src, Path dstZip) throws IOException {
try (ZipOutputStream zos = new
ZipOutputStream(Files.newOutputStream(dstZip));
Stream<Path> walk = Files.walk(src)) {
@@ -299,23 +291,35 @@ class SkillManagerTest {
/**
* Minimal {@link SkillRepository} for lifecycle tests: each instance owns
one fake skill and
- * records whether {@code close()} was invoked. May be configured to throw
a {@link
- * RuntimeException} on close ({@link SkillRepository#close()} declares no
checked exceptions;
- * {@link SkillManager#close()} catches {@code Exception} so this still
exercises the cascade
- * logic).
+ * records whether {@code close()} was invoked. May be configured to fail
on close, exercising
+ * the cascade logic in {@link SkillManager#close()}, and/or to fail from
{@link #getSkills()},
+ * which fails the repo's registration after its {@code open()} has
already succeeded.
+ *
+ * <p>{@link SkillRepository} declares no checked exceptions, so a
configured failure is either
+ * a {@link RuntimeException} or an {@link Error}. Both kinds are needed:
they take different
+ * paths through the handlers in {@link SkillManager}.
*/
private static final class FakeRepo implements SkillRepository {
private final AgentSkill skill;
final AtomicBoolean closed = new AtomicBoolean();
- @javax.annotation.Nullable private final RuntimeException
closeException;
+ @javax.annotation.Nullable private final Throwable closeException;
+ @javax.annotation.Nullable private final Throwable getSkillsException;
FakeRepo(String skillName) {
this(skillName, null);
}
- FakeRepo(String skillName, @javax.annotation.Nullable RuntimeException
closeException) {
+ FakeRepo(String skillName, @javax.annotation.Nullable Throwable
closeException) {
+ this(skillName, closeException, null);
+ }
+
+ FakeRepo(
+ String skillName,
+ @javax.annotation.Nullable Throwable closeException,
+ @javax.annotation.Nullable Throwable getSkillsException) {
this.skill = new AgentSkill(skillName, "fake", "body", null, null,
null);
this.closeException = closeException;
+ this.getSkillsException = getSkillsException;
}
@Override
@@ -325,6 +329,9 @@ class SkillManagerTest {
@Override
public List<AgentSkill> getSkills() {
+ if (getSkillsException != null) {
+ throwUnchecked(getSkillsException);
+ }
return List.of(skill);
}
@@ -337,9 +344,25 @@ class SkillManagerTest {
public void close() {
closed.set(true);
if (closeException != null) {
- throw closeException;
+ throwUnchecked(closeException);
}
}
+
+ /**
+ * Throw a configured failure. Declaring the fields as {@link
Throwable} lets one field
+ * carry either kind of unchecked failure; the interface permits
nothing else, so a checked
+ * exception is a test-setup mistake and fails loudly rather than
being smuggled past the
+ * compiler.
+ */
+ private static void throwUnchecked(Throwable failure) {
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ throw new AssertionError("FakeRepo failures must be unchecked",
failure);
+ }
}
@Test
@@ -394,6 +417,60 @@ class SkillManagerTest {
assertSame(cleanupBoom, ex.getSuppressed()[0]);
}
+ @Test
+ void registrationFailureOfUnwrappedTypeClosesReposAndPropagates() {
+ // A failure whose type is neither IOException nor
IllegalArgumentException reaches the
+ // caller unchanged rather than being wrapped, so only a guard
spanning every failure path
+ // can release the repos. Two repos are owned by the time registration
fails: the earlier
+ // source's, and the one whose registration failed (it is recorded
before registration
+ // runs). Both must be released, and a failure during that release
must ride along as
+ // suppressed instead of replacing the original.
+ IllegalStateException registrationBoom = new
IllegalStateException("registration-boom");
+ RuntimeException cleanupBoom = new RuntimeException("cleanup-boom");
+ FakeRepo first = new FakeRepo("alpha", cleanupBoom);
+ FakeRepo failing = new FakeRepo("beta", null, registrationBoom);
+ SkillSourceRegistry.register("test-register-boom-ok", (params, cl) ->
first);
+ SkillSourceRegistry.register("test-register-boom-fail", (params, cl)
-> failing);
+
+ Skills config =
+ new Skills(
+ List.of(
+ new SkillSourceSpec("test-register-boom-ok",
Map.of()),
+ new SkillSourceSpec("test-register-boom-fail",
Map.of())));
+
+ IllegalStateException ex =
+ assertThrows(IllegalStateException.class, () -> new
SkillManager(config));
+ // Identity, not just type: IllegalStateException is also what the
wrapping catch builds.
+ assertSame(registrationBoom, ex);
+ assertTrue(first.closed.get(), "repo opened before the failure must be
closed");
+ assertTrue(failing.closed.get(), "repo whose registration failed must
be closed");
+ assertEquals(1, ex.getSuppressed().length);
+ assertSame(cleanupBoom, ex.getSuppressed()[0]);
+ }
+
+ @Test
+ void errorDuringRegistrationStillClosesRepoAndSuppressesCloseError() {
+ // An Error is not an Exception, so it survives a load failure only if
both the guard around
+ // the source loop and the guard around that guard's cleanup accept
Throwable. This repo
+ // fails its registration with one Error and then its close() with
another: a load guard
+ // narrowed to Exception would skip the cleanup entirely, and a
cleanup guard narrowed to
+ // Exception would let the close() Error escape and replace the
registration failure.
+ // One source keeps the assertions deterministic — closeRepos()
catches only Exception per
+ // repo, so an Error from any repo's close() ends the iteration over
the remaining ones.
+ Error registrationBoom = new Error("registration-error");
+ Error closeBoom = new Error("close-error");
+ FakeRepo repo = new FakeRepo("alpha", closeBoom, registrationBoom);
+ SkillSourceRegistry.register("test-error-fail", (params, cl) -> repo);
+
+ Skills config = new Skills(List.of(new
SkillSourceSpec("test-error-fail", Map.of())));
+
+ Error ex = assertThrows(Error.class, () -> new SkillManager(config));
+ assertSame(registrationBoom, ex);
+ assertTrue(repo.closed.get(), "the repo owned when registration failed
must be closed");
+ assertEquals(1, ex.getSuppressed().length);
+ assertSame(closeBoom, ex.getSuppressed()[0]);
+ }
+
@Test
void closeAttemptsEveryRepoAndRethrowsFirstFailure() throws Exception {
// Three repos: middle one throws on close. SkillManager.close() must
(a) attempt all
