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 806a7ff23 [Tech Debt][runtime][python] Preserve skill-repository close
failures in SkillManager.close() (#1221)
806a7ff23 is described below
commit 806a7ff23c263e86676c07513460a759c63b4cda
Author: william <[email protected]>
AuthorDate: Mon Oct 12 11:25:41 2026 +0800
[Tech Debt][runtime][python] Preserve skill-repository close failures in
SkillManager.close() (#1221)
---
python/flink_agents/runtime/close_utils.py | 88 +++++++++++
.../flink_agents/runtime/flink_runner_context.py | 17 +-
python/flink_agents/runtime/resource_cache.py | 43 +----
python/flink_agents/runtime/skill/skill_manager.py | 38 ++++-
.../runtime/skill/tests/test_manager_close.py | 176 +++++++++++++++++++++
5 files changed, 309 insertions(+), 53 deletions(-)
diff --git a/python/flink_agents/runtime/close_utils.py
b/python/flink_agents/runtime/close_utils.py
new file mode 100644
index 000000000..360124169
--- /dev/null
+++ b/python/flink_agents/runtime/close_utils.py
@@ -0,0 +1,88 @@
+################################################################################
+# 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.
+#################################################################################
+"""Helpers for "close everything, then surface the first failure" cleanup
paths.
+
+Shared by the runtime components whose ``close()`` must keep going after a
+failure: ``ResourceCache``, ``FlinkRunnerContext`` and ``SkillManager``. One
+implementation is what keeps those three contracts identical rather than merely
+similar.
+
+This module deliberately imports nothing from the project: ``skill_manager``
+cannot reach a helper parked in ``resource_cache``, because
+``skill_manager -> resource_cache -> resource_context -> skill_manager`` is an
+import cycle.
+"""
+
+import logging
+from collections.abc import Callable
+
+_LOG = logging.getLogger(__name__)
+
+
+def failure_of(close: Callable[[], None]) -> Exception | None:
+ """Run ``close``, returning any failure instead of raising it.
+
+ Keeps the caller's cleanup loop free of a ``try`` block so one bad
component
+ cannot end the iteration.
+
+ Parameters
+ ----------
+ close : Callable[[], None]
+ Cleanup callback to run.
+
+ Returns:
+ -------
+ Exception | None
+ The failure raised by ``close``, or ``None`` when it succeeded.
+ """
+ try:
+ close()
+ except Exception as e:
+ return e
+ return None
+
+
+def first_or_logged(
+ failure: Exception | None, previous: Exception | None, what: str
+) -> Exception | None:
+ """Keep the first failure and log any later one.
+
+ The Python analogue of Flink's ``ExceptionUtils.firstOrSuppressed``. Later
+ failures are logged rather than attached, because ``ExceptionGroup``
requires
+ 3.11 and this package supports 3.10.
+
+ Parameters
+ ----------
+ failure : Exception | None
+ Failure produced by the component just closed, if any.
+ previous : Exception | None
+ First failure seen so far, if any.
+ what : str
+ Short description of the component, used in the log message.
+
+ Returns:
+ -------
+ Exception | None
+ The failure to keep: the first one seen.
+ """
+ if failure is None:
+ return previous
+ if previous is None:
+ return failure
+ _LOG.warning("Suppressed failure closing %s.", what, exc_info=failure)
+ return previous
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index bdd37b776..43b6be163 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -46,6 +46,7 @@ from flink_agents.api.runner_context import (
RunnerContext,
)
from flink_agents.api.trace import ExecutionReporter
+from flink_agents.runtime.close_utils import failure_of, first_or_logged
from flink_agents.runtime.durable_exception import (
deserialize_durable_exception,
durable_exception_message,
@@ -66,11 +67,7 @@ from
flink_agents.runtime.memory.internal_base_long_term_memory import (
from flink_agents.runtime.memory.mem0.mem0_long_term_memory import (
Mem0LongTermMemory,
)
-from flink_agents.runtime.resource_cache import (
- ResourceCache,
- _failure_of,
- _first_or_logged,
-)
+from flink_agents.runtime.resource_cache import ResourceCache
from flink_agents.runtime.task_lifecycle_listener import TaskLifecycleListener
logger = logging.getLogger(__name__)
@@ -1720,12 +1717,12 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
ltm = self.__ltm
self.__ltm = None
- first_failure = _failure_of(ltm.close) if ltm is not None else None
+ first_failure = failure_of(ltm.close) if ltm is not None else None
resource_cache = self.__resource_cache
self.__resource_cache = None
- first_failure = _first_or_logged(
- _failure_of(resource_cache.close) if resource_cache is not None
else None,
+ first_failure = first_or_logged(
+ failure_of(resource_cache.close) if resource_cache is not None
else None,
first_failure,
"runner context resource cache",
)
@@ -1737,8 +1734,8 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
scoped_caches = self.__scoped_resource_caches
self.__scoped_resource_caches = {}
for scoped in scoped_caches.values():
- first_failure = _first_or_logged(
- _failure_of(scoped[1].close),
+ first_failure = first_or_logged(
+ failure_of(scoped[1].close),
first_failure,
"sub-agent scope resource cache",
)
diff --git a/python/flink_agents/runtime/resource_cache.py
b/python/flink_agents/runtime/resource_cache.py
index ace542c24..1e0db4ecd 100644
--- a/python/flink_agents/runtime/resource_cache.py
+++ b/python/flink_agents/runtime/resource_cache.py
@@ -15,8 +15,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
-import logging
-from collections.abc import Callable
from typing import Any, Dict
from flink_agents.api.resource import Resource, ResourceType
@@ -24,40 +22,9 @@ from flink_agents.plan.configuration import
AgentConfiguration
from flink_agents.plan.function import JavaFunction
from flink_agents.plan.resource_provider import JavaResourceProvider,
ResourceProvider
from flink_agents.plan.tools.function_tool import FunctionTool
+from flink_agents.runtime.close_utils import failure_of, first_or_logged
from flink_agents.runtime.resource_context import ResourceContextImpl
-_LOG = logging.getLogger(__name__)
-
-
-def _failure_of(close: Callable[[], None]) -> Exception | None:
- """Run ``close``, returning any failure instead of raising it.
-
- Keeps the caller's cleanup loop free of a ``try`` block so one bad
component
- cannot end the iteration.
- """
- try:
- close()
- except Exception as e:
- return e
- return None
-
-
-def _first_or_logged(
- failure: Exception | None, previous: Exception | None, what: str
-) -> Exception | None:
- """Keep the first failure and log any later one.
-
- The Python analogue of Flink's ``ExceptionUtils.firstOrSuppressed``. Later
- failures are logged rather than attached, because ``ExceptionGroup``
requires
- 3.11 and this package supports 3.10.
- """
- if failure is None:
- return previous
- if previous is None:
- return failure
- _LOG.warning("Suppressed failure closing %s.", what, exc_info=failure)
- return previous
-
class ResourceCache:
"""Lazily resolves and caches Resource instances from ResourceProviders.
@@ -166,12 +133,12 @@ class ResourceCache:
first_failure: Exception | None = None
for typed in self._cache.values():
for resource in typed.values():
- first_failure = _first_or_logged(
- _failure_of(resource.close), first_failure, "resource"
+ first_failure = first_or_logged(
+ failure_of(resource.close), first_failure, "resource"
)
self._cache.clear()
- first_failure = _first_or_logged(
- _failure_of(self._resource_context.close), first_failure,
"resource context"
+ first_failure = first_or_logged(
+ failure_of(self._resource_context.close), first_failure, "resource
context"
)
if first_failure is not None:
raise first_failure
diff --git a/python/flink_agents/runtime/skill/skill_manager.py
b/python/flink_agents/runtime/skill/skill_manager.py
index 540d37065..c02a26499 100644
--- a/python/flink_agents/runtime/skill/skill_manager.py
+++ b/python/flink_agents/runtime/skill/skill_manager.py
@@ -15,12 +15,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
-import contextlib
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Dict, List, Mapping
from flink_agents.api.skills import Skills, SkillSourceSpec, redact_skill_url
+from flink_agents.runtime.close_utils import failure_of, first_or_logged
from flink_agents.runtime.skill.agent_skill import AgentSkill, SkillOrigin
from flink_agents.runtime.skill.repository.classpath_repository import (
ClasspathSkillRepository,
@@ -203,7 +203,17 @@ class SkillManager:
# SkillManager reference to clean them up via close() itself, so
# without this their temp dirs / atexit handlers leak until
# interpreter exit.
- self.close()
+ try:
+ self.close()
+ except Exception as close_failure:
+ # close() surfaces repository failures now, so it can raise
here
+ # and would otherwise replace the load failure this cleanup is
+ # running for. Java attaches it with addSuppressed;
ExceptionGroup
+ # needs Python 3.11 and this package supports 3.10, so log it.
+ logger.warning(
+ "Suppressed failure closing skill repositories after a
failed load.",
+ exc_info=close_failure,
+ )
raise
def _register_repo(self, repo: "SkillRepository", origin: SkillOrigin) ->
None:
@@ -225,20 +235,38 @@ class SkillManager:
def close(self) -> None:
"""Close every opened :class:`SkillRepository`, releasing any temp
directory
- materialized for URL / classpath-zip / package sources. Idempotent.
+ materialized for URL / classpath-zip / package sources.
Iterates ``_opened_repos`` rather than ``_repos.values()``: duplicate
skill
names overwrite the earlier repo's reference in ``_repos``, but the
displaced
repo is still owned and must be closed. Dedup by identity in case the
same
repo contributes multiple skills.
+
+ Mirrors the Java ``SkillManager.closeRepos()`` contract from #987 and
the
+ Python ``ResourceCache.close()``: every repository is attempted even
when an
+ earlier one fails, and the first failure is re-raised after the loop,
so a
+ real shutdown bug (locked file, permission denied, disk full) reaches
the
+ caller instead of being swallowed. Later failures are logged rather
than
+ attached, because ``ExceptionGroup`` needs Python 3.11 and this package
+ supports 3.10.
+
+ Idempotent to the extent repository ``close()`` implementations are:
one
+ that fails does so again on a second call, exactly as in Java.
+
+ Only ``Exception`` is caught, matching ``ResourceCache.close()``:
cleanup
+ must not swallow ``KeyboardInterrupt`` or ``SystemExit``.
"""
seen: set[int] = set()
+ first_failure: Exception | None = None
for repo in self._opened_repos:
if id(repo) in seen:
continue
seen.add(id(repo))
- with contextlib.suppress(Exception):
- repo.close()
+ first_failure = first_or_logged(
+ failure_of(repo.close), first_failure, "skill repository"
+ )
+ if first_failure is not None:
+ raise first_failure
def __enter__(self) -> "SkillManager":
return self
diff --git a/python/flink_agents/runtime/skill/tests/test_manager_close.py
b/python/flink_agents/runtime/skill/tests/test_manager_close.py
new file mode 100644
index 000000000..20cebdf4f
--- /dev/null
+++ b/python/flink_agents/runtime/skill/tests/test_manager_close.py
@@ -0,0 +1,176 @@
+################################################################################
+# 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.
+#################################################################################
+"""Close-path contract tests for the Python SkillManager.
+
+These mirror the Java ``SkillManagerTest`` close tests and the Python
+``ResourceCache`` close tests: a failing repository must not strand the
+repositories behind it, and the first failure must reach the caller unchanged
+rather than being swallowed.
+"""
+
+import pytest
+
+from flink_agents.api.skills import Skills, SkillSourceSpec
+from flink_agents.runtime.skill.agent_skill import AgentSkill
+from flink_agents.runtime.skill.skill_manager import SkillManager
+from flink_agents.runtime.skill.skill_repository import SkillRepository
+from flink_agents.runtime.skill.skill_source_handler import SkillSourceHandler
+
+
+class FakeRepo(SkillRepository):
+ """Records whether ``close()`` ran, and optionally fails it."""
+
+ def __init__(self, skill_name: str, failure: Exception | None = None) ->
None:
+ """Create a repository exposing one skill, optionally failing on
close."""
+ self._skill_name = skill_name
+ self._failure = failure
+ self.closed = False
+
+ def get_skill(self, name: str) -> AgentSkill | None:
+ """Return the single skill when it is the one asked for."""
+ return self.get_skills()[0] if name == self._skill_name else None
+
+ def get_skills(self) -> list[AgentSkill]:
+ """Return the single skill held by this repository."""
+ return [AgentSkill(name=self._skill_name, description="dummy",
content="body")]
+
+ def get_resources(self, name: str) -> dict[str, str]:
+ """Return no resources for any skill."""
+ return {}
+
+ def close(self) -> None:
+ """Record the call, then fail if this repository was configured to."""
+ self.closed = True
+ if self._failure is not None:
+ raise self._failure
+
+
+def _manager_with(monkeypatch: pytest.MonkeyPatch, *repos: FakeRepo) ->
SkillManager:
+ """Build a SkillManager whose sources open exactly ``repos``, in order."""
+ opened = list(repos)
+
+ def opener(params) -> SkillRepository:
+ return opened.pop(0)
+
+ monkeypatch.setattr(
+ SkillManager,
+ "_create_handlers",
+ lambda self, bridge: {"test-close": SkillSourceHandler(opener)},
+ )
+ config = Skills(
+ sources=[SkillSourceSpec(scheme="test-close", params={}) for _ in
repos]
+ )
+ return SkillManager(config)
+
+
+def test_close_attempts_every_repo_and_rethrows_first_failure(monkeypatch) ->
None:
+ """A failing repository must not strand the ones behind it.
+
+ The failure reaches the caller unchanged in identity, not wrapped, which is
+ what makes it usable by the caller that owns the shutdown.
+ """
+ failure = RuntimeError("repo close failed")
+ failing = FakeRepo("first", failure=failure)
+ surviving = FakeRepo("second")
+ manager = _manager_with(monkeypatch, failing, surviving)
+
+ with pytest.raises(RuntimeError, match="repo close failed") as excinfo:
+ manager.close()
+
+ assert excinfo.value is failure
+ assert failing.closed
+ assert surviving.closed, "a failing repo must not strand the ones behind
it"
+
+
+def test_close_reports_first_failure_when_several_fail(monkeypatch) -> None:
+ """The first failure is the one raised; later ones do not replace it."""
+ first = RuntimeError("first")
+ second = RuntimeError("second")
+ manager = _manager_with(
+ monkeypatch, FakeRepo("a", failure=first), FakeRepo("b",
failure=second)
+ )
+
+ with pytest.raises(RuntimeError, match="first") as excinfo:
+ manager.close()
+
+ assert excinfo.value is first
+
+
+def test_close_logs_later_failures(
+ monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+ """Later failures are logged, since Python 3.10 cannot attach them."""
+ manager = _manager_with(
+ monkeypatch,
+ FakeRepo("a", failure=RuntimeError("first")),
+ FakeRepo("b", failure=RuntimeError("second")),
+ )
+
+ with pytest.raises(RuntimeError, match="first"):
+ manager.close()
+
+ assert "Suppressed failure closing skill repository." in caplog.text
+ assert "second" in caplog.text
+
+
+def test_close_returns_normally_when_nothing_fails(monkeypatch) -> None:
+ """The healthy path stays a plain no-raise close."""
+ surviving = FakeRepo("only")
+ manager = _manager_with(monkeypatch, surviving)
+
+ manager.close()
+
+ assert surviving.closed
+
+
+def test_load_failure_is_not_replaced_by_close_failure(
+ monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+ """A close failure during the failed-load cleanup is logged, not re-raised.
+
+ The load failure is what the caller needs; letting the cleanup failure
+ replace it would hide the reason the manager could not be built at all.
+ """
+ failing = FakeRepo("skill-1", failure=RuntimeError("close boom"))
+ counter = {"n": 0}
+
+ def opener(params) -> SkillRepository:
+ counter["n"] += 1
+ if counter["n"] == 2:
+ msg = "open boom"
+ raise OSError(msg)
+ return failing
+
+ monkeypatch.setattr(
+ SkillManager,
+ "_create_handlers",
+ lambda self, bridge: {"test-load": SkillSourceHandler(opener)},
+ )
+ config = Skills(
+ sources=[
+ SkillSourceSpec(scheme="test-load", params={}),
+ SkillSourceSpec(scheme="test-load", params={}),
+ ]
+ )
+
+ with pytest.raises(RuntimeError, match="Failed to load skills"):
+ SkillManager(config)
+
+ assert failing.closed
+ assert "Suppressed failure closing skill repositories" in caplog.text
+ assert "close boom" in caplog.text
