"""Regression coverage for the opt-in online-skill metadata cache.""" from __future__ import annotations import concurrent.futures import json import logging import os import subprocess import sys import threading import time from pathlib import Path import pytest from deerflow.config import paths as paths_module from deerflow.sandbox import skill_view from deerflow.skills.storage.local_skill_storage import LocalSkillStorage def _skill(root: Path, directory: str, name: str | None = None) -> Path: location = root / "public" / directory location.mkdir(parents=True, exist_ok=True) (location / "SKILL.md").write_text(f"---\nname: {name or directory}\ndescription: cache test\n---\nSkill body\n", encoding="utf-8") return location def _wait_for_warm_validation_followers(expected: int, timeout: float = 5) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: with skill_view._WARM_VALIDATION_LOCK: if any(flight.followers >= expected for flight in skill_view._WARM_VALIDATION_FLIGHTS.values()): return time.sleep(0.005) raise AssertionError(f"warm validation did not collect {expected} followers") @pytest.fixture def cache_setup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): root = tmp_path / "skills" _skill(root, "online") config = tmp_path / "extensions.json" config.write_text(json.dumps({"skills": {"online": {"enabled": True}}}), encoding="utf-8") monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config)) monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path / "runtime")) monkeypatch.setattr(paths_module, "_paths", None) monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: True) skill_view.reset_skill_sandbox_view_cache() yield root, config, LocalSkillStorage(host_path=str(root)) skill_view.reset_skill_sandbox_view_cache() def test_warm_cache_does_not_walk_parse_copy_or_take_refresh_lock(cache_setup, monkeypatch): _root, _config, storage = cache_setup first = skill_view.refresh_skill_sandbox_view(storage=storage) def unexpected(*args, **kwargs): raise AssertionError("warm cache executed an expensive refresh operation") monkeypatch.setattr(storage, "load_skills", unexpected) monkeypatch.setattr(skill_view, "_scan_tree", unexpected) monkeypatch.setattr(skill_view, "_copy_file_atomic", unexpected) monkeypatch.setattr(skill_view, "_view_lock", unexpected) assert skill_view.refresh_skill_sandbox_view(storage=storage) is first assert first.enabled_roots == (("online", "public", "online"),) def test_offline_and_invalid_config_invalidate_immediately(cache_setup): root, config, storage = cache_setup first = skill_view.refresh_skill_sandbox_view(storage=storage) config.write_text(json.dumps({"skills": {"online": {"enabled": False}}}), encoding="utf-8") offline = skill_view.refresh_skill_sandbox_view(storage=storage) assert not offline.enabled_names assert not (offline.root / "public" / "online").exists() assert (root / "public" / "online" / "SKILL.md").exists() config.write_text("{", encoding="utf-8") with pytest.raises(ValueError, match="not valid JSON"): skill_view.refresh_skill_sandbox_view(storage=storage) assert not (first.root / skill_view.SKILL_VIEW_MARKER).exists() assert not list(first.root.iterdir()) assert not skill_view._VIEW_CACHE def test_external_source_change_same_size_preserved_mtime_and_new_files(cache_setup): root, _config, storage = cache_setup source = root / "public" / "online" / "helper.py" source.write_text("VALUE=1\n", encoding="utf-8") first = skill_view.refresh_skill_sandbox_view(storage=storage) before = source.stat() source.write_text("VALUE=2\n", encoding="utf-8") os.utime(source, ns=(before.st_atime_ns, before.st_mtime_ns)) updated = skill_view.refresh_skill_sandbox_view(storage=storage) assert updated.generation != first.generation assert (updated.root / "public" / "online" / "helper.py").read_text() == "VALUE=2\n" (source.parent / "new-reference.md").write_text("fresh", encoding="utf-8") new_file_view = skill_view.refresh_skill_sandbox_view(storage=storage) assert (new_file_view.root / "public" / "online" / "new-reference.md").read_text() == "fresh" source.unlink() deleted = skill_view.refresh_skill_sandbox_view(storage=storage) assert not (deleted.root / "public" / "online" / "helper.py").exists() def test_nested_new_skill_and_support_manifest_are_detected(cache_setup): root, _config, storage = cache_setup skill_view.refresh_skill_sandbox_view(storage=storage) _skill(root, "nested/new", "new-skill") support = root / "public" / "shared-lib" support.mkdir() (support / "helper.py").write_text("value", encoding="utf-8") manifest = support / skill_view.SKILL_SUPPORT_MANIFEST manifest.write_text(json.dumps({"required_by": ["online"]}), encoding="utf-8") changed = skill_view.refresh_skill_sandbox_view(storage=storage) assert ("new-skill", "public", "nested/new") in changed.enabled_roots assert (changed.root / "public" / "shared-lib" / "helper.py").exists() manifest.write_text(json.dumps({"required_by": ["not-online"]}), encoding="utf-8") removed = skill_view.refresh_skill_sandbox_view(storage=storage) assert not (removed.root / "public" / "shared-lib").exists() def test_destination_tampering_is_repaired_even_with_preserved_size_mtime(cache_setup): root, _config, storage = cache_setup source = root / "public" / "online" / "helper.py" source.write_text("VALUE=1\n", encoding="utf-8") view = skill_view.refresh_skill_sandbox_view(storage=storage) destination = view.root / "public" / "online" / "helper.py" before = destination.stat() destination.write_text("VALUE=9\n", encoding="utf-8") os.utime(destination, ns=(before.st_atime_ns, before.st_mtime_ns)) repaired = skill_view.refresh_skill_sandbox_view(storage=storage) assert (repaired.root / "public" / "online" / "helper.py").read_text() == "VALUE=1\n" rogue = view.root / "public" / "unknown" rogue.mkdir() (rogue / "secret").write_text("unexpected", encoding="utf-8") skill_view.refresh_skill_sandbox_view(storage=storage) assert not rogue.exists() def test_second_process_cache_and_force_rebuild(cache_setup, monkeypatch): _root, _config, storage = cache_setup first = skill_view.refresh_skill_sandbox_view(storage=storage) marker_metadata = (first.root / skill_view.SKILL_VIEW_MARKER).stat() skill_view.reset_skill_sandbox_view_cache() original_load = storage.load_skills count = 0 def counted(**kwargs): nonlocal count count += 1 return original_load(**kwargs) monkeypatch.setattr(storage, "load_skills", counted) second = skill_view.refresh_skill_sandbox_view(storage=storage) assert second == first assert count == 0 assert (first.root / skill_view.SKILL_VIEW_MARKER).stat().st_mtime_ns == marker_metadata.st_mtime_ns child = subprocess.run( [ sys.executable, "-c", """ import sys from deerflow.sandbox import skill_view from deerflow.skills.storage.local_skill_storage import LocalSkillStorage skill_view._skill_view_cache_enabled = lambda: True storage = LocalSkillStorage(host_path=sys.argv[1]) def unexpected(**kwargs): raise AssertionError('another process unnecessarily rediscovered skills') storage.load_skills = unexpected view = skill_view.refresh_skill_sandbox_view(storage=storage) print(view.generation) """, str(_root), ], capture_output=True, text=True, timeout=20, check=True, ) assert child.stdout.strip() == first.generation skill_view.refresh_skill_sandbox_view(storage=storage, force=True) assert count == 1 def test_concurrent_cold_calls_join_one_refresh(cache_setup, monkeypatch, caplog): _root, _config, storage = cache_setup original = storage.load_skills count = 0 barrier = threading.Barrier(8) def counted(**kwargs): nonlocal count count += 1 return original(**kwargs) def call(): barrier.wait(timeout=5) return skill_view.refresh_skill_sandbox_view(storage=storage) monkeypatch.setattr(storage, "load_skills", counted) with caplog.at_level(logging.INFO, logger=skill_view.__name__), concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda _: call(), range(8))) assert count == 1 assert len({result.generation for result in results}) == 1 assert "lock_wait_ms=" in caplog.text assert "refresh_ms=" in caplog.text assert "source_entries=" in caplog.text assert "Skill body" not in caplog.text def test_concurrent_warm_calls_share_only_the_overlapping_validation(cache_setup, monkeypatch): _root, _config, storage = cache_setup first = skill_view.refresh_skill_sandbox_view(storage=storage) original = skill_view._entry_matches entered = threading.Event() release = threading.Event() count = 0 count_lock = threading.Lock() def blocked(entry, skills_root, config_key): nonlocal count with count_lock: count += 1 entered.set() assert release.wait(timeout=5) return original(entry, skills_root, config_key) monkeypatch.setattr(skill_view, "_entry_matches", blocked) with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: leader = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) assert entered.wait(timeout=5) followers = [pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) for _ in range(7)] _wait_for_warm_validation_followers(7) release.set() results = [leader.result(timeout=5), *(future.result(timeout=5) for future in followers)] assert count == 1 assert all(result is first for result in results) assert not skill_view._WARM_VALIDATION_FLIGHTS # There is no TTL/result cache: a caller after the completed batch validates again. assert skill_view.refresh_skill_sandbox_view(storage=storage) is first assert count == 2 def test_warm_validation_failure_reaches_followers_and_next_call_recovers(cache_setup, monkeypatch): _root, _config, storage = cache_setup first = skill_view.refresh_skill_sandbox_view(storage=storage) original = skill_view._entry_matches entered = threading.Event() release = threading.Event() count = 0 def broken(*_args): nonlocal count count += 1 entered.set() assert release.wait(timeout=5) raise RuntimeError("simulated warm validation failure") monkeypatch.setattr(skill_view, "_entry_matches", broken) with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool: leader = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) assert entered.wait(timeout=5) followers = [pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) for _ in range(5)] _wait_for_warm_validation_followers(5) release.set() for future in [leader, *followers]: with pytest.raises(RuntimeError, match="simulated warm validation failure"): future.result(timeout=5) assert count == 1 assert not skill_view._WARM_VALIDATION_FLIGHTS monkeypatch.setattr(skill_view, "_entry_matches", original) assert skill_view.refresh_skill_sandbox_view(storage=storage) is first def test_warm_validation_with_new_config_hash_does_not_join_old_flight(cache_setup, monkeypatch): _root, config, storage = cache_setup skill_view.refresh_skill_sandbox_view(storage=storage) original = skill_view._entry_matches old_key = skill_view._config_key() old_entered = threading.Event() release_old = threading.Event() observed_keys: list[str] = [] observed_lock = threading.Lock() def block_old_config(entry, skills_root, config_key): with observed_lock: observed_keys.append(config_key) if config_key == old_key: old_entered.set() assert release_old.wait(timeout=5) return original(entry, skills_root, config_key) monkeypatch.setattr(skill_view, "_entry_matches", block_old_config) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: old_call = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) assert old_entered.wait(timeout=5) config.write_text(json.dumps({"skills": {"online": {"enabled": False}}}), encoding="utf-8") new_key = skill_view._config_key() assert new_key != old_key new_call = pool.submit(skill_view.refresh_skill_sandbox_view, storage=storage) try: # This must finish while the old-hash leader is still blocked. changed = new_call.result(timeout=5) finally: release_old.set() joined_after_change = old_call.result(timeout=5) assert not changed.enabled_names assert not joined_after_change.enabled_names assert observed_keys.count(old_key) == 1 assert new_key in observed_keys assert not skill_view._WARM_VALIDATION_FLIGHTS def test_disabled_switch_retains_full_refresh_and_legacy_marker(cache_setup, monkeypatch): _root, _config, storage = cache_setup monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: False) original = storage.load_skills count = 0 def counted(**kwargs): nonlocal count count += 1 return original(**kwargs) monkeypatch.setattr(storage, "load_skills", counted) first = skill_view.refresh_skill_sandbox_view(storage=storage) skill_view.refresh_skill_sandbox_view(storage=storage) assert count == 2 assert not first.generation assert "generation" not in json.loads(first.marker_content) assert first.enabled_roots == (("online", "public", "online"),) def test_custom_storage_instances_do_not_share_filesystem_cache(cache_setup): root, _config, _storage = cache_setup _skill(root, "other") class FilteredStorage(LocalSkillStorage): def __init__(self, selected): super().__init__(host_path=str(root)) self.selected = selected def load_skills(self, **kwargs): return [skill for skill in super().load_skills(**kwargs) if skill.name == self.selected] first = skill_view.refresh_skill_sandbox_view(storage=FilteredStorage("online")) second = skill_view.refresh_skill_sandbox_view(storage=FilteredStorage("other")) assert first.enabled_names == {"online"} assert second.enabled_names == {"other"} assert not first.generation and not second.generation def test_config_change_mid_refresh_is_not_published(cache_setup, monkeypatch): _root, config, storage = cache_setup original = skill_view._copy_file_atomic changed = False def copy_and_disable(source, destination): nonlocal changed original(source, destination) if not changed: changed = True config.write_text(json.dumps({"skills": {"online": {"enabled": False}}}), encoding="utf-8") monkeypatch.setattr(skill_view, "_copy_file_atomic", copy_and_disable) view = skill_view.refresh_skill_sandbox_view(storage=storage) assert not view.enabled_names assert not (view.root / "public" / "online").exists() def test_switching_cache_off_and_back_on_does_not_reuse_a_stale_marker(cache_setup, monkeypatch): root, _config, storage = cache_setup original = storage.load_skills count = 0 def counted(**kwargs): nonlocal count count += 1 return original(**kwargs) monkeypatch.setattr(storage, "load_skills", counted) first = skill_view.refresh_skill_sandbox_view(storage=storage) monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: False) (root / "public" / "online" / "reference.md").write_text("new content", encoding="utf-8") legacy = skill_view.refresh_skill_sandbox_view(storage=storage) skill_view.refresh_skill_sandbox_view(storage=storage) assert count == 3 assert not legacy.generation assert "generation" not in json.loads(legacy.marker_content) monkeypatch.setattr(skill_view, "_skill_view_cache_enabled", lambda: True) restored = skill_view.refresh_skill_sandbox_view(storage=storage) assert restored.generation != first.generation assert (restored.root / "public" / "online" / "reference.md").read_text() == "new content" assert count == 4 assert skill_view.refresh_skill_sandbox_view(storage=storage) is restored assert count == 4 def test_marker_damage_and_copy_failure_clear_capability(cache_setup, monkeypatch): root, _config, storage = cache_setup view = skill_view.refresh_skill_sandbox_view(storage=storage) marker = view.root / skill_view.SKILL_VIEW_MARKER marker.write_text("not-a-valid-marker", encoding="utf-8") repaired = skill_view.refresh_skill_sandbox_view(storage=storage) assert marker.read_text() == repaired.marker_content (root / "public" / "online" / "SKILL.md").write_text("---\nname: online\ndescription: changed\n---\n", encoding="utf-8") def broken_copy(*args): raise OSError("simulated disk failure") monkeypatch.setattr(skill_view, "_copy_file_atomic", broken_copy) with pytest.raises(OSError, match="simulated disk failure"): skill_view.refresh_skill_sandbox_view(storage=storage) assert not marker.exists() assert not skill_view._VIEW_CACHE assert not list(view.root.iterdir()) def test_symlink_destination_does_not_modify_external_directory(cache_setup, tmp_path): _root, _config, storage = cache_setup view = skill_view.refresh_skill_sandbox_view(storage=storage) external = tmp_path / "outside" external.mkdir() (external / "keep").write_text("untouched", encoding="utf-8") target = view.root / "public" / "online" import shutil shutil.rmtree(target) try: target.symlink_to(external, target_is_directory=True) except OSError: if os.name != "nt": pytest.skip("directory symlinks are unavailable on this platform") junction = subprocess.run(["cmd", "/c", "mklink", "/J", str(target), str(external)], capture_output=True, timeout=10) if junction.returncode: pytest.skip("directory links are unavailable on this platform") skill_view.refresh_skill_sandbox_view(storage=storage) assert not skill_view._is_view_link(target) assert (external / "keep").read_text() == "untouched" assert not (external / "SKILL.md").exists()