Coverage for src/lilbee/cli/tui/thread_safe.py: 100%

1"""Thread-safe helpers for posting from @work(thread=True) workers to the main thread. 

2 

3Textual's call_from_thread raises OSError when the app's message queue 

4has already been closed during shutdown. Since workers run in daemon 

5threads, they can outlive the app. This module provides a drop-in 

6wrapper that silently drops calls when the app is gone. 

7""" 

8 

9from __future__ import annotations 

10 

11import logging 

12from typing import Any 

13 

14from textual.dom import DOMNode 

15 

16log = logging.getLogger(__name__) 

17 

18 

19def call_from_thread(node: DOMNode, fn: Any, *args: Any, **kwargs: Any) -> None: 

20 """Post *fn* to the main thread via the app. 

21 

22 Drops the call (does not crash the worker) when the target node's app 

23 is no longer reachable, e.g. during shutdown or after a screen was 

24 replaced. Logs at debug so the drop is discoverable without leaking 

25 warning text into the TUI render (textual's log handler routes 

26 stderr into the rendered frame). Long-running workers that must 

27 survive a screen switch should own their state on the app 

28 (TaskBarController pattern in widgets/task_bar.py) rather than 

29 relying on this wrapper. 

30 """ 

31 # Resolving the app is guarded separately from running *fn*. Textual's 

32 # ``node.app`` reads a contextvar that is unset in a plain thread and then 

33 # walks ``node._parent``, which raises AttributeError on a node whose 

34 # MessagePump state is gone: seen in CI as "'ChatScreen' object has no 

35 # attribute '_MessagePump__parent'" from a worker outliving its screen. 

36 # That is the same "app is gone" case this wrapper exists for, but it cannot 

37 # be folded into the except below: an AttributeError raised *inside* fn must 

38 # still propagate, which is what the docstring promises. 

39 try: 

40 app = node.app 

41 except (AttributeError, RuntimeError) as exc: 

42 log.debug("call_from_thread found no app for %s: %s", getattr(fn, "__name__", fn), exc) 

43 return 

44 try: 

45 app.call_from_thread(fn, *args, **kwargs) 

46 except (OSError, RuntimeError) as exc: 

47 # Only the shutdown signals: OSError when the message queue is closed, 

48 # RuntimeError (incl. NoActiveAppError) when the app is no longer running. 

49 # A genuine exception raised inside *fn* propagates so the bug surfaces 

50 # instead of being silently swallowed. 

51 log.debug( 

52 "call_from_thread dropped %s: %s", 

53 getattr(fn, "__name__", fn), 

54 exc, 

55 )