Difference between prefect.engine andprefect flow-run start

[johannesloibl]

Hey there,

i'm currently trying to implement a custom runner but i'm finding different ways on how the runner is actually executing the flow run.
The base flow run command defined in the BaseJobConfiguration is prefect flow-run execute, but the ProcessJobConfiguration is using python -m prefect.engine.

I'm currently using python -m prefect.engine and it's working fine so far, but i don't know the full implications.

Could someone describe where is the difference, pros/cons of the different ways and is one or the other recommended for certain use cases?

[bdalpe]

TL;DR: Summary in one line

engine.py __main__ is the low-level “execute this flow run in this process” entrypoint, while flow_run.execute() is a runner-mediated orchestration command that manages process lifecycle, state transitions, cancellation/rescheduling, storage pulls, and environment setup—often by launching engine.py in a subprocess.

For example, running a Flow on a Kubernetes worker/runner results in a new Pod being created with the container arguments being set to prefect flow-run execute and PREFECT__FLOW_RUN_ID environment set to the GUID of the Flow run ID.

1) Execution model: in-process vs runner-managed subprocess

engine.py __main__ executes the flow directly in-process after loading the flow run and flow, then calling run_flow (sync or async). It’s essentially “run this flow run right now, in this process.”

with handle_engine_signals(flow_run_id):
    from prefect.flow_engine import (
        flow_run_logger,
        load_flow,
        load_flow_run,
        run_flow,
    )

    flow_run: "FlowRun" = load_flow_run(flow_run_id=flow_run_id)
    run_logger: "LoggingAdapter" = flow_run_logger(flow_run=flow_run)

    try:
        flow: "Flow[..., Any]" = load_flow(flow_run)
    except Exception:
        run_logger.error(
            "Unexpected exception encountered when trying to load flow",
            exc_info=True,
        )
        raise

    # run the flow
    if flow.isasync:
        run_coro_as_sync(run_flow(flow, flow_run=flow_run, error_logger=run_logger))
    else:
        run_flow(flow, flow_run=flow_run, error_logger=run_logger)

By contrast, flow_run.execute() delegates to Runner.execute_flow_run, which usually runs a subprocess and monitors it.

runner = Runner()
...
await runner.execute_flow_run(id)

The runner path sets up execution using _run_process, which typically spawns a subprocess running python -m prefect.engine (i.e. your engine.py __main__ entrypoint) and injects PREFECT__FLOW_RUN_ID into the environment.

# Otherwise, we'll need to run a `python -m prefect.engine` command to load and run the flow
if command is None:
    runner_command = [get_sys_executable(), "-m", "prefect.engine"]

...
env.update(
    {
        **{
            "PREFECT__FLOW_RUN_ID": str(flow_run.id),
            "PREFECT__STORAGE_BASE_PATH": str(self._tmp_dir),
            "PREFECT__ENABLE_CANCELLATION_AND_CRASHED_HOOKS": "false",
        },
        **({"PREFECT__FLOW_ENTRYPOINT": entrypoint} if entrypoint else {}),
    }
)
...
process = await run_process(command=runner_command, ...)

So the CLI execute() path is “start a runner, which starts a subprocess that runs prefect.engine.”

---

2) State orchestration & cancellation handling

engine.py uses handle_engine_signals to interpret Abort / Pause / unexpected exceptions into log messages and exit codes. This is local process-level handling.

@contextmanager
def handle_engine_signals(flow_run_id: UUID | None = None):
    try:
        yield
    except Abort:
        ...
        exit(0)
    except Pause:
        ...
        exit(0)
    except Exception:
        ...
        exit(1)
    except BaseException:
        ...
        raise

The runner path coordinates with the API and monitors cancellation and state transitions for you:

• Reads the flow run from the API
• Checks for cancelling/cancelled before start
• Proposes pending state
• Emits heartbeats (optional)
• Logs exit codes and proposes crashed states when needed

flow_run = await self._client.read_flow_run(flow_run_id)

if flow_run.state and flow_run.state.is_cancelling():
    await self._mark_flow_run_as_cancelled(...)
    ...
    return

if flow_run.state and flow_run.state.is_cancelled():
    ...
    return

