Lorenze/imp/skills progressive disclosure (#6675)

* skills progressive disclosure

* skills progressive disclosure

* improving progressive disclosure

* addressed comment

* fix test

---------

Co-authored-by: João Moura <joaomdmoura@gmail.com>

lib/crewai/src/crewai/agent/core.py

@@ -83,7 +83,7 @@ from crewai.mcp.config import MCPServerConfig
 from crewai.rag.embeddings.types import EmbedderConfig
 from crewai.security.fingerprint import Fingerprint
 from crewai.skills.loader import load_skills
-from crewai.skills.models import Skill as SkillModel
+from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
 from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.types.callback import SerializableCallable
@@ -480,9 +480,32 @@ class Agent(BaseAgent):
 
         self.skills = cast(
             list[Path | SkillModel | str] | None,
-            load_skills(items, source=self) or None,
+            load_skills(items, source=self, activate=False) or None,
         )
 
+    def _add_skill_loader_tool(
+        self,
+        tools: list[BaseTool],
+        task: Task | None = None,
+    ) -> list[BaseTool]:
+        """Add the internal loader used for request-scoped skill disclosure."""
+        from crewai.skills.tool import LoadSkillTool, create_skill_loader_tool
+
+        tools = [tool for tool in tools if not isinstance(tool, LoadSkillTool)]
+
+        skill_models = [
+            skill for skill in self.skills or [] if isinstance(skill, SkillModel)
+        ]
+        loader = create_skill_loader_tool(
+            skill_models,
+            source=self,
+            task=task,
+            reserved_names=[tool.name for tool in tools],
+        )
+        if loader is None:
+            return tools
+        return [*tools, loader]
+
     def _is_any_available_memory(self) -> bool:
         """Check if unified memory is available (agent or crew)."""
         if getattr(self, "memory", None):
@@ -557,11 +580,11 @@ class Agent(BaseAgent):
         return apply_training_data(self, task_prompt)
 
     def _emit_skill_usage(self, task: Task) -> None:
-        """Emit one SkillUsedEvent per skill injected into this task's prompt.
+        """Emit usage for always-on skills injected into this task's prompt.
 
-        Skills are agent-scoped and rendered into the prompt on every execution,
-        so this is the runtime usage signal traces need — attributing each skill
-        to the agent and task that used it.
+        Metadata-only skills emit from ``LoadSkillTool`` if the model selects
+        them. This method covers explicitly activated and inline skills, whose
+        instructions are rendered on every execution.
 
         Args:
             task: The task whose prompt the skills are being applied to.
@@ -570,7 +593,10 @@ class Agent(BaseAgent):
             return
 
         for skill in self.skills:
-            if not isinstance(skill, SkillModel):
+            if (
+                not isinstance(skill, SkillModel)
+                or skill.disclosure_level < INSTRUCTIONS
+            ):
                 continue
             crewai_event_bus.emit(
                 self,
@@ -1050,6 +1076,8 @@ class Agent(BaseAgent):
         Returns:
             A tuple of (prompt, stop_words, rpm_limit_fn).
         """
+        from crewai.skills.tool import LoadSkillTool
+
         use_native_tool_calling = self._supports_native_tool_calling(raw_tools)
 
         prompt = Prompts(
@@ -1060,6 +1088,10 @@ class Agent(BaseAgent):
             system_template=self.system_template,
             prompt_template=self.prompt_template,
             response_template=self.response_template,
+            skill_loader_tool_name=next(
+                (tool.name for tool in raw_tools if isinstance(tool, LoadSkillTool)),
+                None,
+            ),
         ).task_execution()
 
         stop_words = [I18N_DEFAULT.slice("observation")]
@@ -1082,7 +1114,8 @@ class Agent(BaseAgent):
         Returns:
             An instance of the CrewAgentExecutor class.
         """
-        raw_tools: list[BaseTool] = tools or self.tools or []
+        configured_tools = tools if tools is not None else self.tools or []
+        raw_tools = self._add_skill_loader_tool(list(configured_tools), task=task)
         parsed_tools = parse_tools(raw_tools)
 
         prompt, stop_words, rpm_limit_fn = self._build_execution_prompt(raw_tools)
@@ -1421,6 +1454,9 @@ class Agent(BaseAgent):
         Returns:
             Tuple of (executor, inputs, agent_info, parsed_tools) ready for execution.
         """
+        if self.tools_handler:
+            self.tools_handler.last_used_tool = None
+
         if self.apps:
             platform_tools = self.get_platform_tools(self.apps)
             if platform_tools:
@@ -1434,7 +1470,7 @@ class Agent(BaseAgent):
                     self.tools = []
                 self.tools.extend(mcps)
 
-        raw_tools: list[BaseTool] = self.tools or []
+        raw_tools = list(self.tools or [])
 
         agent_memory = getattr(self, "memory", None)
         if agent_memory is not None:
@@ -1447,6 +1483,7 @@ class Agent(BaseAgent):
                 if sanitize_tool_name(mt.name) not in existing_names
             )
 
+        raw_tools = self._add_skill_loader_tool(raw_tools)
         parsed_tools = parse_tools(raw_tools)
 
         agent_info = {

lib/crewai/src/crewai/crews/utils.py

@@ -12,8 +12,8 @@ from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.crews.crew_output import CrewOutput
 from crewai.llms.base_llm import BaseLLM
 from crewai.rag.embeddings.types import EmbedderConfig
-from crewai.skills.loader import activate_skill, load_skills
-from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
+from crewai.skills.loader import load_skills
+from crewai.skills.models import Skill as SkillModel
 from crewai.types.streaming import CrewStreamingOutput, FlowStreamingOutput
 from crewai.utilities.file_store import store_files
 from crewai.utilities.streaming import (
@@ -59,13 +59,7 @@ def _resolve_crew_skills(crew: Crew) -> list[SkillModel] | None:
     if not isinstance(crew.skills, list) or not crew.skills:
         return None
 
-    resolved = load_skills(crew.skills)
-    if not resolved:
-        return None
-    return [
-        activate_skill(skill) if skill.disclosure_level < INSTRUCTIONS else skill
-        for skill in resolved
-    ]
+    return load_skills(crew.skills, activate=False) or None
 
 
 def setup_agents(

lib/crewai/src/crewai/events/types/skill_events.py

@@ -66,8 +66,8 @@ class SkillUsedEvent(SkillEvent):
     """Event emitted when an agent uses a skill during task execution.
 
     Discovery/load/activation events describe setup. This one is the runtime
-    signal: it fires each time a skill's context is injected into an agent's
-    prompt for a task, so traces can attribute skill usage to an agent and task.
+    signal: it fires when a metadata skill is selected or an always-on skill's
+    context is injected, so traces can attribute usage to an agent and task.
     """
 
     type: Literal["skill_used"] = "skill_used"

lib/crewai/src/crewai/skills/loader.py

@@ -6,6 +6,7 @@ for agent use, and format skill context for prompt injection.
 
 from __future__ import annotations
 
+from collections import Counter
 from collections.abc import Iterable
 import logging
 from pathlib import Path
@@ -19,7 +20,12 @@ from crewai.events.types.skill_events import (
     SkillLoadFailedEvent,
     SkillLoadedEvent,
 )
-from crewai.skills.models import INSTRUCTIONS, RESOURCES, Skill, SkillFrontmatter
+from crewai.skills.models import (
+    INSTRUCTIONS,
+    RESOURCES,
+    Skill,
+    SkillFrontmatter,
+)
 from crewai.skills.parser import (
     SKILL_FILENAME,
     load_skill_instructions,
@@ -148,25 +154,39 @@ def activate_skill(
 def load_skill(
     skill: Path | Skill | str,
     source: BaseAgent | None = None,
+    *,
+    activate: bool = True,
 ) -> list[Skill]:
     """Load one skill input into Skill objects.
 
     Accepts a pre-loaded Skill object, skill search path, inline SKILL.md
     string, or '@org/name' registry reference. Path inputs can expand to many
-    skills. Path and inline inputs are activated immediately; pre-loaded Skill
-    objects keep their disclosure level.
+    skills. Pre-loaded Skill objects keep their disclosure level. Path and
+    registry inputs are activated by default; callers performing runtime
+    progressive disclosure can leave them at metadata level.
+
+    Args:
+        skill: Skill input to resolve.
+        source: Optional event source for event emission.
+        activate: Whether discovered path and registry skills should load their
+            full instructions immediately.
+
+    Returns:
+        Resolved Skill objects.
     """
     if isinstance(skill, Skill):
         return [skill]
     if isinstance(skill, Path):
-        return [
-            activate_skill(s, source=source)
-            for s in discover_skills(skill, source=source)
-        ]
+        discovered = discover_skills(skill, source=source)
+        if not activate:
+            return discovered
+        return [activate_skill(s, source=source) for s in discovered]
     if isinstance(skill, str) and skill.startswith("@"):
         from crewai.skills.registry import resolve_registry_ref
 
-        return [resolve_registry_ref(skill, source=source)]
+        if activate:
+            return [resolve_registry_ref(skill, source=source)]
+        return [resolve_registry_ref(skill, source=source, activate=False)]
     if isinstance(skill, str) and skill.lstrip().startswith("---\n"):
         frontmatter_dict, body = parse_frontmatter(skill.strip())
         return [
@@ -178,10 +198,10 @@ def load_skill(
             )
         ]
     if isinstance(skill, str):
-        return [
-            activate_skill(s, source=source)
-            for s in discover_skills(Path(skill), source=source)
-        ]
+        discovered = discover_skills(Path(skill), source=source)
+        if not activate:
+            return discovered
+        return [activate_skill(s, source=source) for s in discovered]
 
     msg = f"Unsupported skill input: {skill!r}"
     raise TypeError(msg)
@@ -190,16 +210,27 @@ def load_skill(
 def load_skills(
     skills: Iterable[Path | Skill | str],
     source: BaseAgent | None = None,
+    *,
+    activate: bool = True,
 ) -> list[Skill]:
     """Load skill inputs into de-duplicated Skill objects.
 
     Preserves first-seen order when multiple inputs resolve to the same skill
     name. Registry refs are scoped by org so different orgs can publish skills
     that share a frontmatter name.
+
+    Args:
+        skills: Skill inputs to resolve.
+        source: Optional event source for event emission.
+        activate: Whether discovered path and registry skills should load their
+            full instructions immediately.
+
+    Returns:
+        De-duplicated Skill objects.
     """
     loaded: dict[str, Skill] = {}
     for skill_input in skills:
-        for skill in load_skill(skill_input, source=source):
+        for skill in load_skill(skill_input, source=source, activate=activate):
             dedup_key = skill.name
             if isinstance(skill_input, str) and skill_input.startswith("@"):
                 from crewai.skills.registry import parse_registry_ref
@@ -211,6 +242,42 @@ def load_skills(
     return list(loaded.values())
 
 
+def build_skill_catalog(skills: Iterable[Skill]) -> dict[str, Skill]:
+    """Label the metadata-only skills an execution can still load.
+
+    Skills resolved from different orgs or search paths can share a frontmatter
+    name. Colliding names are qualified by their parent directory — the org for
+    registry skills — so the catalog can address each of them individually.
+
+    Args:
+        skills: Skills to catalog. Skills already at INSTRUCTIONS level are
+            skipped because their instructions are rendered on every execution.
+
+    Returns:
+        Catalog labels mapped to their skill, in first-seen order.
+    """
+    pending = [skill for skill in skills if skill.disclosure_level < INSTRUCTIONS]
+    ambiguous = {
+        name
+        for name, count in Counter(skill.name for skill in pending).items()
+        if count > 1
+    }
+
+    catalog: dict[str, Skill] = {}
+    for skill in pending:
+        label = (
+            f"{skill.path.parent.name}/{skill.name}"
+            if skill.name in ambiguous
+            else skill.name
+        )
+        qualified, attempt = label, 2
+        while label in catalog:
+            label = f"{qualified}-{attempt}"
+            attempt += 1
+        catalog[label] = skill
+    return catalog
+
+
 def load_resources(skill: Skill) -> Skill:
     """Promote a skill to RESOURCES disclosure level.
 
@@ -223,7 +290,7 @@ def load_resources(skill: Skill) -> Skill:
     return load_skill_resources(skill)
 
 
-def format_skill_context(skill: Skill) -> str:
+def format_skill_context(skill: Skill, *, label: str | None = None) -> str:
     """Format skill information for agent prompt injection.
 
     At METADATA level: returns name and description only.
@@ -234,13 +301,16 @@ def format_skill_context(skill: Skill) -> str:
 
     Args:
         skill: The skill to format.
+        label: Name to render in place of the skill's own, used when a catalog
+            has qualified skills that share a frontmatter name.
 
     Returns:
         Formatted skill context string.
     """
+    name = label or skill.name
     if skill.disclosure_level >= INSTRUCTIONS and skill.instructions:
         parts = [
-            f'<skill name="{skill.name}">',
+            f'<skill name="{name}">',
             skill.description,
             "",
             skill.instructions,
@@ -253,4 +323,4 @@ def format_skill_context(skill: Skill) -> str:
                     parts.append(f"- **{dir_name}/**: {', '.join(files)}")
         parts.append("</skill>")
         return "\n".join(parts)
-    return f'<skill name="{skill.name}">\n{skill.description}\n</skill>'
+    return f'<skill name="{name}">\n{skill.description}\n</skill>'

lib/crewai/src/crewai/skills/registry.py

@@ -114,6 +114,8 @@ def parse_registry_ref(ref: str) -> tuple[str, str]:
 def resolve_registry_ref(
     ref: str,
     source: Any = None,
+    *,
+    activate: bool = True,
 ) -> Skill:  # type: ignore[name-defined]  # noqa: F821
     """Resolve a registry reference to a Skill object.
 
@@ -129,9 +131,11 @@ def resolve_registry_ref(
     Args:
         ref: A registry reference, e.g. '@acme/my-skill' or '@acme/my-skill@1.2.0'.
         source: Optional source object passed through to skill loaders (for events).
+        activate: Whether to load the full SKILL.md instructions.
 
     Returns:
-        A Skill loaded at INSTRUCTIONS disclosure level.
+        A Skill loaded at INSTRUCTIONS level by default, or METADATA level
+        when ``activate`` is false.
     """
     from crewai.skills.loader import activate_skill
 
@@ -143,7 +147,7 @@ def resolve_registry_ref(
         # is the only thing a pin can be checked against.
         skill = _load_matching_skill(local_path, version)
         if skill is not None:
-            return activate_skill(skill, source=source)
+            return activate_skill(skill, source=source) if activate else skill
 
     cache = SkillCacheManager()
     cached_path = cache.get_cached_path(org, name, version=version)
@@ -154,9 +158,15 @@ def resolve_registry_ref(
         # metadata.version and re-download it on each resolution.
         skill = _load_skill(cached_path)
         if skill is not None:
-            return activate_skill(skill, source=source)
+            return activate_skill(skill, source=source) if activate else skill
 
-    return download_skill(org, name, source=source, version=version)
+    return download_skill(
+        org,
+        name,
+        source=source,
+        version=version,
+        activate=activate,
+    )
 
 
 def _load_skill(path: Path) -> Skill | None:  # type: ignore[name-defined]  # noqa: F821
@@ -268,6 +278,7 @@ def download_skill(
     source: Any = None,
     *,
     version: str | None = None,
+    activate: bool = True,
 ) -> Skill:  # type: ignore[name-defined]  # noqa: F821
     """Download a skill from the registry and store it in the cache.
 
@@ -276,9 +287,11 @@ def download_skill(
         name: Skill name.
         source: Optional source for event emission.
         version: Optional pinned version; the latest is fetched when omitted.
+        activate: Whether to load the full SKILL.md instructions.
 
     Returns:
-        The downloaded Skill at INSTRUCTIONS level.
+        The downloaded Skill at INSTRUCTIONS level by default, or METADATA
+        level when ``activate`` is false.
 
     Raises:
         ValueError: If *version* is given but blank.
@@ -376,4 +389,4 @@ def download_skill(
             f"Skill archive for {ref!r} downloaded but no SKILL.md found in {skill_dir}"
         )
     skill = load_skill_metadata(skill_dir)
-    return activate_skill(skill, source=source)
+    return activate_skill(skill, source=source) if activate else skill

lib/crewai/src/crewai/skills/tool.py

@@ -0,0 +1,116 @@
+"""Runtime tool for progressively disclosing Agent Skill instructions."""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+from typing import Any, Final
+
+from pydantic import BaseModel, Field
+
+from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.skill_events import SkillUsedEvent
+from crewai.skills.loader import (
+    activate_skill,
+    build_skill_catalog,
+    format_skill_context,
+)
+from crewai.skills.models import Skill
+from crewai.tools.base_tool import BaseTool
+from crewai.utilities.string_utils import sanitize_tool_name
+
+
+LOAD_SKILL_TOOL_NAME: Final[str] = "load_skill"
+
+
+class LoadSkillSchema(BaseModel):
+    """Arguments accepted by the runtime skill loader."""
+
+    skill_name: str = Field(
+        ...,
+        description="Exact name of the available skill whose instructions are needed.",
+    )
+
+
+class LoadSkillTool(BaseTool):
+    """Load one relevant skill's instructions for the current execution."""
+
+    name: str = LOAD_SKILL_TOOL_NAME
+    description: str = (
+        "Load one available skill's full instructions. Most requests need no "
+        "skill, so only call this when an available skill's description "
+        "clearly matches the current request."
+    )
+    args_schema: type[BaseModel] = LoadSkillSchema
+    catalog: dict[str, Skill] = Field(default_factory=dict, exclude=True)
+    source: Any = Field(default=None, exclude=True)
+    task: Any = Field(default=None, exclude=True)
+
+    def _run(self, skill_name: str, **kwargs: Any) -> str:
+        """Load and return one skill's instruction block."""
+        skill = self.catalog.get(skill_name)
+        if skill is None:
+            available = ", ".join(self.catalog)
+            return (
+                f"Skill {skill_name!r} is not available. "
+                f"Available skills: {available or 'none'}."
+            )
+
+        activated = activate_skill(skill, source=self.source)
+        crewai_event_bus.emit(
+            self.source,
+            event=SkillUsedEvent(
+                from_agent=self.source,
+                from_task=self.task,
+                skill_name=skill_name,
+                skill_path=activated.path,
+                disclosure_level=activated.disclosure_level,
+            ),
+        )
+        return format_skill_context(activated, label=skill_name)
+
+
+def resolve_loader_tool_name(reserved_names: Collection[str] = ()) -> str:
+    """Return a loader tool name that none of *reserved_names* already claims.
+
+    A configured tool holding the default name would otherwise be
+    indistinguishable from the loader, so the model could never reach the
+    loader the skill catalog points it at.
+    """
+    taken = {sanitize_tool_name(name) for name in reserved_names}
+    if LOAD_SKILL_TOOL_NAME not in taken:
+        return LOAD_SKILL_TOOL_NAME
+
+    attempt = 2
+    while f"{LOAD_SKILL_TOOL_NAME}_{attempt}" in taken:
+        attempt += 1
+    return f"{LOAD_SKILL_TOOL_NAME}_{attempt}"
+
+
+def create_skill_loader_tool(
+    skills: list[Skill] | None,
+    *,
+    source: Any = None,
+    task: Any = None,
+    reserved_names: Collection[str] = (),
+) -> LoadSkillTool | None:
+    """Create a loader tool when metadata-only skills are available.
+
+    Args:
+        skills: Skills configured on the agent.
+        source: Event source for the skill events the loader emits.
+        task: Task the loader is scoped to.
+        reserved_names: Names of tools already configured on the agent, which
+            the loader must not collide with.
+
+    Returns:
+        The loader tool, or None when every skill is already disclosed.
+    """
+    catalog = build_skill_catalog(skills or [])
+    if not catalog:
+        return None
+    return LoadSkillTool(
+        name=resolve_loader_tool_name(reserved_names),
+        catalog=catalog,
+        source=source,
+        task=task,
+    )

lib/crewai/src/crewai/utilities/prompts.py

@@ -70,6 +70,10 @@ class Prompts(BaseModel):
         description="Whether to use the system prompt when no custom templates are provided",
     )
     agent: Any = Field(description="Reference to the agent using these prompts")
+    skill_loader_tool_name: str | None = Field(
+        default=None,
+        description="Name the runtime skill loader tool is registered under",
+    )
 
     def task_execution(self) -> SystemPromptResult | StandardPromptResult:
         """Generate a standard prompt for task execution.
@@ -115,22 +119,50 @@ class Prompts(BaseModel):
         )
 
     def _build_skill_block(self) -> str:
-        """Render the agent's activated skills as a stable XML block.
+        """Render always-on instructions and the available skill catalog.
 
-        Skills are agent-scoped (do not change per task), so they live in the
-        system prompt where prompt-cache prefixes can survive across calls.
+        Metadata-only skills remain a stable prompt-cache anchor. Their full
+        instructions are disclosed through the runtime loader tool only when the
+        model determines that a skill applies to the current request.
         """
         skills = getattr(self.agent, "skills", None)
         if not skills:
             return ""
 
-        from crewai.skills.loader import format_skill_context
-        from crewai.skills.models import Skill
+        from crewai.skills.loader import build_skill_catalog, format_skill_context
+        from crewai.skills.models import INSTRUCTIONS, Skill
+        from crewai.skills.tool import LOAD_SKILL_TOOL_NAME
 
-        sections = [format_skill_context(s) for s in skills if isinstance(s, Skill)]
-        if not sections:
+        loaded = [skill for skill in skills if isinstance(skill, Skill)]
+        active = [
+            format_skill_context(skill)
+            for skill in loaded
+            if skill.disclosure_level >= INSTRUCTIONS
+        ]
+        catalog = build_skill_catalog(loaded)
+
+        blocks: list[str] = []
+        if active:
+            blocks.append("<skills>\n" + "\n\n".join(active) + "\n</skills>")
+        if catalog:
+            loader = self.skill_loader_tool_name or LOAD_SKILL_TOOL_NAME
+            entries = "\n\n".join(
+                format_skill_context(skill, label=label)
+                for label, skill in catalog.items()
+            )
+            blocks.append(
+                "<available_skills>\n"
+                "These skills are not loaded. Most requests need none of them, "
+                "so answer directly unless one clearly applies to this request. "
+                f"When one does, call `{loader}` with its exact name to load its "
+                "full instructions, and do not follow a skill's instructions "
+                "without loading it first.\n\n"
+                f"{entries}\n"
+                "</available_skills>"
+            )
+        if not blocks:
             return ""
-        return "\n\n<skills>\n" + "\n\n".join(sections) + "\n</skills>"
+        return "\n\n" + "\n\n".join(blocks)
 
     def _build_prompt(
         self,

lib/crewai/tests/events/test_skill_usage_events.py

@@ -12,6 +12,8 @@ import pytest
 from crewai import Agent, Task
 from crewai.events import crewai_event_bus
 from crewai.events.types.skill_events import SkillUsedEvent
+from crewai.skills.loader import activate_skill, discover_skills
+from crewai.skills.tool import LoadSkillTool
 
 
 # Handlers run on the event bus thread pool, so tests must flush before
@@ -29,12 +31,13 @@ def _create_skill_dir(parent: Path, name: str, body: str = "Body.") -> Path:
 
 
 def _agent_with_skills(search_path: Path) -> Agent:
-    # discover_skills scans the SUBdirectories of the given path.
+    """Create an agent with explicitly activated, always-on skills."""
+    skills = [activate_skill(skill) for skill in discover_skills(search_path)]
     return Agent(
         role="Analyst",
         goal="Analyze things",
         backstory="Experienced",
-        skills=[str(search_path)],
+        skills=skills,
         llm="gpt-4o-mini",
     )
 
@@ -44,6 +47,43 @@ def _task_for(agent: Agent) -> Task:
 
 
 class TestSkillUsedEvent:
+    def test_metadata_skill_emits_only_after_runtime_selection(
+        self, tmp_path: Path
+    ) -> None:
+        _create_skill_dir(tmp_path, "alpha")
+        agent = Agent(
+            role="Analyst",
+            goal="Analyze things",
+            backstory="Experienced",
+            skills=[tmp_path],
+            llm="gpt-4o-mini",
+        )
+        task = _task_for(agent)
+        agent.create_agent_executor(task=task)
+        assert agent.agent_executor is not None
+        loader = next(
+            tool
+            for tool in agent.agent_executor.original_tools
+            if isinstance(tool, LoadSkillTool)
+        )
+
+        received: list[SkillUsedEvent] = []
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(SkillUsedEvent)
+            def _handler(source, event: SkillUsedEvent) -> None:  # noqa: ARG001
+                received.append(event)
+
+            agent._emit_skill_usage(task)
+            assert crewai_event_bus.flush(timeout=10)
+            assert received == []
+
+            loader.run(skill_name="alpha")
+            assert crewai_event_bus.flush(timeout=10)
+
+        assert [event.skill_name for event in received] == ["alpha"]
+        assert received[0].task_id == str(task.id)
+
     def test_emits_one_event_per_skill_with_agent_and_task_attribution(
         self, tmp_path: Path
     ) -> None:
@@ -109,7 +149,7 @@ class TestSkillUsedEvent:
         assert len(received) == 2
 
     def test_reports_disclosure_level(self, tmp_path: Path) -> None:
-        """Path-loaded skills are activated, so they report INSTRUCTIONS level."""
+        """Explicitly activated skills report INSTRUCTIONS level."""
         _create_skill_dir(tmp_path, "alpha")
         agent = _agent_with_skills(tmp_path)
         task = _task_for(agent)

lib/crewai/tests/skills/test_integration.py

@@ -144,7 +144,8 @@ class TestSkillDiscoveryAndActivation:
 
         assert agent.skills is not None
         assert [skill.name for skill in agent.skills] == ["path-skill"]
-        assert [skill.instructions for skill in agent.skills] == ["Use the path skill."]
+        assert [skill.disclosure_level for skill in agent.skills] == [METADATA]
+        assert [skill.instructions for skill in agent.skills] == [None]
 
     def test_crew_resolves_inline_skill_string(self) -> None:
         agent = Agent(
@@ -175,7 +176,7 @@ class TestSkillDiscoveryAndActivation:
         assert [skill.name for skill in skills] == ["crew-inline-review"]
         assert [skill.instructions for skill in skills] == ["Apply this to every agent."]
 
-    def test_crew_activates_preloaded_metadata_skill(self, tmp_path: Path) -> None:
+    def test_crew_preserves_preloaded_metadata_skill(self, tmp_path: Path) -> None:
         _create_skill_dir(
             tmp_path,
             "crew-preloaded",
@@ -202,7 +203,5 @@ class TestSkillDiscoveryAndActivation:
 
         assert skills is not None
         assert [skill.name for skill in skills] == ["crew-preloaded"]
-        assert [skill.disclosure_level for skill in skills] == [INSTRUCTIONS]
-        assert [skill.instructions for skill in skills] == [
-            "Apply this crew-level guidance to every agent."
-        ]
+        assert [skill.disclosure_level for skill in skills] == [METADATA]
+        assert [skill.instructions for skill in skills] == [None]

lib/crewai/tests/skills/test_progressive_disclosure.py

@@ -0,0 +1,337 @@
+"""Regression tests for runtime skill progressive disclosure.
+
+Run this focused test file with:
+
+    uv run pytest lib/crewai/tests/skills/test_progressive_disclosure.py -q
+"""
+
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from crewai import Agent, Task
+from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.skill_events import SkillUsedEvent
+from crewai.llms.base_llm import BaseLLM
+from crewai.skills.models import METADATA
+from crewai.skills.parser import load_skill_metadata
+from crewai.skills.tool import LoadSkillTool, create_skill_loader_tool
+from crewai.tools.base_tool import BaseTool
+from crewai.utilities.prompts import Prompts
+
+
+class _ConflictingToolSchema(BaseModel):
+    query: str = Field(default="")
+
+
+class _ConflictingTool(BaseTool):
+    """A user tool that already claims the skill loader's default name."""
+
+    name: str = "load_skill"
+    description: str = "Load something unrelated to skills."
+    args_schema: type[BaseModel] = _ConflictingToolSchema
+
+    def _run(self, query: str = "", **kwargs: Any) -> str:
+        return "user tool result"
+
+
+def _create_skill(
+    parent: Path,
+    name: str,
+    description: str,
+    instructions: str,
+) -> None:
+    skill_dir = parent / name
+    skill_dir.mkdir(parents=True)
+    (skill_dir / "SKILL.md").write_text(
+        f"---\nname: {name}\ndescription: {description}\n---\n{instructions}",
+        encoding="utf-8",
+    )
+
+
+class _SkillChoosingLLM(BaseLLM):
+    """Small deterministic LLM that exercises the real agent tool loop."""
+
+    def call(self, messages: Any, **kwargs: Any) -> str:
+        rendered = str(messages)
+        if "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS" in rendered:
+            return "Thought: I loaded the review skill.\nFinal Answer: python loaded"
+        if "TRAVEL_PRIVATE_INSTRUCTIONS" in rendered:
+            return "Thought: I loaded the travel skill.\nFinal Answer: travel loaded"
+        if "Review this Python change" in rendered:
+            return (
+                "Thought: The Python review skill applies.\n"
+                "Action: load_skill\n"
+                'Action Input: {"skill_name": "python-review"}'
+            )
+        if "Plan a weekend trip" in rendered:
+            return (
+                "Thought: The travel planning skill applies.\n"
+                "Action: load_skill\n"
+                'Action Input: {"skill_name": "travel-planning"}'
+            )
+        return "Thought: No skill applies.\nFinal Answer: no skill"
+
+    def supports_function_calling(self) -> bool:
+        return False
+
+    def supports_stop_words(self) -> bool:
+        return False
+
+    def get_context_window_size(self) -> int:
+        return 8_192
+
+
+def test_agent_directory_skills_do_not_eagerly_disclose_instructions(
+    tmp_path: Path,
+) -> None:
+    """An agent should initially receive only the skill catalog metadata."""
+    _create_skill(
+        tmp_path,
+        "python-review",
+        "Review Python code when the user asks for a code review.",
+        "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS",
+    )
+    _create_skill(
+        tmp_path,
+        "travel-planning",
+        "Plan travel when the user asks for a trip itinerary.",
+        "TRAVEL_PRIVATE_INSTRUCTIONS",
+    )
+
+    agent = Agent(
+        role="Assistant",
+        goal="Help with the current request.",
+        backstory="A general-purpose assistant.",
+        skills=[tmp_path],
+    )
+
+    assert agent.skills is not None
+    assert [skill.disclosure_level for skill in agent.skills] == [
+        METADATA,
+        METADATA,
+    ]
+    assert [skill.instructions for skill in agent.skills] == [None, None]
+
+
+def test_consecutive_kickoffs_select_skills_without_cross_call_leakage(
+    tmp_path: Path,
+) -> None:
+    """Each execution should independently choose its relevant skill."""
+    _create_skill(
+        tmp_path,
+        "python-review",
+        "Review Python code when the user asks for a code review.",
+        "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS",
+    )
+    _create_skill(
+        tmp_path,
+        "travel-planning",
+        "Plan travel when the user asks for a trip itinerary.",
+        "TRAVEL_PRIVATE_INSTRUCTIONS",
+    )
+    agent = Agent(
+        role="Assistant",
+        goal="Help with the current request.",
+        backstory="A general-purpose assistant.",
+        skills=[tmp_path],
+        llm=_SkillChoosingLLM(model="skill-test"),
+        max_iter=3,
+    )
+
+    review_output = agent.kickoff("Review this Python change")
+    travel_output = agent.kickoff("Plan a weekend trip")
+    repeated_review_output = agent.kickoff("Review this Python change")
+
+    assert review_output.raw == "python loaded"
+    assert travel_output.raw == "travel loaded"
+    assert repeated_review_output.raw == "python loaded"
+    assert agent.skills is not None
+    assert [skill.disclosure_level for skill in agent.skills] == [
+        METADATA,
+        METADATA,
+    ]
+    assert [skill.instructions for skill in agent.skills] == [None, None]
+
+    prompt = Prompts(
+        agent=agent,
+        has_tools=False,
+        use_system_prompt=True,
+    ).task_execution()
+    rendered = getattr(prompt, "system", "") or prompt.prompt
+
+    assert "python-review" in rendered
+    assert "travel-planning" in rendered
+    assert "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS" not in rendered
+    assert "TRAVEL_PRIVATE_INSTRUCTIONS" not in rendered
+
+
+def test_each_execution_can_disclose_only_its_relevant_skill(tmp_path: Path) -> None:
+    """Loading one skill must not leak or permanently activate another."""
+    _create_skill(
+        tmp_path,
+        "python-review",
+        "Review Python code when the user asks for a code review.",
+        "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS",
+    )
+    _create_skill(
+        tmp_path,
+        "travel-planning",
+        "Plan travel when the user asks for a trip itinerary.",
+        "TRAVEL_PRIVATE_INSTRUCTIONS",
+    )
+    agent = Agent(
+        role="Assistant",
+        goal="Help with the current request.",
+        backstory="A general-purpose assistant.",
+        skills=[tmp_path],
+    )
+
+    review_task = Task(
+        description="Review this Python change.",
+        expected_output="Review findings.",
+        agent=agent,
+    )
+    agent.create_agent_executor(task=review_task)
+    assert agent.agent_executor is not None
+    review_loader = next(
+        tool
+        for tool in agent.agent_executor.original_tools
+        if isinstance(tool, LoadSkillTool)
+    )
+    review_context = review_loader.run(skill_name="python-review")
+
+    assert "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS" in review_context
+    assert "TRAVEL_PRIVATE_INSTRUCTIONS" not in review_context
+
+    travel_task = Task(
+        description="Plan a weekend trip.",
+        expected_output="A travel itinerary.",
+        agent=agent,
+    )
+    agent.create_agent_executor(task=travel_task)
+    assert agent.agent_executor is not None
+    travel_loader = next(
+        tool
+        for tool in agent.agent_executor.original_tools
+        if isinstance(tool, LoadSkillTool)
+    )
+    travel_context = travel_loader.run(skill_name="travel-planning")
+
+    assert "TRAVEL_PRIVATE_INSTRUCTIONS" in travel_context
+    assert "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS" not in travel_context
+    assert agent.skills is not None
+    assert [skill.disclosure_level for skill in agent.skills] == [
+        METADATA,
+        METADATA,
+    ]
+    assert [skill.instructions for skill in agent.skills] == [None, None]
+
+
+def test_loader_stays_reachable_when_a_tool_claims_its_default_name(
+    tmp_path: Path,
+) -> None:
+    """A configured tool named load_skill must not shadow the skill loader."""
+    _create_skill(
+        tmp_path,
+        "python-review",
+        "Review Python code when the user asks for a code review.",
+        "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS",
+    )
+
+    agent = Agent(
+        role="Assistant",
+        goal="Help with the current request.",
+        backstory="A general-purpose assistant.",
+        skills=[tmp_path],
+        tools=[_ConflictingTool()],
+    )
+    agent.create_agent_executor(
+        task=Task(
+            description="Review this Python change.",
+            expected_output="Review findings.",
+            agent=agent,
+        )
+    )
+
+    assert agent.agent_executor is not None
+    tools = agent.agent_executor.original_tools
+    loader = next(tool for tool in tools if isinstance(tool, LoadSkillTool))
+
+    assert loader.name != "load_skill"
+    assert [tool.name for tool in tools].count(loader.name) == 1
+    assert "PYTHON_REVIEW_PRIVATE_INSTRUCTIONS" in loader.run(
+        skill_name="python-review"
+    )
+
+    prompt = agent.agent_executor.prompt
+    rendered = prompt.get("system") or prompt.get("prompt")
+    assert f"call `{loader.name}`" in rendered
+
+
+def test_same_named_skills_from_different_orgs_stay_individually_loadable(
+    tmp_path: Path,
+) -> None:
+    """Skills sharing a frontmatter name must each be addressable."""
+    _create_skill(
+        tmp_path / "acme",
+        "code-review",
+        "Review code the Acme way.",
+        "ACME_PRIVATE_INSTRUCTIONS",
+    )
+    _create_skill(
+        tmp_path / "globex",
+        "code-review",
+        "Review code the Globex way.",
+        "GLOBEX_PRIVATE_INSTRUCTIONS",
+    )
+
+    loader = create_skill_loader_tool(
+        [
+            load_skill_metadata(tmp_path / "acme" / "code-review"),
+            load_skill_metadata(tmp_path / "globex" / "code-review"),
+        ]
+    )
+
+    assert loader is not None
+    assert sorted(loader.catalog) == ["acme/code-review", "globex/code-review"]
+    assert "ACME_PRIVATE_INSTRUCTIONS" in loader.run(skill_name="acme/code-review")
+    assert "GLOBEX_PRIVATE_INSTRUCTIONS" in loader.run(skill_name="globex/code-review")
+
+
+def test_loaded_block_and_event_keep_the_catalog_label(tmp_path: Path) -> None:
+    """A disambiguated skill must report the label the model was told to load."""
+    _create_skill(
+        tmp_path / "acme",
+        "code-review",
+        "Review code the Acme way.",
+        "ACME_PRIVATE_INSTRUCTIONS",
+    )
+    _create_skill(
+        tmp_path / "globex",
+        "code-review",
+        "Review code the Globex way.",
+        "GLOBEX_PRIVATE_INSTRUCTIONS",
+    )
+
+    loader = create_skill_loader_tool(
+        [
+            load_skill_metadata(tmp_path / "acme" / "code-review"),
+            load_skill_metadata(tmp_path / "globex" / "code-review"),
+        ]
+    )
+    assert loader is not None
+
+    used: list[str] = []
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(SkillUsedEvent)
+        def _record(_source: Any, event: SkillUsedEvent) -> None:
+            used.append(event.skill_name)
+
+        context = loader.run(skill_name="globex/code-review")
+        assert crewai_event_bus.flush(timeout=10)
+
+    assert '<skill name="globex/code-review">' in context
+    assert used == ["globex/code-review"]

lib/crewai/tests/skills/test_registry.py

@@ -10,6 +10,7 @@ from zipfile import ZipFile
 
 from crewai.context import platform_context
 from crewai.skills.cache import SkillCacheManager
+from crewai.skills.models import METADATA
 from crewai.skills.registry import (
     SkillRef,
     download_skill,
@@ -181,6 +182,17 @@ class TestParseSkillRef:
 
 
 class TestResolveRegistryRef:
+    def test_can_resolve_metadata_without_loading_instructions(
+        self, tmp_path: Path
+    ) -> None:
+        _write_local_skill(tmp_path, "my-skill")
+
+        with patch.object(Path, "cwd", return_value=tmp_path):
+            skill = resolve_registry_ref("@acme/my-skill", activate=False)
+
+        assert skill.disclosure_level == METADATA
+        assert skill.instructions is None
+
     def test_prefers_project_local_skill_over_cached_skill(
         self, tmp_path: Path
     ) -> None: