Add structured `QtractorSession` lifecycle

Factor nonblocking `start_qtractor()` from the blocking Qt-loop
adapter and expose guest readiness, completion, outcome, and owned
cancellation through `QtractorSession`.

Replace `MainWindow.closeEvent()` process-wide `SIGINT` with the
session shutdown callback. Publish readiness after `_async_main()`
enters all startup scopes, then restore shared Qt state and globals
after guest completion.

Add a headless regression which starts and closes two real Tractor
runtimes under one pytest-owned `QApplication`.

Prompt-IO: ai/prompt-io/opencode/20260903T205404Z_aadfd1cf_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

ai/prompt-io/opencode/20260903T205404Z_aadfd1cf_prompt_io.md

@@ -0,0 +1,59 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_faae5c497ffeS0ofaTEfbRo2LP
+timestamp: 2026-09-03T20:54:04Z
+git_ref: aadfd1cf
+scope: code
+substantive: true
+raw_file: 20260903T205404Z_aadfd1cf_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue from the completed replay handoff when next steps are clear.
+Honor the prior decision to park replay and begin the live-backend-first
+chart E2E roadmap with the Phase A Qt/Trio lifecycle seam. First perform
+the approved exhaustive local scan, then create
+`wkt/live_chart_qtractor` from exact Phase 1 commit `aadfd1cf` and
+implement nonblocking startup, structured shutdown, observable readiness
+and completion, and provider-independent lifecycle coverage without new
+interlayer mocks.
+
+## Response summary
+
+Creates the isolated worktree after classifying two stale historical
+lifecycle experiments as related but not reusable. Extracts
+`start_qtractor()` and `QtractorSession`, replaces process-wide SIGINT on
+window close with owned Trio cancellation, publishes application
+readiness at the full `_async_main()` startup boundary, and preserves
+`run_qtractor()` as the blocking Qt-loop owner which unwraps the guest
+outcome. A headless regression proves two sequential real Tractor
+runtimes start and stop cleanly under one pytest-owned Qt application.
+
+## Files changed
+
+- `piker/ui/_exec.py` - add nonblocking session startup, lifecycle
+  signals, owned cancellation, outcome capture, Qt cleanup, and the
+  blocking CLI adapter.
+- `piker/ui/_window.py` - inject structured shutdown and remove normal
+  window-close SIGINT while cleaning the global zoom filter.
+- `piker/ui/_app.py` - publish readiness after all application startup
+  scopes are live.
+- `tests/ui/test_qtractor.py` - prove repeated real-Tractor lifecycle
+  startup and close under pytest-qt.
+
+## Human edits
+
+The human chose exhaustive local discovery specifically to catch prior
+Qt/Trio experiments, confirmed the replay branch had been pushed to
+Gitea for backup, and directed continuation without another planning
+pause. The inherited human-approved roadmap parks replay, prioritizes a
+real Kraken chart, forbids additional interlayer mocks without a concrete
+regression, and fixes `aadfd1cf` as this worktree's base. The human also
+explicitly authorized the focused headless test through `nix develop`;
+no real-compositor run was authorized or performed. During final review,
+the human corrected nonconforming boolean layout, requested local type
+annotations wherever helper return types were not obvious at the binding,
+and replaced silent Qt disconnect suppression with Piker-style exception
+logging.

ai/prompt-io/opencode/20260903T205404Z_aadfd1cf_prompt_io.raw.md

