Short answer - The UI sends a POST to the flow run set_state endpoint with a Scheduled state named “AwaitingRetry”. No new flow run is created; the existing run’s state history is updated and its run_count is incremented. Task runs are not modified immediately; when the run executes again, task run retry counters reset as they leave terminal state. What the UI calls under the hood 1) REST endpoint - POST /flow_runs/{flow_run_id}/set_state - Body: a Scheduled state named “AwaitingRetry” (optionally with a scheduled_time) - Handler: server API function

set_flow_run_state(...)

Links to source - Flow run API handler: prefect/server/api/flow_runs.py - Orchestration policies (retry/terminal-state rules): prefect/server/orchestration/core_policy.py - State constructors (including AwaitingRetry): prefect/states.py - Client method that the UI-equivalent would use: prefect/client/orchestration/_flow_runs/client.py Key server-side pieces - API entrypoint: -

prefect.server.api.flow_runs.set_flow_run_state(flow_run_id, state, force=False, ...)

- Model layer: -

prefect.server.models.flow_runs.set_flow_run_state(...)

(invoked by the API handler) - Orchestration policies involved: - HandleFlowTerminalStateTransitions: allows leaving a terminal state and sets retry metadata for a manual retry - CopyScheduledTime, WaitForScheduledTime, PreventDuplicateTransitions, etc. - RetryFailedFlows is for automatic retries, not the manual “Retry” button State transition details - Proposed state:

type=SCHEDULED

,

name="AwaitingRetry"

, optional

scheduled_time

- The flow run’s

run_count

is incremented - The flow run is not recreated; the same

flow_run_id

is reused and a new state record is appended - On the subsequent execution, task runs will transition out of terminal states and their retry counters are reset so task-level retries are available again No new flow run is created - The “Retry” button reschedules the existing flow run. Workers will pick up the rescheduled attempt and execute it with the same

flow_run_id

. How to replicate in a pytest - If you want to simulate exactly what the UI does, call the client’s

set_flow_run_state

with

states.AwaitingRetry(...)

. Example:

import asyncio
from prefect.client.orchestration import get_client
from prefect import states

async def retry_flow_run(flow_run_id):
    async with get_client() as client:
        # Simulate the UI “Retry” button
        orch_result = await client.set_flow_run_state(
            flow_run_id,
            states.AwaitingRetry()  # optionally: states.AwaitingRetry(scheduled_time=pendulum.now().add(seconds=1))
        )
        return orch_result

# In your pytest (async):
# - Arrange: create a flow run and mark it Failed
# - Act: call retry_flow_run(flow_run_id)
# - Assert:
#   - flow run's latest state is Scheduled with name "AwaitingRetry"
#   - flow run's run_count incremented by 1
#   - no new flow run object was created (same id)

- To set the run to a Failed state in your test before retrying, you can do:

from prefect.client.orchestration import get_client
from prefect import states

async def fail_run(flow_run_id):
    async with get_client() as client:
        await client.set_flow_run_state(flow_run_id, states.Failed(name="TestFailed"))

Assertions you can make - GET the flow run and verify: -

flow_run.id

unchanged - Latest state:

type == "SCHEDULED" and name == "AwaitingRetry"

-

run_count

incremented by 1 - If you proceed to let a worker pick up the retry, verify task runs’ retry counters reset as they leave terminal state on the new attempt. Notes

- Resume vs Retry: Paused runs are resumed via a different API path; the “Retry” button explicitly sets the Scheduled/AwaitingRetry state. - Automatic retries use different policy logic; the button is a manual reschedule. If you share how deep you want the test to go (pure unit vs integration against a running Server + worker), I can flesh out a ready-to-run pytest fixture and assertions for your setup.

Great question — with the Prefect test harness, the “Retry” button behavior only changes the flow run’s state to Scheduled (“AwaitingRetry”). It won’t actually execute again unless something is responsible for picking up that scheduled run. You have two workable patterns in tests: Option A — Test engine-managed retries (no worker needed) - Run the flow function with retries configured. The Prefect engine will handle the retry in-process. - This is simplest if you just want to verify “it retried.” Example:

from prefect import flow
from prefect.testing.utilities import prefect_test_harness, get_most_recent_flow_run

calls = {"n": 0}

