fix: properly drain workers in prefect_test_harness when used in async contexts

[devin-ai-integration]

Fixes #19762

When prefect_test_harness is used in an async context, the drain_all() and drain() methods return coroutines that are never awaited, causing RuntimeWarning: coroutine 'wait' was never awaited.

Root cause: The drain() and drain_all() methods in _QueueServiceBase are "dual-mode" - they return awaitables when called from an async context (when get_running_loop() is not None). Since prefect_test_harness is a sync context manager, these returned coroutines were being dropped without being awaited.

Solution: Use a wrapper coroutine passed to run_coro_as_sync that calls the public drain() and drain_all() methods and awaits them if they return awaitables. This ensures the awaitable is created and awaited on the same loop, avoiding cross-loop issues while using public APIs.

Key changes:

• Add a drain_workers() async helper that calls APILogWorker.instance().drain() and EventsWorker.drain_all()
• Check if results are awaitable with inspect.isawaitable() and await them if so
• Execute the wrapper coroutine via run_coro_as_sync() to handle both sync and async contexts
• Add regression test that catches unawaited coroutines using PytestUnraisableExceptionWarning

Human review checklist:

• Verify the wrapper coroutine approach correctly handles both sync and async contexts
• Verify error handling for when APILogWorker hasn't been started
• Verify the regression test properly catches unawaited coroutines

Checklist

• This pull request references any related issue by including "closes <link to issue>"
• If this pull request adds new functionality, it includes unit tests that cover the changes
• If this pull request removes docs files, it includes redirect settings in mint.json.
• If this pull request adds functions or classes, it includes helpful docstrings.

---

Link to Devin run: https://app.devin.ai/sessions/70901d7018934b0f9269d6e42042e4a3
Requested by: Nate Nowack ([email protected]) (@zzstoatzz)

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[codspeed-hq]

CodSpeed Performance Report

Merging #19770 will not alter performance

_{Comparing devin/1765652244-fix-test-harness-async-cleanup (d3141ff) with main (a4ae0c1)}

Summary

✅ 2 untouched