process = await self._runs_task_group.start(
    partial(self._submit_run_and_capture_errors, ...)
)
...
if self.heartbeat_seconds is not None:
    await self._emit_flow_run_heartbeat(flow_run)

And when the process exits, it logs exit status and proposes crashed state if needed.

if exit_code:
    flow_run_logger.log(...)

...
if exit_code != 0 and not self._rescheduling:
    await self._propose_crashed_state(
        flow_run,
        f"Flow run process exited with non-zero status code {exit_code}.",
    )

This extra orchestration does not happen in engine.py’s direct execution path.

---

3) Signal handling differences (SIGTERM behavior)

engine.py doesn’t set any special signal handlers beyond handle_engine_signals (which is about Prefect exceptions, not OS signals).

flow_run.execute() adds an optional SIGTERM handler that reschedules the flow run if PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=reschedule.

def _handle_reschedule_sigterm(_signal: int, _frame: FrameType | None):
    logger.info("SIGTERM received, initiating graceful shutdown...")
    runner.reschedule_current_flow_runs()
    exit_with_success("Flow run successfully rescheduled.")

...
if ... and on_sigterm == "reschedule":
    signal.signal(signal.SIGTERM, _handle_reschedule_sigterm)

The reschedule logic is runner-based and updates state to AwaitingRetry, then SIGTERM’s the child process.

def reschedule_current_flow_runs(self) -> None:
    self._rescheduling = True
    ...
    propose_state_sync(client, AwaitingRetry(), flow_run_id=flow_run.id)
    os.kill(process_info["pid"], signal.SIGTERM)

---

4) Storage & deployment-specific behavior

Runner execution has deployment-aware behavior:

• If the runner already has the flow in its map, it can run directly in a subprocess without spawning python -m prefect.engine.
• Otherwise it starts prefect.engine and sets PREFECT__FLOW_ENTRYPOINT.
• It can perform an ad‑hoc code pull for remote storage before running.

if flow_run.deployment_id is not None:
    flow = self._deployment_flow_map.get(flow_run.deployment_id)
    if flow:
        process = run_flow_in_subprocess(flow, flow_run=flow_run)
        ...
        return process.exitcode

...
storage = self._deployment_storage_map.get(flow_run.deployment_id) ...
if storage and storage.pull_interval:
    ...
    await storage.pull_code()

The engine.py __main__ path doesn’t do any of this orchestration or storage handling by itself; it just loads and runs the flow from the flow run.

---

5) Environment and config propagation

Runner builds a process environment by combining settings + run-specific vars + current OS environment.

env.update(get_current_settings().to_environment_variables(exclude_unset=True))
env.update({
    "PREFECT__FLOW_RUN_ID": str(flow_run.id),
    "PREFECT__STORAGE_BASE_PATH": str(self._tmp_dir),
    "PREFECT__ENABLE_CANCELLATION_AND_CRASHED_HOOKS": "false",
    ...
})
env.update(**os.environ)

engine.py __main__ uses whatever environment it’s launched with.

[johannesloibl]

Thanks Brandon for the clarification,

so can i use option 2 in the following scenario?

• Custom Worker lives inside Kubernetes
• Logs into a Linux compute environment via SSH
• Creates a temporary working directory
• Templates a bash script by using the deployment pull steps and job config as context
• Note: In the compute environment there is no existing Python environment with Prefect installed, such as in a Docker container, so we have to recreate it each time. Currently i'm using python -m prefect.engine because by setting the PREFECT__FLOW_ENTRYPOINT env var i could tell the runner to skip the pull steps and directly execute the flow (because the bash script will set everything up)
• With the temp dir as CWD, schedules the bash script via IBM LSF to a compute node
• The bash script will then:
  • Execute the pull steps (only GIT supported)
  • Prepares UV
  • Sets the PREFECT__FLOW_RUN_ID env var
  • Sets the PREFECT__FLOW_ENTRYPOINT env var
  • Executes uv run python -m prefect.engine within the repo (pyproject.toml/uv.lock specify prefect as dependency)
• The worker then:
  • monitors the LSF job via SSH for crashes and flow-run external logs.
  • polls the flow run for Cancellation state and maybe kills the LSF job (currently i'm using python -m prefect.engine so with prefect flow-run execute i might not need this anymore)

Does this sound right?