@flow(retries=1, retry_delay_seconds=0)
def flaky():
    calls["n"] += 1
    if calls["n"] == 1:
        raise RuntimeError("boom")
    return "ok"

with prefect_test_harness():
    result = flaky()
    assert result == "ok"
    assert calls["n"] == 2

    # Optional: verify persisted state/run_count against the ephemeral API
    fr = get_most_recent_flow_run(flaky)
    assert fr.run_count == 2

Option B — Test the UI “Retry” button behavior (requires a worker) - The harness starts an API, but nothing executes runs. To actually run again after pressing Retry, start a worker in the test and make sure the run is associated with a deployment. Then set the run state to

AwaitingRetry

and let the worker pick it up. - Manual “Retry” (terminal -> Scheduled/AwaitingRetry) sets

empirical_policy.retry_type='reschedule'

under the hood, which workers will pick up. If you craft the state yourself, ensure it’s not

'in_process'

or workers will skip it. Sketch of the flow for a pytest (async) integration test:

import asyncio
import pendulum
import pytest
from prefect import flow, states
from prefect.testing.utilities import prefect_test_harness
from prefect.client.orchestration import get_client
from prefect.workers.process import ProcessWorker

@flow
def flaky():
    # fail so we can click 'retry' and see another attempt
    raise RuntimeError("demo failure")

@pytest.mark.asyncio
async def test_manual_retry_with_worker():
    with prefect_test_harness():
        # 1) Create a deployment the worker can run
        # In Prefect 3.x use the source-based API:
        # NOTE: adjust source/entrypoint to your test module path
        d = flaky.from_source(
            source=__file__,                  # path to this test file
            entrypoint="test_module:flaky",   # module_name:function_name
        ).deploy(
            name="retry-demo",
            work_pool_name="default",         # ensure a process-type pool exists
        )

        async with get_client() as client:
            # 2) Create an initial run and schedule it immediately
            flow_run = await client.create_flow_run_from_deployment(
                d.id,
                state=states.Scheduled(scheduled_time=pendulum.now("UTC"))
            )

            # 3) Start a ProcessWorker to pick up the run
            async with ProcessWorker(work_pool_name="default") as worker:
                # Pull and submit the first (failing) attempt
                await worker.get_and_submit_flow_runs()

                # Wait briefly for the process to run and fail
                await asyncio.sleep(1)

                # 4) Simulate pressing the UI "Retry" button by setting AwaitingRetry
                await client.set_flow_run_state(
                    flow_run.id,
                    states.AwaitingRetry(scheduled_time=pendulum.now("UTC"))
                )

                # 5) Worker picks up the retried run
                await worker.get_and_submit_flow_runs()
                await asyncio.sleep(1)

            # 6) Verify run_count incremented (attempt 2 exists)
            refreshed = await client.read_flow_run(flow_run.id)
            assert refreshed.run_count >= 2

Important notes

- The test harness spins up an API only. For a run to execute after pressing Retry, a worker must be running in your test and the run must be associated with a deployment. - Workers skip in-process retries. They will pick up Scheduled runs with

AwaitingRetry

name when

empirical_policy.retry_type

is not

'in_process'

. Manual retry sets

'reschedule'

, so you’re good if you use the server’s state transition via

set_flow_run_state(AwaitingRetry(...))

. - If you just want to validate “it retried” without a worker, prefer Option A (engine-managed retries). References (source) - Flow run state API: prefect/server/api/flow_runs.py - AwaitingRetry state helper: prefect/states.py - Worker scheduling filters (retry_type handling): server/database/sql/postgres/get-runs-from-worker-queues.sql.jinja - Orchestration policies (manual/auto retry behavior): prefect/server/orchestration/core_policy.py - Test harness utilities: prefect/testing/utilities.py If you share your current test structure (sync vs async, and whether you can spin up a worker in tests), I can tailor a ready-to-run pytest that matches your setup.

Assuming Prefect 3.x. You’ve got two good options: Option 1 — Use the public Events API via the Prefect client (recommended) - Supports rich filtering, pagination, and works the same for Cloud or self-hosted. Quick examples:

import asyncio
from datetime import timedelta
from prefect.client.orchestration import get_client
from prefect.events.filters import EventFilter, EventOccurredFilter, EventNameFilter

