emit events for each deployment pull step execution

[zzstoatzz]

summary

emits events for each deployment pull step execution, enabling the UI to display exact pull step history for flow runs. one event is emitted per step.

changes

event emission

• one event per step (not one aggregate event)
• event types:
  • prefect.flow-run.pull-step.executed - step completed successfully
  • prefect.flow-run.pull-step.failed - step failed
• resource: prefect.flow-run.<flow_run_id>
• related resources: includes deployment as related resource when available
• payload structure (per step):

  {
    "index": 0,
    "qualified_name": "prefect.deployments.steps.run_shell_script",
    "step_name": "run_shell_script",
    "id": "optional-step-id",
    "inputs": {...}
  }

• emits only when PREFECT__FLOW_RUN_ID is set (deployment runs, not local)
• uses get_events_client(checkpoint_every=1) to avoid buffering issues
• logs warning on emission failure without disrupting flow execution

security

• inputs contain pre-templated values (e.g., {{ prefect.blocks.secret.name }})
• secrets are NOT exposed in event payloads
• templating (which resolves secrets) happens after serialization
• added comprehensive test coverage for secret sanitization

implementation

• removed _EnvironmentRunContext class (over-engineered)
• read flow_run_id directly from os.getenv("PREFECT__FLOW_RUN_ID")
• pass flow_run and logger as parameters to run_steps() instead of using context
• inline step serialization directly in run_steps() loop
• keep all inputs including reserved keywords like 'requires'

testing

# run all pull steps tests
uv run pytest tests/deployment/test_steps.py::TestRunSteps -x

# specific tests:
# - test_run_steps_emits_pull_step_events (verifies one event per step)
# - test_run_steps_skips_event_without_flow_run_id
# - test_run_steps_emits_event_on_failure (verifies failed event emission)
# - test_run_steps_does_not_expose_secrets_in_event
# - test_run_steps_includes_deployment_as_related_resource

context

UI wants immutable record of executed pull steps since deployment configuration may change after flow run scheduling. emitting one event per step provides granular visibility into the execution timeline.

[codspeed-hq]

CodSpeed Performance Report

Merging #19339 will not alter performance

_{Comparing pull-step-events-spike (3874fc7) with main (f59e4c8)}

Summary

✅ 2 untouched

[znicholasbrown]

yusss

[zzstoatzz]

yusss

🙏 looking into one potential improvement here but not in a way that'd change the structure of the event

[desertaxle]

I thought we'd emit an event for each step. Is there a reason for emitting them all as one event?

[zzstoatzz]

what's the value in emitting one for each? it seemed unnecessary to do more than one

[kevingrismore]

the case @zzstoatzz made to me yesterday is that if one step fails, the rest of them aren't going to be executed anyway, so the maximum amount of data a user would need fits on a single event. I'm assuming the compromise there is observability, but pull steps are logged too. Is there a case for say, automating on pull steps outside of failure scenarios?

[desertaxle]

I doubt there's a use case for automating outside of a failure, but I think it'd be easier to write an automation against a prefect.pull-step.failed event than writing one against a prefect.pull-step.executed event with failed_step populated in the payload.

I think emitting an event for each pull step is overall more flexible with little downside.

[zzstoatzz]

@znicholasbrown the new event grammar in the PR body has been updated accordingly

[desertaxle]

I doubt there's a use case for automating outside of a failure, but I think it'd be easier to write an automation against a prefect.pull-step.failed event than writing one against a prefect.pull-step.executed event with failed_step populated in the payload.

I think emitting an event for each pull step is overall more flexible with little downside.