@@ -0,0 +1,59 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-09-03T20:54:04Z
+git_ref: aadfd1cf
+diff_cmd: git diff HEAD
+---
+
+## Generated changes
+
+> `git diff HEAD -- piker/ui/_exec.py piker/ui/_window.py`
+
+Extracts nonblocking `start_qtractor()` from the blocking CLI adapter.
+The returned `QtractorSession` exposes the production window and central
+widget, the Trio token and cancellation scope, readiness and completion
+Qt signals, the final `outcome.Outcome`, Qt-loop ownership, and an
+idempotent structured shutdown request. Caller-owned Qt application
+state and installed callbacks are restored when the guest completes.
+`MainWindow.closeEvent()` retains geometry persistence but delegates
+runtime teardown through the injected callback instead of sending
+process-wide SIGINT.
+
+> `git diff HEAD -- piker/ui/_app.py`
+
+Publishes application readiness only after initial symbol loading,
+order-mode startup, search registration, provider-search task startup,
+input handler entry, and startup-status removal.
+
+> `git diff HEAD -- tests/ui/test_qtractor.py`
+
+Adds a provider-independent lifecycle regression which runs two real
+Tractor root runtimes in sequence on pytest's unique registry. Real
+window-close events drive structured cancellation while signal, outcome,
+task-finalization, global-window, runtime, and shared-QApplication
+assertions prove restartable teardown without an interlayer fake.
+
+Human review then brought the generated Python into project style:
+boolean connectives use dedicated vertical lines, opaque helper returns
+and control-flow targets carry local annotations, and Qt disconnect
+failures are logged instead of silently suppressed.
+
+## Verification
+
+- Ruff passes for all four changed Python files.
+- Python static compilation and `git diff --check` pass.
+- The first focused run exposed the test widget's missing production
+  `on_win_resize()` contract; the narrow test widget was corrected.
+- The focused lifecycle regression passes with two sequential real
+  Tractor sessions.
+- The combined UI harness and lifecycle run passes: 3 tests and two
+  upstream Tractor `Registrar.uid` deprecation warnings.
+- A post-pass attempt to disconnect a pre-existing Qt signal proxy
+  segfaulted during PyQt teardown and was fully reverted; it is not part
+  of this patch. The restored lifecycle regression passes again.
+- All automated Qt execution used `--headless` and the offscreen
+  platform.
+- The approved `nix develop` shell hook unexpectedly provisioned the
+  ignored worktree-local `py313/` environment; no tracked dependency or
+  lock files changed.

piker/ui/_app.py

@@ -38,7 +38,10 @@ from ..log import (
     get_console_log,
 )
 from ..service import maybe_spawn_datad
-from ._exec import run_qtractor
+from ._exec import (
+    QtractorSession,
+    run_qtractor,
+)
 
 log = get_logger(__name__)
 
@@ -194,6 +197,14 @@ async def _async_main(
             ):
                 # remove startup status text
                 starting_done()
+                qtractor_session: QtractorSession|None = (
+                    godwidget.window.qtractor_session
+                )
+                if qtractor_session is None:
+                    raise RuntimeError(
+                        'No Qtractor session owns the main window'
+                    )
+                qtractor_session.set_ready()
                 await trio.sleep_forever()
 
 

piker/ui/_exec.py