async def main():
    async with get_client() as client:
        # Last 24h, all flow-run events
        filter = EventFilter(
            occurred=EventOccurredFilter.last(timedelta(hours=24)),
            event=EventNameFilter(prefix=["prefect.flow-run."]),
        )
        page = await client.read_events(filter=filter, limit=100)
        for ev in page.events:
            print(ev.occurred, ev.event, ev.resource.id)

        # Fetch additional pages if present
        while page.next_page:
            page = await client.read_events_page(page.next_page)
            for ev in page.events:
                print(ev.occurred, ev.event)

asyncio.run(main())

Common filters you can combine: - Time window:

EventOccurredFilter(since=..., until=...)

or

.last(timedelta(...))

- Event types:

EventNameFilter(prefix=[...], name=[...], exclude_prefix=[...])

- Resources:

EventResourceFilter(id=[...], id_prefix=[...])

- Related resources:

EventRelatedFilter(id=[...], role=[...])

- Text search:

EventTextFilter(query="error -timeout")

- Sort:

order=EventOrder.DESC

or ASC Count/aggregate examples:

async with get_client() as client:
    # Count events per day (POST /events/count-by/day under the hood)
    counts = await client.request(
        "POST",
        "/events/count-by/day",
        json={"filter": EventFilter().model_dump(mode="json")}
    )
    print(counts)

Docs/source for filters and client: - Filters: prefect/events/filters.py - Client events methods: prefect/client/orchestration Option 2 — Query the database directly (self-hosted) - Useful for custom reporting or heavy aggregation. Tables are typically: -

events

(main event records) -

event_resources

(resource-role mapping per event) Example SQL snippets (PostgreSQL):

-- Recent events for a flow run
SELECT id, occurred, event, resource_id, payload
FROM events
WHERE resource_id = 'prefect.flow-run.<flow_run_id>'
  AND occurred > NOW() - INTERVAL '24 hours'
ORDER BY occurred DESC
LIMIT 100;

-- Count events by type for the last 7 days
SELECT event, COUNT(*) AS count
FROM events
WHERE occurred > NOW() - INTERVAL '7 days'
GROUP BY event
ORDER BY count DESC;

-- Events with payload condition (JSONB)
SELECT id, occurred, event
FROM events
WHERE payload @> '{"severity": "error"}'
  AND occurred > NOW() - INTERVAL '1 day';

When to prefer which: - Prefer the client/API for most use cases: built-in auth, validated filters, pagination. - Use SQL when you need bespoke aggregations or to integrate with your own BI tooling. If you share the filters you need (time range, event types, specific resources), I can sketch the exact client call or SQL for your case.

Short answer: yes, a ProcessWorker can run multiple flows simultaneously. Each flow run is executed in its own subprocess and the worker can submit several at once — as long as it has capacity and is watching the correct work queue(s). Your hang is almost always a capacity or queue mismatch issue. Why your wait_for_flow_run is hanging - Worker capacity deadlock: if the worker’s concurrency is 1 (via worker

limit

, work pool limit, or work queue limit), it will submit the outer flow and won’t submit the inner flow until capacity frees up — but the outer flow is blocked waiting for the inner, so you deadlock. - Queue mismatch: the inner deployment is targeting a different work queue (or pool) than the one your worker is watching, so the inner run never gets picked up. - Scheduling/time: the inner run was scheduled in the future. - Deployment created at runtime: creating/deploying inside the outer flow can end up pointing at a different queue/pool than you expect. How to fix 1) Give the worker >1 slots - Programmatic:

from prefect.workers.process import ProcessWorker

async with ProcessWorker(work_pool_name="default", limit=2) as worker:
    # loop/await while runs are being picked up
    ...

- Or ensure the work pool or work queue concurrency limits are not set to 1. 2) Ensure the inner deployment targets the queue the worker is watching - If your worker watches queue “default”, make sure the inner deployment uses

work_queue_name="default"

(and same work pool). - If you need multiple queues, start the worker with

work_queues=[...]

to watch them or run a second worker. 3) Prefer pre-creating the inner deployment - Create the

other_flow

deployment ahead of time (outside the running flow) to avoid surprises with queue/pool selection inside the runner process. 4) Use run_deployment + wait_for_flow_run - This pattern is supported; just ensure capacity and queues line up:

from prefect import flow
from prefect.deployments.flow_runs import run_deployment
from prefect.flow_runs import wait_for_flow_run

