Run the app and test entry points through one message pump task path

evidence/shared-pump-task-lifetime-20261006/README.md

@@ -0,0 +1,48 @@
+# Shared message-pump task lifetime
+
+Native73 left a real-entry cycle: run_test installed App._task, while run_async
+entered App's replacement _process_messages with no task enrollment. A routed
+pointer's completion could then bubble into its waiting App queue.
+
+MessagePump._process_messages now owns actual task acquisition and release for
+both App entrypoints and scheduled widget pumps. App supplies its existing startup
+body through _process_messages_body. The test launcher retains only its local join
+handle; Screen no longer clears task custody during its exit hook. Widget scheduling
+still enrolls the original task before its coroutine starts, and never reinstalls
+an eager task which has already completed. Timer/Worker/Markdown tasks are distinct
+resources and stay with their original owners.
+
+Original pointer completion, driver coroutine admission, capture, queues, shutdown
+and selection boundaries are unchanged. No missing-task fallback or new queue,
+timer, Toad override or completion exemption is added. App's duplicate body context
+was removed; the shared lifetime borrows its existing polymorphic _context.
+
+Existing refactor-audit Package parsed all 249 native production modules and
+462 before / 463 after test modules, with zero omissions. before.json and after.json
+record declaration and consumer coordinates. All App/Screen pump-task writes were
+deleted; initialization, scheduled enrollment, acquisition and release now belong
+to MessagePump. Independent Timer/Worker/Markdown tasks are included in lexical
+results but retain their distinct owners. Dynamic external subclasses remain
+unresolved explicitly.
+
+## Actual entrypoint qualification
+
+`PYTHONPATH=src PYTHONDONTWRITEBYTECODE=1 /usr/bin/python -m pytest -q
+tests/test_real_app_pump_lifetime.py --tb=short`: **4 passed in 0.69s**.
+
+The actual run_async/HeadlessDriver ingress delivered press, move, release and
+wheel in order, including target-to-App bubbling and capture release. The actual
+pump task matched its caller, cleared at pump exit, and child tasks joined. Other
+cases covered startup failure, cancellation and eager completion without task
+reenrollment. These are source App checks using the original driver and queues.
+They do not use run_test to demonstrate real-entry task ownership.
+
+Original collection refusal and test-assertion refusal are preserved. The latter
+incorrectly assumed an inline run_async caller Task must finish when its borrowed
+pump lifetime finishes. The corrected checks create and join the actual run_async
+Task explicitly; no production or completion rule changed after that refusal.
+
+The old 33 native73 controls and closed physical03 were not rerun. No installed,
+physical, provider, package or public runtime operation occurred. The remaining
+real-entry acceptance is Parent's actual terminal/Toad motion journey against the
+matching source; the original physical03 cause remains unproved without a stack.

evidence/shared-pump-task-lifetime-20261006/caller-task-assertion-refusal.log

@@ -0,0 +1,50 @@
+FF..                                                                     [100%]
+=================================== FAILURES ===================================
+_______ test_run_async_driver_burst_owns_bubbling_pointer_until_terminal _______
+tests/test_real_app_pump_lifetime.py:87: in test_run_async_driver_burst_owns_bubbling_pointer_until_terminal
+    assert app.startup_task.done()
+E   AssertionError: assert False
+E    +  where False = <built-in method done of _asyncio.Task object at 0x7f52d1d663f0>()
+E    +    where <built-in method done of _asyncio.Task object at 0x7f52d1d663f0> = <Task pending name='Task-1' coro=<test_run_async_driver_burst_owns_bubbling_pointer_until_terminal() running at /home/...sts/test_real_app_pump_lifetime.py:87> cb=[_run_until_complete_cb() at /usr/lib/python3.14/asyncio/base_events.py:181]>.done
+E    +      where <Task pending name='Task-1' coro=<test_run_async_driver_burst_owns_bubbling_pointer_until_terminal() running at /home/...sts/test_real_app_pump_lifetime.py:87> cb=[_run_until_complete_cb() at /usr/lib/python3.14/asyncio/base_events.py:181]> = RealEntryApp(title='RealEntryApp', classes={'-dark-mode'}, pseudo_classes={'dark', 'nocolor', 'focus'}).startup_task
+_____________ test_run_async_startup_failure_releases_shared_task ______________
+tests/test_real_app_pump_lifetime.py:106: in test_run_async_startup_failure_releases_shared_task
+    assert app.observed_task.done()
+E   AssertionError: assert False
+E    +  where False = <built-in method done of _asyncio.Task object at 0x7f52d26b4910>()
+E    +    where <built-in method done of _asyncio.Task object at 0x7f52d26b4910> = <Task pending name='Task-16' coro=<test_run_async_startup_failure_releases_shared_task() running at /home/ts/wt/textua...ts/test_real_app_pump_lifetime.py:106> cb=[_run_until_complete_cb() at /usr/lib/python3.14/asyncio/base_events.py:181]>.done
+E    +      where <Task pending name='Task-16' coro=<test_run_async_startup_failure_releases_shared_task() running at /home/ts/wt/textua...ts/test_real_app_pump_lifetime.py:106> cb=[_run_until_complete_cb() at /usr/lib/python3.14/asyncio/base_events.py:181]> = FailedStartup(title='FailedStartup', classes={'-dark-mode'}, pseudo_classes={'dark', 'nocolor', 'focus'}).observed_task
+----------------------------- Captured stderr call -----------------------------
+╭───────────────────── Traceback (most recent call last) ──────────────────────╮
+│ /home/ts/wt/textual-frame-publication-20261006/tests/test_real_app_pump_life │
+│ time.py:98 in on_mount                                                       │
+│                                                                              │
+│    95 │   │   def on_mount(self):                                            │
+│    96 │   │   │   self.observed_task = self.task                             │
+│    97 │   │   │   assert self.task is asyncio.current_task()                 │
+│ ❱  98 │   │   │   raise RuntimeError("original startup failure")             │
+│    99 │                                                                      │
+│   100 │   app = FailedStartup()                                              │
+│   101 │   await asyncio.wait_for(app.run_async(headless=True, size=(20, 4)), │
+│                                                                              │
+│ ╭─────────────────────────── locals ────────────────────────────╮            │
+│ │ @py_assert1 = None                                            │            │
+│ │ @py_assert3 = None                                            │            │
+│ │ @py_assert5 = None                                            │            │
+│ │ @py_assert7 = None                                            │            │
+│ │        self = FailedStartup(                                  │            │
+│ │               │   title='FailedStartup',                      │            │
+│ │               │   classes={'-dark-mode'},                     │            │
+│ │               │   pseudo_classes={                            │            │
+│ │               │   │   'dark',                                 │            │
+│ │               │   │   'nocolor',                              │            │
+│ │               │   │   'focus'                                 │            │
+│ │               │   }                                           │            │
+│ │               )                                               │            │
+│ ╰───────────────────────────────────────────────────────────────╯            │
+╰──────────────────────────────────────────────────────────────────────────────╯
+RuntimeError: original startup failure
+=========================== short test summary info ============================
+FAILED tests/test_real_app_pump_lifetime.py::test_run_async_driver_burst_owns_bubbling_pointer_until_terminal
+FAILED tests/test_real_app_pump_lifetime.py::test_run_async_startup_failure_releases_shared_task
+2 failed, 2 passed in 0.83s

evidence/shared-pump-task-lifetime-20261006/collection-refusal.log

@@ -0,0 +1,16 @@
+
+==================================== ERRORS ====================================
+____________ ERROR collecting tests/test_real_app_pump_lifetime.py _____________
+ImportError while importing test module '/home/ts/wt/textual-frame-publication-20261006/tests/test_real_app_pump_lifetime.py'.
+Hint: make sure your test modules/packages have valid Python names.
+Traceback:
+/usr/lib/python3.14/importlib/__init__.py:88: in import_module
+    return _bootstrap._gcd_import(name[level:], package, level)
+           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+tests/test_real_app_pump_lifetime.py:10: in <module>
+    from test_pointer_capture_order import send_pointer
+E   ModuleNotFoundError: No module named 'test_pointer_capture_order'
+=========================== short test summary info ============================
+ERROR tests/test_real_app_pump_lifetime.py
+!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
+1 error in 0.54s

evidence/shared-pump-task-lifetime-20261006/real-entry.log

@@ -0,0 +1,2 @@
+....                                                                     [100%]
+4 passed in 0.69s

src/textual/app.py

@@ -2208,7 +2208,7 @@ async def run_app(app: App[ReturnType]) -> None:
 
         # Launch the app in the "background"
 
-        self._task = app_task = create_task(run_app(app), name=f"run_test {app}")
+        app_task = create_task(run_app(app), name=f"run_test {app}")
 
         # Wait until the app has performed all startup routines.
         await app_ready_event.wait()
@@ -3381,7 +3381,7 @@ async def _init_devtools(self):
             except DevtoolsConnectionError:
                 self.log.system(f"Couldn't connect to devtools ( {self.devtools.url} )")
 
-    async def _process_messages(
+    async def _process_messages_body(
         self,
         ready_callback: CallbackType | None = None,
         headless: bool = False,
@@ -3492,51 +3492,50 @@ async def invoke_ready_callback() -> None:
                 finally:
                     await Timer._stop_all(self._timers)
 
-        with self._context():
-            if not await app_prelude():
-                return
-            self._running = True
-            try:
-                load_event = events.Load()
-                await self._dispatch_message(load_event)
+        if not await app_prelude():
+            return
+        self._running = True
+        try:
+            load_event = events.Load()
+            await self._dispatch_message(load_event)
 
-                driver = self._driver = self._build_driver(
-                    headless=headless,
-                    inline=inline,
-                    mouse=mouse,
-                    size=terminal_size,
-                )
-                self.log(driver=driver)
+            driver = self._driver = self._build_driver(
+                headless=headless,
+                inline=inline,
+                mouse=mouse,
+                size=terminal_size,
+            )
+            self.log(driver=driver)
 
-                if not self._exit:
-                    driver.start_application_mode()
-                    try:
-                        with redirect_stdout(self._capture_stdout):
-                            with redirect_stderr(self._capture_stderr):
-                                await run_process_messages()
+            if not self._exit:
+                driver.start_application_mode()
+                try:
+                    with redirect_stdout(self._capture_stdout):
+                        with redirect_stderr(self._capture_stderr):
+                            await run_process_messages()
 
-                    finally:
-                        Reactive._clear_watchers(self)
-                        if self._driver.is_inline:
-                            cursor_x, cursor_y = self._previous_cursor_position
+                finally:
+                    Reactive._clear_watchers(self)
+                    if self._driver.is_inline:
+                        cursor_x, cursor_y = self._previous_cursor_position
+                        self._driver.write(
+                            Control.move(-cursor_x, -cursor_y).segment.text
+                        )
+                        self._driver.flush()
+                        if inline_no_clear and not self.app._exit_renderables:
+                            console = Console()
+                            try:
+                                console.print(self.screen._compositor)
+                            except ScreenStackError:
+                                console.print()
+                        else:
                             self._driver.write(
-                                Control.move(-cursor_x, -cursor_y).segment.text
+                                Control.move(0, -self.INLINE_PADDING).segment.text
                             )
-                            self._driver.flush()
-                            if inline_no_clear and not self.app._exit_renderables:
-                                console = Console()
-                                try:
-                                    console.print(self.screen._compositor)
-                                except ScreenStackError:
-                                    console.print()
-                            else:
-                                self._driver.write(
-                                    Control.move(0, -self.INLINE_PADDING).segment.text
-                                )
 
-                        driver.stop_application_mode()
-            except Exception as error:
-                self._handle_exception(error)
+                    driver.stop_application_mode()
+        except Exception as error:
+            self._handle_exception(error)
 
     async def _pre_process(self) -> bool:
         """Special case for the app, which doesn't need the functionality in MessagePump."""

src/textual/message_pump.py

@@ -596,33 +596,23 @@ def _start_messages(self) -> None:
         self._thread_init()
 
         if self.app._running:
-            self._task = create_task(
+            task = create_task(
                 self._process_messages(), name=f"message pump {self}"
             )
+            # Enroll scheduled custody until the coroutine acquires it. An
+            # eager task which already exited has released that custody.
+            if not task.done():
+                self._task = task
         else:
             self._closing = True
             self._closed = True
 
-    async def _process_messages(self) -> None:
-        self._running = True
+    async def _process_messages(self, **kwargs: Any) -> None:
+        """Own the real pump task through subclass startup, dispatch and exit."""
+        self._task = asyncio.current_task()
         try:
             with self._context():
-                try:
-                    if not await self._pre_process():
-                        return
-                    try:
-                        await self._process_messages_loop()
-                    except CancelledError:
-                        pass
-                finally:
-                    self._running = False
-                    try:
-                        if self._timers:
-                            await Timer._stop_all(self._timers)
-                            self._timers.clear()
-                        Reactive._clear_watchers(self)
-                    finally:
-                        await self._message_loop_exit()
+                await self._process_messages_body(**kwargs)
         finally:
             # Bound methods and closures in a quiet publisher may retain this
             # node despite its weak subscriber key. Teardown owns their removal,
@@ -632,6 +622,26 @@ async def _process_messages(self) -> None:
             Reactive._clear_watch_subscriptions(self)
             self._task = None
 
+    async def _process_messages_body(self, **kwargs: Any) -> None:
+        """Default widget startup and teardown inside the shared task lifetime."""
+        self._running = True
+        try:
+            if not await self._pre_process():
+                return
+            try:
+                await self._process_messages_loop()
+            except CancelledError:
+                pass
+        finally:
+            self._running = False
+            try:
+                if self._timers:
+                    await Timer._stop_all(self._timers)
+                    self._timers.clear()
+                Reactive._clear_watchers(self)
+            finally:
+                await self._message_loop_exit()
+
     async def _message_loop_exit(self) -> None:
         """Called when the message loop has completed."""
 

src/textual/screen.py

@@ -1388,7 +1388,6 @@ async def _message_loop_exit(self) -> None:
         self._arrangement_cache.clear()
         self.screen_layout_refresh_signal.unsubscribe(self)
         self._nodes._clear()
-        self._task = None
 
     def _pop_result_callback(self) -> None:
         """Remove the latest result callback from the stack."""