@@ -22,24 +22,22 @@ All global Qt runtime settings are mostly defined here.
 """
 from __future__ import annotations
 from typing import (
-    Callable,
     Any,
-    Type,
+    Awaitable,
+    Callable,
     TYPE_CHECKING,
 )
 import platform
-import traceback
 
 import qdarkstyle
 from qdarkstyle import DarkPalette
 # import qdarkgraystyle  # TODO: play with it
 import trio
-from outcome import Error
+from outcome import Outcome
 
 # Qt version-agnostic
 from .qt import (
     QWidget,
-    QMainWindow,
     QApplication,
     QtCore,
     pyqtRemoveInputHook,
@@ -56,6 +54,7 @@ from . import _style
 
 
 if TYPE_CHECKING:
+    from ._window import MainWindow
     from ._widget import GodWidget
 
 
@@ -70,7 +69,7 @@ _do_overrides()
 # XXX: pretty sure none of this shit works on linux as per:
 # https://bugreports.qt.io/browse/QTBUG-53022
 # it seems to work on windows.. no idea wtf is up.
-is_windows = False
+is_windows: bool = False
 if platform.system() == "Windows":
     is_windows = True
 
@@ -83,24 +82,131 @@ if platform.system() == "Windows":
         QCoreApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True)
 
 
-def run_qtractor(
+class QtractorSession(QtCore.QObject):
-    func: Callable,
+    '''
-    args: tuple,
+    Own one Trio guest run embedded in a Qt application.
-    main_widget_type: Type[QWidget],
+
-    tractor_kwargs: dict[str, Any] = {},
+    '''
-    window_type: QMainWindow = None,
+    application_ready = QtCore.pyqtSignal()
+    completed = QtCore.pyqtSignal(object)
+
+    def __init__(
+        self,
+        app: QApplication,
+        window: MainWindow,
+        main_widget: QWidget,
+        owns_app_exec: bool,
 
     ) -> None:
         '''
-    Run the Qt event loop and embed `trio` via guest mode on it.
+        Initialize state owned by one Qt-hosted Trio guest run.
+
+        '''
+        super().__init__()
+        self.app = app
+        self.window = window
+        self.main_widget = main_widget
+        self.owns_app_exec = owns_app_exec
+        self.trio_token: trio.lowlevel.TrioToken|None = None
+        self.cancel_scope: trio.CancelScope|None = None
+        self.outcome: Outcome[Any]|None = None
+        self.is_ready: bool = False
+        self._shutdown_requested: bool = False
+        self._shutdown_scheduled: bool = False
+
+    def set_ready(self) -> None:
+        '''
+        Publish that the application entered all startup scopes.
+
+        '''
+        if self.is_ready:
+            return
+
+        self.is_ready = True
+        self.application_ready.emit()
+
+    def request_shutdown(self) -> None:
+        '''
+        Cancel the owned Trio guest run through its host token.
+
+        '''
+        if self.outcome is not None:
+            return
+
+        self._shutdown_requested = True
+        token: trio.lowlevel.TrioToken|None = self.trio_token
+        cancel_scope: trio.CancelScope|None = self.cancel_scope
+        if (
+            token is None
+            or
+            cancel_scope is None
+            or
+            self._shutdown_scheduled
+        ):
+            return
+
+        self._shutdown_scheduled = True
+        try:
+            token.run_sync_soon(cancel_scope.cancel)
+        except trio.RunFinishedError:
+            log.debug(
+                'Trio guest run already finished before shutdown'
+            )
+
+    def _guest_started(
+        self,
+        token: trio.lowlevel.TrioToken,
+        cancel_scope: trio.CancelScope,
+
+    ) -> None:
+        self.trio_token = token
+        self.cancel_scope = cancel_scope
+        if self._shutdown_requested:
+            cancel_scope.cancel()
+
+    def _guest_completed(
+        self,
+        final_outcome: Outcome[Any],
+
+    ) -> None:
+        self.outcome = final_outcome
+        self.completed.emit(final_outcome)
+
+
+def start_qtractor(
+    func: Callable[..., Awaitable[Any]],
+    args: tuple[Any, ...],
+    main_widget_type: type[QWidget],
+    tractor_kwargs: dict[str, Any]|None = None,
+    window_type: type[MainWindow]|None = None,
+    app: QApplication|None = None,
+    owns_app_exec: bool = False,
+
+) -> QtractorSession:
+    '''
+    Start a Trio guest run without entering Qt's event loop.
 
     '''
     # avoids annoying message when entering debugger from qt loop
     pyqtRemoveInputHook()
 
-    app = QApplication.instance()
+    running_app: QApplication|None = QApplication.instance()
     if app is None:
-        app = QApplication([])
+        app = running_app or QApplication([])
+    elif (
+        running_app is not None
+        and
+        running_app is not app
+    ):
+        raise RuntimeError(
+            'The requested QApplication is not the active instance'
+        )
+
+    prior_quit_on_last_window: bool = (
+        app.quitOnLastWindowClosed()
+    )
+    prior_cursor_flash_time: int = app.cursorFlashTime()
+    prior_stylesheet: str = app.styleSheet()
 
     # TODO: we might not need this if it's desired
     # to cancel the tractor machinery on Qt loop
@@ -118,39 +224,58 @@ def run_qtractor(
 
     # source gist and credit to njs:
     # https://gist.github.com/njsmith/d996e80b700a339e0623f97f48bcf0cb
-    REENTER_EVENT = QtCore.QEvent.Type(QtCore.QEvent.registerEventType())
+    REENTER_EVENT: QtCore.QEvent.Type = QtCore.QEvent.Type(
+        QtCore.QEvent.registerEventType()
+    )
 
     class ReenterEvent(QtCore.QEvent):
-        pass
+        '''
+        Carry one Trio guest callback through Qt's event queue.
+
+        '''
+        def __init__(
+            self,
+            fn: Callable[[], None],
+
+        ) -> None:
+            '''
+            Initialize the event with its scheduled callback.
+
+            '''
+            super().__init__(REENTER_EVENT)
+            self.fn: Callable[[], None] = fn
 
     class Reenter(QtCore.QObject):
-        def event(self, event):
+        '''
+        Dispatch Trio re-entry events received from Qt.
+
+        '''
+        def event(self, event: QtCore.QEvent) -> bool:
+            '''
+            Run a scheduled callback or defer another event to Qt.
+
+            '''
+            if not isinstance(event, ReenterEvent):
+                return super().event(event)
+
             event.fn()
             return False
 
-    reenter = Reenter()
+    reenter: Reenter = Reenter()
 
-    def run_sync_soon_threadsafe(fn):
+    def run_sync_soon_threadsafe(
-        event = ReenterEvent(REENTER_EVENT)
+        fn: Callable[[], None],
-        event.fn = fn
+
+    ) -> None:
+        '''
+        Post one Trio guest callback onto Qt's event queue.
+
+        '''
+        event: ReenterEvent = ReenterEvent(fn)
         app.postEvent(reenter, event)
 
-    def done_callback(outcome):
-
-        if isinstance(outcome, Error):
-            exc = outcome.error
-
-            if isinstance(outcome.error, KeyboardInterrupt):
-                # make it kinda look like ``trio``
-                print("Terminated!")
-
-            else:
-                traceback.print_exception(type(exc), exc, exc.__traceback__)
-
-        app.quit()
-
     # load dark theme
-    stylesheet = qdarkstyle.load_stylesheet(
+    stylesheet: str = qdarkstyle.load_stylesheet(
         qt_api='pyqt6',
         palette=DarkPalette,
     )
@@ -162,7 +287,7 @@ def run_qtractor(
     if window_type is None:
         window_type = _window.MainWindow
 
-    window = window_type()
+    window: MainWindow = window_type()
 
     # set global app's main window singleton
     _window._qt_win = window
@@ -177,19 +302,66 @@ def run_qtractor(
     instance: GodWidget = main_widget_type()
     instance.window = window
 
+    session = QtractorSession(
+        app=app,
+        window=window,
+        main_widget=instance,
+        owns_app_exec=owns_app_exec,
+    )
+    window.qtractor_session = session
+    window.set_shutdown_callback(session.request_shutdown)
+
     # override tractor's defaults
+    tractor_kwargs = dict(tractor_kwargs or {})
     tractor_kwargs.update(
         get_runtime_vars()
     )
 
     # define tractor entrypoint
-    async def main():
+    async def main() -> None:
+        '''
+        Run the application under its owned Trio cancel scope.
 
+        '''
+        cancel_scope: trio.CancelScope
+        with trio.CancelScope() as cancel_scope:
+            session._guest_started(
+                trio.lowlevel.current_trio_token(),
+                cancel_scope,
+            )
             async with maybe_open_pikerd(
                 **tractor_kwargs,
             ):
                 await func(*((instance,) + args))
 
+    def done_callback(final_outcome: Outcome[Any]) -> None:
+        '''
+        Restore Qt host state and publish guest completion.
+
+        '''
+        try:
+            app.focusChanged.disconnect(window.on_focus_change)
+        except (
+            TypeError,
+            RuntimeError,
+        ):
+            log.exception(
+                'Failed to disconnect '
+                '`MainWindow.on_focus_change()`'
+            )
+        window.uninstall_global_zoom_filter()
+        window.set_shutdown_callback(None)
+        window.qtractor_session = None
+        if _window._qt_win is window:
+            _window._qt_win = None
+        if not owns_app_exec:
+            app.setQuitOnLastWindowClosed(
+                prior_quit_on_last_window
+            )
+            app.setCursorFlashTime(prior_cursor_flash_time)
+            app.setStyleSheet(prior_stylesheet)
+        session._guest_completed(final_outcome)
+
     # guest mode entry
     trio.lowlevel.start_guest_run(
         main,
@@ -208,4 +380,46 @@ def run_qtractor(
 
     # actually render to screen
     window.show()
-    app.exec_()
+    return session
+
+
+def run_qtractor(
+    func: Callable[..., Awaitable[Any]],
+    args: tuple[Any, ...],
+    main_widget_type: type[QWidget],
+    tractor_kwargs: dict[str, Any]|None = None,
+    window_type: type[MainWindow]|None = None,
+
+) -> None:
+    '''
+    Run Qt and unwrap the embedded Trio guest run's result.
+
+    '''
+    session: QtractorSession = start_qtractor(
+        func=func,
+        args=args,
+        main_widget_type=main_widget_type,
+        tractor_kwargs=tractor_kwargs,
+        window_type=window_type,
+        owns_app_exec=True,
+    )
+    session.completed.connect(session.app.quit)
+    try:
+        session.app.exec_()
+    finally:
+        try:
+            session.completed.disconnect(session.app.quit)
+        except (
+            TypeError,
+            RuntimeError,
+        ):
+            log.exception(
+                'Failed to disconnect `QApplication.quit()`'
+            )
+
+    final_outcome: Outcome[Any]|None = session.outcome
+    if final_outcome is None:
+        raise RuntimeError(
+            'Qt exited before the Trio guest run completed'
+        )
+    final_outcome.unwrap()

piker/ui/_window.py

@@ -19,11 +19,10 @@ Qt main window singletons and stuff.
 
 """
 from __future__ import annotations