@flow
def outer_flow():
    # Make sure this deployment uses the same pool/queue the worker is watching
    fr = run_deployment(
        name="other-flow/other-deployment",
        as_subflow=True,                 # default True; keeps parent/child linkage
        # work_queue_name="default",     # optional override if needed
    )
    # Avoid infinite waits; enable state logs to debug where it’s stuck
    result = wait_for_flow_run(
        fr.id,
        timeout=300,
        poll_interval=2,
        log_states=True,
    )
    return result.state.name

5) If you don’t need a separate deployment, call the subflow directly - Running

other_flow()

as a normal Python call executes it in the same process and returns its result without involving a worker or queue. What to check in your test - Worker is running with

limit >= 2

(or pool/queue limit > 1) - The inner deployment’s work pool/queue matches the worker - The inner run is scheduled “now” (or use

scheduled_time=None

in

run_deployment

) - Add

log_states=True

to

wait_for_flow_run

to see if it’s stuck in Scheduled References - ProcessWorker: prefect/workers/process.py - run_deployment: prefect/deployments/flow_runs.py - wait_for_flow_run: prefect/flow_runs.py If you can share the inner deployment’s work pool/queue and the worker’s

limit

(and any pool/queue concurrency limits you’ve set), I can pinpoint exactly where it’s stuck.

You’ve got two straightforward ways to do it in Prefect 3.x: CLI (most common) 1) Ensure the queue exists in your work pool:

prefect work-pool create my-pool --type process
prefect work-queue create default --pool my-pool

2) Start a ProcessWorker that only watches the “default” queue in that pool:

prefect worker start -p my-pool -q default

Programmatic (useful in tests)

import asyncio
from prefect.workers.process import ProcessWorker

async def main():
    async with ProcessWorker(
        work_pool_name="my-pool",
        work_queues=["default"],   # only poll this queue
        limit=2                    # optional: allow >1 concurrent runs
    ) as worker:
        await worker.run()         # continuously poll and submit

asyncio.run(main())

Notes - The “default” queue must belong to the same work pool you pass to the worker. - If you omit

work_queues=[...]

, the worker will poll all active queues in the pool; specifying

["default"]

restricts it to that one. - Make sure your deployment targets the same pool/queue (e.g.,

.deploy(..., work_pool_name="my-pool", work_queue_name="default")

or

run_deployment(..., work_queue_name="default")

). References - ProcessWorker class: prefect/workers/process.py - Work pool CLI: prefect/cli/work_pool.py - Work queue CLI: prefect/cli/work_queue.py

Yes. In Prefect 3.x, a ProcessWorker will auto-create the work pool if it doesn’t exist by default, and that pool creation also creates a “default” work queue. Details: -

ProcessWorker(work_pool_name="my-pool")

uses

create_pool_if_not_found=True

by default. - On first startup (entering the async context), it will: - Create the work pool named

my-pool

with type

process

if missing - Create the pool’s default work queue named

default

- If you pass

create_pool_if_not_found=False

, the worker will not create the pool and will warn if it’s missing. Queues: - The “default” queue is created with the pool automatically. - Custom queues are not auto-created; create them via CLI or API first, then pass

work_queues=[...]

to the worker if you want to restrict polling. Example in pytest:

import asyncio
from prefect.workers.process import ProcessWorker

async def test_worker_creates_pool_by_default():
    async with ProcessWorker(work_pool_name="pytest-pool") as worker:
        # pool "pytest-pool" now exists with a "default" queue
        # worker is ready to poll (all queues in the pool by default)
        ...

If you only want to poll the “default” queue:

async with ProcessWorker(
    work_pool_name="pytest-pool",
    work_queues=["default"],        # only poll this queue
) as worker:
    ...

If you’d rather require the pool to exist already:

async with ProcessWorker(
    work_pool_name="pytest-pool",
    create_pool_if_not_found=False  # do not auto-create
) as worker:
    ...

Short answer: yes, ProcessWorker really does spawn separate OS processes per flow run, even in pytest. If your inner run is hanging at wait_for_flow_run, it’s almost always because the worker isn’t actively polling when the inner run is scheduled, or there’s a capacity/queue/concurrency limit mismatch. Checklist to avoid deadlocks in tests - Keep the worker running continuously: don’t just call get_and_submit_flow_runs() once. Run the worker loop in the background while your outer run schedules the inner run. - Use limit >= 2 on the worker and ensure no work pool/queue concurrency limit is set to 1. - Ensure the inner deployment targets the exact pool/queue the worker is watching. - Don’t define your flows inside a pytest test module as the entrypoint. Prefer a separate importable module for flows; subprocess import of a test module can behave oddly under pytest. - Make sure the inner run is scheduled “now” (no future scheduled_time). - Use log_states=True in wait_for_flow_run to confirm where the state is stuck. A robust pytest pattern

