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_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/agent.py b/veadk/agent.py index 84cf765c2..e6bf66c01 100644 --- a/veadk/agent.py +++ b/veadk/agent.py @@ -410,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: + if self.skills or self.enable_dynamic_load_skills: self.load_skills() if self.enable_skills_checklist: logger.info("Skills checklist enabled") @@ -510,17 +510,7 @@ 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, - 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: @@ -609,58 +599,7 @@ 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: if self.before_agent_callback: 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/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_tool.py b/veadk/tools/skills_tools/skills_tool.py index 41acffe9e..609b77bcb 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,126 +230,73 @@ 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 - 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(): @@ -368,6 +317,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" diff --git a/veadk/tools/skills_tools/skills_toolset.py b/veadk/tools/skills_tools/skills_toolset.py index c7507ff5d..ea53768f6 100644 --- a/veadk/tools/skills_tools/skills_toolset.py +++ b/veadk/tools/skills_tools/skills_toolset.py @@ -56,7 +56,11 @@ 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, + ) -> None: """Initialize the skills toolset. Args: @@ -67,7 +71,12 @@ def __init__(self, skills: Dict[str, Skill], skills_mode: str) -> None: 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), @@ -77,6 +86,32 @@ def __init__(self, skills: Dict[str, Skill], skills_mode: str) -> None: "update_check_list": FunctionTool(update_check_list), } + 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( self, readonly_context: Optional[ReadonlyContext] = None