From 54b3da1c220794c4d40ab3c2ed01dd22bdc82e73 Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Tue, 8 Sep 2026 23:34:07 +0800 Subject: [PATCH 1/5] feat(skills): refresh invocation skill views with external source composition --- tests/skills/test_dynamic_skill_runtime.py | 195 +++++++++++++++++++++ veadk/agent.py | 67 +++++-- veadk/skills/DYNAMIC_SKILLS.md | 57 ++++++ veadk/skills/runtime.py | 166 ++++++++++++++++++ veadk/skills/utils.py | 1 + veadk/tools/skills_tools/skills_toolset.py | 13 +- 6 files changed, 485 insertions(+), 14 deletions(-) create mode 100644 tests/skills/test_dynamic_skill_runtime.py create mode 100644 veadk/skills/DYNAMIC_SKILLS.md create mode 100644 veadk/skills/runtime.py diff --git a/tests/skills/test_dynamic_skill_runtime.py b/tests/skills/test_dynamic_skill_runtime.py new file mode 100644 index 000000000..45cd83c6f --- /dev/null +++ b/tests/skills/test_dynamic_skill_runtime.py @@ -0,0 +1,195 @@ +"""Behavioral coverage for stable skill refresh and external-source composition.""" + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from veadk.skills.runtime import SkillRuntime +from veadk.skills.skill import Skill + + +def remote(name="demo", version="v1", description="description"): + return Skill( + name=name, + description=description, + path=f"skills/s-demo/{version}/demo.zip", + id="s-demo", + version_id=version, + skill_space_id="ss-one", + ) + + +def runtime(dynamic=True, **kwargs): + agent = SimpleNamespace( + skills=["ss-one"], + instruction="Original instruction\nYou have the following skills: user text", + enable_dynamic_load_skills=dynamic, + skills_mode="local", + tools=[object()], + skills_transform=None, + skill_tool_wrapper=None, + skills_refresh_failure_policy="retain", + _skills_with_checklist={}, + **kwargs, + ) + rt = SkillRuntime(agent) + return rt, agent + + +@pytest.mark.asyncio +async def test_no_change_or_reorder_keeps_prompt_and_toolset(): + rt, agent = runtime() + values = [remote(), remote("other")] + with patch( + "veadk.skills.runtime.load_skills_from_cloud", + side_effect=lambda *a, **k: values, + ): + rt.initialize() + prompt, toolset = agent.instruction, rt.toolset + values.reverse() + await rt.prepare(None) + assert agent.instruction == prompt + assert rt.toolset is toolset + assert agent.instruction.startswith(rt.base_instruction) + + +@pytest.mark.asyncio +async def test_version_change_updates_execution_without_prompt_churn(): + rt, agent = runtime() + with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): + rt.initialize() + prompt, toolset = agent.instruction, rt.toolset + with patch( + "veadk.skills.runtime.load_skills_from_cloud", + return_value=[remote(version="v2")], + ): + await rt.prepare(None) + assert agent.skills_dict["demo"].version_id == "v2" + assert rt.toolset is not toolset + assert agent.instruction == prompt + + +@pytest.mark.asyncio +async def test_external_results_survive_refresh_and_can_be_removed(): + rt, agent = runtime() + extra = remote("explicit") + extra.id = "s-explicit" + agent.skills_transform = lambda skills, ctx: skills + [extra] + with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): + rt.initialize() + await rt.prepare(None) + await rt.prepare(None) + assert set(agent.skills_dict) == {"demo", "explicit"} + agent.skills_transform = lambda skills, ctx: skills + await rt.prepare(None) + assert set(agent.skills_dict) == {"demo"} + + +@pytest.mark.asyncio +async def test_failure_retains_but_empty_success_removes_and_instances_are_independent(): + rt, agent = runtime() + other, _ = runtime() + with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): + rt.initialize() + with patch( + "veadk.skills.runtime.load_skills_from_cloud", + side_effect=RuntimeError("private detail"), + ): + await rt.prepare(None) + assert "demo" in agent.skills_dict + assert rt.status()["issues"] == [{"source": "ss-one", "error": "RuntimeError"}] + assert other.sources == {} + with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[]): + await rt.prepare(None) + assert agent.skills_dict == {} + assert agent.instruction == rt.base_instruction + + +@pytest.mark.asyncio +async def test_wrapper_failure_does_not_publish_partial_state(): + rt, agent = runtime() + with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): + rt.initialize() + old = agent.instruction, agent.skills_dict, rt.toolset + agent.skill_tool_wrapper = lambda tool: (_ for _ in ()).throw(ValueError("wrapper")) + with patch( + "veadk.skills.runtime.load_skills_from_cloud", + return_value=[remote(description="new")], + ): + with pytest.raises(ValueError, match="wrapper"): + await rt.prepare(None) + assert (agent.instruction, agent.skills_dict, rt.toolset) == old + + +@pytest.mark.asyncio +async def test_disabled_does_not_reload_sdk_sources(): + rt, agent = runtime(dynamic=False) + with patch( + "veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()] + ) as load: + rt.initialize() + await rt.prepare(None) + assert load.call_count == 1 + + +@pytest.mark.asyncio +async def test_local_change_and_deleted_file(tmp_path): + root = tmp_path / "skills" + skill = root / "local" + skill.mkdir(parents=True) + readme = skill / "SKILL.md" + readme.write_text("---\nname: local\ndescription: first\n---\nbody\n") + rt, agent = runtime() + agent.skills = [str(root)] + rt.initialize() + readme.write_text("---\nname: local\ndescription: second\n---\nbody\n") + await rt.prepare(None) + assert "second" in agent.instruction + readme.unlink() + await rt.prepare(None) + assert not agent.skills_dict + + +@pytest.mark.asyncio +async def test_agent_holds_lock_through_stream_and_releases_on_close(): + import asyncio + from veadk import Agent + from google.adk.agents import LlmAgent + from google.adk.models.base_llm import BaseLlm + + class OfflineModel(BaseLlm): + async def generate_content_async(self, llm_request, stream=False): + raise AssertionError("model should not run in lifecycle test") + yield + + agent = Agent( + name="lifecycle", + model=OfflineModel(model="offline"), + model_api_key="offline-test", + skills_mode="local", + enable_dynamic_load_skills=True, + ) + entered = [] + closed = [] + + async def events(self, context): + entered.append(context) + try: + yield context + await asyncio.Event().wait() + finally: + closed.append(context) + + with patch.object(LlmAgent, "run_async", events): + first = agent.run_async("first") + assert await anext(first) == "first" + second = agent.run_async("second") + pending = asyncio.create_task(anext(second)) + await asyncio.sleep(0) + assert entered == ["first"] + await first.aclose() + assert await asyncio.wait_for(pending, 2) == "second" + await second.aclose() + assert closed == ["first", "second"] + assert not agent._skill_runtime.lock.locked() diff --git a/veadk/agent.py b/veadk/agent.py index 84cf765c2..5afc8a05c 100644 --- a/veadk/agent.py +++ b/veadk/agent.py @@ -16,7 +16,15 @@ import os import warnings -from typing import TYPE_CHECKING, AsyncGenerator, Dict, Literal, Optional, Union +from typing import ( + TYPE_CHECKING, + AsyncGenerator, + Callable, + Dict, + Literal, + Optional, + Union, +) from google.adk.flows.llm_flows.base_llm_flow import BaseLlmFlow @@ -35,7 +43,7 @@ from google.adk.agents.llm_agent import InstructionProvider, ToolUnion from google.adk.agents.run_config import ToolThreadPoolConfig from google.adk.examples.base_example_provider import BaseExampleProvider -from pydantic import ConfigDict, Field +from pydantic import ConfigDict, Field, PrivateAttr from typing_extensions import Any from veadk.config import settings @@ -190,6 +198,18 @@ class Agent(LlmAgent): enable_dataset_gen: bool = False enable_dynamic_load_skills: bool = False + # The transform receives SDK-loaded skills and the invocation context. It may + # asynchronously merge external results; it must return the complete list. + skills_transform: Optional[Callable] = Field(default=None, exclude=True) + skill_tool_wrapper: Optional[Callable] = Field(default=None, exclude=True) + skills_refresh_failure_policy: Literal["retain", "omit"] = "retain" + _skill_runtime: Any = PrivateAttr(default=None) + + @property + def skills_status(self) -> dict: + """Safe per-source refresh status, without performing network requests.""" + return self._skill_runtime.status() if self._skill_runtime else {"issues": []} + enable_skills_checklist: bool = False _skills_with_checklist: Dict[str, Any] = {} @@ -410,7 +430,7 @@ def model_post_init(self, __context: Any) -> None: else: self.after_agent_callback = save_session_to_long_term_memory - if self.skills: + if self.skills or self.skills_transform or self.enable_dynamic_load_skills: self.load_skills() if self.enable_skills_checklist: logger.info("Skills checklist enabled") @@ -512,7 +532,6 @@ def update_model(self, model_name: str): def load_skills(self): from pathlib import Path - from veadk.skills.check_skills_callback import check_skills from veadk.skills.skill import Skill from veadk.skills.utils import ( load_skills_from_cloud, @@ -596,6 +615,15 @@ def load_skills(self): ) logger.info(f"Determined skills_mode: {self.skills_mode}") + if self.skills_mode == "local" and ( + self.enable_dynamic_load_skills or self.skills_transform + ): + from veadk.skills.runtime import SkillRuntime + + self._skill_runtime = SkillRuntime(self) + self._skill_runtime.initialize() + return + if self.skills_mode == "local": warning_message = ( "Agent(skills=..., skills_mode='local') is deprecated for legacy " @@ -663,14 +691,13 @@ def load_skills(self): self.tools.append(SkillsToolset(self.skills_dict, self.skills_mode)) if self.enable_dynamic_load_skills: - if self.before_agent_callback: - if isinstance(self.before_agent_callback, list): - self.before_agent_callback.append(check_skills) - else: - self.before_agent_callback = [ - self.before_agent_callback, - check_skills, - ] + # Preserve the remote-execution modes' existing callback behavior. + from veadk.skills.check_skills_callback import check_skills + + callbacks = self.before_agent_callback + if callbacks: + callbacks = callbacks if isinstance(callbacks, list) else [callbacks] + self.before_agent_callback = [*callbacks, check_skills] else: self.before_agent_callback = check_skills @@ -776,6 +803,22 @@ def _llm_flow(self) -> BaseLlmFlow: return SupervisorAutoFlow(supervised_agent=self) return AutoFlow() + async def run_async(self, parent_context): + """Keep a mutable skill view pinned for the complete invocation stream.""" + from contextlib import aclosing + + runtime = self._skill_runtime + if runtime is None: + async with aclosing(super().run_async(parent_context)) as events: + async for event in events: + yield event + return + async with runtime.lock: + await runtime.prepare(parent_context) + async with aclosing(super().run_async(parent_context)) as events: + async for event in events: + yield event + async def _run_async_impl( self, ctx: "InvocationContext" ) -> AsyncGenerator["Event", None]: diff --git a/veadk/skills/DYNAMIC_SKILLS.md b/veadk/skills/DYNAMIC_SKILLS.md new file mode 100644 index 000000000..8407d6e4d --- /dev/null +++ b/veadk/skills/DYNAMIC_SKILLS.md @@ -0,0 +1,57 @@ +# Dynamic local-execution skills + +`Agent(skills=[local_directory, space_id], skills_mode="local", +enable_dynamic_load_skills=True)` refreshes configured sources before each +invocation. The default remains false. Comma-separated spaces and provider +routing use the existing loader. Space versions are provider-bound versions, +not an independent client-side latest-version policy. + +An instance lock covers the complete async event stream, including callbacks +and closure. Agents do not share refresh state. This is not a cross-process +filesystem lock. Realtime/live execution is not covered. + +Metadata and package-locator changes construct a candidate Toolset before +publication. Unchanged results retain the tools. Names render in stable order; +version-only changes do not add generation IDs or timestamps to the prompt. +Local SKILL.md content changes invalidate the execution view; arbitrary scripts +and concurrent filesystem writes are not snapshotted. The prompt asks the model +to reload instructions each turn; conversation history is not rewritten. Prompt +stability alone does not establish provider-side prefix-cache hits. + +A failed SDK source retains its previous result by default, exposing a sanitized +entry in `agent.skills_status["issues"]`. Use +`skills_refresh_failure_policy="omit"` to omit it instead. A successful empty +result removes the source's skills. Initial failures have nothing to retain. + +## External results and instrumentation + +`skills_transform` accepts a synchronous or asynchronous callable: + +```python +async def merge_skills(sdk_skills, invocation_context): + return combine(sdk_skills, await external_source.refresh()) + +agent = Agent( + name="example", + skills=["/path/to/skills", "ss-example"], + skills_mode="local", + enable_dynamic_load_skills=True, + skills_transform=merge_skills, + skill_tool_wrapper=instrument_tool, +) +``` + +The transform receives copies of SDK Skill objects and returns the complete +merged list before one publication. It runs every invocation, even if SDK-source +refresh is disabled, and owns external-source precedence/errors. Duplicate names +are rejected. Transform or wrapper errors abort preparation before publication. +Construction prepares only SDK sources; external transforms first run at +invocation time. + +`skill_tool_wrapper(tool)` returns a BaseTool and runs on every new Toolset, +keeping instrumentation after refresh. Preserve names, declarations and behavior. +The public SkillsToolset constructor also accepts `tool_wrapper=`. No private +tool dictionary mutation or monkey patch is needed. + +These hooks apply to the legacy local-execution SkillsToolset path, not arbitrary +tools or ADK SkillRegistry implementations. diff --git a/veadk/skills/runtime.py b/veadk/skills/runtime.py new file mode 100644 index 000000000..c83501295 --- /dev/null +++ b/veadk/skills/runtime.py @@ -0,0 +1,166 @@ +"""Invocation-scoped refresh for legacy local-execution skills. + +Space lookup remains in the existing provider; external sources can merge their +results through Agent.skills_transform before a single, consistent publication. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import inspect +import json +from pathlib import Path + +from veadk.skills.skill import Skill +from veadk.skills.utils import load_skills_from_cloud, load_skills_from_directory +from veadk.tools.skills_tools.skills_toolset import SkillsToolset + + +class SkillRuntime: + def __init__(self, agent): + self.agent = agent + self.lock = asyncio.Lock() + self.base_instruction = agent.instruction + self.sources: dict[str, list[Skill]] = {} + self.issues: list[dict[str, str]] = [] + self.fingerprint: str | None = None + self.toolset = None + self.description = "" + + def _load(self): + sources = {} + issues = [] + for source in self.agent.skills: + if not source.strip(): + continue + path = Path(source) + # A comma-separated space input preserves the provider's historical + # behavior. Expand it here to isolate failures per space. + selectors = [source] if path.is_dir() else source.split(",") + for selector in selectors: + selector = selector.strip() + if not selector: + continue + try: + loaded = ( + load_skills_from_directory(Path(selector)) + if Path(selector).is_dir() + else load_skills_from_cloud(selector, raise_on_error=True) + ) + sources[selector] = loaded + except Exception as exc: + issues.append({"source": selector, "error": type(exc).__name__}) + sources[selector] = ( + self.sources.get(selector, []) + if self.agent.skills_refresh_failure_policy == "retain" + else [] + ) + return sources, issues + + def initialize(self): + sources, issues = self._load() + self._apply(self._flatten(sources)) + self.sources, self.issues = sources, issues + + @staticmethod + def _flatten(sources): + # Preserve configured precedence but render in canonical order later. + by_name = {} + for skills in sources.values(): + for skill in skills: + by_name[skill.name] = skill + return list(by_name.values()) + + async def prepare(self, context): + sources, issues = self.sources, self.issues + if self.agent.enable_dynamic_load_skills: + sources, issues = await asyncio.to_thread(self._load) + skills = self._flatten(sources) + transform = self.agent.skills_transform + if transform is not None: + # Copies prevent an extension from mutating our retained source state. + skills = transform([s.model_copy(deep=True) for s in skills], context) + if inspect.isawaitable(skills): + skills = await skills + self._apply(skills) + self.sources, self.issues = sources, issues + + def _apply(self, skills): + by_name = {} + records = [] + for skill in sorted(skills, key=lambda item: item.name): + if skill.name in by_name: + raise ValueError(f"Duplicate skill name: {skill.name}") + by_name[skill.name] = skill + record = skill.model_dump(mode="json") + if not skill.skill_space_id: + readme = Path(skill.path) / "SKILL.md" + record["content_digest"] = hashlib.sha256( + readme.read_bytes() + ).hexdigest() + records.append(record) + fingerprint = hashlib.sha256( + json.dumps(records, sort_keys=True, ensure_ascii=False).encode() + ).hexdigest() + if fingerprint == self.fingerprint: + return + description = self._describe(by_name) + # All potentially failing construction precedes publication. An existing + # toolset and prompt remain valid if a wrapper rejects the candidate. + toolset = SkillsToolset( + by_name, + self.agent.skills_mode, + tool_wrapper=self.agent.skill_tool_wrapper, + ) + instruction = self.base_instruction + if isinstance(instruction, str): + instruction = instruction + description + else: + base = instruction + + async def instruction(context): + value = base(context) + if inspect.isawaitable(value): + value = await value + return value + description + + tools = [tool for tool in self.agent.tools if tool is not self.toolset] + self.agent.instruction = instruction + # Preserve the dictionary identity captured by checklist callbacks. + self.agent._skills_with_checklist.clear() + self.agent._skills_with_checklist.update(by_name) + self.agent.skills_dict = by_name + self.agent.tools = [*tools, toolset] + self.toolset = toolset + self.description = description + self.fingerprint = fingerprint + + @staticmethod + def _describe(skills): + if not skills: + return "" + lines = ["\nYou have the following skills:\n"] + for skill in skills.values(): + lines.append( + f"- name: {skill.name}\n- description: {skill.description}\n\n" + ) + if any(skill.checklist for skill in skills.values()): + lines.append( + "Use `update_check_list` to mark completed skill checklist items.\n" + ) + lines.append( + "Use `skills_tool` to load skill instructions for the current turn. " + "Instructions returned in earlier turns may be outdated.\n" + ) + return "".join(lines) + + def status(self): + return { + "ready": self.fingerprint is not None, + "issues": list(self.issues), + "loaded_skills": [ + {"name": skill.name, "id": skill.id, "version": skill.version_id} + for skill in self.agent.skills_dict.values() + ], + } diff --git a/veadk/skills/utils.py b/veadk/skills/utils.py index 72fdfc58e..b6dc5f962 100644 --- a/veadk/skills/utils.py +++ b/veadk/skills/utils.py @@ -239,6 +239,7 @@ def _build_skill_from_space_item( skill_space_id=skill_space_id, bucket_name=item.get("BucketName"), id=item.get("SkillId"), + version_id=item.get("Version") or item.get("SkillVersion"), ) diff --git a/veadk/tools/skills_tools/skills_toolset.py b/veadk/tools/skills_tools/skills_toolset.py index c7507ff5d..7d5c0f6d1 100644 --- a/veadk/tools/skills_tools/skills_toolset.py +++ b/veadk/tools/skills_tools/skills_toolset.py @@ -14,7 +14,7 @@ from __future__ import annotations -from typing import Dict, List, Optional +from typing import Callable, Dict, List, Optional try: from typing_extensions import override @@ -56,7 +56,13 @@ class SkillsToolset(BaseToolset): file manipulation, and command execution. """ - def __init__(self, skills: Dict[str, Skill], skills_mode: str) -> None: + def __init__( + self, + skills: Dict[str, Skill], + skills_mode: str, + *, + tool_wrapper: Optional[Callable[[BaseTool], BaseTool]] = None, + ) -> None: """Initialize the skills toolset. Args: @@ -77,6 +83,9 @@ def __init__(self, skills: Dict[str, Skill], skills_mode: str) -> None: "update_check_list": FunctionTool(update_check_list), } + if tool_wrapper is not None: + self._tools = {key: tool_wrapper(tool) for key, tool in self._tools.items()} + @override async def get_tools( self, readonly_context: Optional[ReadonlyContext] = None From d5399147e3494a268830e69b4d6e251e247bb054 Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Wed, 9 Sep 2026 09:20:36 +0800 Subject: [PATCH 2/5] fix(skills): add required license headers --- tests/skills/test_dynamic_skill_runtime.py | 14 ++++++++++++++ veadk/skills/runtime.py | 14 ++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/tests/skills/test_dynamic_skill_runtime.py b/tests/skills/test_dynamic_skill_runtime.py index 45cd83c6f..bf617094d 100644 --- a/tests/skills/test_dynamic_skill_runtime.py +++ b/tests/skills/test_dynamic_skill_runtime.py @@ -1,3 +1,17 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + """Behavioral coverage for stable skill refresh and external-source composition.""" from types import SimpleNamespace diff --git a/veadk/skills/runtime.py b/veadk/skills/runtime.py index c83501295..9e2d26be1 100644 --- a/veadk/skills/runtime.py +++ b/veadk/skills/runtime.py @@ -1,3 +1,17 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + """Invocation-scoped refresh for legacy local-execution skills. Space lookup remains in the existing provider; external sources can merge their From b6c1862027584c85da3d869524aa4a99601c4a9c Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Wed, 9 Sep 2026 09:34:22 +0800 Subject: [PATCH 3/5] fix(skills): normalize downloaded archives and clean temporary packages --- tests/skills/test_skill_archive_install.py | 141 +++++++++++++ veadk/skills/DYNAMIC_SKILLS.md | 19 ++ veadk/tools/skills_tools/skills_tool.py | 234 ++++++++++++--------- 3 files changed, 300 insertions(+), 94 deletions(-) create mode 100644 tests/skills/test_skill_archive_install.py diff --git a/tests/skills/test_skill_archive_install.py b/tests/skills/test_skill_archive_install.py new file mode 100644 index 000000000..3bc6f26ed --- /dev/null +++ b/tests/skills/test_skill_archive_install.py @@ -0,0 +1,141 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch +import zipfile + +import pytest +from veadk.skills.skill import Skill +from veadk.tools.skills_tools.skills_tool import SkillsTool + + +@pytest.fixture +def invoke(tmp_path): + (tmp_path / "skills").mkdir() + + def run(files, name="hello", source="skillspace", fail_publish=False): + skill = Skill( + name=name, + description="test", + path="sample.zip", + skill_space_id="ss-test", + source_type=source, + ) + tool = SkillsTool({name: skill}) + + def download(self, skill, save_path, region): + if files is None: + save_path.write_bytes(b"partial download") + raise RuntimeError("download interrupted") + if isinstance(files, bytes): + save_path.write_bytes(files) + else: + with zipfile.ZipFile(save_path, "w") as archive: + for path, value in files.items(): + archive.writestr(path, value) + + original = Path.rename + + def rename(path, target): + if fail_publish and "extracted" in path.parts: + raise OSError("publish interrupted") + return original(path, target) + + with ( + patch( + "veadk.tools.skills_tools.skills_tool.get_session_path", + return_value=tmp_path, + ), + patch.object(SkillsTool, "_download_space_archive", download), + patch.object(Path, "rename", rename), + ): + return tool._invoke_skill( + name, SimpleNamespace(session=SimpleNamespace(id="test")) + ) + + return run + + +@pytest.mark.parametrize("source", ["skillspace", "skillhub"]) +@pytest.mark.parametrize("prefix", ["", "hello/", "different-wrapper/"]) +def test_layouts_install_under_name_and_remove_zips(tmp_path, invoke, source, prefix): + legacy = tmp_path / "skills/hello.zip" + legacy.write_bytes(b"old zip") + result = invoke( + {prefix + "SKILL.md": "instructions", prefix + "scripts/run.py": "sample"}, + source=source, + ) + assert not result.startswith("Error"), result + assert (tmp_path / "skills/hello/SKILL.md").read_text() == "instructions" + assert (tmp_path / "skills/hello/scripts/run.py").read_text() == "sample" + assert sorted(p.name for p in (tmp_path / "skills").iterdir()) == ["hello"] + assert sorted(p.name for p in tmp_path.iterdir()) == ["skills"] + + +def test_separate_skills_and_successful_update(tmp_path, invoke): + invoke({"SKILL.md": "one", "obsolete": "old"}) + invoke({"SKILL.md": "two"}, name="other") + invoke({"hello/SKILL.md": "new"}) + assert (tmp_path / "skills/hello/SKILL.md").read_text() == "new" + assert not (tmp_path / "skills/hello/obsolete").exists() + assert (tmp_path / "skills/other/SKILL.md").read_text() == "two" + assert not (tmp_path / "skills/SKILL.md").exists() + + +@pytest.mark.parametrize( + "files", + [ + None, + b"invalid zip", + {"file": "missing readme"}, + {"../escaped": "unsafe", "SKILL.md": "bad"}, + {"SKILL.md": b"\xff"}, + ], +) +def test_bad_updates_preserve_old_skill_and_clean_temp(tmp_path, invoke, files): + invoke({"SKILL.md": "old", "resource": "retain"}) + assert invoke(files).startswith("Error") + assert (tmp_path / "skills/hello/SKILL.md").read_text() == "old" + assert (tmp_path / "skills/hello/resource").read_text() == "retain" + assert sorted(p.name for p in tmp_path.iterdir()) == ["skills"] + assert not list((tmp_path / "skills").glob("*.zip")) + + +def test_failed_publish_rolls_back(tmp_path, invoke): + invoke({"SKILL.md": "old"}) + assert invoke({"SKILL.md": "new"}, fail_publish=True).startswith("Error") + assert (tmp_path / "skills/hello/SKILL.md").read_text() == "old" + assert sorted(p.name for p in tmp_path.iterdir()) == ["skills"] + + +def test_ambiguous_nested_layout_is_deterministic_and_logged(tmp_path, invoke): + with patch("veadk.tools.skills_tools.skills_tool.logger") as logger: + result = invoke( + {"z/SKILL.md": "z", "a/SKILL.md": "a", "a/deeper/SKILL.md": "deep"} + ) + assert not result.startswith("Error") + assert (tmp_path / "skills/hello/SKILL.md").read_text() == "a" + assert any( + "3 SKILL.md candidates" in str(c) and "a/SKILL.md" in str(c) + for c in logger.warning.call_args_list + ) + + +def test_deep_layout_falls_back_and_root_takes_precedence(tmp_path, invoke): + assert not invoke({"wrapper/nested/SKILL.md": "nested"}).startswith("Error") + assert (tmp_path / "skills/hello/SKILL.md").read_text() == "nested" + invoke({"SKILL.md": "root", "example/SKILL.md": "example"}) + assert (tmp_path / "skills/hello/SKILL.md").read_text() == "root" diff --git a/veadk/skills/DYNAMIC_SKILLS.md b/veadk/skills/DYNAMIC_SKILLS.md index 8407d6e4d..1db4507c6 100644 --- a/veadk/skills/DYNAMIC_SKILLS.md +++ b/veadk/skills/DYNAMIC_SKILLS.md @@ -55,3 +55,22 @@ tool dictionary mutation or monkey patch is needed. These hooks apply to the legacy local-execution SkillsToolset path, not arbitrary tools or ADK SkillRegistry implementations. + +## Downloaded archive layout + +Space and SkillHub archives are downloaded and extracted in a temporary directory. +Both a root `SKILL.md` and a wrapping directory containing `SKILL.md` install as +`skills//SKILL.md`, with resources alongside it. The root file takes +precedence; otherwise the shallowest candidate wins, with path order breaking +ties. Multiple candidates and deeper layouts produce a warning rather than a +rejection. Only the selected skill root and its descendants are installed. + +Extraction and UTF-8 readability checks finish before an existing installation +is moved aside. A failed replacement restores the old installation. ZIPs and +staging files are cleaned on success and failure; successful installs also remove +the legacy `skills/.zip`. Layout selection, destination, fallback, +success, and failure are logged. A failed rollback retains a backup and logs its +location. This does not provide cross-process atomicity or crash recovery. + +Skill names remain the installation key; this does not change duplicate-name +selection or remove files spilled into the shared skills directory by old versions. diff --git a/veadk/tools/skills_tools/skills_tool.py b/veadk/tools/skills_tools/skills_tool.py index 41acffe9e..512cda84f 100644 --- a/veadk/tools/skills_tools/skills_tool.py +++ b/veadk/tools/skills_tools/skills_tool.py @@ -14,6 +14,8 @@ from __future__ import annotations import os +import shutil +import tempfile from pathlib import Path from typing import Any, Dict @@ -228,107 +230,36 @@ def _invoke_skill(self, skill_name: str, tool_context: ToolContext) -> str: f"Attempting to download skill '{skill_name}' from skill space..." ) try: - save_path = skill_dir / f"{skill_name}.zip" - - if skill.source_type == "skillhub": - from veadk.skills.utils import download_skillhub_skill - - success = download_skillhub_skill(skill, save_path) - else: - from veadk.integrations.ve_tos.ve_tos import VeTOS - from veadk.skills.utils import _get_cloud_credentials - - access_key, secret_key, session_token = _get_cloud_credentials() - - tos_bucket, tos_path = skill.bucket_name, skill.path - - cloud_provider = (os.getenv("CLOUD_PROVIDER") or "").lower() - if cloud_provider == "vestack": - success = self._download_skill_via_vestack( - skill=skill, - tos_path=tos_path, - cloud_provider=cloud_provider, - access_key=access_key, - secret_key=secret_key, - session_token=session_token, - skill_name=skill_name, - save_path=save_path, - ) - else: - # Initialize VeTOS client - tos_client = VeTOS( - ak=access_key, - sk=secret_key, - session_token=session_token, - bucket_name=tos_bucket, - region=region, - ) - - success = tos_client.download( - bucket_name=tos_bucket, - object_key=tos_path, - save_path=save_path, - ) - - if not success: - source_desc = ( - "SkillHub" if skill.source_type == "skillhub" else "TOS" + if Path(skill_name).name != skill_name or skill_name in {".", ".."}: + raise ValueError("Skill name must be a single directory name") + # Stage on the same filesystem as the destination. ZIPs never + # enter the public skill directory and are removed on failure too. + with tempfile.TemporaryDirectory( + prefix=".skill-install-", dir=working_dir + ) as temp: + stage = Path(temp) + save_path = stage / "package.zip" + logger.info( + f"Downloading skill '{skill_name}' (id={skill.id}, version={skill.version_id})" ) - return f"Error: Failed to download skill '{skill_name}' from {source_desc}." - - # Extract downloaded zip into the skill directory - import zipfile - import shutil - - # Remove existing skill directory to ensure clean extraction - target_skill_dir = skill_dir / skill_name - if target_skill_dir.exists(): - try: - shutil.rmtree(target_skill_dir) - logger.info( - f"Removed existing skill directory: {target_skill_dir}" - ) - except Exception as e: - logger.warning( - f"Failed to remove existing skill directory {target_skill_dir}: {e}" - ) - + self._download_space_archive(skill, save_path, region) + self._install_skill_archive(save_path, skill_dir, skill_name) + legacy_zip = skill_dir / f"{skill_name}.zip" try: - if skill.source_type == "skillhub": - # SkillHub zips may contain files at archive root. - # Extract them into the skill-specific directory so - # they do not spill into the shared session skills dir. - target_skill_dir.mkdir(parents=True, exist_ok=True) - extract_dir = target_skill_dir - else: - # Legacy skill-space zips already include their - # top-level skill directory; keep the previous - # extraction location to avoid changing behavior. - extract_dir = skill_dir - self._safe_extract_zip(save_path, extract_dir) - except zipfile.BadZipFile: - logger.error( - f"Downloaded file for '{skill_name}' is not a valid zip" - ) - return f"Error: Downloaded file for skill '{skill_name}' is not a valid zip archive." - except Exception as e: - logger.error( - f"Failed to extract skill zip for '{skill_name}': {e}" + legacy_zip.unlink(missing_ok=True) + except OSError as exc: + logger.warning( + f"Skill '{skill_name}' installed but legacy ZIP cleanup failed: {type(exc).__name__}" ) - return f"Error: Failed to extract skill '{skill_name}' from zip: {e}" - logger.info( - f"Successfully downloaded skill '{skill_name}' from skill space" + f"Installed skill '{skill_name}' at {skill_dir / skill_name}; temporary ZIP cleaned" ) - - except Exception as e: + except Exception as exc: logger.error( - f"Failed to download skill '{skill_name}' from skill space: {e}" - ) - return ( - f"Error: Skill '{skill_name}' not found locally and failed to download from skill space: {e}. " - f"Check the available skills list in the tool description." + f"Failed to install skill '{skill_name}': {type(exc).__name__}: {exc}" ) + return f"Error: Failed to install skill '{skill_name}': {exc}" + else: # 3. Use the local skill # Create symlink to skills directory @@ -368,6 +299,121 @@ def _invoke_skill(self, skill_name: str, tool_context: ToolContext) -> str: logger.error(f"Failed to invoke skill {skill_name}: {e}") return f"Error invoking skill '{skill_name}': {e}" + def _download_space_archive( + self, skill: Skill, save_path: Path, region: str + ) -> None: + skill_name = skill.name + if skill.source_type == "skillhub": + from veadk.skills.utils import download_skillhub_skill + + success = download_skillhub_skill(skill, save_path) + else: + from veadk.integrations.ve_tos.ve_tos import VeTOS + from veadk.skills.utils import _get_cloud_credentials + + access_key, secret_key, session_token = _get_cloud_credentials() + + tos_bucket, tos_path = skill.bucket_name, skill.path + + cloud_provider = (os.getenv("CLOUD_PROVIDER") or "").lower() + if cloud_provider == "vestack": + success = self._download_skill_via_vestack( + skill=skill, + tos_path=tos_path, + cloud_provider=cloud_provider, + access_key=access_key, + secret_key=secret_key, + session_token=session_token, + skill_name=skill_name, + save_path=save_path, + ) + else: + # Initialize VeTOS client + tos_client = VeTOS( + ak=access_key, + sk=secret_key, + session_token=session_token, + bucket_name=tos_bucket, + region=region, + ) + + success = tos_client.download( + bucket_name=tos_bucket, + object_key=tos_path, + save_path=save_path, + ) + if not success: + raise RuntimeError("Skill archive download failed") + + def _install_skill_archive( + self, zip_path: Path, skill_dir: Path, skill_name: str + ) -> None: + extracted = zip_path.parent / "extracted" + extracted.mkdir() + self._safe_extract_zip(zip_path, extracted) + root_readme = extracted / "SKILL.md" + if root_readme.is_file(): + selected = root_readme + layout = "root" + else: + candidates = sorted( + (p for p in extracted.rglob("SKILL.md") if p.is_file()), + key=lambda p: (len(p.relative_to(extracted).parts), str(p)), + ) + if not candidates: + raise ValueError("Skill archive has no SKILL.md file") + selected = candidates[0] + layout = ( + "wrapped" + if len(selected.relative_to(extracted).parts) == 2 + else "nested fallback" + ) + if len(candidates) > 1 or layout == "nested fallback": + logger.warning( + f"Skill '{skill_name}' package fallback: {len(candidates)} SKILL.md candidates; " + f"selected {selected.relative_to(extracted)} by depth and path" + ) + # Validate readability before moving the existing installation aside. + selected.read_text(encoding="utf-8") + logger.info( + f"Skill '{skill_name}' package layout={layout}; selected={selected.relative_to(extracted)}; " + f"destination={skill_dir / skill_name}" + ) + target = skill_dir / skill_name + backup_root = None + if target.exists() or target.is_symlink(): + backup_root = Path( + tempfile.mkdtemp(prefix=".skill-backup-", dir=skill_dir.parent) + ) + try: + target.rename(backup_root / "previous") + except BaseException: + backup_root.rmdir() + raise + try: + selected.parent.rename(target) + except BaseException: + if backup_root is not None: + try: + (backup_root / "previous").rename(target) + except OSError: + logger.error( + f"Skill '{skill_name}' rollback failed; previous installation retained at {backup_root}" + ) + raise + backup_root.rmdir() + logger.warning( + f"Skill '{skill_name}' installation failed; previous installation restored" + ) + raise + if backup_root is not None: + try: + shutil.rmtree(backup_root) + except OSError as exc: + logger.warning( + f"Skill '{skill_name}' installed but backup cleanup failed at {backup_root}: {type(exc).__name__}" + ) + def _find_skill_file(self, skill_dir: Path, skill_name: str) -> Path: skill_root = skill_dir / skill_name skill_file = skill_root / "SKILL.md" From ceb10f29acaf85a033e8882bf5803691bf6605cf Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Wed, 9 Sep 2026 10:25:18 +0800 Subject: [PATCH 4/5] fix(skills): narrow dynamic loading to the existing callback --- tests/skills/test_agent_adk_skill_toolset.py | 4 +- tests/skills/test_check_skills_callback.py | 251 +++++++++ tests/skills/test_dynamic_skill_runtime.py | 209 -------- veadk/agent.py | 130 +---- veadk/skills/DYNAMIC_SKILLS.md | 92 ++-- veadk/skills/check_skills_callback.py | 503 ++++++------------- veadk/skills/runtime.py | 180 ------- veadk/tools/skills_tools/skills_tool.py | 48 +- veadk/tools/skills_tools/skills_toolset.py | 38 +- 9 files changed, 527 insertions(+), 928 deletions(-) create mode 100644 tests/skills/test_check_skills_callback.py delete mode 100644 tests/skills/test_dynamic_skill_runtime.py delete mode 100644 veadk/skills/runtime.py diff --git a/tests/skills/test_agent_adk_skill_toolset.py b/tests/skills/test_agent_adk_skill_toolset.py index 0bc30939e..2d2a9c94c 100644 --- a/tests/skills/test_agent_adk_skill_toolset.py +++ b/tests/skills/test_agent_adk_skill_toolset.py @@ -23,7 +23,7 @@ from veadk import Agent from veadk.prompts.agent_default_prompt import DEFAULT_INSTRUCTION -from veadk.skills import utils as skill_utils +from veadk.skills import check_skills_callback as skill_utils from veadk.skills.skill import Skill as VeADKSkill from veadk.tools.skills_tools.skills_toolset import SkillsToolset @@ -81,7 +81,7 @@ def test_sandbox_agent_skills_path_does_not_warn_as_deprecated( monkeypatch.setattr( skill_utils, "load_skills_from_cloud", - lambda source: [remote_skill], + lambda source, **kwargs: [remote_skill], ) with warnings.catch_warnings(record=True) as caught: diff --git a/tests/skills/test_check_skills_callback.py b/tests/skills/test_check_skills_callback.py new file mode 100644 index 000000000..ff340db90 --- /dev/null +++ b/tests/skills/test_check_skills_callback.py @@ -0,0 +1,251 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from veadk import Agent +from veadk.skills import check_skills_callback as refresh +from veadk.skills.skill import Skill + + +def remote(name="demo", version="v1", description="same"): + return Skill( + name=name, + description=description, + path=f"{version}.zip", + skill_space_id="ss-one", + id="s-demo", + version_id=version, + ) + + +def make_agent(sources=None, instruction="User instruction", dynamic=True): + return Agent( + name="test", + skills=["ss-one"] if sources is None else sources, + skills_mode="local", + enable_dynamic_load_skills=dynamic, + instruction=instruction, + model_api_key="offline-test", + ) + + +async def turn(agent): + await refresh.check_skills( + SimpleNamespace(_invocation_context=SimpleNamespace(agent=agent)) + ) + + +@pytest.mark.asyncio +async def test_package_change_on_first_callback_keeps_prompt_and_callbacks(): + with patch.object(refresh, "load_skills_from_cloud", return_value=[remote()]): + agent = make_agent() + prompt = agent.instruction + toolset = agent.tools[-1] + previous = toolset._tools["skills"] + agent.instruction += "\nOther callback suffix" + with patch.object( + refresh, "load_skills_from_cloud", return_value=[remote(version="v2")] + ): + await turn(agent) + assert agent.instruction == prompt + "\nOther callback suffix" + assert agent.skills_dict["demo"].path == "v2.zip" + assert toolset._tools["skills"] is not previous + assert agent.before_agent_callback is refresh.check_skills + + +@pytest.mark.asyncio +async def test_agents_have_independent_baselines_and_reorder_is_stable(): + with patch.object( + refresh, "load_skills_from_cloud", return_value=[remote(), remote("other")] + ): + one, two = make_agent(), make_agent() + prompt = one.instruction + original = one.tools[-1]._tools + with patch.object( + refresh, "load_skills_from_cloud", return_value=[remote("other"), remote()] + ): + await turn(one) + assert one.tools[-1]._tools is original + assert one.instruction == prompt + with patch.object( + refresh, "load_skills_from_cloud", return_value=[remote(version="v2")] + ): + await turn(one) + await turn(two) + assert one.skills_dict == two.skills_dict + + +@pytest.mark.asyncio +async def test_failure_retains_successful_empty_and_removed_sources_delete(): + with patch.object(refresh, "load_skills_from_cloud", return_value=[remote()]): + agent = make_agent() + with patch.object( + refresh, "load_skills_from_cloud", side_effect=RuntimeError("offline") + ): + await turn(agent) + assert "demo" in agent.skills_dict + assert agent.tools[-1].status()["issues"] + with patch.object(refresh, "load_skills_from_cloud", return_value=[]): + await turn(agent) + assert agent.skills_dict == {} + assert agent.instruction == "User instruction" + with patch.object(refresh, "load_skills_from_cloud", return_value=[remote()]): + await turn(agent) + agent.skills = [] + with patch.object(refresh, "load_skills_from_cloud") as load: + await turn(agent) + load.assert_not_called() + assert agent.skills_dict == {} + + +@pytest.mark.asyncio +async def test_failed_removed_source_does_not_reappear_and_same_name_falls_back(): + with patch.object( + refresh, + "load_skills_from_cloud", + side_effect=lambda source, **kw: [remote(version=source)], + ): + agent = make_agent(["ss-one", "ss-two"]) + assert agent.skills_dict["demo"].path == "ss-two.zip" + agent.skills = ["ss-one"] + await turn(agent) + assert agent.skills_dict["demo"].path == "ss-one.zip" + agent.skills = ["ss-three"] + with patch.object(refresh, "load_skills_from_cloud", side_effect=RuntimeError): + await turn(agent) + assert not agent.skills_dict + + +@pytest.mark.asyncio +async def test_local_content_and_same_content_source_replacement(tmp_path): + roots = [tmp_path / "one", tmp_path / "two"] + for root in roots: + (root / "demo").mkdir(parents=True) + (root / "demo/SKILL.md").write_text( + "---\nname: demo\ndescription: same\n---\nfirst" + ) + agent = make_agent([str(roots[0])]) + prompt = agent.instruction + old_tools = agent.tools[-1]._tools + (roots[0] / "demo/SKILL.md").write_text( + "---\nname: demo\ndescription: same\n---\nsecond" + ) + await turn(agent) + assert agent.tools[-1]._tools is not old_tools + assert agent.instruction == prompt + agent.skills = [str(roots[1])] + await turn(agent) + assert agent.skills_dict["demo"].path == str(roots[1] / "demo") + assert agent.instruction == prompt + + +@pytest.mark.asyncio +async def test_empty_initial_sources_can_gain_skills_and_disabled_does_not_reload(): + agent = make_agent([]) + agent.skills = ["ss-one"] + with patch.object( + refresh, "load_skills_from_cloud", return_value=[remote()] + ) as load: + await turn(agent) + assert load.call_count == 1 + assert "demo" in agent.skills_dict + with patch.object(refresh, "load_skills_from_cloud", return_value=[remote()]): + disabled = make_agent(dynamic=False) + with patch.object(refresh, "load_skills_from_cloud") as load: + await turn(disabled) + load.assert_not_called() + + +@pytest.mark.asyncio +async def test_description_refresh_preserves_user_text_and_checklist_identity(): + base = "User text: You have the following skills: keep all this" + with patch.object(refresh, "load_skills_from_cloud", return_value=[remote()]): + agent = make_agent(instruction=base) + checklist = agent._skills_with_checklist + agent.instruction += "\nOther callback suffix" + with patch.object( + refresh, "load_skills_from_cloud", return_value=[remote(description="updated")] + ): + await turn(agent) + assert agent.instruction.startswith(base) + assert agent.instruction.endswith("\nOther callback suffix") + assert "description: updated" in agent.instruction + assert agent._skills_with_checklist is checklist + assert checklist["demo"].description == "updated" + + +@pytest.mark.asyncio +async def test_callable_instruction_and_failed_candidate(): + async def base(context): + return "Callable base" + + with patch.object(refresh, "load_skills_from_cloud", return_value=[remote()]): + agent = make_agent(instruction=base) + old = agent.instruction + with patch.object( + refresh, "load_skills_from_cloud", return_value=[remote(version="v2")] + ): + await turn(agent) + assert agent.instruction is old + assert (await agent.instruction(None)).startswith("Callable base") + toolset = agent.tools[-1] + tools = toolset._tools + with ( + patch.object( + refresh, "load_skills_from_cloud", return_value=[remote(description="new")] + ), + patch.object(toolset, "build_tools", side_effect=RuntimeError), + ): + await turn(agent) + assert toolset._tools is tools + assert agent.instruction is old + assert agent.skills_dict["demo"].description == "same" + + +def test_agent_has_no_new_runtime_or_callable_parameters(): + assert ( + not {"skills_transform", "skill_tool_wrapper", "skills_refresh_failure_policy"} + & Agent.model_fields.keys() + ) + assert "run_async" not in Agent.__dict__ + assert "_skill_runtime" not in Agent.__private_attributes__ + + +@pytest.mark.asyncio +async def test_source_switch_changes_actual_local_tool_output(tmp_path): + from veadk.tools.skills_tools.skills_tool import SkillsTool + + for version in ("v1", "v2"): + root = tmp_path / version / "demo" + root.mkdir(parents=True) + (root / "SKILL.md").write_text( + f"---\nname: demo\ndescription: same\n---\n{version}" + ) + session = tmp_path / "session" + (session / "skills").mkdir(parents=True) + agent = make_agent([str(tmp_path / "v1")]) + context = SimpleNamespace(session=SimpleNamespace(id="test")) + with patch( + "veadk.tools.skills_tools.skills_tool.get_session_path", return_value=session + ): + assert "v1" in SkillsTool(agent.skills_dict)._invoke_skill("demo", context) + agent.skills = [str(tmp_path / "v2")] + await turn(agent) + assert "v2" in agent.tools[-1]._tools["skills"]._invoke_skill("demo", context) + assert (session / "skills/demo").resolve() == tmp_path / "v2/demo" + assert not list(session.glob(".skill-link-*")) diff --git a/tests/skills/test_dynamic_skill_runtime.py b/tests/skills/test_dynamic_skill_runtime.py deleted file mode 100644 index bf617094d..000000000 --- a/tests/skills/test_dynamic_skill_runtime.py +++ /dev/null @@ -1,209 +0,0 @@ -# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. -# -# Licensed 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. - -"""Behavioral coverage for stable skill refresh and external-source composition.""" - -from types import SimpleNamespace -from unittest.mock import patch - -import pytest - -from veadk.skills.runtime import SkillRuntime -from veadk.skills.skill import Skill - - -def remote(name="demo", version="v1", description="description"): - return Skill( - name=name, - description=description, - path=f"skills/s-demo/{version}/demo.zip", - id="s-demo", - version_id=version, - skill_space_id="ss-one", - ) - - -def runtime(dynamic=True, **kwargs): - agent = SimpleNamespace( - skills=["ss-one"], - instruction="Original instruction\nYou have the following skills: user text", - enable_dynamic_load_skills=dynamic, - skills_mode="local", - tools=[object()], - skills_transform=None, - skill_tool_wrapper=None, - skills_refresh_failure_policy="retain", - _skills_with_checklist={}, - **kwargs, - ) - rt = SkillRuntime(agent) - return rt, agent - - -@pytest.mark.asyncio -async def test_no_change_or_reorder_keeps_prompt_and_toolset(): - rt, agent = runtime() - values = [remote(), remote("other")] - with patch( - "veadk.skills.runtime.load_skills_from_cloud", - side_effect=lambda *a, **k: values, - ): - rt.initialize() - prompt, toolset = agent.instruction, rt.toolset - values.reverse() - await rt.prepare(None) - assert agent.instruction == prompt - assert rt.toolset is toolset - assert agent.instruction.startswith(rt.base_instruction) - - -@pytest.mark.asyncio -async def test_version_change_updates_execution_without_prompt_churn(): - rt, agent = runtime() - with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): - rt.initialize() - prompt, toolset = agent.instruction, rt.toolset - with patch( - "veadk.skills.runtime.load_skills_from_cloud", - return_value=[remote(version="v2")], - ): - await rt.prepare(None) - assert agent.skills_dict["demo"].version_id == "v2" - assert rt.toolset is not toolset - assert agent.instruction == prompt - - -@pytest.mark.asyncio -async def test_external_results_survive_refresh_and_can_be_removed(): - rt, agent = runtime() - extra = remote("explicit") - extra.id = "s-explicit" - agent.skills_transform = lambda skills, ctx: skills + [extra] - with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): - rt.initialize() - await rt.prepare(None) - await rt.prepare(None) - assert set(agent.skills_dict) == {"demo", "explicit"} - agent.skills_transform = lambda skills, ctx: skills - await rt.prepare(None) - assert set(agent.skills_dict) == {"demo"} - - -@pytest.mark.asyncio -async def test_failure_retains_but_empty_success_removes_and_instances_are_independent(): - rt, agent = runtime() - other, _ = runtime() - with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): - rt.initialize() - with patch( - "veadk.skills.runtime.load_skills_from_cloud", - side_effect=RuntimeError("private detail"), - ): - await rt.prepare(None) - assert "demo" in agent.skills_dict - assert rt.status()["issues"] == [{"source": "ss-one", "error": "RuntimeError"}] - assert other.sources == {} - with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[]): - await rt.prepare(None) - assert agent.skills_dict == {} - assert agent.instruction == rt.base_instruction - - -@pytest.mark.asyncio -async def test_wrapper_failure_does_not_publish_partial_state(): - rt, agent = runtime() - with patch("veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()]): - rt.initialize() - old = agent.instruction, agent.skills_dict, rt.toolset - agent.skill_tool_wrapper = lambda tool: (_ for _ in ()).throw(ValueError("wrapper")) - with patch( - "veadk.skills.runtime.load_skills_from_cloud", - return_value=[remote(description="new")], - ): - with pytest.raises(ValueError, match="wrapper"): - await rt.prepare(None) - assert (agent.instruction, agent.skills_dict, rt.toolset) == old - - -@pytest.mark.asyncio -async def test_disabled_does_not_reload_sdk_sources(): - rt, agent = runtime(dynamic=False) - with patch( - "veadk.skills.runtime.load_skills_from_cloud", return_value=[remote()] - ) as load: - rt.initialize() - await rt.prepare(None) - assert load.call_count == 1 - - -@pytest.mark.asyncio -async def test_local_change_and_deleted_file(tmp_path): - root = tmp_path / "skills" - skill = root / "local" - skill.mkdir(parents=True) - readme = skill / "SKILL.md" - readme.write_text("---\nname: local\ndescription: first\n---\nbody\n") - rt, agent = runtime() - agent.skills = [str(root)] - rt.initialize() - readme.write_text("---\nname: local\ndescription: second\n---\nbody\n") - await rt.prepare(None) - assert "second" in agent.instruction - readme.unlink() - await rt.prepare(None) - assert not agent.skills_dict - - -@pytest.mark.asyncio -async def test_agent_holds_lock_through_stream_and_releases_on_close(): - import asyncio - from veadk import Agent - from google.adk.agents import LlmAgent - from google.adk.models.base_llm import BaseLlm - - class OfflineModel(BaseLlm): - async def generate_content_async(self, llm_request, stream=False): - raise AssertionError("model should not run in lifecycle test") - yield - - agent = Agent( - name="lifecycle", - model=OfflineModel(model="offline"), - model_api_key="offline-test", - skills_mode="local", - enable_dynamic_load_skills=True, - ) - entered = [] - closed = [] - - async def events(self, context): - entered.append(context) - try: - yield context - await asyncio.Event().wait() - finally: - closed.append(context) - - with patch.object(LlmAgent, "run_async", events): - first = agent.run_async("first") - assert await anext(first) == "first" - second = agent.run_async("second") - pending = asyncio.create_task(anext(second)) - await asyncio.sleep(0) - assert entered == ["first"] - await first.aclose() - assert await asyncio.wait_for(pending, 2) == "second" - await second.aclose() - assert closed == ["first", "second"] - assert not agent._skill_runtime.lock.locked() diff --git a/veadk/agent.py b/veadk/agent.py index 5afc8a05c..e6bf66c01 100644 --- a/veadk/agent.py +++ b/veadk/agent.py @@ -16,15 +16,7 @@ import os import warnings -from typing import ( - TYPE_CHECKING, - AsyncGenerator, - Callable, - Dict, - Literal, - Optional, - Union, -) +from typing import TYPE_CHECKING, AsyncGenerator, Dict, Literal, Optional, Union from google.adk.flows.llm_flows.base_llm_flow import BaseLlmFlow @@ -43,7 +35,7 @@ from google.adk.agents.llm_agent import InstructionProvider, ToolUnion from google.adk.agents.run_config import ToolThreadPoolConfig from google.adk.examples.base_example_provider import BaseExampleProvider -from pydantic import ConfigDict, Field, PrivateAttr +from pydantic import ConfigDict, Field from typing_extensions import Any from veadk.config import settings @@ -198,18 +190,6 @@ class Agent(LlmAgent): enable_dataset_gen: bool = False enable_dynamic_load_skills: bool = False - # The transform receives SDK-loaded skills and the invocation context. It may - # asynchronously merge external results; it must return the complete list. - skills_transform: Optional[Callable] = Field(default=None, exclude=True) - skill_tool_wrapper: Optional[Callable] = Field(default=None, exclude=True) - skills_refresh_failure_policy: Literal["retain", "omit"] = "retain" - _skill_runtime: Any = PrivateAttr(default=None) - - @property - def skills_status(self) -> dict: - """Safe per-source refresh status, without performing network requests.""" - return self._skill_runtime.status() if self._skill_runtime else {"issues": []} - enable_skills_checklist: bool = False _skills_with_checklist: Dict[str, Any] = {} @@ -430,7 +410,7 @@ def model_post_init(self, __context: Any) -> None: else: self.after_agent_callback = save_session_to_long_term_memory - if self.skills or self.skills_transform or self.enable_dynamic_load_skills: + if self.skills or self.enable_dynamic_load_skills: self.load_skills() if self.enable_skills_checklist: logger.info("Skills checklist enabled") @@ -530,16 +510,7 @@ def update_model(self, model_name: str): ) def load_skills(self): - from pathlib import Path - - from veadk.skills.skill import Skill - from veadk.skills.utils import ( - load_skills_from_cloud, - load_skills_from_directory, - ) - from veadk.tools.skills_tools.skills_toolset import SkillsToolset - - self.skills_dict: Dict[str, Skill] = {} + from veadk.skills.check_skills_callback import check_skills, initialize_skills # Determine skills_mode if not set if not self.skills_mode: @@ -615,15 +586,6 @@ def load_skills(self): ) logger.info(f"Determined skills_mode: {self.skills_mode}") - if self.skills_mode == "local" and ( - self.enable_dynamic_load_skills or self.skills_transform - ): - from veadk.skills.runtime import SkillRuntime - - self._skill_runtime = SkillRuntime(self) - self._skill_runtime.initialize() - return - if self.skills_mode == "local": warning_message = ( "Agent(skills=..., skills_mode='local') is deprecated for legacy " @@ -637,67 +599,17 @@ def load_skills(self): warnings.warn(warning_message, DeprecationWarning, stacklevel=2) logger.warning(warning_message) - for item in self.skills: - if not item or str(item).strip() == "": - continue - path = Path(item) - if path.exists() and path.is_dir(): - for skill in load_skills_from_directory(path): - self.skills_dict[skill.name] = skill - else: - for skill in load_skills_from_cloud(item): - self.skills_dict[skill.name] = skill - if self.skills_dict: - self.instruction += "\nYou have the following skills:\n" - - self._skills_with_checklist = self.skills_dict - - has_checklist = False - for skill in self.skills_dict.values(): - self.instruction += ( - f"- name: {skill.name}\n- description: {skill.description}\n\n" - ) - if skill.checklist: - has_checklist = True - - if has_checklist: - self.instruction += ( - "Some skills have a checklist that you must complete step by step. " - "Use the `update_check_list` tool to mark each item as completed.\n\n" - ) - - if self.skills_mode not in [ - "skills_sandbox", - "aio_sandbox", - "local", - ]: - raise ValueError( - f"Unsupported skill mode {self.skills_mode}, use `skills_sandbox`, `aio_sandbox` or `local` instead." - ) - - if self.skills_mode == "skills_sandbox": - self.instruction += ( - "You can use the skills by calling the `execute_skills` tool.\n\n" - ) - - if self.skills_mode == "local": - self.instruction += ( - "You can use the skills by calling the `skills_tool` tool.\n\n" - ) - - else: - logger.warning("No skills loaded.") - - self.tools.append(SkillsToolset(self.skills_dict, self.skills_mode)) + initialize_skills(self) if self.enable_dynamic_load_skills: - # Preserve the remote-execution modes' existing callback behavior. - from veadk.skills.check_skills_callback import check_skills - - callbacks = self.before_agent_callback - if callbacks: - callbacks = callbacks if isinstance(callbacks, list) else [callbacks] - self.before_agent_callback = [*callbacks, check_skills] + if self.before_agent_callback: + if isinstance(self.before_agent_callback, list): + self.before_agent_callback.append(check_skills) + else: + self.before_agent_callback = [ + self.before_agent_callback, + check_skills, + ] else: self.before_agent_callback = check_skills @@ -803,22 +715,6 @@ def _llm_flow(self) -> BaseLlmFlow: return SupervisorAutoFlow(supervised_agent=self) return AutoFlow() - async def run_async(self, parent_context): - """Keep a mutable skill view pinned for the complete invocation stream.""" - from contextlib import aclosing - - runtime = self._skill_runtime - if runtime is None: - async with aclosing(super().run_async(parent_context)) as events: - async for event in events: - yield event - return - async with runtime.lock: - await runtime.prepare(parent_context) - async with aclosing(super().run_async(parent_context)) as events: - async for event in events: - yield event - async def _run_async_impl( self, ctx: "InvocationContext" ) -> AsyncGenerator["Event", None]: diff --git a/veadk/skills/DYNAMIC_SKILLS.md b/veadk/skills/DYNAMIC_SKILLS.md index 1db4507c6..1cda00eb8 100644 --- a/veadk/skills/DYNAMIC_SKILLS.md +++ b/veadk/skills/DYNAMIC_SKILLS.md @@ -1,60 +1,38 @@ -# Dynamic local-execution skills - -`Agent(skills=[local_directory, space_id], skills_mode="local", -enable_dynamic_load_skills=True)` refreshes configured sources before each -invocation. The default remains false. Comma-separated spaces and provider -routing use the existing loader. Space versions are provider-bound versions, -not an independent client-side latest-version policy. - -An instance lock covers the complete async event stream, including callbacks -and closure. Agents do not share refresh state. This is not a cross-process -filesystem lock. Realtime/live execution is not covered. - -Metadata and package-locator changes construct a candidate Toolset before -publication. Unchanged results retain the tools. Names render in stable order; -version-only changes do not add generation IDs or timestamps to the prompt. -Local SKILL.md content changes invalidate the execution view; arbitrary scripts -and concurrent filesystem writes are not snapshotted. The prompt asks the model -to reload instructions each turn; conversation history is not rewritten. Prompt -stability alone does not establish provider-side prefix-cache hits. - -A failed SDK source retains its previous result by default, exposing a sanitized -entry in `agent.skills_status["issues"]`. Use -`skills_refresh_failure_policy="omit"` to omit it instead. A successful empty -result removes the source's skills. Initial failures have nothing to retain. - -## External results and instrumentation - -`skills_transform` accepts a synchronous or asynchronous callable: - -```python -async def merge_skills(sdk_skills, invocation_context): - return combine(sdk_skills, await external_source.refresh()) - -agent = Agent( - name="example", - skills=["/path/to/skills", "ss-example"], - skills_mode="local", - enable_dynamic_load_skills=True, - skills_transform=merge_skills, - skill_tool_wrapper=instrument_tool, -) -``` - -The transform receives copies of SDK Skill objects and returns the complete -merged list before one publication. It runs every invocation, even if SDK-source -refresh is disabled, and owns external-source precedence/errors. Duplicate names -are rejected. Transform or wrapper errors abort preparation before publication. -Construction prepares only SDK sources; external transforms first run at -invocation time. - -`skill_tool_wrapper(tool)` returns a BaseTool and runs on every new Toolset, -keeping instrumentation after refresh. Preserve names, declarations and behavior. -The public SkillsToolset constructor also accepts `tool_wrapper=`. No private -tool dictionary mutation or monkey patch is needed. - -These hooks apply to the legacy local-execution SkillsToolset path, not arbitrary -tools or ADK SkillRegistry implementations. +# Dynamic legacy skills + +The existing `Agent(skills=[local_directory, space_id], skills_mode="local", +enable_dynamic_load_skills=True)` option registers `check_skills` as a before-agent +callback. Its default remains false. It works without a sandbox subclass. + +Each callback reads the current `Agent.skills`, including replaced directories, +added/removed spaces, and an empty list. Comma-separated spaces use the existing +provider routing. Provider versions remain authoritative; there is no additional +client-side latest-version policy. + +State is held by the existing SkillsToolset instance, with its baseline recorded +during initialization. Changes include local SKILL.md content, source identity, +package path, bucket, version and metadata. A path-only change replaces execution +tools without rewriting an otherwise identical skills prompt. Stable name ordering +avoids prompt churn when a provider reorders results. The base instruction and +other callback text are preserved, including callable instructions. + +A failed source retains its previous successful result; successful empty responses +remove it. Sources removed from configuration cannot reappear from retained state. +`SkillsToolset.status()` reports safe error types and loaded names/IDs/versions. +The next successful refresh clears failures. Tool construction finishes before the +new skill dictionary, checklist view and tools are published. + +Applications can subclass SkillsToolset's named `prepare_skills`, `wrap_tool`, +and `on_source_error` methods to adapt results and instrumentation. Rebuilding tools +uses the same instance, so instrumentation is preserved. There are no new Agent +callable parameters, execution overrides, or runtime binding/discovery components. + +This callback does **not** lock the full execution of a shared Agent. Applications +that run a mutable Agent concurrently must coordinate that execution themselves. +The Playground sandbox does so in its own SkillSandboxAgent. Realtime/live flows, +arbitrary local script writes, and remote objects overwritten without any locator +or metadata change are not snapshotted. Prompt stability is not a guarantee of +provider-side prefix-cache hits, and earlier conversation messages are not rewritten. ## Downloaded archive layout diff --git a/veadk/skills/check_skills_callback.py b/veadk/skills/check_skills_callback.py index 1252054ff..76f7509ce 100644 --- a/veadk/skills/check_skills_callback.py +++ b/veadk/skills/check_skills_callback.py @@ -12,361 +12,180 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""Refresh legacy skills through the existing before-agent callback. + +State belongs to each SkillsToolset. This callback does not serialize concurrent +invocations of a shared Agent; applications sharing mutable agents own that lock. +""" + +import asyncio import hashlib +import inspect +import json +from dataclasses import dataclass, field from pathlib import Path -from typing import Optional, Dict, List -from google.adk.agents.callback_context import CallbackContext -from google.genai import types -from veadk.skills.skill import Skill -from veadk.skills.utils import load_skill_from_directory, load_skills_from_cloud + +from veadk.skills.utils import load_skills_from_cloud, load_skills_from_directory from veadk.utils.logger import get_logger logger = get_logger(__name__) -# Cache for storing skill states to detect changes -# Key format: "local:{path}" or "cloud:{skill_space_id}:{skill_name}" -skill_cache: Dict[str, str] = {} - - -def get_local_skill_hash(skill_directory: Path) -> str: - """Calculate hash value for local skill directory to detect changes""" - skill_readme = skill_directory / "SKILL.md" - if not skill_readme.exists(): - return "" - - content = skill_readme.read_text(encoding="utf-8") - return hashlib.md5(content.encode("utf-8")).hexdigest() - - -def get_cloud_skill_hash(skill: Skill) -> str: - """Calculate hash value for cloud skill based on name and description - - Args: - skill: Skill object - - Returns: - MD5 hash of skill name and description - """ - content = f"{skill.name}|{skill.description}" - return hashlib.md5(content.encode("utf-8")).hexdigest() - - -def detect_skill_changes( - current_skills_dict: Dict[str, Skill], reloaded_skills_dict: Dict[str, Skill] -) -> tuple[List[Skill], List[Skill], List[str]]: - """Detect new, modified, and deleted skills by comparing current and reloaded skills - - Args: - current_skills_dict: Current skills dictionary from agent - reloaded_skills_dict: Newly reloaded skills dictionary - - Returns: - Tuple of (new_skills, modified_skills, deleted_skill_names) - """ - global skill_cache - - new_skills = [] - modified_skills = [] - deleted_skill_names = [] - - # Check for new and modified skills - for skill_name, reloaded_skill in reloaded_skills_dict.items(): - if skill_name not in current_skills_dict: - # New skill detected - new_skills.append(reloaded_skill) - logger.info(f"Detected new skill: {skill_name}") - - # Initialize cache for new skill - if reloaded_skill.skill_space_id: - # Cloud skill - use name and description hash - cache_key = f"cloud:{reloaded_skill.skill_space_id}:{skill_name}" - skill_cache[cache_key] = get_cloud_skill_hash(reloaded_skill) - else: - # Local skill - use SKILL.md file hash - cache_key = f"local:{reloaded_skill.path}" - skill_cache[cache_key] = get_local_skill_hash(Path(reloaded_skill.path)) - else: - # Existing skill - check if modified - current_skill = current_skills_dict[skill_name] - - # Determine if skill is modified based on source type - is_modified = False - - if reloaded_skill.skill_space_id: - # Cloud skill - check using name and description - cache_key = f"cloud:{reloaded_skill.skill_space_id}:{skill_name}" - current_hash = get_cloud_skill_hash(reloaded_skill) - previous_hash = skill_cache.get(cache_key, "") - if previous_hash == "": - # First time seeing this skill, initialize cache but don't mark as modified - skill_cache[cache_key] = current_hash - logger.debug(f"Initialized cache for cloud skill: {skill_name}") - elif current_hash != previous_hash: - # Hash changed, skill is modified - is_modified = True - skill_cache[cache_key] = current_hash - else: - # Local skill - check using file hash - cache_key = f"local:{reloaded_skill.path}" - current_hash = get_local_skill_hash(Path(reloaded_skill.path)) - previous_hash = skill_cache.get(cache_key, "") - - if previous_hash == "": - # First time seeing this skill, initialize cache but don't mark as modified - skill_cache[cache_key] = current_hash - logger.debug(f"Initialized cache for local skill: {skill_name}") - elif current_hash != previous_hash: - # Hash changed, skill is modified - is_modified = True - skill_cache[cache_key] = current_hash - - if is_modified: - modified_skills.append(reloaded_skill) - logger.info(f"Detected modified skill: {skill_name}") - - # Check for deleted skills - for skill_name in current_skills_dict.keys(): - if skill_name not in reloaded_skills_dict: - deleted_skill_names.append(skill_name) - logger.info(f"Detected deleted skill: {skill_name}") - - # Remove from cache - current_skill = current_skills_dict[skill_name] - if current_skill.skill_space_id: - cache_key = f"cloud:{current_skill.skill_space_id}:{skill_name}" - else: - cache_key = f"local:{current_skill.path}" - skill_cache.pop(cache_key, None) - - return new_skills, modified_skills, deleted_skill_names - - -def reload_skills_from_config(skills_config: List[str]) -> Dict[str, Skill]: - """Reload all skills from configuration (both local and cloud) - - Args: - skills_config: List of skill paths/IDs from agent configuration - - Returns: - Dictionary mapping skill names to Skill objects - """ - all_skills: Dict[str, Skill] = {} - - for item in skills_config: - if not item or str(item).strip() == "": +@dataclass +class _RefreshState: + sources: dict = field(default_factory=dict) + fingerprints: dict = field(default_factory=dict) + issues: list = field(default_factory=list) + section: str = "" + instruction: object = None + base_instruction: object = None + + +def _fingerprints(skills): + result = {} + for name, skill in skills.items(): + record = skill.model_dump(mode="json") + if not skill.skill_space_id: + record["readme"] = hashlib.sha256( + (Path(skill.path) / "SKILL.md").read_bytes() + ).hexdigest() + result[name] = json.dumps(record, sort_keys=True, ensure_ascii=False) + return result + + +def _load_sources(config, toolset): + sources, issues = {}, [] + previous = toolset._refresh_state.sources + for value in config: + if not str(value).strip(): continue - - path = Path(item) - - # Check if it's a local directory - if path.exists() and path.is_dir(): - logger.debug(f"Reloading skills from local directory: {path}") + selectors = [str(value)] if Path(value).is_dir() else str(value).split(",") + for source in map(str.strip, selectors): + if not source: + continue try: - for skill_dir in path.iterdir(): - if skill_dir.is_dir(): - skill = load_skill_from_directory(skill_dir) - # Only add skill if it loaded successfully - if skill is not None: - all_skills[skill.name] = skill - else: - logger.warning(f"Skipped failed skill from {skill_dir}") - except Exception as e: - logger.error( - f"Failed to reload skills from local directory {path}: {e}" + sources[source] = ( + load_skills_from_directory(Path(source)) + if Path(source).is_dir() + else load_skills_from_cloud(source, raise_on_error=True) ) - else: - # Treat as cloud skill space ID - logger.debug(f"Reloading skills from cloud space: {item}") - try: - cloud_skills = load_skills_from_cloud(item) - for skill in cloud_skills: - all_skills[skill.name] = skill - except Exception as e: - logger.error(f"Failed to reload skills from cloud space {item}: {e}") - - return all_skills - - -def rebuild_instruction_with_skills( - current_instruction: str, skills_dict: Dict[str, Skill] -) -> str: - """Rebuild instruction with updated skill list - - Args: - current_instruction: Current agent instruction - skills_dict: All current skills - - Returns: - Updated instruction string - """ - new_instruction_parts = [] - - # Find the skills section start - skills_section_start = current_instruction.find("You have the following skills:") - - if skills_section_start == -1: - # No existing skills section, append new one - new_instruction_parts.append(current_instruction) - new_instruction_parts.append("\nYou have the following skills:\n") - else: - # Keep content before skills section - new_instruction_parts.append(current_instruction[:skills_section_start]) - new_instruction_parts.append("You have the following skills:\n") - - # Add all current skills - for skill in skills_dict.values(): - new_instruction_parts.append( - f"- name: {skill.name}\n- description: {skill.description}\n\n" - ) - - # Determine the tool instruction based on skills_mode from agent - if "skills_tool" in current_instruction: - tool_instruction = ( - "You can use the skills by calling the `skills_tool` tool.\n\n" - ) - elif "execute_skills" in current_instruction: - tool_instruction = ( - "You can use the skills by calling the `execute_skills` tool.\n\n" - ) - else: - tool_instruction = "You can use the skills by calling the appropriate tool.\n\n" - - new_instruction_parts.append(tool_instruction) - - return "".join(new_instruction_parts) - - -def update_skills_toolset( - callback_context: CallbackContext, updated_skills_dict: Dict[str, Skill] -) -> None: - """Remove old SkillsToolset and add new one with updated skills - - Args: - callback_context: Callback context containing agent information - updated_skills_dict: Updated skills dictionary - """ - try: - from veadk.tools.skills_tools.skills_toolset import SkillsToolset - - agent = callback_context._invocation_context.agent - - # Find and remove existing SkillsToolset - tools_to_remove = [] - for i, tool in enumerate(agent.tools): - if isinstance(tool, SkillsToolset): - tools_to_remove.append(i) - logger.debug(f"Found SkillsToolset at index {i}, will remove it") - - # Remove in reverse order to avoid index shifting issues - for i in reversed(tools_to_remove): - agent.tools.pop(i) - logger.info("Removed old SkillsToolset from agent tools") - - # Get skills_mode from agent - skills_mode = getattr(agent, "skills_mode", "local") - - # Add new SkillsToolset with updated skills - new_toolset = SkillsToolset(updated_skills_dict, skills_mode) - agent.tools.append(new_toolset) - logger.info(f"Added new SkillsToolset with {len(updated_skills_dict)} skills") - - except Exception as e: - logger.error(f"Failed to update SkillsToolset: {e}", exc_info=True) - - -def check_skills(callback_context: CallbackContext) -> Optional[types.Content]: - """Check for skill changes and update agent instruction and toolset dynamically - - This callback checks both local directory skills and cloud space skills for changes, - including new skills, modified skills, and deleted skills. When changes are detected, - it updates the agent's instruction and reloads the SkillsToolset. - - The detection process: - 1. Reload all skills from the original configuration - 2. Compare with current skills_dict to detect: - - New skills: present in reloaded but not in current - - Modified skills: present in both but content changed (via hash comparison) - * Local skills: compare SKILL.md file hash - * Cloud skills: compare name and description hash - - Deleted skills: present in current but not in reloaded - 3. Update agent.skills_dict with the reloaded skills - 4. Rebuild instruction with updated skill list - 5. Replace SkillsToolset with new instance using updated skills - - Note: On first run when cache is empty, skills are initialized in cache but not - marked as modified to avoid false positives. - - Args: - callback_context: Callback context containing agent information - - Returns: - None (updates agent instruction, skills_dict and tools in-place) - """ - global skill_cache - - try: - agent = callback_context._invocation_context.agent - - # Get current skills_dict from agent - if not hasattr(agent, "skills_dict"): - logger.debug("Agent has no skills_dict attribute, skip checking") - return None - - current_skills_dict = agent.skills_dict - - # Get skills configuration from agent - if not hasattr(agent, "skills") or not agent.skills: - logger.debug("Agent has no skills configuration, skip checking") - return None - - # Reload skills from original configuration - reloaded_skills_dict = reload_skills_from_config(agent.skills) - - # If both are empty, skip - if not current_skills_dict and not reloaded_skills_dict: - logger.debug("No skills found in both current and reloaded, skip checking") - return None - - # Detect changes - new_skills, modified_skills, deleted_skill_names = detect_skill_changes( - current_skills_dict, reloaded_skills_dict - ) + except Exception as exc: + issues.append({"source": source, "error": type(exc).__name__}) + logger.warning( + "Skill source refresh failed: source=%s error=%s", + source, + type(exc).__name__, + ) + sources[source] = toolset.on_source_error( + source, exc, previous.get(source, []) + ) + return sources, issues - # If no changes detected, return early - if not new_skills and not modified_skills and not deleted_skill_names: - logger.debug("No skill changes detected") - return None - # Log changes - if new_skills: - logger.info(f"New skills: {[s.name for s in new_skills]}") - if modified_skills: - logger.info(f"Modified skills: {[s.name for s in modified_skills]}") - if deleted_skill_names: - logger.info(f"Deleted skills: {deleted_skill_names}") +def _flatten(sources): + return {skill.name: skill for skills in sources.values() for skill in skills} - # Update agent.skills_dict with reloaded skills - agent.skills_dict = reloaded_skills_dict - agent._skills_with_checklist = reloaded_skills_dict - logger.info( - f"Updated agent.skills_dict with {len(reloaded_skills_dict)} skills" - ) - # Rebuild instruction with updated skills - current_instruction = agent.instruction - new_instruction = rebuild_instruction_with_skills( - current_instruction, reloaded_skills_dict +def _describe(skills, mode): + if not skills: + return "" + lines = ["\nYou have the following skills:\n"] + for name in sorted(skills): + skill = skills[name] + lines.append(f"- name: {skill.name}\n- description: {skill.description}\n\n") + if any(skill.checklist for skill in skills.values()): + lines.append( + "Some skills have a checklist that you must complete step by step. " + "Use the `update_check_list` tool to mark each item as completed.\n\n" ) - - # Update agent instruction - agent.instruction = new_instruction - logger.info("Agent instruction updated with skill changes") - - # Update SkillsToolset with new skills_dict - update_skills_toolset(callback_context, reloaded_skills_dict) - - except Exception as e: - logger.error(f"Error checking skills: {e}", exc_info=True) - + tool = {"local": "skills_tool", "skills_sandbox": "execute_skills"}.get(mode) + if tool: + lines.append(f"You can use the skills by calling the `{tool}` tool.\n\n") + return "".join(lines) + + +def _instruction(agent, state, section): + current = agent.instruction + if isinstance(current, str): + # Replace only our exact previous section, preserving appended content + # from other callbacks. Never truncate at a generic user-visible marker. + if state.section and state.section in current: + return current.replace(state.section, section, 1) + return current + section + base = state.base_instruction if current is state.instruction else current + + async def instruction(context): + value = base(context) + if inspect.isawaitable(value): + value = await value + return value + section + + return instruction + + +def _apply(agent, toolset, skills, sources, issues): + state = toolset._refresh_state + skills = dict(sorted(skills.items())) + fingerprints = _fingerprints(skills) + section = _describe(skills, agent.skills_mode) + if fingerprints != state.fingerprints or state.instruction is None: + # Construct before publication; failure leaves the prior view intact. + tools = toolset.build_tools(skills) + instruction = agent.instruction + if section != state.section: + instruction = _instruction(agent, state, section) + if callable(agent.instruction) and agent.instruction is not state.instruction: + state.base_instruction = agent.instruction + agent.instruction = instruction + agent._skills_with_checklist.clear() + agent._skills_with_checklist.update(skills) + agent.skills_dict = skills + toolset._tools = tools + state.fingerprints = fingerprints + state.section = section + state.instruction = instruction + logger.info("Skills refreshed: count=%d", len(skills)) + state.sources, state.issues = sources, issues + + +def initialize_skills(agent): + from veadk.tools.skills_tools.skills_toolset import SkillsToolset + + if agent.skills_mode not in {"local", "skills_sandbox", "aio_sandbox"}: + raise ValueError(f"Unsupported skill mode {agent.skills_mode}") + toolset = SkillsToolset({}, agent.skills_mode) + toolset._refresh_state = _RefreshState(base_instruction=agent.instruction) + sources, issues = _load_sources(agent.skills, toolset) + _apply(agent, toolset, _flatten(sources), sources, issues) + agent.tools.append(toolset) + + +async def check_skills(callback_context): + from veadk.tools.skills_tools.skills_toolset import SkillsToolset + + agent = callback_context._invocation_context.agent + if not agent.enable_dynamic_load_skills: + return + for toolset in agent.tools: + if not isinstance(toolset, SkillsToolset) or toolset._refresh_state is None: + continue + try: + sources, issues = await asyncio.to_thread( + _load_sources, agent.skills, toolset + ) + skills = await toolset.prepare_skills( + { + name: skill.model_copy(deep=True) + for name, skill in _flatten(sources).items() + }, + callback_context, + ) + _apply(agent, toolset, skills, sources, issues) + except Exception as exc: + toolset._refresh_state.issues = [ + {"source": "refresh", "error": type(exc).__name__} + ] + logger.warning( + "Skills refresh retained previous view: error=%s", type(exc).__name__ + ) return None diff --git a/veadk/skills/runtime.py b/veadk/skills/runtime.py deleted file mode 100644 index 9e2d26be1..000000000 --- a/veadk/skills/runtime.py +++ /dev/null @@ -1,180 +0,0 @@ -# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. -# -# Licensed 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. - -"""Invocation-scoped refresh for legacy local-execution skills. - -Space lookup remains in the existing provider; external sources can merge their -results through Agent.skills_transform before a single, consistent publication. -""" - -from __future__ import annotations - -import asyncio -import hashlib -import inspect -import json -from pathlib import Path - -from veadk.skills.skill import Skill -from veadk.skills.utils import load_skills_from_cloud, load_skills_from_directory -from veadk.tools.skills_tools.skills_toolset import SkillsToolset - - -class SkillRuntime: - def __init__(self, agent): - self.agent = agent - self.lock = asyncio.Lock() - self.base_instruction = agent.instruction - self.sources: dict[str, list[Skill]] = {} - self.issues: list[dict[str, str]] = [] - self.fingerprint: str | None = None - self.toolset = None - self.description = "" - - def _load(self): - sources = {} - issues = [] - for source in self.agent.skills: - if not source.strip(): - continue - path = Path(source) - # A comma-separated space input preserves the provider's historical - # behavior. Expand it here to isolate failures per space. - selectors = [source] if path.is_dir() else source.split(",") - for selector in selectors: - selector = selector.strip() - if not selector: - continue - try: - loaded = ( - load_skills_from_directory(Path(selector)) - if Path(selector).is_dir() - else load_skills_from_cloud(selector, raise_on_error=True) - ) - sources[selector] = loaded - except Exception as exc: - issues.append({"source": selector, "error": type(exc).__name__}) - sources[selector] = ( - self.sources.get(selector, []) - if self.agent.skills_refresh_failure_policy == "retain" - else [] - ) - return sources, issues - - def initialize(self): - sources, issues = self._load() - self._apply(self._flatten(sources)) - self.sources, self.issues = sources, issues - - @staticmethod - def _flatten(sources): - # Preserve configured precedence but render in canonical order later. - by_name = {} - for skills in sources.values(): - for skill in skills: - by_name[skill.name] = skill - return list(by_name.values()) - - async def prepare(self, context): - sources, issues = self.sources, self.issues - if self.agent.enable_dynamic_load_skills: - sources, issues = await asyncio.to_thread(self._load) - skills = self._flatten(sources) - transform = self.agent.skills_transform - if transform is not None: - # Copies prevent an extension from mutating our retained source state. - skills = transform([s.model_copy(deep=True) for s in skills], context) - if inspect.isawaitable(skills): - skills = await skills - self._apply(skills) - self.sources, self.issues = sources, issues - - def _apply(self, skills): - by_name = {} - records = [] - for skill in sorted(skills, key=lambda item: item.name): - if skill.name in by_name: - raise ValueError(f"Duplicate skill name: {skill.name}") - by_name[skill.name] = skill - record = skill.model_dump(mode="json") - if not skill.skill_space_id: - readme = Path(skill.path) / "SKILL.md" - record["content_digest"] = hashlib.sha256( - readme.read_bytes() - ).hexdigest() - records.append(record) - fingerprint = hashlib.sha256( - json.dumps(records, sort_keys=True, ensure_ascii=False).encode() - ).hexdigest() - if fingerprint == self.fingerprint: - return - description = self._describe(by_name) - # All potentially failing construction precedes publication. An existing - # toolset and prompt remain valid if a wrapper rejects the candidate. - toolset = SkillsToolset( - by_name, - self.agent.skills_mode, - tool_wrapper=self.agent.skill_tool_wrapper, - ) - instruction = self.base_instruction - if isinstance(instruction, str): - instruction = instruction + description - else: - base = instruction - - async def instruction(context): - value = base(context) - if inspect.isawaitable(value): - value = await value - return value + description - - tools = [tool for tool in self.agent.tools if tool is not self.toolset] - self.agent.instruction = instruction - # Preserve the dictionary identity captured by checklist callbacks. - self.agent._skills_with_checklist.clear() - self.agent._skills_with_checklist.update(by_name) - self.agent.skills_dict = by_name - self.agent.tools = [*tools, toolset] - self.toolset = toolset - self.description = description - self.fingerprint = fingerprint - - @staticmethod - def _describe(skills): - if not skills: - return "" - lines = ["\nYou have the following skills:\n"] - for skill in skills.values(): - lines.append( - f"- name: {skill.name}\n- description: {skill.description}\n\n" - ) - if any(skill.checklist for skill in skills.values()): - lines.append( - "Use `update_check_list` to mark completed skill checklist items.\n" - ) - lines.append( - "Use `skills_tool` to load skill instructions for the current turn. " - "Instructions returned in earlier turns may be outdated.\n" - ) - return "".join(lines) - - def status(self): - return { - "ready": self.fingerprint is not None, - "issues": list(self.issues), - "loaded_skills": [ - {"name": skill.name, "id": skill.id, "version": skill.version_id} - for skill in self.agent.skills_dict.values() - ], - } diff --git a/veadk/tools/skills_tools/skills_tool.py b/veadk/tools/skills_tools/skills_tool.py index 512cda84f..609b77bcb 100644 --- a/veadk/tools/skills_tools/skills_tool.py +++ b/veadk/tools/skills_tools/skills_tool.py @@ -261,24 +261,42 @@ def _invoke_skill(self, skill_name: str, tool_context: ToolContext) -> str: return f"Error: Failed to install skill '{skill_name}': {exc}" else: - # 3. Use the local skill - # Create symlink to skills directory - skills_mount = Path(skill.path) + # Refresh an existing link when a configured local source moves. + skills_mount = Path(skill.path).resolve() skills_link = skill_dir / skill_name - if skills_mount.exists() and not skills_link.exists(): - try: - skills_link.symlink_to(skills_mount) - logger.debug( - f"Created symlink: {skills_link} -> {skills_mount}" + try: + if Path(skill_name).name != skill_name or skill_name in {".", ".."}: + raise ValueError("Skill name must be a single directory name") + if not skills_mount.is_dir(): + raise FileNotFoundError( + "Configured local skill directory is missing" ) - except FileExistsError: - # Symlink already exists (race condition from concurrent session setup) - pass - except Exception as e: - # Log but don't fail - skills can still be accessed via absolute path - logger.warning( - f"Failed to create skills symlink for {str(skills_mount)}: {e}" + if skills_link.exists() and not skills_link.is_symlink(): + raise FileExistsError( + "Skill destination is not a managed symlink" ) + if ( + not skills_link.is_symlink() + or skills_link.resolve() != skills_mount + ): + with tempfile.TemporaryDirectory( + prefix=".skill-link-", dir=working_dir + ) as temp: + staging = Path(temp) / "link" + staging.symlink_to(skills_mount) + os.replace(staging, skills_link) + logger.info( + "Updated local skill link: %s -> %s", + skills_link, + skills_mount, + ) + except Exception as exc: + logger.warning( + "Failed to link local skill '%s': %s", + skill_name, + type(exc).__name__, + ) + return f"Error: Failed to link local skill '{skill_name}': {exc}" skill_file = self._find_skill_file(skill_dir, skill_name) if not skill_file.exists(): diff --git a/veadk/tools/skills_tools/skills_toolset.py b/veadk/tools/skills_tools/skills_toolset.py index 7d5c0f6d1..ea53768f6 100644 --- a/veadk/tools/skills_tools/skills_toolset.py +++ b/veadk/tools/skills_tools/skills_toolset.py @@ -14,7 +14,7 @@ from __future__ import annotations -from typing import Callable, Dict, List, Optional +from typing import Dict, List, Optional try: from typing_extensions import override @@ -60,8 +60,6 @@ def __init__( self, skills: Dict[str, Skill], skills_mode: str, - *, - tool_wrapper: Optional[Callable[[BaseTool], BaseTool]] = None, ) -> None: """Initialize the skills toolset. @@ -73,7 +71,12 @@ def __init__( self.skills_mode = skills_mode - self._tools = { + self._refresh_state = None + self._tools = self.build_tools(skills) + + def build_tools(self, skills): + """Build a candidate tool collection without changing the active one.""" + tools = { "skills": SkillsTool(skills), "read_file": FunctionTool(read_file_tool), "write_file": FunctionTool(write_file_tool), @@ -83,8 +86,31 @@ def __init__( "update_check_list": FunctionTool(update_check_list), } - if tool_wrapper is not None: - self._tools = {key: tool_wrapper(tool) for key, tool in self._tools.items()} + return {key: self.wrap_tool(tool) for key, tool in tools.items()} + + async def prepare_skills(self, skills, callback_context): + """Adapt loaded results before publishing the refreshed skill view.""" + return skills + + def on_source_error(self, source, error, previous): + """Retain the last successful result of a configured, failing source.""" + return previous + + def status(self): + state = self._refresh_state + skills = self._tools["skills"].skills if state is not None else {} + return { + "ready": state is not None and state.instruction is not None, + "issues": list(state.issues) if state is not None else [], + "loaded_skills": [ + {"name": skill.name, "id": skill.id, "version": skill.version_id} + for skill in skills.values() + ], + } + + def wrap_tool(self, tool: BaseTool) -> BaseTool: + """Override to instrument tools while preserving their declarations.""" + return tool @override async def get_tools( From eebb2e0da36637016888e806e5033837e2e27ec0 Mon Sep 17 00:00:00 2001 From: "hanzhi.421" Date: Wed, 9 Sep 2026 10:46:36 +0800 Subject: [PATCH 5/5] docs(skills): remove standalone dynamic skills notes --- veadk/skills/DYNAMIC_SKILLS.md | 54 ---------------------------------- 1 file changed, 54 deletions(-) delete mode 100644 veadk/skills/DYNAMIC_SKILLS.md diff --git a/veadk/skills/DYNAMIC_SKILLS.md b/veadk/skills/DYNAMIC_SKILLS.md deleted file mode 100644 index 1cda00eb8..000000000 --- a/veadk/skills/DYNAMIC_SKILLS.md +++ /dev/null @@ -1,54 +0,0 @@ -# Dynamic legacy skills - -The existing `Agent(skills=[local_directory, space_id], skills_mode="local", -enable_dynamic_load_skills=True)` option registers `check_skills` as a before-agent -callback. Its default remains false. It works without a sandbox subclass. - -Each callback reads the current `Agent.skills`, including replaced directories, -added/removed spaces, and an empty list. Comma-separated spaces use the existing -provider routing. Provider versions remain authoritative; there is no additional -client-side latest-version policy. - -State is held by the existing SkillsToolset instance, with its baseline recorded -during initialization. Changes include local SKILL.md content, source identity, -package path, bucket, version and metadata. A path-only change replaces execution -tools without rewriting an otherwise identical skills prompt. Stable name ordering -avoids prompt churn when a provider reorders results. The base instruction and -other callback text are preserved, including callable instructions. - -A failed source retains its previous successful result; successful empty responses -remove it. Sources removed from configuration cannot reappear from retained state. -`SkillsToolset.status()` reports safe error types and loaded names/IDs/versions. -The next successful refresh clears failures. Tool construction finishes before the -new skill dictionary, checklist view and tools are published. - -Applications can subclass SkillsToolset's named `prepare_skills`, `wrap_tool`, -and `on_source_error` methods to adapt results and instrumentation. Rebuilding tools -uses the same instance, so instrumentation is preserved. There are no new Agent -callable parameters, execution overrides, or runtime binding/discovery components. - -This callback does **not** lock the full execution of a shared Agent. Applications -that run a mutable Agent concurrently must coordinate that execution themselves. -The Playground sandbox does so in its own SkillSandboxAgent. Realtime/live flows, -arbitrary local script writes, and remote objects overwritten without any locator -or metadata change are not snapshotted. Prompt stability is not a guarantee of -provider-side prefix-cache hits, and earlier conversation messages are not rewritten. - -## Downloaded archive layout - -Space and SkillHub archives are downloaded and extracted in a temporary directory. -Both a root `SKILL.md` and a wrapping directory containing `SKILL.md` install as -`skills//SKILL.md`, with resources alongside it. The root file takes -precedence; otherwise the shallowest candidate wins, with path order breaking -ties. Multiple candidates and deeper layouts produce a warning rather than a -rejection. Only the selected skill root and its descendants are installed. - -Extraction and UTF-8 readability checks finish before an existing installation -is moved aside. A failed replacement restores the old installation. ZIPs and -staging files are cleaned on success and failure; successful installs also remove -the legacy `skills/.zip`. Layout selection, destination, fallback, -success, and failure are logged. A failed rollback retains a backup and logs its -location. This does not provide cross-process atomicity or crash recovery. - -Skill names remain the installation key; this does not change duplicate-name -selection or remove files spilled into the shared skills directory by old versions.