import asyncio
from contextlib import suppress
from prefect.workers.process import ProcessWorker
from prefect.client.orchestration import get_client
from prefect.flow_runs import wait_for_flow_run
from prefect import states
import pendulum

async def run_worker(pool="my-pool", queues=["default"], limit=2):
    async with ProcessWorker(work_pool_name=pool, work_queues=queues, limit=limit) as worker:
        await worker.run()  # continuous polling

@pytest.mark.asyncio
async def test_nested_flows_run_concurrently():
    async with get_client() as client:
        # Assume outer and other_flow deployments already exist and target my-pool/default
        outer_dep = await client.read_deployment_by_name("outer-flow/outer-deployment")

        # Start the worker in the background so it can pick up both runs
        worker_task = asyncio.create_task(run_worker(pool="my-pool", queues=["default"], limit=2))
        try:
            # Kick off the outer run now
            outer_run = await client.create_flow_run_from_deployment(
                outer_dep.id,
                state=states.Scheduled(scheduled_time=pendulum.now("UTC"))
            )

            # Wait for the outer run to finish; during its execution it will schedule the inner run
            # If you want to explicitly wait on the inner run, capture its id in logs or return it
            await wait_for_flow_run(outer_run.id, timeout=300, poll_interval=2, log_states=True)

        finally:
            worker_task.cancel()
            with suppress(asyncio.CancelledError):
                await worker_task

Common gotchas that cause “hangs” - Worker not running when the inner run is created: If you only “submit once” before the inner run exists, that run never gets picked up. Keep the worker loop running. - Concurrency set to 1 somewhere else: Even with ProcessWorker(limit=2), a work pool or work queue concurrency limit of 1 will still cause a deadlock for parent/child patterns. - Queue mismatch: The inner deployment silently targets a different queue/pool than the worker. - Flows defined inside test files: The subprocess imports the test module as the entrypoint and gets stuck. Move flows to a non-test module and use that module path as the deployment entrypoint. If you can share: - How you start the worker in the test (single submit vs continuous run) - Whether the outer run is created via deployment (and the inner deployment’s pool/queue) - Any pool/queue concurrency limits I can help pinpoint the exact reason your wait_for_flow_run is stalling. References: ProcessWorker and process infra prefect/workers/process.py, wait_for_flow_run prefect/flow_runs.py, subprocess launcher prefect/processutils.py.

That error means your API doesn’t have a work pool named “default-work-pool” at the time you make the request. In pytest, the pool will only be auto-created after the ProcessWorker enters its async context and syncs with the backend. If you reference the pool before the worker is started (e.g., when creating a deployment or calling run_deployment), you’ll get “Work pool not found”. Two reliable fixes: A) Start the worker first so it auto-creates the pool (and default queue)

import asyncio
from prefect.workers.process import ProcessWorker

async def start_worker():
    async with ProcessWorker(
        work_pool_name="default-work-pool",   # will be auto-created
        work_queues=["default"],              # polls the default queue
        limit=2
    ) as worker:
        await worker.run()

# In your test, schedule this before interacting with the pool
worker_task = asyncio.create_task(start_worker())
# ... now create deployments/run_deployment targeting default-work-pool/default
# ... when done:
worker_task.cancel()

B) Explicitly create the pool/queue in your test setup before use

import asyncio
from prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound

async def ensure_pool_and_queue(pool="default-work-pool", queue="default"):
    async with get_client() as client:
        try:
            await client.read_work_pool(pool)
        except ObjectNotFound:
            await client.create_work_pool(name=pool, type="process")
        queues = await client.read_work_queues(work_pool_name=pool)
        if queue not in [q.name for q in queues]:
            await client.create_work_queue(work_pool_name=pool, name=queue)

# In your test
await ensure_pool_and_queue()
# Now safe to create deployments/run_deployment pointing to default-work-pool/default