-import os
-import signal
 import time
 from typing import (
     Callable,
+    TYPE_CHECKING,
     Union,
 )
 import uuid
@@ -51,6 +50,10 @@ from ._style import (
 from ._widget import GodWidget
 
 
+if TYPE_CHECKING:
+    from ._exec import QtractorSession
+
+
 log = get_logger(__name__)
 
 
@@ -291,15 +294,45 @@ class MainWindow(QMainWindow):
 
         # event filter for global zoom shortcuts
         self._zoom_filter: GlobalZoomEventFilter | None = None
+        self._shutdown_callback: Callable[[], None]|None = None
+        self.qtractor_session: QtractorSession|None = None
 
     def install_global_zoom_filter(self) -> None:
         '''Install application-level event filter for global UI zoom shortcuts.'''
         if self._zoom_filter is None:
             self._zoom_filter = GlobalZoomEventFilter(self)
-            app = QApplication.instance()
+            app: QApplication|None = QApplication.instance()
+            if app is None:
+                raise RuntimeError(
+                    'Cannot install zoom filter without a QApplication'
+                )
             app.installEventFilter(self._zoom_filter)
             log.info('Installed global zoom shortcuts: Ctrl+Shift+Plus/Minus/0')
 
+    def uninstall_global_zoom_filter(self) -> None:
+        '''
+        Remove this window's application-level zoom filter.
+
+        '''
+        if self._zoom_filter is None:
+            return
+
+        app: QApplication|None = QApplication.instance()
+        if app is not None:
+            app.removeEventFilter(self._zoom_filter)
+        self._zoom_filter = None
+
+    def set_shutdown_callback(
+        self,
+        callback: Callable[[], None]|None,
+
+    ) -> None:
+        '''
+        Set the structured runtime-shutdown request callback.
+
+        '''
+        self._shutdown_callback = callback
+
     @property
     def mode_label(self) -> QLabel:
 
@@ -341,11 +374,10 @@ class MainWindow(QMainWindow):
         settings.setValue('windowGeometry', self.saveGeometry())
         log.debug('Saved window geometry for next session')
 
-        # raising KBI seems to get intercepted by by Qt so just use the system.
+        if self._shutdown_callback is not None:
-        os.kill(
+            self._shutdown_callback()
-            os.getpid(),
+
-            signal.SIGINT,
+        super().closeEvent(event)
-        )
 
     @property
     def status_bar(self) -> QStatusBar:

tests/ui/test_qtractor.py

@@ -0,0 +1,148 @@
+from types import SimpleNamespace
+from typing import Any
+
+from PyQt6.QtGui import QResizeEvent
+from PyQt6.QtWidgets import (
+    QApplication,
+    QWidget,
+)
+from pytestqt.qtbot import QtBot
+from pytestqt.wait_signal import SignalBlocker
+import tractor
+import trio
+
+from piker.ui import _window
+from piker.ui._exec import (
+    QtractorSession,
+    start_qtractor,
+)
+from piker.ui._window import MainWindow
+
+
+class QtractorTestWidget(QWidget):
+    '''
+    Supply the central-widget resize hook expected by `MainWindow`.
+
+    '''
+    window: MainWindow
+
+    def on_win_resize(self, event: QResizeEvent) -> None:
+        '''
+        Accept production `MainWindow.resizeEvent()` notifications.
+
+        '''
+
+
+async def run_until_window_close(
+    main_widget: QtractorTestWidget,
+    exited: list[QtractorTestWidget],
+
+) -> None:
+    '''
+    Publish readiness and retain the guest task until cancellation.
+
+    '''
+    session: QtractorSession|None = (
+        main_widget.window.qtractor_session
+    )
+    if session is None:
+        raise RuntimeError(
+            'No Qtractor session owns the test window'
+        )
+    session.set_ready()
+    try:
+        await trio.sleep_forever()
+    finally:
+        exited.append(main_widget)
+
+
+def test_qtractor_restarts_after_structured_window_close(
+    isolated_ui_state: SimpleNamespace,
+    qapp: QApplication,
+    qtbot: QtBot,
+    reg_addr: tuple[str, int],
+
+) -> None:
+    '''
+    Repeated Qt sessions must not signal or strand the Trio runtime.
+
+    `MainWindow.closeEvent()` previously sent process-wide SIGINT
+    while `run_qtractor()` hid the guest outcome behind blocking
+    `app.exec_()`. A caller-owned pytest-qt application therefore
+    could not observe a clean startup or close one session without
+    interrupting pytest.
+
+    Start two real Tractor root runtimes in sequence on the
+    test-owned registry. Each guest publishes readiness only after
+    entering its task, then a real window-close event requests
+    cancellation through `QtractorSession`. Completion, the guest
+    `finally` block, the empty runtime slot, and successful second
+    startup prove structured teardown completed without quitting the
+    shared Qt application.
+
+    '''
+    assert isolated_ui_state.qapp is qapp
+    exited: list[QtractorTestWidget] = []
+    qt_state: tuple[bool, int, str] = (
+        qapp.quitOnLastWindowClosed(),
+        qapp.cursorFlashTime(),
+        qapp.styleSheet(),
+    )
+    tractor_kwargs: dict[str, Any] = {
+        'registry_addrs': [reg_addr],
+        'tractor_runtime_overrides': {
+            'piker_test_dir': str(
+                isolated_ui_state.test_config_dir
+            ),
+        },
+    }
+
+    _: int
+    for _ in range(2):
+        session: QtractorSession = start_qtractor(
+            func=run_until_window_close,
+            args=(exited,),
+            main_widget_type=QtractorTestWidget,
+            tractor_kwargs=tractor_kwargs,
+            app=qapp,
+        )
+        qtbot.addWidget(session.window)
+
+        assert session.app is qapp
+        assert not session.owns_app_exec
+        assert session.window.godwidget is session.main_widget
+        assert session.window.isVisible()
+        assert session.outcome is None
+
+        ready: SignalBlocker = qtbot.waitSignal(
+            session.application_ready,
+            timeout=30_000,
+        )
+        ready.wait()
+        assert session.is_ready
+        assert session.trio_token is not None
+        assert session.cancel_scope is not None
+
+        completed: SignalBlocker
+        with qtbot.waitSignal(
+            session.completed,
+            timeout=30_000,
+        ) as completed:
+            session.window.close()
+
+        assert completed.args == [session.outcome]
+        assert session.outcome is not None
+        assert session.outcome.unwrap() is None
+        assert session.cancel_scope.cancel_called
+        assert exited[-1] is session.main_widget
+        assert session.window.qtractor_session is None
+        assert _window._qt_win is None
+        assert tractor.current_actor(
+            err_on_no_runtime=False,
+        ) is None
+        assert not qapp.closingDown()
+        assert (
+            qapp.quitOnLastWindowClosed(),
+            qapp.cursorFlashTime(),
+            qapp.styleSheet(),
+        ) == qt_state