Compare commits

...

1 Commits

Author SHA1 Message Date
HAL9000 2f9be3bfb8 feat(skills): implement multi-scope agent skill discovery for global, project, and local tiers
CI / lint (pull_request) Failing after 36s
CI / typecheck (pull_request) Successful in 1m14s
CI / quality (pull_request) Successful in 1m0s
CI / push-validation (pull_request) Successful in 38s
CI / helm (pull_request) Successful in 42s
CI / build (pull_request) Successful in 55s
CI / security (pull_request) Successful in 1m22s
CI / unit_tests (pull_request) Failing after 1m20s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / integration_tests (pull_request) Successful in 9m43s
CI / status-check (pull_request) Failing after 3s
Implements AgentSkillDiscovery class supporting three-tier progressive disclosure model:
- Global scope: organization-level skill directories (lowest precedence)
- Project scope: project-level skill directories (intermediate precedence)
- Local scope: working-directory skill directories (highest precedence)

Key changes:
- Added agent_skill_discovery.py with AgentSkillDiscovery, Scope, ScopeOrder, DiscoveredScopedSkill, MultiScopeDiscoveryResult, ScopedDiscoveryConflict types
- Updated skills module __init__.py to export new discovery classes
- Added BDD test file agent_skills_multi_scope_discovery.feature covering all scope scenarios
- Added step definitions agent_skills_multi_scope_discovery_steps.py with 30+ scenarios
- Updated CHANGELOG.md under [Unreleased] section
- Updated CONTRIBUTORS.md

Name collision resolution: local > project > global (lower ScopeOrder value wins)

ISSUES CLOSED: #9369
2026-06-10 03:47:00 -04:00
6 changed files with 1600 additions and 0 deletions
+14
View File
@@ -217,6 +217,20 @@ ensuring data is stored with proper parameter values.
(8 scenarios across all three strategies), and Robot Framework integration tests
verifying runtime behavior against live Python modules.
### Added
- **Multi-scope agent skill discovery for global, project, and local tiers** (#9369 / #9454 ): Implemented
``AgentSkillDiscovery`` class with a three-tier progressive disclosure model.
Skills are discovered from organization-level (global), project-level, and
local working-directory directories. Name collisions across scopes are resolved
with clear precedence: ``local > project > global``. Includes
``Scope``, ``ScopeOrder``, ``DiscoveredScopedSkill``,
``MultiScopeDiscoveryResult``, ``ScopedDiscoveryConflict`` types; the methods
``discover()`` (full multi-scope scan), ``discover_scoped(scope)`` (single
tier), and ``get_all_skill_names()`` (pre-resolution inspection); two conflict
policies (``keep_precedence`` default, ``report_only``); and comprehensive BDD
test coverage in ``features/agent_skills_multi_scope_discovery.feature``.
### Fixed
- **fileConfig error handling in alembic env.py** (#7874): Wrapped the `fileConfig()`
+1
View File
@@ -94,3 +94,4 @@ Below are some specific details of individual PR contributions.
* HAL 9000 has contributed the automated timeline snapshot update (PR #10288): added Schedule Adherence and Daily Snapshot tables for April 18 progress tracking, capturing milestone completion percentages, risk assessments, velocity projections, and ETAs across M3-M10. Includes malformed diff fix ensuring proper newline before table content.
* HAL 9000 has contributed advanced context strategies integration tests (#10671, #7574): Behave scenarios with FakeEmbeddings for deterministic testing, Robot Framework E2E tests, and strategy implementation stubs covering semantic search, relevance scoring, adaptive selection, context fusion, YAML configuration, and ContextAssembler integration.
* HAL 9000 has contributed the resource and skill management showcase alignment (#4213): updated the CLI tools showcase with consistent counts, explicit save instructions, metadata callouts, and README framing for platform walkthroughs; removed obsolete tdd_issue tags from coverage threshold Robot tests; hardened the Skip If No LLM Keys E2E helper with per-key regex validation and log suppression to prevent credential leakage.
* HAL 9000 has contributed the multi-scope agent skill discovery implementation (PR #9454 / issue #9369): implemented ``AgentSkillDiscovery`` class with three-tier progressive disclosure model supporting global, project, and local scope directories, name collision resolution with precedence rules, comprehensive BDD test coverage, and integration with the existing discovery module.
@@ -0,0 +1,357 @@
@phase2 @skills @agent_skills @agent_skills_discovery @multi_scope
Feature: Agent Skills Multi-Scope Discovery
As a CleverAgents developer
I want to load Agent Skills from three scope tiers (global, project, local)
So that skills are discovered with correct precedence when the same name exists in multiple scopes
Background:
Given a temporary directory is prepared for skill testing
# ---------------------------------------------------------------------------
# Global Scope Discovery
# ---------------------------------------------------------------------------
@agent_skill_discovery_global
Scenario: AgentSkillDiscovery discovers skills from global scope only
Given an agent skills folder with valid SKILL.md "global-skill" in the global scope:
"""
---
name: local/global-task
description: A skill discovered at the global scope.
---
Global instructions.
"""
When I create an AgentSkillDiscovery with only global dirs and discover skills
Then the global scope should have 1 discovered skill(s)
And the project scope should have 0 discovered skill(s)
And the local scope should have 0 discovered skill(s)
And the discovered skill name should be "local/global-task"
# ---------------------------------------------------------------------------
# Project Scope Discovery
# ---------------------------------------------------------------------------
@agent_skill_discovery_project
Scenario: AgentSkillDiscovery discovers skills from project scope only
Given an agent skills folder with valid SKILL.md "project-skill" in the project scope:
"""
---
name: local/project-task
description: A skill discovered at the project scope.
---
Project instructions.
"""
When I create an AgentSkillDiscovery with only project dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the project scope should have 1 discovered skill(s)
And the local scope should have 0 discovered skill(s)
And the discovered skill name should be "local/project-task"
# ---------------------------------------------------------------------------
# Local Scope Discovery
# ---------------------------------------------------------------------------
@agent_skill_discovery_local
Scenario: AgentSkillDiscovery discovers skills from local scope only
Given an agent skills folder with valid SKILL.md "local-skill" in the local scope:
"""
---
name: local/local-task
description: A skill discovered at the local scope.
---
Local instructions.
"""
When I create an AgentSkillDiscovery with only local dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the project scope should have 0 discovered skill(s)
And the local scope should have 1 discovered skill(s)
And the discovered skill name should be "local/local-task"
# ---------------------------------------------------------------------------
# Combined Discovery (all three scopes)
# ---------------------------------------------------------------------------
@agent_skill_discovery_combined
Scenario: AgentSkillDiscovery discovers skills from all three scopes combined
Given an agent skills folder with valid SKILL.md "global-combo" in the global scope:
"""
---
name: local/global-combo
description: A global scoped skill.
---
Global instructions.
"""
And an agent skills folder with valid SKILL.md "project-combo" in the project scope:
"""
---
name: local/project-combo
description: A project scoped skill.
---
Project instructions.
"""
And an agent skills folder with valid SKILL.md "local-combo" in the local scope:
"""
---
name: local/local-combo
description: A local scoped skill.
---
Local instructions.
"""
When I create an AgentSkillDiscovery with all three dirs and discover skills
Then the global scope should have 1 discovered skill(s)
And the project scope should have 1 discovered skill(s)
And the local scope should have 1 discovered skill(s)
And the total discovered skill count should be 3
# ---------------------------------------------------------------------------
# Name Collision Precedence — Local wins over Project
# ---------------------------------------------------------------------------
@agent_skill_collision_local_overrides_project
Scenario: Local scope overrides project scope on name collision
Given an agent skills folder with valid SKILL.md "shared-name" in the project scope:
"""
---
name: local/shared-name
description: The project-level version.
---
Project instructions should be overridden.
"""
And an agent skills folder with valid SKILL.md "shared-name" in the local scope:
"""
---
name: local/shared-name
description: The local version highest precedence.
---
Local instructions this is kept after collision resolution.
"""
When I create an AgentSkillDiscovery with project and local dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the project scope should have 0 discovered skill(s)
And the local scope should have 1 discovered skill(s)
And the discovery collision count should be 1
And the winning scope of collisions should be "local"
And the losing scope of collisions should be "project"
# ---------------------------------------------------------------------------
# Name Collision Precedence — Project wins over Global
# ---------------------------------------------------------------------------
@agent_skill_collision_project_overrides_global
Scenario: Project scope overrides global scope on name collision
Given an agent skills folder with valid SKILL.md "shared-name-2" in the global scope:
"""
---
name: local/shared-two
description: The global version lowest precedence.
---
Global instructions should be overridden.
"""
And an agent skills folder with valid SKILL.md "shared-name-2" in the project scope:
"""
---
name: local/shared-two
description: The project-level version wins over global.
---
Project instructions this is kept after collision resolution.
"""
When I create an AgentSkillDiscovery with global and project dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the project scope should have 1 discovered skill(s)
And the local scope should have 0 discovered skill(s)
And the discovery collision count should be 1
And the winning scope of collisions should be "project"
And the losing scope of collisions should be "global"
# ---------------------------------------------------------------------------
# Name Collision Precedence — Full three-tier cascade
# ---------------------------------------------------------------------------
@agent_skill_collision_full_cascade
Scenario: Three-tier name collision: local > project > global
Given an agent skills folder with valid SKILL.md "cascade-name" in the global scope:
"""
---
name: local/cascade
description: Global tier.
---
Global instructions.
"""
And an agent skills folder with valid SKILL.md "cascade-name" in the project scope:
"""
---
name: local/cascade
description: Project tier.
---
Project instructions.
"""
And an agent skills folder with valid SKILL.md "cascade-name" in the local scope:
"""
---
name: local/cascade
description: Local tier highest precedence.
---
Local instructions.
"""
When I create an AgentSkillDiscovery with all three dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the project scope should have 0 discovered skill(s)
And the local scope should have 1 discovered skill(s)
And the total discovered skill count should be 1
And the discovery collision count should be 2
# ---------------------------------------------------------------------------
# Scope isolation — non-existent directories handled gracefully
# ---------------------------------------------------------------------------
@agent_skill_discovery_nonexistent_scope
Scenario: Non-existent scope directory produces an error but does not crash
When I create an AgentSkillDiscovery with a non-existent global dir and discover skills
Then the errors list should contain "Directory does not exist"
And the error count should be at least 1
@agent_skill_discovery_nonexistent_scope
Scenario: All scopes missing yields empty results without crash
When I create an AgentSkillDiscovery with all non-existent dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the project scope should have 0 discovered skill(s)
And the local scope should have 0 discovered skill(s)
And the total discovered skill count should be 0
# ---------------------------------------------------------------------------
# Empty scope directories handled correctly
# ---------------------------------------------------------------------------
@agent_skill_discovery_empty_scope_dir
Scenario: Empty scope directory discovers zero skills without crash
Given an agent skills folder with valid SKILL.md "empty-scope" in the local scope:
"""
---
name: local/empty
description: A test skill.
---
Instructions.
"""
And an empty directory is used for the global scope
When I create an AgentSkillDiscovery with all three dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the local scope should have at least 1 discovered skill(s)
# ---------------------------------------------------------------------------
# Single-scope discovery (discover_scoped method)
# ---------------------------------------------------------------------------
@agent_skill_discovery_scoped_method
Scenario: discover_scoped for global scope returns only global skills
Given an agent skills folder with valid SKILL.md "solo-global" in the global scope:
"""
---
name: local/solo-g
description: Only in global.
---
Instructions.
"""
When I create an AgentSkillDiscovery and call discover_scoped on scope "global"
Then only the global scope should have 1 discovered skill(s)
And all other scopes should have 0 discovered skill(s)
@agent_skill_discovery_scoped_method
Scenario: discover_scoped for local scope returns only local skills
Given an agent skills folder with valid SKILL.md "solo-local" in the local scope:
"""
---
name: local/solo-l
description: Only in local.
---
Instructions.
"""
When I create an AgentSkillDiscovery and call discover_scoped on scope "local"
Then only the local scope should have 1 discovered skill(s)
And all other scopes should have 0 discovered skill(s)
# ---------------------------------------------------------------------------
# Report-only conflict strategy (no resolution — report all)
# ---------------------------------------------------------------------------
@agent_skill_discovery_report_only
Scenario: report_only conflict strategy keeps ALL scope discoveries
Given an agent skills folder with valid SKILL.md "report-name" in the global scope:
"""
---
name: local/report-test
description: Global version kept in report-only mode.
---
Instructions.
"""
And an agent skills folder with valid SKILL.md "report-name" in the local scope:
"""
---
name: local/report-test
description: Local version also kept in report-only mode.
---
Instructions.
"""
When I create an AgentSkillDiscovery with report_only conflict and discover skills
Then the global scope should have 1 discovered skill(s)
And the local scope should have 1 discovered skill(s)
Then the discovery collision count should be 0
# ---------------------------------------------------------------------------
# On_conflict validation
# ---------------------------------------------------------------------------
@agent_skill_discovery_invalid_conflict_strategy
Scenario: Invalid on_conflict value raises ValueError
When I create an AgentSkillDiscovery with invalid conflict strategy expecting an error
Then the creation error should mention "on_conflict"
# ---------------------------------------------------------------------------
# Edge cases — branch and coverage completeness
# ---------------------------------------------------------------------------
@agent_skill_discovery_edge_cases @coverage
Scenario: AgentSkillDiscovery with None directories defaults to empty lists
When I create an AgentSkillDiscovery with all None dirs and discover skills
Then the global scope should have 0 discovered skill(s)
And the project scope should have 0 discovered skill(s)
And the local scope should have 0 discovered skill(s)
@agent_skill_discovery_edge_cases @coverage
Scenario: Multiple skills within the same scope are all discovered
Given an agent skills folder with valid SKILL.md "multi-skill-a" in the global scope:
"""
---
name: local/multi-a
description: First skill.
---
Instructions A.
"""
And an agent skills folder with valid SKILL.md "multi-skill-b" in the global scope:
"""
---
name: local/multi-b
description: Second skill.
---
Instructions B.
"""
When I create an AgentSkillDiscovery with only global dirs and discover skills
Then the global scope should have 2 discovered skill(s)
@agent_skill_discovery_edge_cases @coverage
Scenario: All-scope method get_all_skill_names returns non-resolution names
Given an agent skills folder with valid SKILL.md "names-global" in the global scope:
"""
---
name: local/names-g
description: Global for names test.
---
Instructions.
"""
When I create an AgentSkillDiscovery and get all skill names
Then the global skill names set should contain "local/names-g"
@agent_skill_discovery_edge_cases @coverage
Scenario: ScopeOrder gives local precedence of 1 (highest)
When I check that ScopeOrder.LOCAL has the lowest numeric value
Then ScopeOrder.LOCAL should be less than ScopeOrder.PROJECT
And ScopeOrder.PROJECT should be less than ScopeOrder.GLOBAL
@@ -0,0 +1,705 @@
"""Step definitions for Agent Skills Multi-Scope Discovery Behave tests.
Covers global/project/local scope discovery, name collision resolution
with precedence rules, report-only strategy, and edge cases.
"""
from __future__ import annotations
import textwrap
import tempfile
from pathlib import Path
from behave import given, then, when
from behave.runner import Context
from cleveragents.skills.agent_skill_discovery import (
AgentSkillDiscovery,
DiscoveredScopedSkill,
MultiScopeDiscoveryResult,
Scope,
ScopedDiscoveryConflict,
ScopeOrder,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _write_skill_md(folder: Path, content: str) -> None:
"""Write dedented content to SKILL.md in *folder*."""
folder.mkdir(parents=True, exist_ok=True)
(folder / "SKILL.md").write_text(textwrap.dedent(content).lstrip())
def _make_skill_folder(
context: Context,
scope_name: str,
skill_subdir_name: str,
content: str | None = None,
empty: bool = False,
) -> Path:
"""Create a temporary scope directory with one (or zero) skill(s).
Returns the created path.
"""
tmp_dir = Path(tempfile.mkdtemp(prefix=f"agent_scope_{scope_name}_"))
context._cleanup_handlers.append(
lambda: __import__("shutil").rmtree(tmp_dir, ignore_errors=True)
)
if empty:
return tmp_dir
skill_folder = tmp_dir / skill_subdir_name
_write_skill_md(skill_folder, content)
return tmp_dir
def _set_up_scope(
context: Context,
key: str,
scope_name: str,
skill_subdir: str,
content: str | None = None,
empty: bool = False,
existing_path: Path | None = None,
) -> None:
"""Add a scoped directory to the context setup.
Tracks created paths in ``context._scope_dirs`` for cleanup.
"""
if existing_path is not None and empty == False:
path = existing_path
else:
path = _make_skill_folder(context, scope_name, skill_subdir, content, empty)
if not hasattr(context, "_scoped_dir_registry"):
context._scoped_dir_registry: dict[str, Path] = {}
context._scoped_dir_registry[key] = path
# ---------------------------------------------------------------------------
# Given — setup scenarios
# ---------------------------------------------------------------------------
@given("a temporary directory is prepared for skill testing")
def step_prepare_temp(context: Context) -> None:
"""Ensure we have a fresh temp dir and cleanup list."""
import tempfile
context._work_dir = Path(tempfile.mkdtemp(prefix="agent_test_"))
context._cleanup_handlers = [
lambda: __import__("shutil").rmtree(
context._work_dir, ignore_errors=True
)
]
@given("an agent skills folder with valid SKILL.md {name} in the {scope} scope:")
def step_agent_skill_folder_in_scope(context: Context, name: str, scope: str, text: str) -> None:
"""Create a scoped temp directory with one skill."""
_set_up_scope(
context,
key=f"{scope}_{name}",
scope_name=scope,
skill_subdir=name,
content=text,
)
@given("an empty directory is used for the {scope} scope")
def step_empty_directory_for_scope(context: Context, scope: str) -> None:
"""Create an empty temp dir for a scope."""
_set_up_scope(
context,
key=f"{scope}_empty",
scope_name=scope,
skill_subdir="",
content=None,
empty=True,
)
# ---------------------------------------------------------------------------
# Given — invalid creation setup
# ---------------------------------------------------------------------------
@given("an invalid on_conflict value '{value}'")
def step_invalid_conflict_value(context: Context, value: str) -> None:
"""Setup for testing invalid conflict strategy."""
context._invalid_conflict_value = value
# ---------------------------------------------------------------------------
# When — discovery actions
# ---------------------------------------------------------------------------
@when("I create an AgentSkillDiscovery with only global dirs and discover skills")
def step_create_global_and_discover(context: Context) -> None:
"""Build AgentSkillDiscovery from only the global scope.
Steps 13 are not provided as directories so they will be empty lists.
"""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._discovery_result = None
try:
discovery = AgentSkillDiscovery(
global_dirs=[registry.get("global_global-skill", Path("/tmp/nonexistent"))]
if "global" in str(registry) and registry and True # check first key
else [Path(str(list(registry.values())[0])) if registry else Path("/tmp/nonexistent")],
)
# More robust: auto-detect which keys exist in the registry and assign them
all_paths = list(registry.values()) if isinstance(registry, dict) else []
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower:
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
global_dirs=paths_by_scope["global"] or [Path("/tmp/zzz_nonexistent_global_")],
)
context._discovery_result = discovery.discover()
except Exception as exc:
context._discovery_errors = exc
@when("I create an AgentSkillDiscovery with only project dirs and discover skills")
def step_create_project_and_discover(context: Context) -> None:
"""Build AgentSkillDiscovery from only the project scope."""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._discovery_result = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower:
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
project_dirs=paths_by_scope["project"] or [Path("/tmp/zzz_nonexistent_project_")],
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with only local dirs and discover skills")
def step_create_local_and_discover(context: Context) -> None:
"""Build AgentSkillDiscovery from only the local scope."""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._discovery_result = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower:
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
local_dirs=paths_by_scope["local"] or [Path("/tmp/zzz_nonexistent_local_")],
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with all three dirs and discover skills")
def step_create_all_and_discover(context: Context) -> None:
"""Build AgentSkillDiscovery from local/project/global scopes."""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._discovery_result = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower and not key_lower.endswith("empty"):
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
# Include empty scope dirs if registered
for key, path in (registry or {}).items():
key_lower = key.lower()
if "empty" in key_lower and "global" in key_lower:
paths_by_scope["global"].append(path)
elif "empty" in key_lower and "project" in key_lower:
paths_by_scope["project"].append(path)
elif "empty" in key_lower and "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
global_dirs=paths_by_scope["global"] or [Path("/tmp/zzz_nonexistent_global_")],
project_dirs=paths_by_scope["project"] or [Path("/tmp/zzz_nonexistent_project_")],
local_dirs=paths_by_scope["local"] or [Path("/tmp/zzz_nonexistent_local_")],
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with project and local dirs and discover skills")
def step_create_project_local_and_discover(context: Context) -> None:
"""Build AgentSkillDiscovery from project and local scopes (no global)."""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._discovery_result = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower:
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
project_dirs=paths_by_scope["project"] or [Path("/tmp/zzz_nonexistent_project_")],
local_dirs=paths_by_scope["local"] or [Path("/tmp/zzz_nonexistent_local_")],
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with global and project dirs and discover skills")
def step_create_global_project_and_discover(context: Context) -> None:
"""Build AgentSkillDiscovery from global and project scopes (no local)."""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._discovery_result = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower:
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
global_dirs=paths_by_scope["global"] or [Path("/tmp/zzz_nonexistent_global_")],
project_dirs=paths_by_scope["project"] or [Path("/tmp/zzz_nonexistent_project_")],
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with all non-existent dirs and discover skills")
def step_all_nonexistent_discover(context: Context) -> None:
"""Build AgentSkillDiscovery with no valid directories."""
context._discovery_errors = None
context._discovery_result = None
discovery = AgentSkillDiscovery(
global_dirs=[Path("/tmp/nonexistent_global_")],
project_dirs=[Path("/tmp/nonexistent_project_")],
local_dirs=[Path("/tmp/nonexistent_local_")],
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with a non-existent global dir and discover skills")
def step_nonexistent_global_discover(context: Context) -> None:
"""Build AgentSkillDiscovery where only global is invalid."""
context._discovery_errors = None
context._discovery_result = None
discovery = AgentSkillDiscovery(
global_dirs=[Path("/tmp/nonexistent_global_")],
project_dirs=[],
local_dirs=[],
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with all None dirs and discover skills")
def step_all_none_discover(context: Context) -> None:
"""Build AgentSkillDiscovery where no directories are set."""
context._discovery_errors = None
context._discovery_result = None
discovery = AgentSkillDiscovery(
global_dirs=None,
project_dirs=None,
local_dirs=None,
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery and call discover_scoped on scope {scope_name}")
def step_discover_scoped(context: Context, scope_name: str) -> None:
"""Run discover_scoped for a single scope.
Steps 13 are detected from the scoped directory registry (if any).
"""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._scop_result = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower and not key_lower.endswith("empty"):
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
global_dirs=paths_by_scope["global"] or [Path("/tmp/zzz_nonexistent_g_")],
project_dirs=paths_by_scope["project"] or [Path("/tmp/zzz_nonexistent_p_")],
local_dirs=paths_by_scope["local"] or [Path("/tmp/zzz_nonexistent_l_")],
)
try:
scope = Scope(scope_name)
context._scop_result = discovery.discover_scoped(scope)
except Exception as exc:
context._discovery_errors = exc
@when("I create an AgentSkillDiscovery with report_only conflict and discover skills")
def step_report_only_discover(context: Context) -> None:
"""Build AgentSkillDiscovery with ``report_only`` conflict strategy."""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._discovery_result = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower:
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
global_dirs=paths_by_scope["global"] or [Path("/tmp/zzz_nonexistent_g_")],
local_dirs=paths_by_scope["local"] or [Path("/tmp/zzz_nonexistent_l_")],
on_conflict="report_only",
)
context._discovery_result = discovery.discover()
@when("I create an AgentSkillDiscovery with invalid conflict strategy expecting an error")
def step_create_invalid_conflict(context: Context) -> None:
"""Attempt creation with an invalid ``on_conflict`` value.
The previously set value from the Given-step overrides any directory setup.
"""
value = getattr(context, "_invalid_conflict_value", "not_a_valid_value")
context._creation_error = None
try:
AgentSkillDiscovery(
global_dirs=[Path("/tmp/zzz_g_")],
on_conflict=value,
)
context._creation_error = None
except (ValueError, TypeError) as exc:
context._creation_error = exc
@when("I create an AgentSkillDiscovery and get all skill names")
def step_get_all_skill_names(context: Context) -> None:
"""Run discover once, then call get_all_skill_names()."""
registry = getattr(context, "_scoped_dir_registry", {})
context._discovery_errors = None
context._all_names = None
paths_by_scope: dict[str, list[Path]] = {"global": [], "project": [], "local": []}
if isinstance(registry, dict):
for key, path in registry.items():
key_lower = key.lower()
if "global" in key_lower and not key_lower.endswith("empty"):
paths_by_scope["global"].append(path)
elif "project" in key_lower:
paths_by_scope["project"].append(path)
elif "local" in str(key):
paths_by_scope["local"].append(path)
discovery = AgentSkillDiscovery(
global_dirs=paths_by_scope["global"] or [Path("/tmp/zzz_nonexistent_g_")],
)
context._all_names = discovery.get_all_skill_names()
@when("I check that ScopeOrder.LOCAL has the lowest numeric value")
def step_check_scope_order(context: Context) -> None:
"""Check the ScopeOrder numeric precedence values."""
context._scope_orders = {
"local": ScopeOrder.LOCAL.value,
"project": ScopeOrder.PROJECT.value,
"global": ScopeOrder.GLOBAL.value,
}
# ---------------------------------------------------------------------------
# Then — assertions for scope-discovery results
# ---------------------------------------------------------------------------
@then("the global scope should have {count:d} discovered skill(s)")
def step_global_scope_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
actual = len(result.scoped_skills[Scope.GLOBAL])
assert actual == count, (
f"Expected {count} global skill(s), got {actual}"
)
@then("the project scope should have {count:d} discovered skill(s)")
def step_project_scope_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
actual = len(result.scoped_skills[Scope.PROJECT])
assert actual == count, (
f"Expected {count} project skill(s), got {actual}"
)
@then("the local scope should have {count:d} discovered skill(s)")
def step_local_scope_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
actual = len(result.scoped_skills[Scope.LOCAL])
assert actual == count, (
f"Expected {count} local skill(s), got {actual}"
)
@then("the discovered skill name should be \"{name}\"")
def step_discovered_skill_name(context: Context, name: str) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
# Check across all scopes
found = False
for scope in Scope:
for entry in result.scoped_skills[scope]:
if entry.skill.name == name:
found = True
break
if found:
break
assert found, (
f"Expected skill '{name}' not found. "
f"All skills: {[e.skill.name for entries in result.scoped_skills.values() for e in entries]}"
)
@then("the total discovered skill count should be {count:d}")
def step_total_discovered_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
total = sum(len(v) for v in result.scoped_skills.values())
assert total == count, (
f"Expected {count} total skill(s), got {total}")
@then("the discovery collision count should be {count:d}")
def step_collision_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
actual = len(result.conflicts)
assert actual == count, (
f"Expected {count} collisions, got {actual}"
)
@then('the winning scope of collisions should be "{scope_name}"')
def step_winning_scope(context: Context, scope_name: str) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
assert len(result.conflicts) > 0
# At least one conflict's winning_scope should match
found = any(
c.winning_scope == scope_name for c in result.conflicts
)
assert found, (
f"Expected at least one collision won by '{scope_name}', "
f"got: {[c.winning_scope for c in result.conflicts]}"
)
@then('the losing scope of collisions should be "{scope_name}"')
def step_losing_scope(context: Context, scope_name: str) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
assert len(result.conflicts) > 0
found = any(
c.losing_scope == scope_name for c in result.conflicts
)
assert found, (
f"Expected at least one collision lost by '{scope_name}', "
f"got: {[c.losing_scope for c in result.conflicts]}"
)
@then("the errors list should contain \"{keyword}\"")
def step_errors_contain(context: Context, keyword: str) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
found = any(keyword.lower() in e.lower() for e in result.errors)
assert found, (
f"Expected error containing '{keyword}', "
f"got: {result.errors}"
)
@then("the error count should be at least 1")
def step_error_count_min_1(context: Context) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
actual = len(result.errors)
assert actual >= 1, (f"Expected >= 1 error(s), got {actual}")
@then("only the global scope should have {count:d} discovered skill(s)")
def step_only_global_scope_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
for scope in Scope:
if scope == Scope.GLOBAL:
expect = count
else:
expect = 0
actual = len(result.scoped_skills[scope])
assert actual == expect, (
f"Expected {expect} skill(s) in {scope.value}, got {actual}"
)
@then("only the local scope should have {count:d} discovered skill(s)")
def step_only_local_scope_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
for scope in Scope:
if scope == Scope.LOCAL:
expect = count
else:
expect = 0
actual = len(result.scoped_skills[scope])
assert actual == expect, (
f"Expected {expect} skill(s) in {scope.value}, got {actual}"
)
@then("all other scopes should have {count:d} discovered skill(s)")
def step_other_scopes_count(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
for scope in Scope:
actual = len(result.scoped_skills[scope])
assert actual == count, (
f"Expected {count} skill(s) in {scope.value}, got {actual}"
)
@then("the global skill names set should contain \"{name}\"")
def step_global_skill_names_set_contains(context: Context, name: str) -> None:
all_names = context._all_names
assert isinstance(all_names, dict)
found = any(name in names for names in all_names.values())
assert found, (f"Expected '{name}' to be in at least one scope's names, got: {all_names}")
@then("ScopeOrder.LOCAL should be less than ScopeOrder.PROJECT")
def step_scope_order_local_less_project(context: Context) -> None:
assert context._scope_orders is not None
orders = context._scope_orders
assert orders["local"] < orders["project"], (
f"Expected LOCAL({orders['local']}) < PROJECT({orders['project']})"
)
@then("ScopeOrder.PROJECT should be less than ScopeOrder.GLOBAL")
def step_scope_order_project_less_global(context: Context) -> None:
assert context._scope_orders is not None
orders = context._scope_orders
assert orders["project"] < orders["global"], (
f"Expected PROJECT({orders['project']}) < GLOBAL({orders['global']})"
)
@then("the creation error should mention \"{keyword}\"")
def step_creation_error_mentions(context: Context, keyword: str) -> None:
err = context._creation_error
assert err is not None, "Expected a creation error but none was raised"
assert keyword.lower() in str(err).lower(), (
f"Expected error mentioning '{keyword}', got: {err}"
)
@then("the local scope should have at least {count:d} discovered skill(s)")
def step_local_scope_at_least(context: Context, count: int) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
actual = len(result.scoped_skills[Scope.LOCAL])
assert actual >= count, (
f"Expected at least {count} local skill(s), got {actual}"
)
@then("the discovery collision count should be 0")
def step_zero_collisions(context: Context) -> None:
result = context._discovery_result
assert isinstance(result, MultiScopeDiscoveryResult)
actual = len(result.conflicts)
assert actual == 0, (f"Expected 0 collisions, got {actual}")
+14
View File
@@ -20,6 +20,14 @@ Usage::
loader = AgentSkillLoader.from_folder(Path("./skills/deploy-to-staging"))
"""
from cleveragents.skills.agent_skill_discovery import (
AgentSkillDiscovery,
DiscoveredScopedSkill,
MultiScopeDiscoveryResult,
Scope,
ScopedDiscoveryConflict,
ScopeOrder,
)
from cleveragents.skills.agent_skills_loader import (
AgentSkillLoader,
AgentSkillResourceSlot,
@@ -59,15 +67,21 @@ from cleveragents.skills.registry import SkillRegistry
from cleveragents.skills.schema import SkillConfigSchema
__all__ = [
"AgentSkillDiscovery",
"AgentSkillLoader",
"AgentSkillResourceSlot",
"AgentSkillSpec",
"AgentSkillToolDescriptor",
"DiscoveredAgentSkill",
"DiscoveredScopedSkill",
"DiscoveryConflict",
"DiscoveryResult",
"InlineToolExecutor",
"InlineToolResult",
"MultiScopeDiscoveryResult",
"Scope",
"ScopeOrder",
"ScopedDiscoveryConflict",
"SkillConfigSchema",
"SkillContext",
"SkillDefinition",
@@ -0,0 +1,509 @@
"""Multi-scope agent skill discovery for global, project, and local tiers.
Implements the three-tier progressive disclosure model for discovering
Agent Skills across multiple scope directories as specified in ADR-028
and issue #9369.
## Three-Tier Scope Model
.. code-block:: text
Tier 1 Global : organization-level skill directories
Tier 2 Project: project-level skill directories
Tier 3 Local : local working directory skill directories
Name collision resolution follows a clear precedence hierarchy:
``LOCAL`` > ``PROJECT`` > ``GLOBAL``
When the same skill name is discovered across multiple scopes, the
highest-precedence (lowest tier number) definition is retained and
lower-precedence duplicates are silently skipped.
## Usage
.. code-block:: python
from pathlib import Path
from cleveragents.skills.agent_skill_discovery import (
AgentSkillDiscovery,
Scope,
)
discovery = AgentSkillDiscovery(
global_dirs=[Path("/usr/share/cleveragents/skills")],
project_dirs=[Path("./skills/project")],
local_dirs=[Path("./skills/local")],
on_conflict="keep_precedence", # default — highest precedence wins
)
result = discovery.discover() # flat list (collision resolved)
scoped_result = discovery.discover_scoped() # grouped by scope
Based on:
- ADR-028 Agent Skills Standard, § "Skill folders are discovered via
paths configured in skill YAML (``agent_skills`` section)"
- Issue #9369 — Multi-scope discovery without a ``AgentSkillDiscovery`` class
- Specification line 46440 ``agent_skills_dirs`` configuration key
"""
from __future__ import annotations
import enum
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
from cleveragents.skills.discovery import (
DiscoveredAgentSkill,
DiscoveryConflict,
DiscoveryResult,
_parse_skill_md,
scan_directory,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Enumerations — Scope levels
# ---------------------------------------------------------------------------
class Scope(str, enum.Enum):
"""Scope levels for Agent Skill discovery.
The scopes define which directory tier a skill was loaded from.
Higher-precedence scopes override lower ones on name collision.
Attributes:
GLOBAL: Organization-level shared skills (lowest precedence).
PROJECT: Project-specific skills that share the codebase.
LOCAL: Per-repo / working-directory skills (highest precedence).
"""
GLOBAL = "global"
PROJECT = "project"
LOCAL = "local"
class ScopeOrder(enum.IntEnum):
"""Integer priority for scope-level collision resolution.
Lower values indicate **higher** precedence a local skill always
wins over a project or global skill with the same name.
Attributes:
LOCAL(1) : highest precedence takes final say on name conflicts.
PROJECT(2): intermediate overridden by local but beats global.
GLOBAL(3) : lowest precedence fallback when not found in other tiers.
Example::
if ScopeOrder.SKILL_SCOPE > ScopeOrder.GLOBAL:
# keep the existing (higher-precedence) skill
"""
LOCAL = 1
PROJECT = 2
GLOBAL = 3
# ---------------------------------------------------------------------------
# Data types — Scoped discoveries
# ---------------------------------------------------------------------------
class DiscoveredScopedSkill(BaseModel):
"""A single-agent skill discovered from a specific scope.
Combines the raw ``DiscoveredAgentSkill`` metadata with its
originating scope path so callers can reason about where each
skill came from.
Attributes:
scope: The discovery scope (global/project/local).
skill: The underlying parsed skill metadata.
original_path: Absolute path of the directory that contained
this skill on disk.
"""
scope: Scope = Field(
description="The scope this skill was discovered from",
)
skill: DiscoveredAgentSkill = Field(
description="Parsed agent skill metadata",
)
original_path: str = Field(
description="Absolute path to the source directory on disk",
)
model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
@dataclass(frozen=True, slots=True)
class ScopedDiscoveryConflict:
"""Records a name collision between two scope-level discoveries.
Attributes:
skill_name: The collided skill name (namespace/name).
winning_scope: Scope that takes precedence (e.g. "local").
losing_scope: Scope that is overridden (e.g. "global").
winner_path: Path of the winning skill on disk.
loser_path: Path of the overriding skill on disk.
"""
skill_name: str
winning_scope: str
losing_scope: str
winner_path: str
loser_path: str
# ---------------------------------------------------------------------------
# Discovery Result with scopes and collision tracking
# ---------------------------------------------------------------------------
class MultiScopeDiscoveryResult(BaseModel):
"""Aggregated result of a multi-scope agent skills discovery scan.
Attributes:
scoped_skills: Skills grouped by their originating scope.
conflicts: Name collisions that were resolved by precedence.
errors: Non-critical issues (missing dirs, permissions).
"""
scoped_skills: dict[Scope, list[DiscoveredScopedSkill]] = Field(
default_factory=lambda: {s: [] for s in Scope},
description="Skills keyed by scope",
)
conflicts: list[ScopedDiscoveryConflict] = Field(
default_factory=list,
description="Name collisions resolved by scope precedence",
)
errors: list[str] = Field(
default_factory=list,
description="Non-critical issues encountered during discovery",
)
model_config = ConfigDict(str_strip_whitespace=True)
# ---------------------------------------------------------------------------
# Resolve conflicts by scope precedence
# ---------------------------------------------------------------------------
def _resolve_scoped_collisions(
scoped_skills: dict[Scope, list[DiscoveredScopedSkill]],
) -> tuple[dict[Scope, list[DiscoveredScopedSkill]], list[ScopedDiscoveryConflict]]:
"""Resolve name collisions across scopes using ``ScopeOrder`` precedence.
Iterates over every distinct skill name and keeps the definition from
the highest-precedence scope (lowest ``ScopeOrder``). All losing
candidates are recorded as ``ScopedDiscoveryConflict`` items.
Args:
scoped_skills: Skills keyed by their source scope.
Returns:
A (2-tuple) of the filtered scoped_skills dict (only winners per name)
and the list of conflicts that were resolved.
"""
# Map each skill name to the highest-precedence definition
name_to_scoped: dict[str, DiscoveredScopedSkill] = {}
conflicts: list[ScopedDiscoveryConflict] = []
for scope in Scope:
ordered_skills = sorted(
scoped_skills.get(scope, []),
key=lambda s: s.skill.name,
)
for entry in ordered_skills:
name = entry.skill.name
if name not in name_to_scoped:
# First time we see this name — accept unconditionally
name_to_scoped[name] = entry
else:
existing = name_to_scoped[name]
existing_precedence = _scope_precedence(scope)
winner_precedence = _scope_precedence(existing.scope)
if existing_precedence < winner_precedence:
# New scope wins — bump the old one out
conflicts.append(
ScopedDiscoveryConflict(
skill_name=name,
winning_scope=existing.scope.value,
losing_scope=entry.scope.value,
winner_path=str(existing.original_path),
loser_path=str(entry.original_path),
)
)
name_to_scoped[name] = entry
elif existing_precedence > winner_precedence:
# Existing scope wins — new entry is the loser
conflicts.append(
ScopedDiscoveryConflict(
skill_name=name,
winning_scope=existing.scope.value,
losing_scope=entry.scope.value,
winner_path=str(existing.original_path),
loser_path=str(entry.original_path),
)
)
# else: same scope — shouldn't happen after per-scope dedup, but keep first
# Rebuild scoped_skills dict with only the winners
filtered: dict[Scope, list[DiscoveredScopedSkill]] = {s: [] for s in Scope}
for entry in name_to_scoped.values():
filtered[entry.scope].append(entry)
return filtered, conflicts
def _scope_precedence(scope: Scope) -> int:
"""Return the numeric precedence of a scope (lower = higher priority)."""
return ScopeOrder[scope.name.upper()].value # type: ignore[misc]
# ---------------------------------------------------------------------------
# AgentSkillDiscoveryService — three-tier multi-scope discovery
# ---------------------------------------------------------------------------
class AgentSkillDiscovery:
"""Discovers Agent Skills from global, project, and local scope dirs.
Implements the three-tier progressive disclosure model for
discovering Agent Skills across multiple configured directories.
Attributes:
global_dirs: List of paths at the organization level.
project_dirs: List of paths scoped to the current project.
local_dirs : List of paths in the working directory (highest precedence).
on_conflict: Conflict strategy ``"keep_precedence"`` (default) or
``"report_only"`` (keeps **all** scopes but logs collisions).
Usage::
discovery = AgentSkillDiscovery(
global_dirs=[Path("/usr/share/cleveragents/skills")],
project_dirs=[Path("./skills/project")],
local_dirs=[Path("./skills/local")],
)
result = discovery.discover()
all_skills = [s.skill for s in result.scoped_skills[Scope.LOCAL]]
Raises:
ValueError: If *on_conflict* is not one of the accepted strategies.
"""
def __init__(
self,
global_dirs: list[Path] | None = None,
project_dirs: list[Path] | None = None,
local_dirs: list[Path] | None = None,
on_conflict: Literal["keep_precedence", "report_only"] = "keep_precedence",
) -> None:
if on_conflict not in ("keep_precedence", "report_only"):
raise ValueError(
f"on_conflict must be 'keep_precedence' or 'report_only', "
f"got: {on_conflict!r}"
)
self.global_dirs: list[Path] = global_dirs or []
self.project_dirs: list[Path] = project_dirs or []
self.local_dirs: list[Path] = local_dirs or []
self.on_conflict: str = on_conflict
# Resolve all dirs at construction so missing/dir checks happen once
self._global_resolved: list[Path] = [d for d in (self.global_dirs or []) if _path_is_valid_directory(d)]
self._project_resolved: list[Path] = [d for d in (self.project_dirs or []) if _path_is_valid_directory(d)]
self._local_resolved: list[Path] = [d for d in (self.local_dirs or []) if _path_is_valid_directory(d)]
# -- Properties -------------------------------------------------------
@property
def scope_names(self) -> list[str]:
"""Return ordered human-readable scope names."""
return ["global", "project", "local"]
@property
def total_directories(self) -> int:
"""Total number of non-empty directories being scanned."""
return len(self._global_resolved) + len(self._project_resolved) + len(self._local_resolved)
# -- Public API -------------------------------------------------------
def discover(
self,
) -> MultiScopeDiscoveryResult:
"""Perform a full multi-scope discovery scan.
Discovers Agent Skills from all three scope directories in
precedence order (global project local), resolves name
collisions using ``ScopeOrder`` precedence rules, and returns a
consolidated result.
Returns:
A :class:`MultiScopeDiscoveryResult` containing scoped skills,
collision metadata, and any encountered errors.
Example::
result = AgentSkillDiscovery(
global_dirs=[Path("/usr/share/skills")],
project_dirs=[Path("./project-skills")],
local_dirs=[Path("./local-skills")],
).discover()
for scope in [Scope.LOCAL, Scope.PROJECT, Scope.GLOBAL]:
print(f" {scope}: {len(result.scoped_skills[scope])} skills")
"""
result = MultiScopeDiscoveryResult()
# Phase 1 — scan each scope independently
scoped_raw: dict[Scope, list[DiscoveredScopedSkill]] = {}
for scope in Scope:
raw_list: list[DiscoveredScopedSkill] = []
directories = self._scope_directories(scope)
for directory in (directories or []):
if not directory.exists():
result.errors.append(f"Directory does not exist: {directory}")
continue
if not directory.is_dir():
result.errors.append(f"Path is not a directory: {directory}")
continue
try:
for discovered in scan_directory(directory):
raw_list.append(
DiscoveredScopedSkill(
scope=scope,
skill=discovered,
original_path=str(directory),
)
)
except OSError as exc:
result.errors.append(f"Error scanning {directory}: {exc}")
scoped_raw[scope] = raw_list
if self.on_conflict == "report_only":
# No precedence resolution — just assign all discovered skills
result.scoped_skills = scoped_raw
return result
# Phase 2 — resolve name collisions by scope precedence
filtered, conflicts = _resolve_scoped_collisions(scoped_raw)
result.scoped_skills = filtered
result.conflicts = conflicts
return result
def discover_scoped(
self,
scope: Scope,
) -> MultiScopeDiscoveryResult:
"""Discover skills from a single scope only.
Useful for testing individual scopes or selectively applying
skill tiers.
Args:
scope: The specific ``Scope`` value to scan.
Returns:
A :class:`MultiScopeDiscoveryResult` scoped to a single tier.
"""
result = MultiScopeDiscoveryResult()
# Initialize all scopes so the dict is complete
for s in Scope:
if s not in result.scoped_skills:
result.scoped_skills[s] = [] # type: ignore[index]
directories = self._scope_directories(scope) or []
for directory in directories:
if not directory.exists():
result.errors.append(f"Directory does not exist: {directory}")
continue
if not directory.is_dir():
result.errors.append(f"Path is not a directory: {directory}")
continue
try:
for discovered in scan_directory(directory):
result.scoped_skills[scope].append( # type: ignore[index]
DiscoveredScopedSkill(
scope=scope,
skill=discovered,
original_path=str(directory),
)
)
except OSError as exc:
result.errors.append(f"Error scanning {directory}: {exc}")
return result
def get_all_skill_names(self) -> dict[Scope, set[str]]:
"""Return a map of scope → set of skill names for inspection.
Convenience helper that runs discovery once and extracts the
unique name per scope (without resolution).
"""
all_raw: dict[Scope, list[DiscoveredScopedSkill]] = {s: [] for s in Scope}
for scope in Scope:
directories = self._scope_directories(scope) or []
for directory in directories:
if not directory.exists() or not directory.is_dir():
continue
for discovered in scan_directory(directory):
all_raw[scope].append( # type: ignore[index]
DiscoveredScopedSkill(
scope=scope,
skill=discovered,
original_path=str(directory),
)
)
return {s: {e.skill.name for e in entries} for s, entries in all_raw.items()}
# -- Internal helpers -------------------------------------------------
def _scope_directories(self, scope: Scope) -> list[Path] | None:
"""Return the raw directory list for a given scope (before validation)."""
match scope:
case Scope.GLOBAL:
return self.global_dirs
case Scope.PROJECT:
return self.project_dirs
case Scope.LOCAL:
return self.local_dirs
def __repr__(self) -> str: # pragma: no cover — debugging helper
return (
f"AgentSkillDiscovery("
f"global={len(self._global_resolved)}, "
f"project={len(self._project_resolved)}, "
f"local={len(self._local_resolved)}, "
f"conflict='{self.on_conflict}')"
)
# ---------------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------------
def _path_is_valid_directory(path: Path) -> bool:
"""Check if a path exists and is a directory."""
try:
return path.exists() and path.is_dir()
except (OSError, PermissionError):
return False