refactor: make connection startup asynchronous

Replace GUI connection startup waits with a cancellable, staged Qt transaction. Observe core endpoints, DNS resolution, TUN device readiness, and runtime survival without nested event-loop waits while preserving the synchronous plugin compatibility path.

Commit runtimes only after every required stage succeeds, reject stale generations, and cover readiness, cancellation, rollback, platform ordering, and controller integration.

Signed-off-by: Loren Eteval <loren.eteval@proton.me>

Furious/Backends/Hysteria1/Plugin.py

@@ -94,6 +94,7 @@ class Hysteria1CoreRuntimeFactory(CoreRuntimeFactory):
                 Hysteria1.mmdb(routingObject.get('mmdb', '')),
             ),
             options=request.options,
+            startup=CoreRuntimeStartup(endpoint=config.httpProxy()),
         )
 
     def prepareDownloadTest(self, config, port: int):

Furious/Backends/Hysteria2/Plugin.py

@@ -211,6 +211,7 @@ class Hysteria2CoreRuntimeFactory(CoreRuntimeFactory):
             runtime,
             request.configuration,
             options=request.options,
+            startup=CoreRuntimeStartup(endpoint=request.configuration.httpProxy()),
         )
 
     def prepareDownloadTest(self, config, port: int):

Furious/Backends/Xray/Plugin.py

@@ -327,7 +327,12 @@ class XrayCoreRuntimeFactory(CoreRuntimeFactory):
         )
         runtime.xrayStatsTarget = statsTarget
 
-        return CoreRuntimeLaunch(runtime, config, options=request.options)
+        return CoreRuntimeLaunch(
+            runtime,
+            config,
+            options=request.options,
+            startup=CoreRuntimeStartup(endpoint=config.httpProxy()),
+        )
 
     def prepareDownloadTest(self, config, port: int):
         """Create an Xray configuration with one local HTTP test inbound."""

Furious/Controllers/ConnectionController.py

@@ -24,6 +24,7 @@ from Furious.Interface import *
 from Furious.Models import ServerProfile
 from Furious.Plugins import getPluginRegistry
 from Furious.Qt.DynamicTranslate import gettext as _
+from Furious.Qt.Signals import connectWeakly
 from Furious.Repository import Storage
 from Furious.Service import (
     CORE_LOG_CATEGORY,
@@ -102,6 +103,8 @@ class ConnectionController(QtCore.QObject):
         self._state = ConnectionState.Disconnected
         self._activeProfile = None
         self._lastError = None
+        self._startOperation = None
+        self._pendingHttpProxy = ''
 
         self._actionTimer = QtCore.QTimer(self)
         self._actionTimer.timeout.connect(self._callActionFromQueue)
@@ -187,6 +190,8 @@ class ConnectionController(QtCore.QObject):
 
     def _reset(self):
         """Restore disconnected state after all runtime resources stop."""
+        self._startOperation = None
+        self._pendingHttpProxy = ''
         self.progressFinished.emit(True)
         self._setActiveProfile(None)
 
@@ -269,12 +274,57 @@ class ConnectionController(QtCore.QObject):
 
         self._lastError = None
         self._setActiveProfile(configuration)
+        self._pendingHttpProxy = httpProxy
         self._startConnecting()
 
         logManager = AppLogManager()
         # Retain application diagnostics while starting a fresh runtime log.
         logManager.clear(runtimeOnly=True)
 
+        startAsync = getattr(self._coreManager, 'startAsync', None)
+
+        if callable(startAsync):
+            try:
+                operation = startAsync(
+                    configuration,
+                    routing=AppSettings.get('Routing'),
+                    exitCallback=self.coreExitCallback,
+                    msgCallbackCore=logManager.callback(CORE_LOG_CATEGORY),
+                    msgCallbackTUN_=logManager.callback(
+                        TUN2SOCKS_LOG_CATEGORY,
+                        source='Tun2socks',
+                    ),
+                )
+            except Exception as ex:
+                logger.error(f'failed to schedule core manager startup: {ex}')
+
+                return self._failConnection(
+                    f'{configuration.coreName()}: ' + _('Unknown error'),
+                    str(ex),
+                )
+
+            self._startOperation = operation
+            connectWeakly(
+                operation.succeeded,
+                self,
+                '_connectionStartSucceeded',
+                sender=operation,
+            )
+            connectWeakly(
+                operation.failed,
+                self,
+                '_connectionStartFailed',
+                sender=operation,
+            )
+            connectWeakly(
+                operation.cancelled,
+                self,
+                '_connectionStartCancelled',
+                sender=operation,
+            )
+
+            return True
+
         startExceptionDetails = ''
 
         try:
@@ -315,6 +365,10 @@ class ConnectionController(QtCore.QObject):
                 startExceptionDetails,
             )
 
