You’re seeing this because in Prefect 3 a flow is marked “completed” if the flow function finishes without raising. In your loop you’re catching the task exception and only logging it, so the flow completes. You have two good options: 1) Fail immediately on the first failing task - Re-raise the exception you catch so it bubbles up to the flow:

for executor in executors:
    checks: List[BaseCheck] = executor.rule.checks
    for check in checks:
        try:
            tasks.run_check(check=check, executor=executor, severity=severity)
        except Exception as e:
            logger.error(f"Error running check [{check.name}] with severity [{check.severity}]: {e}")
            raise

2) Run all checks, then fail the flow if any failed - Submit tasks to run (optionally concurrently) and only raise after collecting results:

futures = []

for executor in executors:
    checks: List[BaseCheck] = executor.rule.checks
    for check in checks:
        # submit returns a PrefectFuture
        fut = tasks.run_check.submit(check=check, executor=executor, severity=severity)
        futures.append((check, fut))

errors = []
for check, fut in futures:
    try:
        # raises on failure by default (raise_on_failure=True)
        fut.result()
    except Exception as exc:
        logger.error(f"Check failed [{check.name}] with severity [{getattr(check, 'severity', 'n/a')}]: {exc}")
        errors.append(exc)

if errors:
    # raise a summary error so the flow is marked failed
    raise Exception(f"{len(errors)} check(s) failed")

Notes: - Using

.submit(...)

gives you `PrefectFuture`s;

.result()

will raise if the task run failed which is the easiest reliable way to propagate task failures to the flow. - If you prefer to work with states explicitly, you can call tasks with

return_state=True

and use

prefect.states.raise_state_exception(state)

on any failed state at the end to fail the flow. Helpful docs: - Futures overview - Task.submit - PrefectFuture.result - raise_state_exception