Common causes in pytest - Order of operations: creating a deployment or running it before the worker has created the pool. - Separate API contexts: your worker and your deployment code may be pointing at different APIs. In tests, verify PREFECT_API_URL and keep everything inside the same

prefect_test_harness()

scope. - Ephemeral DB reset: each test/harness run starts with a fresh DB; you must create the pool/queue inside each test or in a fixture. References - ProcessWorker (runs subprocesses, auto-creates pool on enter): prefect/workers/process.py - BaseWorker pool sync/creation logic (create_pool_if_not_found defaults to True): prefect/workers/base.py - Default “default” queue created with new pool (server): prefect/server/models/workers.py If you share where the error is raised (deployment creation vs run_deployment vs flow run creation), I can suggest the minimal change to your test order.

You’re very close — the issue is that you’re calling

get_and_submit_flow_runs()

only once. That’s a one-shot fetch/submit. When the outer run creates the inner run later, the worker is no longer polling, so the inner run sits in Scheduled and your

wait_for_flow_run

hangs. Fix: keep the worker’s polling loop running in the background for the duration of the test. Also use the async client context. Try this pattern:

import asyncio
import pendulum
from contextlib import suppress
from prefect.testing.utilities import prefect_test_harness
from prefect.client.orchestration import get_client
from prefect.workers.process import ProcessWorker
from prefect import states
from prefect.flow_runs import wait_for_flow_run

with prefect_test_harness(server_startup_timeout=60):
    async def test_run():
        async with get_client() as client:
            async with ProcessWorker(
                work_pool_name="default-work-pool",
                work_queues=["default"],   # ensure it polls the right queue
                limit=5
            ) as worker:
                # Run the worker continuously in the background
                worker_task = asyncio.create_task(worker.run())
                try:
                    # Create the outer run; it will create the inner run when it executes
                    flow_run = await client.create_flow_run_from_deployment(
                        deployment_id,
                        state=states.Scheduled(scheduled_time=pendulum.now("UTC"))
                    )

                    # Wait for the outer to complete; during its execution, the inner run
                    # will be created and picked up because the worker is still polling
                    await wait_for_flow_run(flow_run.id, timeout=300, poll_interval=2, log_states=True)

                finally:
                    worker_task.cancel()
                    with suppress(asyncio.CancelledError):
                        await worker_task

    asyncio.run(test_run())

Other things to double-check: - Queue/pool match: make sure the inner deployment targets

default-work-pool

and the

default

queue (or include that queue in

work_queues

). - Schedule “now”: don’t schedule the inner run in the future. - Concurrency limits: ensure the work pool or work queue don’t have a concurrency limit of 1. Even with

limit=5

on the worker, a pool/queue limit of 1 will serialize execution and can create a parent/child deadlock. If you prefer not to run the full loop, you can poll in a small loop instead of once:

while True:
    submitted = await worker.get_and_submit_flow_runs()
    if not submitted:
        await asyncio.sleep(0.5)

…but the background

worker.run()

is simpler and more robust. Reference: ProcessWorker implementation and run loop - prefect/workers/process.py

You’re right to call that out — my earlier suggestion to call

worker.run()

directly was wrong. On workers,

run(flow_run, configuration)

is the “submit a single run” method and is not the polling loop. Use

await worker.start(...)

instead. That’s the public method that runs the polling loop and submits flow runs. In tests, you typically want

run_once=True

to do a single poll/submit cycle, or run it in a background task to keep polling while your outer run creates the inner run. Adjusted test pattern - Keep the worker alive and polling while the outer run schedules the inner run. - Use the async client context. Example:

import asyncio
import pendulum
from contextlib import suppress
from prefect.testing.utilities import prefect_test_harness
from prefect.client.orchestration import get_client
from prefect.workers.process import ProcessWorker
from prefect import states
from prefect.flow_runs import wait_for_flow_run

with prefect_test_harness(server_startup_timeout=60):
    async def main():
        async with get_client() as client:
            async with ProcessWorker(
                work_pool_name="default-work-pool",
                work_queues=["default"],
                limit=5,
            ) as worker:
                # Keep polling in the background
                polling = asyncio.create_task(worker.start())  # continuous loop

                try:
                    # Kick off the outer run
                    fr = await client.create_flow_run_from_deployment(
                        deployment_id,
                        state=states.Scheduled(scheduled_time=pendulum.now("UTC")),
                    )

                    # Wait for the outer run to finish; inner run will be picked up
                    await wait_for_flow_run(fr.id, timeout=300, poll_interval=2, log_states=True)

                finally:
                    polling.cancel()
                    with suppress(asyncio.CancelledError):
                        await polling

    asyncio.run(main())