+        return self._finishConnection(configuration, httpProxy)
+
+    def _finishConnection(self, configuration, httpProxy) -> bool:
+        """Commit system integration after manager runtime ownership commits."""
         settings = AppSettings.get('CustomProxyBypass')
 
         proxyServerBypass = (
@@ -352,14 +406,80 @@ class ConnectionController(QtCore.QObject):
 
         return True
 
+    @QtCore.Slot(object)
+    def _connectionStartSucceeded(self, operation):
+        """Finish the exact manager generation that committed successfully."""
+        if operation is not self._startOperation or not self.isConnecting():
+            return
+
+        self._startOperation = None
+        self._emitRuntimesChanged()
+
+        while not self._actionQueue.empty():
+            self._callActionFromQueue()
+
+        if not self.isConnecting():
+            return
+
+        configuration = self.activeProfile
+
+        if configuration is None:
+            self._failConnection(_('Unknown error'))
+
+            return
+
+        self._finishConnection(configuration, self._pendingHttpProxy)
+
+    @QtCore.Slot(object, str, str)
+    def _connectionStartFailed(self, operation, message, details):
+        """Return one failed manager generation to stable disconnected state."""
+        if operation is not self._startOperation:
+            return
+
+        self._startOperation = None
+        self._emitRuntimesChanged()
+
+        configuration = self.activeProfile
+        coreName = configuration.coreName() if configuration is not None else ''
+        startError = message or getattr(self._coreManager, 'lastStartError', '')
+        displayMessage = (f'{coreName}: ' if coreName else '') + (
+            _(startError) if startError else _('Unknown error')
+        )
+
+        self._failConnection(displayMessage, details)
+
+    @QtCore.Slot(object)
+    def _connectionStartCancelled(self, operation):
+        """Ignore stale cancellation or reset an externally cancelled start."""
+        if operation is not self._startOperation:
+            return
+
+        self._startOperation = None
+        self._emitRuntimesChanged()
+        self._reset()
+
     def startDisconnection(self, notification: str = '') -> bool:
         """Stop the active runtime and optionally request a notification."""
         if self.state is ConnectionState.Disconnected:
             return False
 
+        operation = self._startOperation
+
+        self._startOperation = None
         self._setState(ConnectionState.Disconnecting)
         self._actionTimer.stop()
 
+        if operation is not None:
+            cancelStart = getattr(self._coreManager, 'cancelStart', None)
+
+            if callable(cancelStart):
+                try:
+                    cancelStart(operation)
+                except Exception as ex:
+                    # Any non-exit exceptions
+
+                    logger.error(f'failed to cancel connection startup: {ex}')
+
         try:
             SystemProxy.off()
         except Exception as ex:
@@ -390,13 +510,10 @@ class ConnectionController(QtCore.QObject):
 
     def startReconnection(self, notification: str = '') -> bool:
         """Restart the active repository profile when lifecycle state permits."""
-        if self.state in (
-            ConnectionState.Connecting,
-            ConnectionState.Disconnecting,
-        ):
+        if self.state is ConnectionState.Disconnecting:
             return False
 
-        if self.isConnected():
+        if self.isConnected() or self.isConnecting():
             self.startDisconnection(notification)
 
         return self.startConnection()

Furious/Core/CoreProcessWorker.py

@@ -64,7 +64,12 @@ class CoreProcessState(Enum):
 
 @dataclass
 class CoreLaunchSpec:
-    """Describe the parameters required by a core launch operation."""
+    """Describe the parameters required by a core launch operation.
+
+    'waitCore' preserves the synchronous plugin compatibility path. Normal GUI
+    connections launch with it disabled and confirm readiness through the
+    connection manager's event-driven startup transaction.
+    """
 
     DefaultWaitTime: ClassVar[int] = 2500
 
@@ -437,20 +442,35 @@ class CoreProcessWorker(CoreProcessMonitor, ABC):
         self.msgQueue.setTimeout(MsgQueue.ACTIVE_DRAIN_INTERVAL)
         self.msgQueue.startTimer()
 
+        # Monitor the process during asynchronous startup as well. A child that
+        # exits before readiness must fail the owning transaction instead of
+        # waiting for a later connected-state health check.
+        self.daemon.start(CORE_CHECK_ALIVE_INTERVAL)
+
         if launchSpec.waitCore:
             # Wait for the core to start up completely
             PySide6Legacy.eventLoopWait(launchSpec.waitTime)
 
         if self.queryIsAlive():
+            # Preserve the low-level compatibility contract: a successfully
+            # spawned process is Running even when its caller disables the
+            # historical grace wait. The connection transaction separately
+            # owns semantic readiness and final commit.
             self.setState(CoreProcessState.Running)
 
-            # Start core daemon
-            self.daemon.start(CORE_CHECK_ALIVE_INTERVAL)
-
             return True
         else:
             return False
 
+    def confirmStartup(self) -> bool:
+        """Move a live externally observed process into the running state."""
+        if not self.isAlive():
+            return False
+
+        self.setState(CoreProcessState.Running)
+
+        return True
+
     def stop(self):
         """Stop the core process worker."""
         self.msgQueue.stopTimer()

Furious/Interface/Runtime.py

@@ -69,6 +69,14 @@ class CoreRuntime(ABC):
         if callable(self._exitCallback):
             self._exitCallback(self, exitcode)
 
+    def setExitCallback(self, callback):
+        """Transfer termination reporting to the runtime's current owner."""
+        self._exitCallback = callback
+
+    def confirmStartup(self) -> bool:
+        """Confirm readiness after an external startup observer succeeds."""
+        return True
+
     def startError(self) -> str:
         """Return the most recent concise startup failure, if any."""
         return self._startError

Furious/Plugins/API.py

@@ -30,6 +30,7 @@ __all__ = [
     'CoreRuntimeFactory',
     'CoreRuntimeLaunch',
     'CoreRuntimeRequest',
+    'CoreRuntimeStartup',
     'FuriousPlugin',
     'NavigationPageDescriptor',
     'NavigationPageProvider',
@@ -401,6 +402,15 @@ class CoreRuntimeRequest:
     options: Mapping[str, Any] = field(default_factory=dict)
 
 
+@dataclass(frozen=True)
+class CoreRuntimeStartup:
+    """Describe an optional event-driven readiness contract for one runtime."""
+
+    endpoint: str = ''
+    timeout: int = 2500
+    retryInterval: int = 50
+
+
 @dataclass(frozen=True)
 class CoreRuntimeLaunch:
     """Bind a constructed core runtime to its prepared start arguments."""
@@ -409,14 +419,18 @@ class CoreRuntimeLaunch:
     configuration: Any
     arguments: Tuple[Any, ...] = tuple()
     options: Mapping[str, Any] = field(default_factory=dict)
+    startup: Optional[CoreRuntimeStartup] = None
 
-    def start(self) -> bool:
+    def start(self, **optionOverrides) -> bool:
         """Start the prepared core runtime."""
+        options = dict(self.options)
+        options.update(optionOverrides)
+
         return bool(
             self.runtime.start(
                 self.configuration,
                 *self.arguments,
-                **dict(self.options),
+                **options,
             )
         )
 

Furious/Plugins/__init__.py

@@ -26,6 +26,7 @@ from .API import (
     CoreRuntimeFactory,
     CoreRuntimeLaunch,
     CoreRuntimeRequest,
+    CoreRuntimeStartup,
     FuriousPlugin,
     NavigationPageDescriptor,
     NavigationPageProvider,
@@ -74,6 +75,7 @@ __all__ = [
     'CoreRuntimeFactory',
     'CoreRuntimeLaunch',
     'CoreRuntimeRequest',
+    'CoreRuntimeStartup',
     'FuriousPlugin',
     'NavigationPageDescriptor',
     'NavigationPageProvider',

Furious/Service/AGENTS.md

@@ -12,8 +12,11 @@
 ## Runtime and asynchronous invariants
 
 - `ConnectionManager` consumes an attempt-scoped copy, asks the selected factory for native-TUN/application-tun2socks
-  policy, and owns exact runtimes only after commit. On failure it rolls back only resources acquired by that attempt and
-  restores host routing/DNS state through those owners.
+  policy, and owns exact runtimes only after commit. Normal built-in GUI startup belongs to one generation-checked
+  `ConnectionStartOperation`: launch, local-endpoint readiness, asynchronous DNS, TUN-device observation, host mutation,
+  rollback, and commit advance through owned Qt timers/signals without a nested event loop. The synchronous `start()`
+  path is a compatibility boundary for legacy plugins/non-interactive callers, not the preferred GUI path. On failure
+  or cancellation, roll back only resources acquired by that attempt and restore host routing/DNS through those owners.
 - Every async workflow defines supersession: generation/version checks for stale completion or exact-object identity for
   independent requests. Terminal paths release context, finish/abort once, and schedule each Qt reply for deletion.
   Callbacks cannot retain a shut-down manager; worker results cross into the manager's Qt thread before mutation.

Furious/Service/DnsResolver.py

@@ -22,6 +22,7 @@ from __future__ import annotations
 from Furious.Frozenlib import *
 from Furious.Models import *
 from Furious.Qt.HttpGetManager import *
+from Furious.Qt.Signals import connectWeakly
 
 from PySide6 import QtCore
 from PySide6.QtNetwork import *
@@ -30,11 +31,104 @@ from typing import Tuple
 
 import logging
 
-__all__ = ['DnsResolver']
+__all__ = ['DnsResolutionOperation', 'DnsResolver']
 
 logger = logging.getLogger(__name__)
 
 
+class DnsResolutionOperation(QtCore.QObject):
+    """Observe one recursive DNS request without nesting the Qt event loop."""
+
+    finished = QtCore.Signal(bool, object)
+
+    def __init__(self, resolver, domain, timeout=30000, parent=None):
+        """Initialize an idle resolution operation."""
+        super().__init__(parent)
+
+        self._resolver = resolver
+        self._domain = domain
+        self._timeout = max(int(timeout), 1)
+        self._resultMap = resolver._newResultMap(domain)
+        self._terminal = False
+
+        self._elapsed = QtCore.QElapsedTimer()
+        self._timer = QtCore.QTimer(self)
+        self._timer.setInterval(20)
+
+        connectWeakly(self._timer.timeout, self, '_poll')
+
+    def start(self):
+        """Start the DNS request and its event-driven completion observer."""
+        if self._terminal or self._timer.isActive():
+            return
+
+        try:
+            self._resolver._beginResolve(self._resultMap)
+        except Exception as ex:
+            # Any non-exit exceptions
+
+            logger.error(f'failed to start DNS resolution for {self._domain!r}: {ex}')
+
+            self._resultMap['error'] = True
+            self._finish()
+
+            return
+
+        self._elapsed.start()
+        self._timer.start()
+        self._poll()
+
+    def _poll(self):
+        """Finish when recursion drains, or abort this request at its deadline."""
+        if self._terminal:
+            return
+
+        if self._resultMap['depth'] == 0:
+            self._finish()
+
+            return
+
+        if self._elapsed.isValid() and self._elapsed.elapsed() >= self._timeout:
+            logger.error(
+                f'DNS resolution for {self._domain!r} reached timeout '
+                f'{self._timeout // 1000}s'
+            )
+
+            self._resultMap['error'] = True
+            self._abortReplies()
+            self._finish()
+
+    def _abortReplies(self):
+        """Abort only network replies acquired by this resolution."""
+        for networkReply in self._resultMap['reference']:
+            if (
+                isinstance(networkReply, QNetworkReply)
+                and not networkReply.isFinished()
+            ):
+                networkReply.abort()
+
+    def _finish(self):
+        """Publish exactly one terminal result."""
+        if self._terminal:
+            return
+
+        self._terminal = True
+        self._timer.stop()
+        self.finished.emit(
+            bool(self._resultMap['error']),
+            list(self._resultMap['result'].keys()),
+        )
+
+    def cancel(self):
+        """Cancel without publishing a stale result."""
+        if self._terminal:
+            return
+
+        self._terminal = True
+        self._timer.stop()
+        self._abortReplies()
+
+
 class DnsResolver(HttpGetManager):
     """Represent DNS resolver."""
 
@@ -199,17 +293,25 @@ class DnsResolver(HttpGetManager):
         resultMap['error'] = True
         resultMap['depth'] -= 1
 
-    def resolve(self, domain, timeout=30000) -> Tuple[bool, list[str]]:
-        """Resolve the DNS resolver."""
-        resultMap = {
+    @staticmethod
+    def _newResultMap(domain):
+        """Return mutable state for one recursive DNS resolution."""
+        normalizedDomain = str(domain).rstrip('.').strip().casefold()
+
+        return {
             'domain': domain,
             'depth': 0,
             'error': False,
             'reference': [],
             'result': {},
-            'visited': {str(domain).rstrip('.').strip().casefold()},
+            'visited': {normalizedDomain},
         }
 
+    def _beginResolve(self, resultMap):
+        """Start the root request for one prepared resolution state."""
+        domain = resultMap['domain']
+        normalizedDomain = str(domain).rstrip('.').strip().casefold()
+
         resultMap['depth'] += 1
 
         networkReply = self.webGET(
@@ -218,15 +320,29 @@ class DnsResolver(HttpGetManager):
             domain=domain,
             resultMap=resultMap,
             referenceDepth=0,
-            ancestry=(str(domain).rstrip('.').strip().casefold(),),
+            ancestry=(normalizedDomain,),
         )
 
         resultMap['reference'].append(networkReply)
 
+    def resolve(self, domain, timeout=30000) -> Tuple[bool, list[str]]:
+        """Resolve the DNS resolver."""
+        resultMap = self._newResultMap(domain)
+
+        self._beginResolve(resultMap)
         self.wait(resultMap, timeout=timeout)
 
         return resultMap['error'], list(resultMap['result'].keys())
 
+    def resolveAsync(self, domain, timeout=30000, parent=None):
+        """Return an event-driven DNS operation; the caller starts and owns it."""
+        return DnsResolutionOperation(
+            self,
+            domain,
+            timeout=timeout,
+            parent=parent,
+        )
+
     @staticmethod
     def wait(resultMap, startCounter=0, timeout=30000, step=100):
         """Wait for the DNS resolver operation to complete."""

Furious/Service/__init__.py

@@ -19,7 +19,11 @@
 
 from __future__ import annotations
 
-from .ConnectionManager import ConnectionManager
+from .ConnectionManager import (
+    ConnectionManager,
+    ConnectionStartOperation,
+    ConnectionStartStage,
+)
 from .ConnectivityManager import ConnectivityManager
 from .DnsResolver import DnsResolver
 from .EndpointInfoService import (
@@ -84,6 +88,8 @@ from .UpdateManager import UpdateManager
 
 __all__ = [
     'ConnectionManager',
+    'ConnectionStartOperation',
+    'ConnectionStartStage',
     'ConnectivityManager',
     'DnsResolver',
     'EndpointInfo',

tests/README.md

@@ -40,7 +40,7 @@ strategy in an individual test.
 | Configuration, profiles, migration, repositories | `test_models_and_services.py`, `test_repository_contracts.py` |
 | Generation log invariants, model fuzzing, concurrency, reclamation, complexity, and opt-in soak/latency probes | `test_log_manager_generation.py` |
 | Low-level application, runtime, editor, and storage contracts | `test_interface.py` |
-| Application composition, startup rollback, connection ownership, entry-point and crash boundaries | `test_architecture_refactors.py`, `test_application_process.py` |
+| Application composition, asynchronous readiness/TUN startup, rollback, connection ownership, entry-point and crash boundaries | `test_architecture_refactors.py`, `test_connection_startup_async.py`, `test_application_process.py` |
 | Plugin registration, capability dispatch, factories, rollback, and Hysteria1 ownership | `test_plugin_architecture.py`, `test_hysteria1_protocol.py` |
 | Controller state and error transitions with injected runtimes | `test_controllers.py` |
 | SOCKS and SIP002 Shadowsocks codecs and generated round trips | `test_socks_uri.py`, `test_shadowsocks_uri.py` |
@@ -109,7 +109,7 @@ Then run the desired test tier.
 python -m unittest discover -s tests -v
 
 # Regular logic, persistence, plugin, controller, codec, and UI regressions
-python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_architecture_refactors tests.test_plugin_architecture tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_controllers tests.test_subscription_manager tests.test_subscription_sync tests.test_profile_test_jobs tests.test_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract tests.test_xray_asset_download tests.test_native_tun_semantics tests.test_metrics_behavior tests.test_endpoint_info tests.test_service_runtime tests.test_frozenlib tests.test_isolation_and_navigation tests.test_main_window_geometry tests.test_dialog_geometry tests.test_ui_behavior tests.test_qt_interactions tests.test_stylesheet_states tests.test_theme_transition tests.test_public_api -v
+python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_architecture_refactors tests.test_connection_startup_async tests.test_plugin_architecture tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_controllers tests.test_subscription_manager tests.test_subscription_sync tests.test_profile_test_jobs tests.test_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract tests.test_xray_asset_download tests.test_native_tun_semantics tests.test_metrics_behavior tests.test_endpoint_info tests.test_service_runtime tests.test_frozenlib tests.test_isolation_and_navigation tests.test_main_window_geometry tests.test_dialog_geometry tests.test_ui_behavior tests.test_qt_interactions tests.test_stylesheet_states tests.test_theme_transition tests.test_public_api -v
 
 # Direct Qt/process integration and destruction/lifetime checks
 python -m unittest tests.test_application_process tests.test_external_core tests.test_layout_matrix tests.test_qt_lifetime -v
@@ -126,7 +126,7 @@ python -m unittest tests.test_very_heavy -v
 python -m unittest tests.test_log_manager_generation.VeryHeavyGenerationLogManagerTest -v
 
 # Shared-state order-independence spot check
-python -m unittest tests.test_public_api tests.test_theme_transition tests.test_stylesheet_states tests.test_qt_interactions tests.test_ui_behavior tests.test_dialog_geometry tests.test_main_window_geometry tests.test_isolation_and_navigation tests.test_frozenlib tests.test_service_runtime tests.test_endpoint_info tests.test_metrics_behavior tests.test_native_tun_semantics tests.test_xray_asset_download tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri tests.test_profile_test_jobs tests.test_subscription_sync tests.test_subscription_manager tests.test_controllers tests.test_hysteria2_compatibility tests.test_hysteria1_protocol tests.test_plugin_architecture tests.test_architecture_refactors tests.test_repository_contracts tests.test_models_and_services tests.test_interface -v
+python -m unittest tests.test_public_api tests.test_theme_transition tests.test_stylesheet_states tests.test_qt_interactions tests.test_ui_behavior tests.test_dialog_geometry tests.test_main_window_geometry tests.test_isolation_and_navigation tests.test_frozenlib tests.test_service_runtime tests.test_endpoint_info tests.test_metrics_behavior tests.test_native_tun_semantics tests.test_xray_asset_download tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri tests.test_profile_test_jobs tests.test_subscription_sync tests.test_subscription_manager tests.test_controllers tests.test_hysteria2_compatibility tests.test_hysteria1_protocol tests.test_plugin_architecture tests.test_connection_startup_async tests.test_architecture_refactors tests.test_repository_contracts tests.test_models_and_services tests.test_interface -v
 python -m unittest discover -s tests -v
 ```
 

tests/test_connection_startup_async.py

@@ -0,0 +1,786 @@
+# Copyright (C) 2024–present  Loren Eteval & contributors <loren.eteval@proton.me>
+#
+# This file is part of Furious.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""Exercise event-driven connection startup with real Qt event delivery."""
+
+from __future__ import annotations
+
+from Furious.Interface import CoreRuntime
+from Furious.Plugins import CoreRuntimeLaunch, CoreRuntimeStartup
+from Furious.Qt.Signals import singleShotWeakly
+from Furious.Service.ConnectionManager import (
+    ConnectionManager,
+    ConnectionStartStage,
+)
+from Furious.Service.DnsResolver import DnsResolutionOperation
+
+from PySide6 import QtCore, QtNetwork
+
+from tests.support import application, processQtEvents, waitFor
+
+import importlib
+import unittest
+
+from unittest import TestCase, mock
+
+
+class _Configuration:
+    """Provide the endpoint values used by manager-only tests."""
+
+    def httpProxy(self):
+        """Return a deterministic local proxy endpoint."""
+        return '127.0.0.1:18080'
+
+    def socksProxy(self):
+        """Return a deterministic local SOCKS endpoint."""
+        return '127.0.0.1:18081'
+
+    def remoteAddress(self):
+        """Avoid DNS during mocked TUN tests."""
+        return '192.0.2.1'
+
+
+class _Runtime(CoreRuntime):
+    """Provide an observable in-process stand-in for a core runtime."""
+
+    def __init__(self, exitCallback=None):
+        """Initialize an idle runtime and lifecycle counters."""
+        super().__init__(exitCallback)
+
+        self.alive = False
+        self.startOptions = []
+        self.stopCount = 0
+        self.disposeCount = 0
+        self.confirmCount = 0
+
+    @staticmethod
+    def name():
+        """Return the fixture runtime name."""
+        return 'Async Fixture'
+
+    @staticmethod
+    def version():
+        """Return a fixture version."""
+        return '1.0'
+
+    def start(self, _configuration=None, *_args, **kwargs):
+        """Become live and retain launch options."""
+        self.startOptions.append(dict(kwargs))
+        self.alive = True
+
+        return True
+
+    def stop(self):
+        """Stop this exact runtime."""
+        self.stopCount += 1
+        self.alive = False
+
+    def dispose(self):
+        """Record final resource disposal."""
+        self.disposeCount += 1
+        self._exitCallback = None
+
+    def isAlive(self):
+        """Return the controlled process-survival state."""
+        return self.alive
+
+    def confirmStartup(self):
+        """Record readiness confirmation."""
+        if not self.alive:
+            return False
+
+        self.confirmCount += 1
+
+        return True
+
+    def fail(self, exitcode=61):
+        """Simulate an early child-process exit."""
+        self.alive = False
+        self.callExitCallback(exitcode)
+
+
+class _Registry:
+    """Return prepared launches in deterministic order."""
+
+    def __init__(self, launches):
+        """Retain a launch queue."""
+        self.launches = list(launches)
+
+    def createCoreRuntime(self, *_args, **_kwargs):
+        """Return the next prepared runtime launch."""
+        return self.launches.pop(0)
+
+
+class _ResolverFixture(QtCore.QObject):
+    """Complete one recursive-resolution state through a real Qt timer."""
+
+    def __init__(self):
+        """Initialize without a pending result."""
+        super().__init__()
+
+        self.resultMap = None
+
+    @staticmethod
+    def _newResultMap(domain):
+        """Return the minimum state required by the observer."""
+        return {
+            'domain': domain,
+            'depth': 0,
+            'error': False,
+            'reference': [],
+            'result': {},
+        }
+
+    def _beginResolve(self, resultMap):
+        """Schedule asynchronous completion."""
+        self.resultMap = resultMap
+        resultMap['depth'] = 1
+        singleShotWeakly(0, self, '_complete')
+
+    def _complete(self):
+        """Publish one deterministic address through shared state."""
+        self.resultMap['result']['192.0.2.10'] = True
+        self.resultMap['depth'] = 0
+
+
+class ConnectionStartupAsyncTest(TestCase):
+    """Verify readiness, cancellation, rollback, and compatibility."""
+
+    def setUp(self):
+        """Ensure a Qt application exists for real timer/socket delivery."""
+        self.app = application()
+        self.servers = []
+        self.managers = []
+
+    def tearDown(self):
+        """Release every exact manager and local listener."""
+        for manager in self.managers:
+            manager.cleanup()
+
+        for server in self.servers:
+            server.close()
+            server.deleteLater()
+
+        processQtEvents()
+
+    def _manager(self):
+        """Return a manager with host TUN disabled at the test boundary."""
+        manager = ConnectionManager()
+        manager._prepareTUNPolicy = mock.Mock(return_value=(False, False))
+        self.managers.append(manager)
+
+        return manager
+
+    def _server(self):
+        """Listen on one real loopback endpoint."""
+        server = QtNetwork.QTcpServer(self.app)
+        self.assertTrue(
+            server.listen(
+                QtNetwork.QHostAddress.SpecialAddress.LocalHost,
+                0,
+            )
+        )
+        self.servers.append(server)
+
+        return server
+
+    def _operation(self, manager, launch):
+        """Start one operation through an injected registry."""
+        registry = _Registry([launch])
+        patcher = mock.patch(
+            'Furious.Service.ConnectionManager.getPluginRegistry',
+            return_value=registry,
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+        return manager.startAsync(
+            _Configuration(),
+            'Global',
+            deepcopy=False,
+        )
+
+    def testLocalEndpointReadinessCompletesEarlyAndKeepsQtResponsive(self):
+        """Commit as soon as a real listener accepts while timers still run."""
+        server = self._server()
+        endpoint = f'127.0.0.1:{server.serverPort()}'
+        runtime = _Runtime()
+        launch = CoreRuntimeLaunch(
+            runtime,
+            _Configuration(),
+            startup=CoreRuntimeStartup(endpoint=endpoint, timeout=1500),
+        )
+        manager = self._manager()
+        operation = self._operation(manager, launch)
+        succeeded = []
+        timerEvents = []
+        operation.succeeded.connect(succeeded.append)
+        QtCore.QTimer.singleShot(0, lambda: timerEvents.append(True))
+
+        self.assertEqual(operation.stage, ConnectionStartStage.Pending)
+        self.assertTrue(waitFor(lambda: bool(succeeded), timeout=0.5))
+        self.assertEqual(timerEvents, [True])
+        self.assertEqual(manager.runtimes, [runtime])
+        self.assertEqual(runtime.confirmCount, 1)
+        self.assertFalse(runtime.startOptions[0]['waitCore'])
+
+    def testReadinessTimeoutRollsBackTheExactRuntime(self):
+        """Fail a live process whose promised local endpoint never appears."""
+        server = self._server()
+        port = server.serverPort()
+        server.close()
+        runtime = _Runtime()
+        manager = self._manager()
+        operation = self._operation(
+            manager,
+            CoreRuntimeLaunch(
+                runtime,
+                _Configuration(),
+                startup=CoreRuntimeStartup(
+                    endpoint=f'127.0.0.1:{port}',
+                    timeout=60,
+                    retryInterval=5,
+                ),
+            ),
+        )
+        failures = []
+        operation.failed.connect(lambda *_args: failures.append(_args))
+
+        self.assertTrue(waitFor(lambda: bool(failures)))
+        self.assertEqual(manager.runtimes, [])
+        self.assertEqual(runtime.stopCount, 1)
+        self.assertEqual(runtime.disposeCount, 1)
+
+    def testEarlyRuntimeExitFailsOnceAndIgnoresLateProbeEvents(self):
+        """Let process termination own the terminal result before timeout."""
+        runtime = _Runtime()
+        manager = self._manager()
+        operation = self._operation(
+            manager,
+            CoreRuntimeLaunch(
+                runtime,
+                _Configuration(),
+                startup=CoreRuntimeStartup(
+                    endpoint='127.0.0.1:1',
+                    timeout=500,
+                ),
+            ),
+        )
+        failures = []
+        operation.failed.connect(lambda *_args: failures.append(_args))
+
+        processQtEvents()
+        singleShotWeakly(0, runtime, 'fail')
+
+        self.assertTrue(waitFor(lambda: bool(failures)))
+        processQtEvents(5)
+        self.assertEqual(len(failures), 1)
+        self.assertEqual(runtime.stopCount, 1)
+        self.assertEqual(runtime.disposeCount, 1)
+
+    def testCancellationAndReplacementCannotCommitAStaleGeneration(self):
+        """Cancel the first generation before a second ready launch commits."""
+        server = self._server()
+        first = _Runtime()
+        second = _Runtime()
+        manager = self._manager()
+        registry = _Registry(
+            [
+                CoreRuntimeLaunch(
+                    first,
+                    _Configuration(),
+                    startup=CoreRuntimeStartup(
+                        endpoint='127.0.0.1:1',
+                        timeout=5000,
+                    ),
+                ),
+                CoreRuntimeLaunch(
+                    second,
+                    _Configuration(),
+                    startup=CoreRuntimeStartup(
+                        endpoint=f'127.0.0.1:{server.serverPort()}',
+                        timeout=500,
+                    ),
+                ),
+            ]
+        )
+
+        with mock.patch(
+            'Furious.Service.ConnectionManager.getPluginRegistry',
+            return_value=registry,
+        ):
+            firstOperation = manager.startAsync(
+                _Configuration(),
+                'Global',
+                deepcopy=False,
+            )
+            cancelled = []
+            firstOperation.cancelled.connect(cancelled.append)
+            processQtEvents()
+
+            secondOperation = manager.startAsync(
+                _Configuration(),
+                'Global',
+                deepcopy=False,
+            )
+            succeeded = []
+            secondOperation.succeeded.connect(succeeded.append)
+
+            self.assertTrue(waitFor(lambda: bool(succeeded)))
+
+        self.assertEqual(cancelled, [firstOperation])
+        self.assertEqual(first.stopCount, 1)
+        self.assertEqual(first.disposeCount, 1)
+        self.assertEqual(manager.runtimes, [second])
+
+    def testLegacyLaunchRetainsSynchronousStartOptions(self):
+        """Do not inject waitCore into a plugin without async capability."""
+        runtime = _Runtime()
+        manager = self._manager()
+        operation = self._operation(
+            manager,
+            CoreRuntimeLaunch(runtime, _Configuration()),
+        )
+        succeeded = []
+        operation.succeeded.connect(succeeded.append)
+
+        self.assertTrue(waitFor(lambda: bool(succeeded)))
+        self.assertEqual(runtime.startOptions, [{}])
+        self.assertEqual(runtime.confirmCount, 0)
+        self.assertEqual(manager.runtimes, [runtime])
+
+    def testDnsResolutionOperationCompletesAndCancelsWithoutNestedWait(self):
+        """Observe recursive DNS state through timers and suppress stale cancel."""
+        resolver = _ResolverFixture()
+        operation = DnsResolutionOperation(
+            resolver,
+            'example.test',
+            timeout=200,
+        )
+        results = []
+        operation.finished.connect(
+            lambda error, addresses: results.append((error, addresses))
+        )
+        operation.start()
+
+        self.assertTrue(waitFor(lambda: bool(results)))
+        self.assertEqual(results, [(False, ['192.0.2.10'])])
+
+        cancelled = DnsResolutionOperation(
+            resolver,
+            'cancelled.test',
+            timeout=200,
+        )
+        staleResults = []
+        cancelled.finished.connect(
+            lambda error, addresses: staleResults.append((error, addresses))
+        )
+        cancelled.start()
+        cancelled.cancel()
+        processQtEvents(5)
+
+        self.assertEqual(staleResults, [])
+        operation.deleteLater()
+        cancelled.deleteLater()
+        resolver.deleteLater()
+
+    def testLinuxTunPreservesDeviceBeforeRuntimeOrderingWithoutNestedWait(self):
+        """Create and observe the Linux device before launching tun2socks."""
+        module = importlib.import_module('Furious.Service.ConnectionManager')
+        server = self._server()
+        primary = _Runtime()
+        tun = _Runtime()
+        tun.cleanup = None
+        events = []
+        deviceChecks = 0
+
+        def tunFactory(**kwargs):
+            tun.setExitCallback(kwargs.get('exitCallback'))
+
+            return tun
+
+        originalTunStart = tun.start
+
+        def startTun(*args, **kwargs):
+            events.append('tun-start')
+
+            return originalTunStart(*args, **kwargs)
+
+        tun.start = startTun
+
+        def findDevice(_name):
+            nonlocal deviceChecks
+
+            deviceChecks += 1
+            events.append('find-device')
+
+            return deviceChecks >= 2
+
+        def executeScript(*_args, **_kwargs):
+            events.append('script')
+
+            return True
+
+        manager = ConnectionManager()
+        manager._prepareTUNPolicy = mock.Mock(return_value=(False, True))
+        self.managers.append(manager)
+        registry = _Registry(
+            [
+                CoreRuntimeLaunch(
+                    primary,
+                    _Configuration(),
+                    startup=CoreRuntimeStartup(
+                        endpoint=f'127.0.0.1:{server.serverPort()}',
+                        timeout=500,
+                    ),
+                )
+            ]
+        )
+        shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5)
+
+        with (
+            mock.patch.object(module, 'PLATFORM', 'Linux'),
+            mock.patch.object(module, 'Tun2socks', side_effect=tunFactory),
+            mock.patch.object(
+                module,
+                'CoreRuntimeStartup',
+                return_value=shortSurvival,
+            ),
+            mock.patch.object(
+                module,
+                'getPluginRegistry',
+                return_value=registry,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'managedRoutes',
+                [],
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'getDefaultGateway',
+                return_value=[('192.0.2.254', 'eth0')],
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'LinuxFindTUNDevice',
+                side_effect=findDevice,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'LinuxGetIpRoute',
+                return_value='',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'LinuxExecutePrivilegedScript',
+                side_effect=executeScript,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'LinuxDeleteTUNDevice',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'deleteRelations',
+            ),
+            mock.patch.object(
+                module.SystemRuntime,
+                'flatpakID',
+                return_value='',
+            ),
+            mock.patch.object(
+                module.PySide6Legacy,
+                'eventLoopWait',
+                side_effect=AssertionError('nested event loop used'),
+            ),
+        ):
+            operation = manager.startAsync(
+                _Configuration(),
+                'Global',
+                deepcopy=False,
+            )
+            succeeded = []
+            operation.succeeded.connect(succeeded.append)
+
+            self.assertTrue(waitFor(lambda: bool(succeeded)))
+
+        self.assertEqual(
+            events[:4],
+            ['find-device', 'script', 'find-device', 'tun-start'],
+        )
+        self.assertEqual(manager.runtimes, [primary, tun])
+        self.assertFalse(tun.startOptions[0]['waitCore'])
+
+    def testWindowsTunStartsRuntimeBeforeObservingAndMutatingDevice(self):
+        """Keep the Windows launch, device, then host-mutation sequence."""
+        module = importlib.import_module('Furious.Service.ConnectionManager')
+        server = self._server()
+        primary = _Runtime()
+        tun = _Runtime()
+        tun.cleanup = None
+        events = []
+
+        def tunFactory(**kwargs):
+            tun.setExitCallback(kwargs.get('exitCallback'))
+
+            return tun
+
+        originalTunStart = tun.start
+
+        def startTun(*args, **kwargs):
+            events.append('tun-start')
+
+            return originalTunStart(*args, **kwargs)
+
+        tun.start = startTun
+
+        def findDevice(_name):
+            events.append('find-device')
+
+            return True
+
+        def addRelations():
+            events.append('add-relations')
+
+        manager = ConnectionManager()
+        manager._prepareTUNPolicy = mock.Mock(return_value=(False, True))
+        self.managers.append(manager)
+        registry = _Registry(
+            [
+                CoreRuntimeLaunch(
+                    primary,
+                    _Configuration(),
+                    startup=CoreRuntimeStartup(
+                        endpoint=f'127.0.0.1:{server.serverPort()}',
+                        timeout=500,
+                    ),
+                )
+            ]
+        )
+
+        with (
+            mock.patch.object(module, 'PLATFORM', 'Windows'),
+            mock.patch.object(module, 'Tun2socks', side_effect=tunFactory),
+            mock.patch.object(
+                module,
+                'getPluginRegistry',
+                return_value=registry,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'managedRoutes',
+                [],
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'delete',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'getDefaultGateway',
+                return_value=[('192.0.2.254', '192.0.2.10')],
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'WIN32IpconfigFindContent',
+                side_effect=findDevice,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'WIN32GetInterfaceAliasByIP',
+                return_value='Ethernet',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'WIN32SetInterfaceDNS',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'WIN32FlushDNSCache',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'setDeviceGateway',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'addRelations',
+                side_effect=addRelations,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'deleteRelations',
+            ),
+            mock.patch.object(
+                module.PySide6Legacy,
+                'eventLoopWait',
+                side_effect=AssertionError('nested event loop used'),
+            ),
+        ):
+            operation = manager.startAsync(
+                _Configuration(),
+                'Global',
+                deepcopy=False,
+            )
+            succeeded = []
+            operation.succeeded.connect(succeeded.append)
+
+            self.assertTrue(waitFor(lambda: bool(succeeded)))
+
+        self.assertEqual(
+            events[:3],
+            ['tun-start', 'find-device', 'add-relations'],
+        )
+        self.assertEqual(manager.runtimes, [primary, tun])
+
+    def testDarwinTunSurvivalPrecedesDnsAndRouteMutation(self):
+        """Observe tun2socks survival before applying macOS host networking."""
+        module = importlib.import_module('Furious.Service.ConnectionManager')
+        server = self._server()
+        primary = _Runtime()
+        tun = _Runtime()
+        tun.cleanup = None
+        events = []
+
+        def tunFactory(**kwargs):
+            tun.setExitCallback(kwargs.get('exitCallback'))
+
+            return tun
+
+        originalTunStart = tun.start
+
+        def startTun(*args, **kwargs):
+            events.append('tun-start')
+
+            return originalTunStart(*args, **kwargs)
+
+        tun.start = startTun
+
+        def dnsServers():
+            events.append('read-dns')
+
+            return [('Wi-Fi', ['192.0.2.53'])]
+
+        manager = ConnectionManager()
+        manager._prepareTUNPolicy = mock.Mock(return_value=(False, True))
+        self.managers.append(manager)
+        registry = _Registry(
+            [
+                CoreRuntimeLaunch(
+                    primary,
+                    _Configuration(),
+                    startup=CoreRuntimeStartup(
+                        endpoint=f'127.0.0.1:{server.serverPort()}',
+                        timeout=500,
+                    ),
+                )
+            ]
+        )
+        shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5)
+
+        with (
+            mock.patch.object(module, 'PLATFORM', 'Darwin'),
+            mock.patch.object(module, 'Tun2socks', side_effect=tunFactory),
+            mock.patch.object(
+                module,
+                'CoreRuntimeStartup',
+                return_value=shortSurvival,
+            ),
+            mock.patch.object(
+                module,
+                'getPluginRegistry',
+                return_value=registry,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'managedRoutes',
+                [],
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'getDefaultGateway',
+                return_value=['192.0.2.254'],
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'DarwinGetDNSServers',
+                side_effect=dnsServers,
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'DarwinSetDNSServers',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'setDeviceGateway',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'addRelations',
+            ),
+            mock.patch.object(
+                module.SystemRoutingTable,
+                'deleteRelations',
+            ),
+            mock.patch.object(
+                module.PySide6Legacy,
+                'eventLoopWait',
+                side_effect=AssertionError('nested event loop used'),
+            ),
+        ):
+            operation = manager.startAsync(
+                _Configuration(),
+                'Global',
+                deepcopy=False,
+            )
+            succeeded = []
+            operation.succeeded.connect(succeeded.append)
+
+            self.assertTrue(waitFor(lambda: bool(succeeded)))
+
+        self.assertEqual(events, ['tun-start', 'read-dns'])
+        self.assertEqual(manager.runtimes, [primary, tun])
+
+    def testRepeatedCancellationReleasesOperationTimersAndRuntimes(self):
+        """Keep repeated startup cancellation bounded and independently owned."""
+        manager = self._manager()
+
+        for _index in range(12):
+            runtime = _Runtime()
+            operation = self._operation(
+                manager,
+                CoreRuntimeLaunch(
+                    runtime,
+                    _Configuration(),
+                    startup=CoreRuntimeStartup(
+                        endpoint='127.0.0.1:1',
+                        timeout=5000,
+                    ),
+                ),
+            )
+            processQtEvents()
+            self.assertTrue(manager.cancelStart(operation))
+            processQtEvents()
+            self.assertEqual(runtime.stopCount, 1)
+            self.assertEqual(runtime.disposeCount, 1)
+            self.assertIsNone(manager._activeStartOperation)
+
+
+if __name__ == '__main__':
+    application()
+    unittest.main()

tests/test_controllers.py

@@ -29,6 +29,8 @@ from Furious.Frozenlib import AppBinarySettings, AppSettings
 from Furious.Models import CoreConfiguration, ServerProfile
 from Furious.Service.LogManager import LogManager
 
+from PySide6 import QtCore
+
 from tests.support import application, isolatedSettings, processQtEvents
 
 import unittest
@@ -87,6 +89,53 @@ class FixtureCoreManager:
             raise self.stopException
 
 
+class FixtureStartOperation(QtCore.QObject):
+    """Publish controllable asynchronous startup outcomes."""
+
+    succeeded = QtCore.Signal(object)
+    failed = QtCore.Signal(object, str, str)
+    cancelled = QtCore.Signal(object)
+
+    def succeed(self):
+        """Publish one successful manager commit."""
+        self.succeeded.emit(self)
+
+    def fail(self, message='fixture failure', details=''):
+        """Publish one failed manager transaction."""
+        self.failed.emit(self, message, details)
+
+
+class FixtureAsyncCoreManager:
+    """Record controller use of the asynchronous manager boundary."""
+
+    def __init__(self):
+        """Initialize empty operation and runtime history."""
+        self.lastStartError = ''
+        self.runtimes = []
+        self.operations = []
+        self.cancelCalls = []
+        self.stopCalls = 0
+
+    def startAsync(self, configuration, **kwargs):
+        """Return one idle operation for the controller to observe."""
+        operation = FixtureStartOperation()
+        self.operations.append((operation, configuration, kwargs))
+
+        return operation
+
+    def cancelStart(self, operation):
+        """Cancel one exact operation."""
+        self.cancelCalls.append(operation)
+        operation.cancelled.emit(operation)
+
+        return True
+
+    def stopAll(self):
+        """Record stable-runtime cleanup."""
+        self.stopCalls += 1
+        self.runtimes.clear()
+
+
 class FixtureUpdatesManager:
     """Record update hooks without contacting any update service."""
 
@@ -313,6 +362,103 @@ class ConnectionControllerTest(unittest.TestCase):
 
             controller.deleteLater()
 
+    def testAsyncManagerCommitsSystemProxyOnlyAfterSuccess(self):
+        """Remain Connecting until the manager transaction commits."""
+        with isolatedSettings():
+            core = FixtureAsyncCoreManager()
+            controller = ConnectionController(
+                coreManager=core,
+                updatesManager=FixtureUpdatesManager(),
+            )
+
+            with (
+                mock.patch(
+                    'Furious.Controllers.ConnectionController.SystemProxy.set'
+                ) as proxySet,
+                mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'),
+                mock.patch.object(controller, '_runPostConnectTasksOnce'),
+            ):
+                self.assertTrue(controller.startConnection(self.profile))
+                self.assertTrue(controller.isConnecting())
+                proxySet.assert_not_called()
+
+                operation = core.operations[0][0]
+                core.runtimes.append(object())
+                operation.succeed()
+
+                self.assertTrue(controller.isConnected())
+                proxySet.assert_called_once()
+                self.assertTrue(controller.startDisconnection())
+
+            controller.deleteLater()
+
+    def testDisconnectCancelsAnInFlightStartupGeneration(self):
+        """Cancel partial startup without ever enabling the system proxy."""
+        with isolatedSettings():
+            core = FixtureAsyncCoreManager()
+            controller = ConnectionController(
+                coreManager=core,
+                updatesManager=FixtureUpdatesManager(),
+            )
+
+            with (
+                mock.patch(
+                    'Furious.Controllers.ConnectionController.SystemProxy.set'
+                ) as proxySet,
+                mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'),
+            ):
+                self.assertTrue(controller.startConnection(self.profile))
+                operation = core.operations[0][0]
+                self.assertTrue(controller.startDisconnection())
+
+            self.assertEqual(core.cancelCalls, [operation])
+            self.assertTrue(controller.state is ConnectionState.Disconnected)
+            proxySet.assert_not_called()
+
+            controller.deleteLater()
+
+    def testReconnectCancelsConnectingGenerationBeforeReplacement(self):
+        """Replace a Connecting attempt without accepting stale completion."""
+        with isolatedSettings():
+            core = FixtureAsyncCoreManager()
+            controller = ConnectionController(
+                coreManager=core,
+                updatesManager=FixtureUpdatesManager(),
+            )
+
+            with (
+                mock.patch(
+                    'Furious.Controllers.ConnectionController.Storage.UserServers',
+                    return_value=[self.profile],
+                ),
+                mock.patch(
+                    'Furious.Controllers.ConnectionController.Storage.UserActivatedItemIndex',
+                    return_value=0,
+                ),
+                mock.patch(
+                    'Furious.Controllers.ConnectionController.SystemProxy.set'
+                ) as proxySet,
+                mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'),
+            ):
+                self.assertTrue(controller.startConnection(self.profile))
+                first = core.operations[0][0]
+                self.assertTrue(controller.startReconnection())
+                second = core.operations[1][0]
+
+                first.succeed()
+                self.assertTrue(controller.isConnecting())
+                proxySet.assert_not_called()
+
+                core.runtimes.append(object())
+                second.succeed()
+                self.assertTrue(controller.isConnected())
+                proxySet.assert_called_once()
+                controller.startDisconnection()
+
+            self.assertEqual(core.cancelCalls, [first])
+
+            controller.deleteLater()
+
     def testStartAndProxyExceptionsReturnToStableDisconnectedState(self):
         """Clean every partially acquired resource after injected failures."""
         for core, proxySideEffect in (