This diff moves the copy_intermediate_from_prev_run calls for retried pipelines from
the pipeline setup to the engine execution.
This is so that we can check the intermediates and skip steps for which we already have outputs.
Previously, we reused step outputs from the previous run, but still ran each step execution and
thus had to deal with any new failures.
Because we reconstitute the previous runs step output based on object_store operations, we also
need to emit object store operations for any events that are skipped (for future retries).
Moving the copy operations means that we transparently expose these operations as DagsterEvents
when we retry.
The downside of moving the copy operation to the engine is that every engine implementation must
now be aware of retries and handle this behavior explicitly.
This diff does not test for nor explicitly handle retries in the Dask engine.