Handle the loop closing between ThreadsafeProxy's is_closed() check and dispatch

[zigpy-review-bot]

Closes #740. Follow-up to #727 and #739 — this closes out the RuntimeError: Event loop is closed teardown-race family, and, with the newest commit, the silent-hang variant of it reported on this PR (see also #745, which describes the same window from the uart.py side).

ThreadsafeProxy.func_wrapper snapshots the loop object at construction, so #739's ordering fix (publish self.loop = None before closing) can't reach it: the worker thread can close the loop between the proxy's loop.is_closed() check and the dispatch via asyncio.run_coroutine_threadsafe() / loop.call_soon_threadsafe(), and both raise RuntimeError: Event loop is closed at the caller. This wraps both dispatches in try/except RuntimeError and treats close-during-dispatch exactly like already-closed: log the existing "Attempted to use a closed event loop" warning and drop the call. The except is narrow — both try blocks wrap only the scheduling call (for the async path, call() merely constructs the coroutine; its body never runs there), so no RuntimeError from user code can be swallowed.

Two findings from pre-open review (details below) are folded in:

• Disconnected async proxy calls are now awaitable. The pre-existing is_closed() branch returned bare None for coroutine methods too, so real callers that await through a proxy (await self._gw.reset(), await self._gw.disconnect(), await self._gw.send_data(...)) got TypeError: object NoneType can't be used in 'await' expression when disconnected — trading one teardown exception for another. Both disconnected paths (already-closed and closed-mid-dispatch) now return an already-completed future for coroutine methods, so await proxy.method() resolves cleanly to None; sync calls still return None as before. On the race path the never-to-be-scheduled coroutine is explicitly closed, so it doesn't warn as never-awaited.
• EventLoopThread.run_coroutine_threadsafe() raises legibly when the loop is gone. After Clear EventLoopThread.loop before closing it #739 the shutdown-window failure there was AttributeError: 'NoneType' object has no attribute 'call_soon_threadsafe'; it now snapshots self.loop and raises RuntimeError("Event loop is not running"). The sole caller (uart.connect) already handles this via except Exception.

Two follow-up commits, from post-open review of the same family:

