diff --git a/evidence/reactive-callback-retirement-20261007/README.md b/evidence/reactive-callback-retirement-20261007/README.md new file mode 100644 index 0000000000..fe3bbdbfdc --- /dev/null +++ b/evidence/reactive-callback-retirement-20261007/README.md @@ -0,0 +1,45 @@ +# Deferred watcher ownership + +Async watcher results were created by `invoke_watcher`, then hidden in a +`partial(await_watcher, ...)`. Dropping an undelivered callback did not close +the coroutine. A receiver can also be constructed and abandoned before its +message task ever starts; task exit alone cannot cover that lifetime. + +`WatcherCallback`, an original Callback subtype, now owns that result's await +and disposal. `Message._discard` supplies the delivery resource hook. +MessagePump releases rejected messages, abandoned callback batches, and pending +messages on task exit or never-started receiver destruction. Actual watcher +invocation stays immediate, arguments retain their original values, and compute +still follows successful completion. Disposing a delivery closes native +coroutines; it does not cancel independently owned Tasks or Futures. + +The source pass covered Reactive invocation/reset/subscriptions, Callback +dispatch, Message completion/bubbling/replacement, pump admission/startup/close/ +batch cancellation, and the call_next consumers in App, Widget, Screen, Signal, +Lazy, AwaitMount, AwaitRemove and AwaitComplete. No dispatch supply, directory +watcher, Toad caller, or frame publication behavior was replaced. +The existing Package AST reader parsed 250 native modules with no omissions; +the source search found no competing native or current Toad `__del__` owner. + +## Verification + +- Existing reactive and message-pump checks plus five lifetime checks: + **46 passed, 1.19s**. These cover immediate synchronous invocation, borrowed + Task ownership, running/pending watcher cancellation, closed receiver + rejection, never-mounted DirectoryTree disposal, and real tree removal. +- Original `tests/sidebar_retirement_pilot.py`, unchanged, using the released + sidebar-eviction author runtime with this native source: **exit 0, empty + stderr**. All three cycles restored actual selected filenames and scroll + positions, collected retired panel graphs, and completed source replacement + and hydration cancellation assertions. No provider inputs or package changes. + +Logs: `/home/ts/.cache/agent-scratch/native-watcher-retirement-20261007/`. +`native-final.log`, `app02.stdout.log`, and `app02.stderr.log` are the final +results. `app.stdout.log` / `app.stderr.log` preserve the initial App negative: +restoration passed but the 161-byte unawaited-coroutine warning remained before +never-started receiver disposal was added. An attempted callback-only finalizer +also failed the unmounted-tree check; that competing finalizer was deleted. + +This is native source plus affected source-App acceptance. No installed wheel, +live UI, or performance improvement is claimed. The original app fixture owns +its private child cleanup; exact child birth records were not collected here. diff --git a/src/textual/message.py b/src/textual/message.py index 25b4b9c804..46c48d8b9b 100644 --- a/src/textual/message.py +++ b/src/textual/message.py @@ -62,6 +62,9 @@ def __post_init__(self) -> None: def __rich_repr__(self) -> rich.repr.Result: yield from () + def _discard(self) -> None: + """Release resources owned by an abandoned or completed delivery.""" + def __init_subclass__( cls, bubble: bool | None = True, diff --git a/src/textual/message_pump.py b/src/textual/message_pump.py index e72f03fd34..07ff18deb4 100644 --- a/src/textual/message_pump.py +++ b/src/textual/message_pump.py @@ -552,7 +552,9 @@ def call_later(self, callback: Callback, *args: Any, **kwargs: Any) -> bool: message = events.Callback(callback=partial(callback, *args, **kwargs)) return self.post_message(message) - def call_next(self, callback: Callback, *args: Any, **kwargs: Any) -> None: + def call_next( + self, callback: Callback | events.Callback, *args: Any, **kwargs: Any + ) -> None: """Schedule a callback to run immediately after processing the current message. Args: @@ -561,11 +563,38 @@ def call_next(self, callback: Callback, *args: Any, **kwargs: Any) -> None: **kwargs: Keyword arguments to pass to the callable. """ assert callback is not None, "Callback must not be None" - callback_message = events.Callback(callback=partial(callback, *args, **kwargs)) + if isinstance(callback, events.Callback): + if args or kwargs: + raise TypeError("A callback event already owns its arguments") + callback_message = callback + else: + callback_message = events.Callback(callback=partial(callback, *args, **kwargs)) + if self._closing or self._closed: + callback_message._discard() + return callback_message._prevent.update(self._get_prevented_messages()) self._next_callbacks.append(callback_message) self.check_idle() + def _discard_next_callbacks(self) -> None: + for callback in self.__dict__.pop("_next_callbacks", ()): + callback._discard() + + def _discard_pending_messages(self) -> None: + """Release undelivered messages after their receiver has stopped.""" + self._discard_next_callbacks() + queue = self.__dict__.get("_message_queue") + if queue is not None: + while not queue.empty(): + message = queue.get_nowait() + if message is not None: + message._discard() + + def __del__(self) -> None: + # Constructor watchers may enqueue work before a widget is mounted. + # An abandoned, never-started receiver still owns those deliveries. + self._discard_pending_messages() + def _on_invoke_later(self, message: messages.InvokeLater) -> None: # Forward InvokeLater message to the Screen if self.app._running: @@ -582,6 +611,7 @@ async def _close_messages(self, wait: bool = True) -> None: return self._closing = True self._message_pump_closing() + self._discard_next_callbacks() self._clear_signal_subscriptions() Reactive._clear_watch_subscriptions(self) if self._timers: @@ -617,6 +647,7 @@ def _start_messages(self) -> None: self._closing = True self._closed = True self._message_pump_closing() + self._discard_pending_messages() async def _process_messages(self, **kwargs: Any) -> None: """Own the real pump task through subclass startup, dispatch and exit.""" @@ -629,6 +660,7 @@ async def _process_messages(self, **kwargs: Any) -> None: # node despite its weak subscriber key. Teardown owns their removal, # including cancellation and failed pre-processing. self._running = False + self._discard_pending_messages() self._clear_signal_subscriptions() Reactive._clear_watch_subscriptions(self) self._task = None @@ -745,6 +777,7 @@ async def _process_messages_loop(self) -> None: break finally: message._complete_dispatch(self, result) + message._discard() # Subscription access initializes this optional signal. Do not # allocate its weak-key subscriber machinery for quiet pumps. message_signal = self.__dict__.get("message_signal") @@ -787,13 +820,17 @@ async def _flush_next_callbacks(self) -> None: """Invoke pending callbacks in next callbacks queue.""" callbacks = self._next_callbacks.copy() self._next_callbacks.clear() - for callback in callbacks: - try: - with self.prevent(*callback._prevent): - await invoke(callback.callback) - except Exception as error: - self.app._handle_exception(error) - break + try: + for callback in callbacks: + try: + with self.prevent(*callback._prevent): + await invoke(callback.callback) + except Exception as error: + self.app._handle_exception(error) + break + finally: + for callback in callbacks: + callback._discard() async def _dispatch_message(self, message: Message) -> Message | None: """Dispatch a message received from the message queue. @@ -1006,8 +1043,10 @@ def post_message(self, message: Message) -> bool: "Message is missing attributes; did you forget to call super().__init__() ?" ) if self._closing or self._closed: + message._discard() return False if not self.check_message_enabled(message): + message._discard() return False # Add a copy of the prevented message types to the message # This is so that prevented messages are honoured by the event's handler diff --git a/src/textual/reactive.py b/src/textual/reactive.py index f45594a03e..f258fe0188 100644 --- a/src/textual/reactive.py +++ b/src/textual/reactive.py @@ -6,7 +6,7 @@ from __future__ import annotations from functools import partial -from inspect import isawaitable +from inspect import isawaitable, iscoroutine from operator import methodcaller from weakref import WeakKeyDictionary from typing import ( @@ -81,12 +81,31 @@ def __call__(self, obj: ReactableType) -> ReactiveType: return self.callback(obj) -async def await_watcher(obj: Reactable, awaitable: Awaitable[object]) -> None: - """Coroutine to await an awaitable returned from a watcher""" - _rich_traceback_omit = True - await awaitable - # Watcher may have changed the state, so run compute again - obj.post_message(events.Callback(callback=partial(Reactive._compute, obj))) +class WatcherCallback(events.Callback): + """Own a watcher's already-created result through deferred delivery.""" + + def __init__(self, obj: Reactable, awaitable: Awaitable[object]) -> None: + self.obj = obj + self.awaitable = awaitable + super().__init__(self._await_result) + + async def _await_result(self) -> None: + await self.awaitable + # Watcher may have changed the state, so run compute again. + self.obj.post_message( + events.Callback(callback=partial(Reactive._compute, self.obj)) + ) + + def _discard(self) -> None: + # A freshly returned coroutine belongs to this deferred delivery. + # Tasks / Futures have an independent execution owner; do not cancel + # them merely because this callback was never admitted. + if iscoroutine(self.awaitable): + self.awaitable.close() + + def __rich_repr__(self) -> rich.repr.Result: + yield "obj", self.obj + yield "awaitable", self.awaitable class _StoredReactiveAccess: @@ -172,9 +191,7 @@ def invoke_watcher( watch_result = cast(WatchCallbackNoArgsType, watch_function)() if isawaitable(watch_result): # Result is awaitable, so we need to await it within an async context - watcher_object.call_next( - partial(await_watcher, watcher_object, watch_result) - ) + watcher_object.call_next(WatcherCallback(watcher_object, watch_result)) @rich.repr.auto diff --git a/tests/test_reactive.py b/tests/test_reactive.py index 4211244d52..f236ea63f4 100644 --- a/tests/test_reactive.py +++ b/tests/test_reactive.py @@ -1,6 +1,8 @@ from __future__ import annotations import asyncio +import gc +from inspect import CORO_CLOSED, getcoroutinestate import pytest @@ -9,11 +11,127 @@ from textual.message_pump import MessagePump from textual.reactive import Initialize, Reactive, TooManyComputesError, reactive, var from textual.widget import Widget +from textual.widgets import DirectoryTree OLD_VALUE = 5_000 NEW_VALUE = 1_000_000 +async def test_abandoned_watcher_results_keep_invocation_and_borrowed_task_lifetime(): + """Retirement closes owned coroutines without cancelling borrowed execution.""" + results = [] + calls = [] + release = asyncio.Event() + + async def work(): + await release.wait() + + class WatchWidget(Widget): + count = var(0, init=False) + + def watch_count(self, value): + calls.append(value) + result = work() if value == 1 else asyncio.create_task(work()) + results.append(result) + return result + + async with App().run_test(): + widget = WatchWidget() + widget.count = 1 + widget.count = 2 + assert calls == [1, 2] + await widget._close_messages(wait=False) + assert getcoroutinestate(results[0]) == CORO_CLOSED + assert not results[1].cancelled() + release.set() + await results[1] + + +async def test_cancelled_watcher_batch_closes_active_and_unstarted_results(): + started = asyncio.Event() + results = [] + + async def work(): + started.set() + await asyncio.Event().wait() + + class WatchWidget(Widget): + count = var(0, init=False) + + def watch_count(self): + result = work() + results.append(result) + return result + + async with App().run_test(): + widget = WatchWidget() + widget.count = 1 + widget.count = 2 + batch = asyncio.create_task(widget._flush_next_callbacks()) + await started.wait() + batch.cancel() + with pytest.raises(asyncio.CancelledError): + await batch + assert [getcoroutinestate(result) for result in results] == [ + CORO_CLOSED, + CORO_CLOSED, + ] + + +async def test_watcher_created_after_retirement_is_disposed(): + results = [] + + async def work(): + raise AssertionError("A closed receiver must not run its watcher") + + class WatchWidget(Widget): + count = var(0, init=False) + + def watch_count(self): + result = work() + results.append(result) + return result + + async with App().run_test(): + widget = WatchWidget() + await widget._close_messages(wait=False) + widget.count = 1 + assert getcoroutinestate(results[0]) == CORO_CLOSED + + +async def test_unmounted_directory_tree_releases_initial_watcher(tmp_path, recwarn): + async with App().run_test(): + tree = DirectoryTree(tmp_path) + del tree + gc.collect() + assert not [ + warning for warning in recwarn if "was never awaited" in str(warning.message) + ] + + +async def test_directory_tree_path_changes_and_removal_release_watchers( + tmp_path, recwarn +): + """Exercise the real directory watcher and removal, including pending delivery.""" + first = tmp_path / "first" + second = tmp_path / "second" + first.mkdir() + second.mkdir() + (first / "one.txt").write_text("one") + (second / "two.txt").write_text("two") + + async with App().run_test() as pilot: + for _ in range(3): + tree = DirectoryTree(first) + await pilot.app.mount(tree) + tree.path = second + await tree.remove() + gc.collect() + assert not [ + warning for warning in recwarn if "was never awaited" in str(warning.message) + ] + + async def test_initialize(): """Test that the default accepts an Initialize instance."""