depends on D8896
introducing a generic pipeline_sensor so users can define sensors that react to pipeline statuses like starting, started, success, etc
@pipeline_sensor(pipeline_run_status=PipelineRunStatus.STARTED) def my_pipeline_started_sensor(_): ...
it takes pipeline_run_status arg which internally will be mapped to a dagster event type (lookup in D8896)
changes on implementation details:
- created PipelineSensorDefinition
- built pipeline_failure_sensor on top of this generic