• The passed coroutine is closed on both of run_coroutine_threadsafe()'s failure paths. The None guard closes it before raising, and when the worker closes the loop after the snapshot (so asyncio.run_coroutine_threadsafe() itself raises RuntimeError), it is now closed before re-raising too — mirroring the proxy's async dispatch path — so neither failure mode leaks a "coroutine ... was never awaited" RuntimeWarning at teardown.
• inspect.iscoroutinefunction() replaces the deprecated asyncio.iscoroutinefunction() (deprecated since Python 3.14, removal slated for 3.16; these were bellows' only two call sites). Drop-in for every case here: real coroutine functions, plain callables, functools.partial, AsyncMock.

The third window: stopped, but not yet closed

jetliuzhe reported a third member of the same family on this PR, and it is the worst of the three because it is completely silent. Between EventLoopThread.force_stop() (which schedules loop.stop()) and _thread_main's loop.close(), the worker loop reports is_closed() == False and asyncio.run_coroutine_threadsafe() accepts work without raising — but the loop never runs it, and closing drops the pending task. The caller (await self._gw.disconnect(), which has no timeout) then waits on a future that is never resolved: no exception, no log line, no retry. Reproduced against bellows itself — a proxy call dispatched right after force_stop() never resolves, with Task was destroyed but it is pending! at teardown.

The fix is an explicit flag rather than another loop-state check, because every loop-state check is racy here: is_closed() stays False for the whole window, and is_running() stays True until run_forever() returns, so both only narrow it. force_stop() now publishes stopping = True before it schedules anything; ThreadsafeProxy takes the owning EventLoopThread (uart._connect() passes it through, so the use_thread=False path and the api proxy keep their old behavior) and treats stopping exactly like an already-closed loop: sync calls dropped, async calls resolved to None, with a warning naming the shutdown. The check sits before the coroutine is constructed, so nothing leaks, and EventLoopThread.run_coroutine_threadsafe() applies the same guard.

Dropping those calls costs nothing: by the time force_stop() has run, connection_lost has already been through zigpy's SerialProtocol, so Gateway.disconnect() (close() + wait_until_closed()) is a no-op — _transport is None and the disconnected event is already set. All that mattered was that the call must neither hang nor raise, in any of the three windows.

Known residual, deliberately not fixed here. The proxy reads stopping and then dispatches; if force_stop() lands between those two steps, the coroutine can still reach a loop that stops before running it. That is a few bytecodes wide instead of the entire shutdown — the reported failure, a dispatch well after force_stop(), is gone — but it is not zero, and it fails the same silent way. Closing it completely needs something that resolves the caller's future without depending on the loop running again, e.g. tracking dispatched futures and resolving them when the worker thread exits; a re-check on the worker loop's own thread before the task is created narrows it further, but still depends on that callback getting to run. Happy to add either here or as a follow-up if you want it.

Tests

Ten new tests. test_proxy_loop_closed_during_async_dispatch and test_proxy_loop_closed_during_sync_dispatch pin the race window by making loop.call_soon_threadsafe raise RuntimeError while is_closed() still returns False — a faithful stand-in, since a genuinely closed loop would trip the earlier guard. test_proxy_loop_closed_async pins the awaitable disconnected result on the already-closed branch; test_thread_run_coroutine_threadsafe_loop_not_running pins the RuntimeError on the None path; test_thread_run_coroutine_threadsafe_loop_closed_mid_dispatch pins the coroutine being closed (CORO_CLOSED) on the post-snapshot race path. All five fail on dev and pass with the change.

Five more come with the stopping-loop commit. test_proxy_loop_stopping_async pins the no-hang behavior end-to-end on a real worker thread: the thread is wedged so the stopping-but-not-closed window is reliably open, and the callee cannot finish inside it (as Gateway.disconnect() cannot), so on the unpatched code the test fails with a TimeoutError — the reported hang itself. test_proxy_loop_stopping_sync pins the dropped sync call and the warning; test_thread_run_coroutine_threadsafe_stopping pins the RuntimeError and the closed coroutine; test_proxy_stopping_ignored_without_thread and test_thread_start_clears_stopping pin the unchanged no-thread path and the restart path.

Verification

• pytest tests/ → 451 passed (Python 3.14.5), clean under -W error::RuntimeWarning -W error::DeprecationWarning (pins both the no-leaked-coroutine behavior and the deprecation removal); the 3 remaining AsyncMock warnings in tests/test_application.py reproduce identically on dev
• With bellows/thread.py reverted to dev: the five original tests fail (leaked RuntimeError ×2, TypeError on await None, AttributeError, leaked never-awaited coroutine), rest pass
• With only the stopping branch removed from func_wrapper: test_proxy_loop_stopping_async fails with TimeoutError, i.e. it pins the actual hang rather than just the new code path; tests/test_thread.py ran 10× green for determinism (the wedge is what makes the window reliable)
• The stopped-but-not-closed window was reproduced directly against bellows.thread before the fix (proxy call after force_stop() never resolves; Task was destroyed but it is pending!) and re-run after it (resolves to None, warning logged, no leaked coroutine under -W error::RuntimeWarning)
• Pre-open review: an Opus subagent review (confirmed the except RuntimeError breadth is safe and flagged the coroutine leak on the raise path) and a Copilot CLI second opinion (GPT-5.6 Sol; flagged the await None TypeError) — both findings addressed as described
• The stopping-loop commit went through two more independent review rounds (a Fable 5 subagent and Copilot GPT-5.6 Sol, both read-only): round 1 produced three test/comment nits, all folded in, plus the residual window now named in the description; round 2 came back clean from both
• pre-commit: autoflake, black, flake8, isort, codespell, ruff pass; pyupgrade/mypy fail on untouched files in the local hook env only (same Python 3.14 breakage as noted on Clear EventLoopThread.loop before closing it #739) — CI is the real check

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 99.55%. Comparing base (553ec35) to head (7fb32e7).
⚠️ Report is 1 commits behind head on dev.

Additional details and impacted files

@@           Coverage Diff           @@
##              dev     #741   +/-   ##
=======================================
  Coverage   99.55%   99.55%           
=======================================
  Files          64       64           
  Lines        4263     4292   +29     
=======================================
+ Hits         4244     4273   +29     
  Misses         19       19           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[jetliuzhe]

There is one more member of the family that this PR does not close: the window where the
worker loop is stopped but not yet closed.

EventLoopThread.force_stop() schedules loop.stop(); loop.close() only happens
afterwards in _thread_main's finally. In between:

• loop.is_closed() is False, so the is_closed() guard lets the call through
• asyncio.run_coroutine_threadsafe() does not raise, so the new
  try/except RuntimeError does not fire either
• the loop never runs again, so the coroutine is never executed and the wrapped future
  never resolves

The caller (await self._gw.disconnect()) has no timeout, so it waits forever. Unlike the
TypeError and RuntimeError variants, this one is completely silent: no exception, no
log line, no retry.

Self-contained reproduction (Python 3.14, no bellows import needed):

import asyncio, threading, time, contextlib

loop = asyncio.new_event_loop()
threading.Thread(target=lambda: (asyncio.set_event_loop(loop), loop.run_forever()),
                 daemon=True).start()
time.sleep(0.3)

loop.call_soon_threadsafe(loop.stop)          # what force_stop() does
time.sleep(0.3)
print("is_closed:", loop.is_closed())         # False  -> guard lets it through

async def victim(): return "ran"

fut = asyncio.run_coroutine_threadsafe(victim(), loop)   # does NOT raise
try:
    print("got:", fut.result(timeout=3))
except TimeoutError:
    print("HANG: never resolves; caller has no timeout")

with contextlib.suppress(Exception):
    loop.close()
try:
    asyncio.run_coroutine_threadsafe(victim(), loop)
except RuntimeError as e:
    print("after close:", e)                  # this branch IS handled by this PR

Output:

is_closed: False
HANG: never resolves; caller has no timeout
after close: Event loop is closed

It also emits the same warning that shows up in the field, from the same place:

/usr/local/lib/python3.14/asyncio/base_events.py:744: RuntimeWarning:
coroutine 'victim' was never awaited
  self._ready.clear()

I hit this variant in production on an EZSP-over-TCP coordinator (bellows 0.49.0,
HA 2026.4.1). After the coordinator lost power, ZHA logged the never awaited warning for
SerialProtocol.disconnect and then went completely silent. I sampled for 90 seconds:
zero reconnect attempts, and ss -tn showed zero TCP connection attempts from the host to
the coordinator port. The config-entry reload never completed. Only restarting Home
Assistant recovered it.

Returning an already-completed future in the disconnected branches (which this PR already
does) would cover this case too — provided the branch is reached. So the missing piece is
just that is_closed() is not sufficient to decide "unusable": there is a live window
where the loop is neither closed nor ever going to run again.

I do not have a strong opinion on the fix, and the obvious cheap check is not quite right:
loop.is_running() is still True between force_stop() scheduling loop.stop() and
run_forever() actually returning, so it only narrows the window rather than closing it.
An explicit flag set by force_stop() (and consulted by the proxy) would not have that
race, since it is set before anything is scheduled. A bounded wait on the
run_coroutine_threadsafe result would also convert the hang into a loud failure, though
it treats the symptom.

Happy to test a patch — I can reproduce the real-world case on demand by cutting power to
the coordinator.

[TheJulianJES]

@zigpy-review-bot Review above comment, possibly implement here, or open new PR. Do review rounds, also on this PR. Also check #745

[zigpy-review-bot]

Confirmed, and fixed here in 7fb32e7 rather than in a new PR — it is the same window as the rest of this PR, one step earlier.

I reproduced it against bellows.thread itself, not just the standalone snippet: a proxy call dispatched right after force_stop() never resolves, and teardown prints Task was destroyed but it is pending!. The three-row table is exactly right, including that this row is the worst one — the caller has no timeout, so it is an unbounded wait with no exception and no log line.

You were also right that is_closed() is not the property the proxy actually needs, and that the cheap loop-state checks do not close it: is_running() is True until run_forever() returns, so it only trims the tail of the window. So force_stop() now publishes an explicit stopping = True before it schedules anything, and ThreadsafeProxy — which now gets the owning EventLoopThread, passed through by uart._connect() — treats stopping exactly like an already-closed loop: sync calls dropped, async calls resolved to None, warning logged. Being set before anything is scheduled is what makes the flag not have the is_running() race. The use_thread=False path and the api proxy pass no thread and are unchanged.

One thing worth stating explicitly, since "the disconnect still does not actually run" is the reasonable worry about resolving to None: in this path nothing is lost. By the time force_stop() has run, connection_lost has already been through zigpy's SerialProtocol, so Gateway.disconnect() (close() + wait_until_closed()) is a no-op — _transport is None and the disconnected event is already set. What mattered was only that the call must neither hang nor raise.

What is not fully closed, and is now named in the PR description: the proxy reads stopping and then dispatches, so a force_stop() landing between those two steps can still hand a coroutine to a loop that stops before running it. That is a few bytecodes wide rather than the whole shutdown, and your case — a dispatch well after force_stop() — is gone, but it fails the same silent way if it ever hits. Closing it completely needs something that resolves the caller's future without depending on the loop running again (e.g. tracking dispatched futures and resolving them when the worker thread exits); a re-check on the worker loop's own thread narrows it further but still needs that callback to run. I left it out to keep this PR to one mechanism — glad to add it here or as a follow-up if the maintainers prefer.

Your offer to test is very welcome: the branch is zigpy-bot/threadsafe-proxy-closed-loop-race, and a power cut on the coordinator is precisely the case the tests can only approximate. What you should see instead of silence is Attempted to use an event loop that is shutting down followed by the reconnect proceeding normally.

I answered #745 separately — short version: the ordering you identified is real, but flipping those two lines would not serialize the teardown against the application's cleanup, because the notification at :89 is itself an asynchronous cross-thread dispatch.

Verification

• The pre-fix reproduction against bellows.thread: EventLoopThread started, ThreadsafeProxy built on its loop, force_stop(), then an awaited proxy call — is_closed() was False, is_running() was True, and the call never resolved (bounded only by a test timeout); after the fix the same probe returns None immediately, logs the warning, and is clean under -W error::RuntimeWarning.
• test_proxy_loop_stopping_async pins it end-to-end: on the unpatched code it fails with TimeoutError — the hang itself, not just the missing branch. The worker thread is wedged so the stopping-but-not-closed window is reliably open, and the callee cannot finish inside it, as Gateway.disconnect() cannot.
• pytest tests/ → 451 passed on Python 3.14.5, clean under -W error::RuntimeWarning -W error::DeprecationWarning; tests/test_thread.py ran 10× green for determinism.
• Two independent read-only review rounds on the new commit (a Fable 5 subagent and Copilot GPT-5.6 Sol): round 1 gave three test/comment nits, all folded in, and the residual window above; round 2 came back clean from both.