feat: emit FlowFailedEvent when a flow execution fails (#6718 )

* feat: emit FlowFailedEvent when a flow execution fails A failed flow never emitted a terminal lifecycle event, so the `flow_started` scope stayed open and consumers such as tracing closed the root span with a generic orphaned message instead of the real error. `kickoff_async` and the resume path now emit `FlowFailedEvent`, paired with `flow_started` and carrying the exception, after draining pending handlers and background memory writes. The resume path also emits the `MethodExecutionStartedEvent` it was missing for the method being resumed, so its finished or failed event pairs with its own scope instead of popping the flow's. * fix: skip FlowFailedEvent when the run never opened a scope The `kickoff_async` try block starts before `FlowStartedEvent` is emitted, so an abort in the execution-start hooks, in input handling or in state restore emitted a `flow_failed` with no opener, which pops an unrelated scope and warns about an empty scope stack. The failure event is now gated on the flow scope actually being open, either from this kickoff's `flow_started` or from a restored deferred session scope.
Lorenze/imp/skills progressive disclosure (#6675 )
2026-07-29 18:49:21 +00:00 · 2026-07-29 14:44:55 -04:00 · 2026-07-28 09:33:43 -07:00
25 changed files with 1179 additions and 93 deletions
--- a/docs/edge/ar/concepts/event-listener.mdx
+++ b/docs/edge/ar/concepts/event-listener.mdx
@@ -157,6 +157,7 @@ class MyCustomCrew:
 - **FlowCreatedEvent**: يُرسل عند إنشاء تدفق
 - **FlowStartedEvent**: يُرسل عند بدء تنفيذ تدفق
 - **FlowFinishedEvent**: يُرسل عند اكتمال تنفيذ تدفق
+- **FlowFailedEvent**: يُرسل عند فشل تنفيذ تدفق. يحتوي على اسم التدفق والاستثناء الذي أنهى التنفيذ.
 - **FlowPausedEvent**: يُرسل عند إيقاف تدفق مؤقتًا بانتظار ملاحظات بشرية

 ### أحداث LLM
--- a/docs/edge/en/concepts/event-listener.mdx
+++ b/docs/edge/en/concepts/event-listener.mdx
@@ -256,6 +256,7 @@ CrewAI provides a wide range of events that you can listen for:
 - **FlowCreatedEvent**: Emitted when a Flow is created
 - **FlowStartedEvent**: Emitted when a Flow starts execution
 - **FlowFinishedEvent**: Emitted when a Flow completes execution
+- **FlowFailedEvent**: Emitted when a Flow execution fails. Contains the flow name and the exception that ended the execution.
 - **FlowPausedEvent**: Emitted when a Flow is paused waiting for human feedback. Contains the flow name, flow ID, method name, current state, message shown when requesting feedback, and optional list of possible outcomes for routing.
 - **FlowPlotEvent**: Emitted when a Flow is plotted
 - **MethodExecutionStartedEvent**: Emitted when a Flow method starts execution
--- a/docs/edge/en/concepts/skills.mdx
+++ b/docs/edge/en/concepts/skills.mdx
@@ -9,7 +9,10 @@ mode: "wide"

 Skills are self-contained directories that provide agents with **domain-specific instructions, guidelines, and reference material**. Each skill is defined by a `SKILL.md` file with YAML frontmatter and a markdown body.

-When activated, a skill's instructions are injected directly into the agent's task prompt — giving the agent expertise without requiring any code changes.
+Agents first receive each configured skill's name and description. When a
+description applies to the current request, the agent loads that skill's full
+instructions for that execution. This keeps unrelated instructions out of the
+context while giving the agent the relevant expertise without code changes.

 <Note type="info" title="Skills vs Tools — The Key Distinction">
 **Skills are NOT tools.** This is the most common point of confusion.
@@ -79,12 +82,13 @@ reviewer = Agent(
    role="Senior Code Reviewer",
    goal="Review pull requests for quality and security issues",
    backstory="Staff engineer with expertise in secure coding practices.",
-    skills=["./skills"],                          # Injects review guidelines
+    skills=["./skills"],                          # Discovers review skills
    tools=[GithubSearchTool(), FileReadTool()],   # Lets agent read code
 )
 ```

-The agent now has both **expertise** (from the skill) and **capabilities** (from the tools).
+The agent now has both **expertise** (loaded from the relevant skill when
+needed) and **capabilities** (from the tools).

 ---

@@ -324,7 +328,8 @@ The directory name must match the `name` field in `SKILL.md`. The `scripts/`, `r

 ## Pre-loading Skills

-For more control, you can discover and activate skills programmatically:
+For more control, you can discover and activate skills programmatically.
+Passing an activated skill makes its instructions always-on:

 ```python
 from pathlib import Path
@@ -351,12 +356,21 @@ agent = Agent(

 Skills use **progressive disclosure** — only loading what's needed at each stage:

-| Stage      | What's loaded                         | When                |
-| :--------- | :------------------------------------ | :------------------ |
-| Discovery  | Name, description, frontmatter fields | `discover_skills()` |
-| Activation | Full SKILL.md body text               | `activate_skill()`  |
+| Stage      | What's loaded                         | When                                      |
+| :--------- | :------------------------------------ | :---------------------------------------- |
+| Discovery  | Name, description, frontmatter fields | Agent setup or `discover_skills()`        |
+| Activation | Full SKILL.md body text               | Relevant runtime request or `activate_skill()` |
+| Resources  | Resource directory catalog            | Explicit `load_resources()` call          |

-During normal agent execution (passing directory paths via `skills=["./skills"]`), skills are automatically discovered and activated. The progressive loading only matters when using the programmatic API.
+With `skills=["./skills"]`, the directory is discovered at setup but the full
+instructions are not placed in every prompt. The agent reviews the metadata on
+each execution and loads only a skill that applies. The loaded instructions are
+scoped to that execution, so skills selected for earlier calls do not accumulate
+on the agent.
+
+Inline skill strings and `Skill` objects already activated with
+`activate_skill()` remain always-on. This provides an explicit opt-in when the
+instructions should apply to every request.

 ---

--- a/docs/edge/ko/concepts/event-listener.mdx
+++ b/docs/edge/ko/concepts/event-listener.mdx
@@ -255,6 +255,7 @@ CrewAI는 여러분이 청취할 수 있는 다양한 이벤트를 제공합니
 - **FlowCreatedEvent**: Flow가 생성될 때 발생
 - **FlowStartedEvent**: Flow가 실행을 시작할 때 발생
 - **FlowFinishedEvent**: Flow가 실행을 완료할 때 발생
+- **FlowFailedEvent**: Flow 실행이 실패할 때 발생합니다. Flow 이름과 실행을 종료시킨 예외를 포함합니다.
 - **FlowPausedEvent**: 사람의 피드백을 기다리며 Flow가 일시 중지될 때 발생합니다. Flow 이름, Flow ID, 메서드 이름, 현재 상태, 피드백 요청 시 표시되는 메시지, 라우팅을 위한 선택적 결과 목록을 포함합니다.
 - **FlowPlotEvent**: Flow가 플롯될 때 발생
 - **MethodExecutionStartedEvent**: Flow 메서드가 실행을 시작할 때 발생
--- a/docs/edge/pt-BR/concepts/event-listener.mdx
+++ b/docs/edge/pt-BR/concepts/event-listener.mdx
@@ -256,6 +256,7 @@ O CrewAI fornece uma ampla variedade de eventos para escuta:
 - **FlowCreatedEvent**: Emitido ao criar um Flow
 - **FlowStartedEvent**: Emitido ao iniciar a execução de um Flow
 - **FlowFinishedEvent**: Emitido ao concluir a execução de um Flow
+- **FlowFailedEvent**: Emitido quando a execução de um Flow falha. Contém o nome do flow e a exceção que encerrou a execução.
 - **FlowPausedEvent**: Emitido quando um Flow é pausado aguardando feedback humano. Contém o nome do flow, ID do flow, nome do método, estado atual, mensagem exibida ao solicitar feedback e lista opcional de resultados possíveis para roteamento.
 - **FlowPlotEvent**: Emitido ao plotar um Flow
 - **MethodExecutionStartedEvent**: Emitido ao iniciar a execução de um método do Flow
--- a/lib/crewai/src/crewai/agent/core.py
+++ b/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 = {
--- a/lib/crewai/src/crewai/crews/utils.py
+++ b/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(
--- a/lib/crewai/src/crewai/events/init.py
+++ b/lib/crewai/src/crewai/events/init.py
@@ -68,6 +68,7 @@ if TYPE_CHECKING:
        ConversationTurnStartedEvent,
        FlowCreatedEvent,
        FlowEvent,
+        FlowFailedEvent,
        FlowFinishedEvent,
        FlowPlotEvent,
        FlowStartedEvent,
@@ -194,6 +195,7 @@ _LAZY_EVENT_MAPPING: dict[str, str] = {
    "ConversationTurnStartedEvent": "crewai.events.types.flow_events",
    "FlowCreatedEvent": "crewai.events.types.flow_events",
    "FlowEvent": "crewai.events.types.flow_events",
+    "FlowFailedEvent": "crewai.events.types.flow_events",
    "FlowFinishedEvent": "crewai.events.types.flow_events",
    "FlowPlotEvent": "crewai.events.types.flow_events",
    "FlowStartedEvent": "crewai.events.types.flow_events",
@@ -329,6 +331,7 @@ __all__ = [
    "Depends",
    "FlowCreatedEvent",
    "FlowEvent",
+    "FlowFailedEvent",
    "FlowFinishedEvent",
    "FlowPlotEvent",
    "FlowStartedEvent",
--- a/lib/crewai/src/crewai/events/event_context.py
+++ b/lib/crewai/src/crewai/events/event_context.py
@@ -269,6 +269,7 @@ SCOPE_STARTING_EVENTS: frozenset[str] = frozenset(
 SCOPE_ENDING_EVENTS: frozenset[str] = frozenset(
    {
        "flow_finished",
+        "flow_failed",
        "flow_paused",
        "method_execution_finished",
        "method_execution_failed",
@@ -320,6 +321,7 @@ SCOPE_ENDING_EVENTS: frozenset[str] = frozenset(

 VALID_EVENT_PAIRS: dict[str, str] = {
    "flow_finished": "flow_started",
+    "flow_failed": "flow_started",
    "flow_paused": "flow_started",
    "method_execution_finished": "method_execution_started",
    "method_execution_failed": "method_execution_started",
--- a/lib/crewai/src/crewai/events/event_listener.py
+++ b/lib/crewai/src/crewai/events/event_listener.py
@@ -43,6 +43,7 @@ from crewai.events.types.env_events import (
 from crewai.events.types.flow_events import (
    ConversationTurnCompletedEvent,
    FlowCreatedEvent,
+    FlowFailedEvent,
    FlowFinishedEvent,
    FlowPausedEvent,
    FlowStartedEvent,
@@ -318,6 +319,15 @@ class EventListener(BaseEventListener):
                    source.flow_id,
                )

+        @crewai_event_bus.on(FlowFailedEvent)
+        def on_flow_failed(source: Any, event: FlowFailedEvent) -> None:
+            if not getattr(source, "suppress_flow_events", False):
+                self.formatter.handle_flow_status(
+                    event.flow_name,
+                    source.flow_id,
+                    "failed",
+                )
+
        @crewai_event_bus.on(ConversationTurnCompletedEvent)
        def on_conversation_turn_completed(
            _: Any, event: ConversationTurnCompletedEvent
--- a/lib/crewai/src/crewai/events/event_types.py
+++ b/lib/crewai/src/crewai/events/event_types.py
@@ -58,6 +58,7 @@ from crewai.events.types.flow_events import (
    ConversationTurnCompletedEvent,
    ConversationTurnFailedEvent,
    ConversationTurnStartedEvent,
+    FlowFailedEvent,
    FlowFinishedEvent,
    FlowStartedEvent,
    MethodExecutionFailedEvent,
@@ -170,6 +171,7 @@ EventTypes = (
    | ConversationTurnStartedEvent
    | FlowStartedEvent
    | FlowFinishedEvent
+    | FlowFailedEvent
    | MethodExecutionStartedEvent
    | MethodExecutionFinishedEvent
    | MethodExecutionFailedEvent
--- a/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py
+++ b/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py
@@ -66,6 +66,7 @@ from crewai.events.types.flow_events import (
    ConversationMessageAddedEvent,
    ConversationRouteSelectedEvent,
    FlowCreatedEvent,
+    FlowFailedEvent,
    FlowFinishedEvent,
    FlowPlotEvent,
    FlowStartedEvent,
@@ -274,6 +275,10 @@ class TraceCollectionListener(BaseEventListener):
        def on_flow_finished(source: Any, event: FlowFinishedEvent) -> None:
            self._handle_trace_event("flow_finished", source, event)

+        @event_bus.on(FlowFailedEvent)
+        def on_flow_failed(source: Any, event: FlowFailedEvent) -> None:
+            self._handle_trace_event("flow_failed", source, event)
+
        @event_bus.on(FlowPlotEvent)
        def on_flow_plot(source: Any, event: FlowPlotEvent) -> None:
            self._handle_action_event("flow_plot", source, event)
--- a/lib/crewai/src/crewai/events/types/flow_events.py
+++ b/lib/crewai/src/crewai/events/types/flow_events.py
@@ -94,6 +94,24 @@ class FlowFinishedEvent(FlowEvent):
    state: dict[str, Any] | BaseModel


+class FlowFailedEvent(FlowEvent):
+    """Event emitted when a flow execution fails.
+
+    Attributes:
+        flow_name: Name of the flow that failed.
+        error: The exception that ended the execution.
+    """
+
+    error: Exception
+    type: Literal["flow_failed"] = "flow_failed"
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    @field_serializer("error")
+    def _serialize_error(self, error: Exception) -> str:
+        return str(error)
+
+
 class FlowPausedEvent(FlowEvent):
    """Event emitted when a flow is paused waiting for human feedback.

--- a/lib/crewai/src/crewai/events/types/skill_events.py
+++ b/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"
--- a/lib/crewai/src/crewai/flow/runtime/init.py
+++ b/lib/crewai/src/crewai/flow/runtime/init.py
@@ -67,6 +67,7 @@ from crewai.events.listeners.tracing.utils import (
 )
 from crewai.events.types.flow_events import (
    FlowCreatedEvent,
+    FlowFailedEvent,
    FlowFinishedEvent,
    FlowPausedEvent,
    FlowPlotEvent,
@@ -1341,6 +1342,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
        except Exception as e:
            if not hook_state["end_dispatched"]:
                self._dispatch_execution_end_failure(e)
+            await self._emit_flow_failed(e, respect_suppression=True)
            raise
        finally:
            # Match kickoff_async: drain pending handlers so the resumed
@@ -1382,35 +1384,68 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
            )
        emit = context.emit

-        # The serialized context carries the full LLM config (a dict, or a
-        # legacy model string) — the single source for cross- and same-process
-        # resume.
-        result = await self._finalize_human_feedback(
-            method_name=context.method_name,
-            method_output=context.method_output,
-            raw_feedback=feedback,
-            emit=emit,
-            default_outcome=context.default_outcome,
-            llm=context.llm,
-            metadata=context.metadata,
-        )
-        collapsed_outcome = result.outcome
-        resumed_method_output = (
-            result.output
-            if emit and isinstance(result, HumanFeedbackResult)
-            else result
-        )
+        if not self.suppress_flow_events:
+            # Opens the scope the finished event below closes; without it that
+            # event pops the enclosing ``flow_started`` instead.
+            future = crewai_event_bus.emit(
+                self,
+                MethodExecutionStartedEvent(
+                    type="method_execution_started",
+                    flow_name=self._definition.name,
+                    method_name=context.method_name,
+                    state=self._copy_and_serialize_state(),
+                ),
+            )
+            if future and isinstance(future, Future):
+                try:
+                    await asyncio.wrap_future(future)
+                except Exception:
+                    logger.warning(
+                        "MethodExecutionStartedEvent handler failed", exc_info=True
+                    )

-        self._completed_methods.add(FlowMethodName(context.method_name))
+        try:
+            # The serialized context carries the full LLM config (a dict, or a
+            # legacy model string) — the single source for cross- and
+            # same-process resume.
+            result = await self._finalize_human_feedback(
+                method_name=context.method_name,
+                method_output=context.method_output,
+                raw_feedback=feedback,
+                emit=emit,
+                default_outcome=context.default_outcome,
+                llm=context.llm,
+                metadata=context.metadata,
+            )
+            collapsed_outcome = result.outcome
+            resumed_method_output = (
+                result.output
+                if emit and isinstance(result, HumanFeedbackResult)
+                else result
+            )

-        await asyncio.to_thread(
-            self._persist_method_completion, FlowMethodName(context.method_name)
-        )
+            self._completed_methods.add(FlowMethodName(context.method_name))

-        self._pending_feedback_context = None
+            await asyncio.to_thread(
+                self._persist_method_completion, FlowMethodName(context.method_name)
+            )

-        if self.persistence is not None:
-            self.persistence.clear_pending_feedback(context.flow_id)
+            self._pending_feedback_context = None
+
+            if self.persistence is not None:
+                self.persistence.clear_pending_feedback(context.flow_id)
+        except Exception as e:
+            if not self.suppress_flow_events:
+                crewai_event_bus.emit(
+                    self,
+                    MethodExecutionFailedEvent(
+                        type="method_execution_failed",
+                        flow_name=self._definition.name,
+                        method_name=context.method_name,
+                        error=e,
+                    ),
+                )
+            raise

        if not self.suppress_flow_events:
            crewai_event_bus.emit(
@@ -2096,6 +2131,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
        # EXECUTION_START/EXECUTION_END dispatch independently.
        execution_start_dispatched = False
        execution_end_dispatched = False
+        # Guards the failure event: everything between here and the
+        # ``flow_started`` emission below (hooks, input handling, state
+        # restore) can raise, and a ``flow_failed`` with no opener would pop
+        # an unrelated scope.
+        flow_scope_open = False

        try:
            from crewai.hooks.contexts import (
@@ -2235,6 +2275,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                and get_current_parent_id() is None
            ):
                restore_event_scope(((deferred_started_event_id, "flow_started"),))
+                flow_scope_open = True
            elif get_current_parent_id() is None:
                reset_emission_counter()
                reset_last_event_id()
@@ -2249,6 +2290,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                    inputs=inputs,
                )
                future = crewai_event_bus.emit(self, started_event)
+                flow_scope_open = True
                if future:
                    try:
                        await asyncio.wrap_future(future)
@@ -2440,6 +2482,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
            # not (exactly-once per invocation).
            if execution_start_dispatched and not execution_end_dispatched:
                self._dispatch_execution_end_failure(e)
+            if flow_scope_open:
+                await self._emit_flow_failed(e)
            raise
        finally:
            # Safety net for the exception path; the success path already
@@ -2487,6 +2531,67 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
        except Exception:  # noqa: S110 - aborting an already-failed execution is meaningless
            pass

+    async def _emit_flow_failed(
+        self, error: Exception, *, respect_suppression: bool = False
+    ) -> None:
+        """Emit ``FlowFailedEvent`` and close out the trace batch for a failed run.
+
+        Mirrors the terminal block of the success path: drain pending event
+        handlers and background memory saves so their spans close before the
+        flow span does, then emit and finalize the trace batch. Never raises,
+        so the original exception propagates unchanged.
+
+        Args:
+            error: The exception that ended the execution.
+            respect_suppression: Skip the emission for suppressed flows. Only
+                the resume path gates its lifecycle events on
+                ``suppress_flow_events``; ``kickoff_async`` emits them either
+                way and lets listeners filter.
+        """
+        if self._should_defer_trace_finalization():
+            return
+        if respect_suppression and self.suppress_flow_events:
+            return
+
+        try:
+            if self._event_futures:
+                await asyncio.gather(
+                    *[asyncio.wrap_future(f) for f in self._event_futures],
+                    return_exceptions=True,
+                )
+                self._event_futures.clear()
+
+            await asyncio.to_thread(self._drain_memory_writes)
+            await asyncio.to_thread(crewai_event_bus.flush)
+            future = crewai_event_bus.emit(
+                self,
+                FlowFailedEvent(
+                    type="flow_failed",
+                    flow_name=self._definition.name,
+                    error=error,
+                ),
+            )
+            if future and isinstance(future, Future):
+                try:
+                    await asyncio.wrap_future(future)
+                except Exception:
+                    logger.warning("FlowFailedEvent handler failed", exc_info=True)
+
+            trace_listener = TraceCollectionListener()
+            if (
+                trace_listener.batch_manager.batch_owner_type == "flow"
+                and current_flow_id.get() == self.flow_id
+                and not trace_listener.batch_manager.defer_session_finalization
+                and not current_flow_defer_trace_finalization.get()
+            ):
+                if trace_listener.first_time_handler.is_first_time:
+                    trace_listener.first_time_handler.mark_events_collected()
+                    trace_listener.first_time_handler.handle_execution_completion()
+                else:
+                    trace_listener.batch_manager.finalize_batch()
+        except Exception:
+            logger.warning("Failed to signal flow failure", exc_info=True)
+
    async def akickoff(
        self,
        inputs: dict[str, Any] | None = None,
--- a/lib/crewai/src/crewai/skills/loader.py
+++ b/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>'
--- a/lib/crewai/src/crewai/skills/registry.py
+++ b/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
--- a/lib/crewai/src/crewai/skills/tool.py
+++ b/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,
+    )
--- a/lib/crewai/src/crewai/state/checkpoint_config.py
+++ b/lib/crewai/src/crewai/state/checkpoint_config.py
@@ -38,6 +38,7 @@ CheckpointEventType = Literal[
    "flow_created",
    "flow_started",
    "flow_finished",
+    "flow_failed",
    "flow_paused",
    "method_execution_started",
    "method_execution_finished",
--- a/lib/crewai/src/crewai/utilities/prompts.py
+++ b/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,
--- a/lib/crewai/tests/events/test_skill_usage_events.py
+++ b/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)
--- a/lib/crewai/tests/skills/test_integration.py
+++ b/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]
--- a/lib/crewai/tests/skills/test_progressive_disclosure.py
+++ b/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"]
--- a/lib/crewai/tests/skills/test_registry.py
+++ b/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:
--- a/lib/crewai/tests/utilities/test_events.py
+++ b/lib/crewai/tests/utilities/test_events.py
@@ -23,6 +23,7 @@ from crewai.events.types.crew_events import (
 )
 from crewai.events.types.flow_events import (
    FlowCreatedEvent,
+    FlowFailedEvent,
    FlowFinishedEvent,
    FlowStartedEvent,
    HumanFeedbackReceivedEvent,
@@ -46,8 +47,11 @@ from crewai.events.types.tool_usage_events import (
    ToolUsageErrorEvent,
    ToolUsageFinishedEvent,
 )
+from crewai.flow.async_feedback.types import PendingFeedbackContext
 from crewai.flow.flow import Flow, listen, start
 from crewai.flow.human_feedback import human_feedback
+from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
+from crewai.hooks.dispatch import HookAborted, InterceptionPoint, clear_all, on
 from crewai.llm import LLM
 from crewai.task import Task
 from crewai.tools.base_tool import BaseTool
@@ -556,6 +560,274 @@ def test_flow_emits_finish_event():
    assert result == "completed"


+def test_flow_emits_failed_event_paired_with_started_event():
+    started: list[FlowStartedEvent] = []
+    failed: list[FlowFailedEvent] = []
+
+    class BoomFlow(Flow[dict]):
+        @start()
+        def begin(self):
+            raise RuntimeError("boom")
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(FlowStartedEvent)
+        def handle_flow_started(source, event):
+            started.append(event)
+
+        @crewai_event_bus.on(FlowFailedEvent)
+        def handle_flow_failed(source, event):
+            failed.append(event)
+
+        with pytest.raises(RuntimeError, match="boom"):
+            BoomFlow().kickoff()
+        wait_for_event_handlers()
+
+    assert len(failed) == 1
+    assert failed[0].type == "flow_failed"
+    assert failed[0].flow_name == "BoomFlow"
+    assert isinstance(failed[0].error, RuntimeError)
+    assert str(failed[0].error) == "boom"
+    assert failed[0].started_event_id == started[0].event_id
+
+
+def test_suppressed_flow_failure_matches_finished_event_emission():
+    finished: list[FlowFinishedEvent] = []
+    failed: list[FlowFailedEvent] = []
+
+    class SuppressedFlow(Flow):
+        suppress_flow_events: bool = True
+
+        @start()
+        def begin(self):
+            return "ok"
+
+    class SuppressedBoomFlow(Flow):
+        suppress_flow_events: bool = True
+
+        @start()
+        def begin(self):
+            raise RuntimeError("boom")
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(FlowFinishedEvent)
+        def handle_flow_finished(source, event):
+            finished.append(event)
+
+        @crewai_event_bus.on(FlowFailedEvent)
+        def handle_flow_failed(source, event):
+            failed.append(event)
+
+        SuppressedFlow().kickoff()
+        with pytest.raises(RuntimeError, match="boom"):
+            SuppressedBoomFlow().kickoff()
+        wait_for_event_handlers()
+
+    assert len(finished) == 1
+    assert len(failed) == 1
+
+
+def test_abort_before_flow_started_emits_no_failed_event():
+    started: list[FlowStartedEvent] = []
+    failed: list[FlowFailedEvent] = []
+
+    class BlockedFlow(Flow):
+        @start()
+        def begin(self) -> str:
+            return "never runs"
+
+    clear_all()
+    try:
+
+        @on(InterceptionPoint.EXECUTION_START)
+        def block(_ctx):
+            raise HookAborted(reason="blocked by policy")
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(FlowStartedEvent)
+            def handle_flow_started(source, event):
+                started.append(event)
+
+            @crewai_event_bus.on(FlowFailedEvent)
+            def handle_flow_failed(source, event):
+                failed.append(event)
+
+            with pytest.raises(HookAborted):
+                BlockedFlow().kickoff()
+            wait_for_event_handlers()
+    finally:
+        clear_all()
+
+    assert started == []
+    assert failed == []
+
+
+def test_resume_emits_failed_event_paired_with_resume_started_event(tmp_path):
+    started: list[FlowStartedEvent] = []
+    failed: list[FlowFailedEvent] = []
+
+    class ResumeBoomFlow(Flow):
+        @start()
+        def begin(self) -> str:
+            return "content"
+
+        @listen(begin)
+        def after_feedback(self, _feedback):
+            raise RuntimeError("boom on resume")
+
+    persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
+    flow_id = "resume-failure-test"
+    persistence.save_pending_feedback(
+        flow_uuid=flow_id,
+        context=PendingFeedbackContext(
+            flow_id=flow_id,
+            flow_class="ResumeBoomFlow",
+            method_name="begin",
+            method_output="content",
+            message="Review:",
+        ),
+        state_data={"id": flow_id},
+    )
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(FlowStartedEvent)
+        def handle_flow_started(source, event):
+            started.append(event)
+
+        @crewai_event_bus.on(FlowFailedEvent)
+        def handle_flow_failed(source, event):
+            failed.append(event)
+
+        flow = ResumeBoomFlow.from_pending(flow_id, persistence)
+        with pytest.raises(RuntimeError, match="boom on resume"):
+            flow.resume("ok")
+        wait_for_event_handlers()
+
+    assert len(started) == 1
+    assert len(failed) == 1
+    assert failed[0].flow_name == "ResumeBoomFlow"
+    assert str(failed[0].error) == "boom on resume"
+    assert failed[0].started_event_id == started[0].event_id
+
+
+def test_resume_pairs_resumed_method_events_with_their_own_scope(tmp_path):
+    started: list[FlowStartedEvent] = []
+    finished: list[FlowFinishedEvent] = []
+    method_started: list[MethodExecutionStartedEvent] = []
+    method_finished: list[MethodExecutionFinishedEvent] = []
+
+    class ResumeFlow(Flow):
+        @start()
+        def begin(self) -> str:
+            return "content"
+
+        @listen(begin)
+        def after_feedback(self, _feedback):
+            return "done"
+
+    persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
+    flow_id = "resume-pairing-test"
+    persistence.save_pending_feedback(
+        flow_uuid=flow_id,
+        context=PendingFeedbackContext(
+            flow_id=flow_id,
+            flow_class="ResumeFlow",
+            method_name="begin",
+            method_output="content",
+            message="Review:",
+        ),
+        state_data={"id": flow_id},
+    )
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(FlowStartedEvent)
+        def handle_flow_started(source, event):
+            started.append(event)
+
+        @crewai_event_bus.on(FlowFinishedEvent)
+        def handle_flow_finished(source, event):
+            finished.append(event)
+
+        @crewai_event_bus.on(MethodExecutionStartedEvent)
+        def handle_method_started(source, event):
+            method_started.append(event)
+
+        @crewai_event_bus.on(MethodExecutionFinishedEvent)
+        def handle_method_finished(source, event):
+            method_finished.append(event)
+
+        ResumeFlow.from_pending(flow_id, persistence).resume("ok")
+        wait_for_event_handlers()
+
+    resumed_started = next(e for e in method_started if e.method_name == "begin")
+    resumed_finished = next(e for e in method_finished if e.method_name == "begin")
+
+    assert resumed_finished.started_event_id == resumed_started.event_id
+    assert finished[0].started_event_id == started[0].event_id
+
+
+def test_resume_failing_before_method_finishes_keeps_flow_pairing(tmp_path):
+    started: list[FlowStartedEvent] = []
+    failed: list[FlowFailedEvent] = []
+    method_failed: list[MethodExecutionFailedEvent] = []
+
+    class ResumeFlow(Flow):
+        @start()
+        def begin(self) -> str:
+            return "content"
+
+        @listen(begin)
+        def after_feedback(self, _feedback):
+            return "done"
+
+    persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
+    flow_id = "resume-finalize-failure-test"
+    persistence.save_pending_feedback(
+        flow_uuid=flow_id,
+        context=PendingFeedbackContext(
+            flow_id=flow_id,
+            flow_class="ResumeFlow",
+            method_name="begin",
+            method_output="content",
+            message="Review:",
+        ),
+        state_data={"id": flow_id},
+    )
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(FlowStartedEvent)
+        def handle_flow_started(source, event):
+            started.append(event)
+
+        @crewai_event_bus.on(FlowFailedEvent)
+        def handle_flow_failed(source, event):
+            failed.append(event)
+
+        @crewai_event_bus.on(MethodExecutionFailedEvent)
+        def handle_method_failed(source, event):
+            method_failed.append(event)
+
+        flow = ResumeFlow.from_pending(flow_id, persistence)
+        with patch.object(
+            Flow,
+            "_finalize_human_feedback",
+            side_effect=RuntimeError("feedback collapse failed"),
+        ):
+            with pytest.raises(RuntimeError, match="feedback collapse failed"):
+                flow.resume("ok")
+        wait_for_event_handlers()
+
+    assert len(method_failed) == 1
+    assert method_failed[0].method_name == "begin"
+    assert len(failed) == 1
+    assert failed[0].started_event_id == started[0].event_id
+
+
 def test_flow_emits_method_execution_started_event():
    received_events = []
    lock = threading.Lock()