emit events for each deployment pull step execution

src/prefect/deployments/steps/core.py

@@ -19,12 +19,15 @@
 import warnings
 from copy import deepcopy
 from importlib import import_module
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any
+from uuid import UUID
 
 from prefect._internal.compatibility.deprecated import PrefectDeprecationWarning
 from prefect._internal.concurrency.api import Call, from_async
 from prefect._internal.installation import install_packages
 from prefect._internal.integrations import KNOWN_EXTRAS_FOR_PACKAGES
+from prefect.events.clients import get_events_client
+from prefect.events.schemas.events import Event, RelatedResource, Resource
 from prefect.logging.loggers import get_logger
 from prefect.settings import PREFECT_DEBUG_MODE
 from prefect.utilities.importtools import import_object
@@ -65,7 +68,7 @@ def _strip_version(requirement: str) -> str:
 
 
 def _get_function_for_step(
-    fully_qualified_name: str, requires: Union[str, List[str], None] = None
+    fully_qualified_name: str, requires: str | list[str] | None = None
 ):
     if not isinstance(requires, list):
         packages = [requires] if requires else []
@@ -103,7 +106,7 @@ def _get_function_for_step(
 
 
 async def run_step(
-    step: dict[str, Any], upstream_outputs: Optional[dict[str, Any]] = None
+    step: dict[str, Any], upstream_outputs: dict[str, Any] | None = None
 ) -> dict[str, Any]:
     """
     Runs a step, returns the step's output.
@@ -141,17 +144,34 @@ async def run_step(
 
 
 async def run_steps(
-    steps: List[Dict[str, Any]],
-    upstream_outputs: Optional[Dict[str, Any]] = None,
+    steps: list[dict[str, Any]],
+    upstream_outputs: dict[str, Any] | None = None,
     print_function: Any = print,
+    deployment: Any | None = None,
+    flow_run: Any | None = None,
+    logger: Any | None = None,
 ) -> dict[str, Any]:
     upstream_outputs = deepcopy(upstream_outputs) if upstream_outputs else {}
-    for step in steps:
+    for step_index, step in enumerate(steps):
         if not step:
             continue
         fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)
         step_name = fqn.split(".")[-1]
         print_function(f" > Running {step_name} step...")
+
+        # SECURITY: Serialize inputs BEFORE running the step (and thus before templating).
+        # This ensures that the event payload contains template strings like
+        # "{{ prefect.blocks.secret.api-key }}" rather than resolved secret values.
+        # Templating (which resolves blocks, variables, and env vars) happens inside
+        # run_step(), so by serializing here we prevent secrets from leaking in events.
+        serialized_step = {
+            "index": step_index,
+            "qualified_name": fqn,
+            "step_name": step_name,
+            "id": inputs.get("id"),
+            "inputs": inputs,  # Keep all inputs including reserved keywords like 'requires'
+        }
+
         try:
             # catch warnings to ensure deprecation warnings are printed
             with warnings.catch_warnings(record=True) as w:
@@ -190,11 +210,88 @@ async def run_steps(
             if inputs.get("id"):
                 upstream_outputs[inputs.get("id")] = step_output
             upstream_outputs.update(step_output)
+
+            # Emit success event for this step
+            await _emit_pull_step_event(
+                serialized_step,
+                event_type="prefect.flow-run.pull-step.executed",
+                deployment=deployment,
+                flow_run=flow_run,
+                logger=logger,
+            )
         except Exception as exc:
+            # Emit failure event for this step
+            await _emit_pull_step_event(
+                serialized_step,
+                event_type="prefect.flow-run.pull-step.failed",
+                deployment=deployment,
+                flow_run=flow_run,
+                logger=logger,
+            )
             raise StepExecutionError(f"Encountered error while running {fqn}") from exc
+
     return upstream_outputs
 
 
-def _get_step_fully_qualified_name_and_inputs(step: Dict) -> Tuple[str, Dict]:
+def _get_step_fully_qualified_name_and_inputs(step: dict) -> tuple[str, dict]:
     step = deepcopy(step)
     return step.popitem()
+
+
+async def _emit_pull_step_event(
+    serialized_step: dict[str, Any],
+    *,
+    event_type: str,
+    deployment: Any | None = None,
+    flow_run: Any | None = None,
+    logger: Any | None = None,
+) -> None:
+    # Get flow_run_id from flow_run param or environment
+    flow_run_id = None
+    if flow_run:
+        flow_run_id = flow_run.id
+    else:
+        # Read directly from environment variable
+        flow_run_id_str = os.getenv("PREFECT__FLOW_RUN_ID")
+        if flow_run_id_str:
+            flow_run_id = UUID(flow_run_id_str)
+
+    if not flow_run_id:
+        return
+
+    # Build related resources
+    related: list[RelatedResource] = []
+    if deployment:
+        related.append(
+            RelatedResource(
+                {
+                    "prefect.resource.id": f"prefect.deployment.{deployment.id}",
+                    "prefect.resource.role": "deployment",
+                }
+            )
+        )
+
+    try:
+        # Use events client directly with checkpoint_every=1 to avoid buffering issues
+        async with get_events_client(checkpoint_every=1) as events_client:
+            await events_client.emit(
+                Event(
+                    event=event_type,
+                    resource=Resource(
+                        {
+                            "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
+                        }
+                    ),
+                    related=related,
+                    payload=serialized_step,
+                )
+            )
+    except Exception:
+        if logger:
+            logger.warning(
+                "Failed to emit pull-step event for flow run %s", flow_run_id
+            )
+        else:
+            get_logger(__name__).warning(
+                "Failed to emit pull-step event for flow run %s", flow_run_id
+            )

src/prefect/flows.py

@@ -2814,7 +2814,11 @@ async def load_flow_from_flow_run(
 
         try:
             output = await run_steps(
-                deployment.pull_steps, print_function=run_logger.info
+                deployment.pull_steps,
+                print_function=run_logger.info,
+                deployment=deployment,
+                flow_run=flow_run,
+                logger=run_logger,
             )
         except StepExecutionError as e:
             e = e.__cause__ or e