If you prefer to avoid a continuous loop, you can emulate it with run_once in a small loop:

while True:
    await worker.start(run_once=True)   # polls and submits once
    await asyncio.sleep(0.5)

Why your original test deadlocked -

get_and_submit_flow_runs()

is a single pass. After it submits the outer run, the worker stops polling. When the outer run creates the inner run later, nothing is polling anymore to submit it. Other sanity checks - Make sure the inner deployment targets the same pool/queue (

default-work-pool

/

default

). - Ensure no work pool or queue concurrency limit is set to 1. - Avoid defining flows inside the test module if those flows are executed in subprocesses; use an importable module for the deployment entrypoint. References - BaseWorker.start (polling loop): prefect/workers/base.py - ProcessWorker class: prefect/workers/process.py - wait_for_flow_run: prefect/flow_runs.py

You’re exactly right: pytest/monkeypatch mocks live in the test process and won’t carry over to the separate Python processes that a ProcessWorker spawns. There isn’t a “worker override” that injects your in-memory mock into those child processes. Use one of these patterns to get your mock into the worker subprocess: 1) Feature-flag seam + env var - Add a small seam in your flow code that checks an env var and switches to a fake implementation. - Set that env var for the subprocess via deployment job variables or the work pool’s base job template. Example seam:

# mypkg/service.py
import os

def call_real(*args, **kwargs):
    ...

def call_fake(*args, **kwargs):
    return {"ok": True}

def call(*args, **kwargs):
    if os.getenv("PREFECT_TEST_MOCKS") == "1":
        return call_fake(*args, **kwargs)
    return call_real(*args, **kwargs)

Set env for runs at deploy time:

deployment = my_flow.from_source(
    source="...",
    entrypoint="mypkg.flows:my_flow",
).deploy(
    name="test-deploy",
    work_pool_name="default-work-pool",
    work_queue_name="default",
    job_variables={
        "env": {"PREFECT_TEST_MOCKS": "1"}
    },
)

2) sitecustomize-based monkeypatch (auto-runs in every Python process) - Create a directory with a sitecustomize.py that mutates the target import when Python starts. - Put that directory on PYTHONPATH for the worker subprocess. tests/mocks/sitecustomize.py:

# Runs automatically on interpreter startup if on sys.path
import mypkg.service as svc
def _fake(*a, **k): return {"ok": True}
svc.call = _fake

Add to PYTHONPATH on the deployment so worker subprocesses load it:

from pathlib import Path, PurePosixPath
import os

mocks_dir = Path(__file__).parent / "mocks"
deployment = my_flow.from_source(...).deploy(
    name="test-deploy",
    work_pool_name="default-work-pool",
    work_queue_name="default",
    job_variables={
        "env": {
            # Prepend mocks dir; use os.pathsep to handle Windows vs *nix
            "PYTHONPATH": f"{mocks_dir}{os.pathsep}" + os.environ.get("PYTHONPATH","")
        }
    },
)

Now every flow subprocess launched by the worker will import sitecustomize and your patch will apply. 3) Module shadowing via PYTHONPATH - Create a mock module that shadows the real one with the same import path, and put that directory first on PYTHONPATH via job variables or work pool template. This avoids monkeypatching; Python just imports your mock module. 4) Set env at the work pool level (applies to all runs from that pool) - If you prefer a pool-wide setting for tests, set env in the work pool’s base job template (via CLI or API) so every job inherits it. For one-off tests, deployment-level job_variables is simpler. Why not monkeypatch in the test? - The ProcessWorker launches a new interpreter. In-memory patches in the pytest process aren’t visible to that interpreter. You must influence the child process by code (feature flag) or environment (PYTHONPATH/sitecustomize/env vars). If you share the module path of the function you want to mock, I can sketch the exact sitecustomize or shadow-module structure to match your import and the deploy/job_variables snippet you can drop into your test. References - ProcessWorker and job configuration (env support via job variables): prefect/workers/process.py - run_deployment job_variables param (useful for injecting env per run): prefect/deployments/flow_runs.py