Preserve runtime ownership through cancellation and cleanup

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

Furious/Backends/ExternalCore/Process.py

@@ -204,8 +204,6 @@ class ExternalCoreProcess(CoreRuntime):
         with self._lock:
             readers = tuple(self._readerThreads)
 
-            self._readerThreads.clear()
-
         current = threading.current_thread()
 
         for thread in readers:
@@ -228,6 +226,11 @@ class ExternalCoreProcess(CoreRuntime):
                 if thread is not current:
                     thread.join(self.ForcedShutdownTimeout)
 
+        with self._lock:
+            self._readerThreads = [
+                thread for thread in self._readerThreads if thread.is_alive()
+            ]
+
     def _watch(self, process: subprocess.Popen):
         """Reap the process and report an unexpected exit exactly once."""
         exitCode = process.wait()
@@ -509,21 +512,28 @@ class ExternalCoreProcess(CoreRuntime):
         exitCode = process.poll()
 
         if exitCode is None:
-            logger.error('external core process could not be reaped')
-        else:
-            logger.info(f'external core process stopped with code {exitCode}')
+            # Readers and the watcher still need this exact child and its pipes.
+            raise RuntimeError('External core process could not be reaped')
+
+        logger.info(f'external core process stopped with code {exitCode}')
 
         self._joinReaders(process)
 
         with self._lock:
             watcher = self._watcherThread
 
-            self._watcherThread = None
-
         if watcher is not None and watcher is not threading.current_thread():
             watcher.join(self.ForcedShutdownTimeout)
 
         with self._lock:
+            if watcher is not None and not watcher.is_alive():
+                self._watcherThread = None
+
+            if self._readerThreads or self._watcherThread is not None:
+                self.setState(RuntimeState.Stopping)
+
+                raise RuntimeError('External core reader or watcher did not stop')
+
             self._lastExitCode = exitCode
             self.setState(RuntimeState.Exited)
             self._process = None

Furious/Controllers/ConnectionController.py

@@ -107,6 +107,10 @@ class ConnectionController(QtCore.QObject):
         self._lastError = None
         self._startOperation = None
         self._pendingHttpProxy = ''
+        self._connectionGeneration = 0
+        self._startAdmissionPending = False
+        self._shuttingDown = False
+        self._runtimeCleanupFailed = False
 
         self._actionTimer = QtCore.QTimer(self)
         self._actionTimer.timeout.connect(self._callActionFromQueue)
@@ -158,9 +162,14 @@ class ConnectionController(QtCore.QObject):
 
         interactionWasEnabled = self.interactionEnabled
 
+        generation = self._connectionGeneration
+
         self._state = state
         self.stateChanged.emit(state)
 
+        if generation != self._connectionGeneration or self._state is not state:
+            return
+
         if self.interactionEnabled != interactionWasEnabled:
             self.interactionEnabledChanged.emit(self.interactionEnabled)
 
@@ -202,26 +211,48 @@ class ConnectionController(QtCore.QObject):
 
         self._setState(ConnectionState.Disconnected)
 
+    def _isCurrentConnection(self, generation):
+        """Reject continuations from an earlier, synchronously replaced lifecycle."""
+        return generation == self._connectionGeneration and not self._shuttingDown
+
     def _startConnecting(self):
         """Enter the connecting state and request progress presentation."""
+        generation = self._connectionGeneration
+
         self._setState(ConnectionState.Connecting)
-        self.progressStarted.emit()
+
+        if self._isCurrentConnection(generation) and self.isConnecting():
+            self.progressStarted.emit()
 
     def _finishConnecting(self):
-        """Enter the connected state and notify connection-aware consumers."""
+        """Publish connection completion only while this lifecycle stays current."""
+        generation = self._connectionGeneration
+
         self.progressFinished.emit(True)
 
+        if not self._isCurrentConnection(generation) or not self.isConnecting():
+            return False
+
         AppSettings.turnON_('Connect')
 
         self._setState(ConnectionState.Connected)
 
+        if not self._isCurrentConnection(generation) or not self.isConnected():
+            return False
+
         Mixins.ConnectionAware.callConnectedCallback()
 
+        return self._isCurrentConnection(generation) and self.isConnected()
+
     def startConnection(self, configuration=None) -> bool:
         """Start *configuration* or the active repository profile."""
         # QObject already exposes a legacy ``connect`` attribute in PySide.
         # Using an explicit operation name avoids shadowing Qt signal plumbing.
-        if self.state is not ConnectionState.Disconnected:
+        if (
+            self.state is not ConnectionState.Disconnected
+            or self._startAdmissionPending
+            or self._shuttingDown
+        ):
             return False
 
         if configuration is None:
@@ -275,17 +306,33 @@ class ConnectionController(QtCore.QObject):
 
             return False
 
+        self._connectionGeneration += 1
+
+        generation = self._connectionGeneration
+
+        self._startAdmissionPending = True
         self._lastError = None
-        self._setActiveProfile(configuration)
         self._pendingHttpProxy = httpProxy
+        self._setActiveProfile(configuration)
+
+        if not self._isCurrentConnection(generation):
+            return False
 
         self._startConnecting()
 
+        if not self._isCurrentConnection(generation) or not self.isConnecting():
+            return False
+
+        self._startAdmissionPending = False
+
         logManager = AppLogManager()
 
         # Retain application diagnostics while starting a fresh runtime log.
         logManager.clear(runtimeOnly=True)
 
+        if not self._isCurrentConnection(generation) or not self.isConnecting():
+            return False
+
         startAsync = getattr(self._coreManager, 'startAsync', None)
 
         if callable(startAsync):
@@ -310,6 +357,11 @@ class ConnectionController(QtCore.QObject):
                     str(ex),
                 )
 
+            if not self._isCurrentConnection(generation) or not self.isConnecting():
+                self._coreManager.cancelStart(operation)
+
+                return False
+
             self._startOperation = operation
 
             connectWeakly(
@@ -362,6 +414,9 @@ class ConnectionController(QtCore.QObject):
 
             return False
 
+        if not self._isCurrentConnection(generation) or not self.isConnecting():
+            return False
+
         if not success:
             logger.error('failed to start core manager')
 
@@ -377,6 +432,8 @@ class ConnectionController(QtCore.QObject):
 
     def _finishConnection(self, configuration, httpProxy) -> bool:
         """Commit system integration after manager runtime ownership commits."""
+        generation = self._connectionGeneration
+
         settings = AppSettings.get('CustomProxyBypass')
 
         proxyServerBypass = (
@@ -401,12 +458,19 @@ class ConnectionController(QtCore.QObject):
                 f'{configuration.coreName()}: ' + _('Unknown error'), str(ex)
             )
 
-        self._finishConnecting()
+        if not self._isCurrentConnection(generation) or not self.isConnecting():
+            return False
+
+        if not self._finishConnecting():
+            return False
 
         self.notificationRequested.emit(
             f'{configuration.coreName()}: ' + _('Connected')
         )
 
+        if not self._isCurrentConnection(generation) or not self.isConnected():
+            return False
+
         interval = CORE_CHECK_ALIVE_INTERVAL
 
         if AppSettings.isStateON_('PowerSaveMode'):
@@ -424,6 +488,8 @@ class ConnectionController(QtCore.QObject):
         if operation is not self._startOperation or not self.isConnecting():
             return
 
+        generation = self._connectionGeneration
+
         self._startOperation = None
 
         self._emitRuntimesChanged()
@@ -431,7 +497,7 @@ class ConnectionController(QtCore.QObject):
         while not self._actionQueue.empty():
             self._callActionFromQueue()
 
-        if not self.isConnecting():
+        if not self._isCurrentConnection(generation) or not self.isConnecting():
             return
 
         configuration = self.activeProfile
@@ -449,10 +515,15 @@ class ConnectionController(QtCore.QObject):
         if operation is not self._startOperation:
             return
 
+        generation = self._connectionGeneration
+
         self._startOperation = None
 
         self._emitRuntimesChanged()
 
+        if not self._isCurrentConnection(generation):
+            return
+
         configuration = self.activeProfile
         coreName = configuration.coreName() if configuration is not None else ''
         startError = message or getattr(self._coreManager, 'lastStartError', '')
@@ -468,17 +539,31 @@ class ConnectionController(QtCore.QObject):
         if operation is not self._startOperation:
             return
 
+        generation = self._connectionGeneration
+
         self._startOperation = None
 
         self._emitRuntimesChanged()
 
+        if not self._isCurrentConnection(generation):
+            return
+
         self._reset()
 
     def startDisconnection(self, notification: str = '') -> bool:
         """Stop the active runtime and optionally request a notification."""
-        if self.state is ConnectionState.Disconnected:
+        if self.isDisconnecting() or (
+            self.state is ConnectionState.Disconnected
+            and not self._startAdmissionPending
+        ):
             return False
 
+        self._connectionGeneration += 1
+
+        generation = self._connectionGeneration
+
+        self._startAdmissionPending = False
+
         operation = self._startOperation
 
         self._startOperation = None
@@ -504,6 +589,8 @@ class ConnectionController(QtCore.QObject):
 
             logger.error(f'failed to turn off system proxy: {ex}')
 
+        self._runtimeCleanupFailed = False
+
         try:
             self._coreManager.stopAll()
         except Exception as ex:
@@ -511,11 +598,16 @@ class ConnectionController(QtCore.QObject):
 
             # Always complete the state transition. A cleanup failure must not
             # strand every connection UI in the disabled Disconnecting state.
+            self._runtimeCleanupFailed = True
+
             logger.error(f'failed to stop connection runtime: {ex}')
 
         self._emitRuntimesChanged()
         self._reset()
 
+        if generation != self._connectionGeneration:
+            return True
+
         while not self._actionQueue.empty():
             try:
                 self._actionQueue.get_nowait()
@@ -552,11 +644,25 @@ class ConnectionController(QtCore.QObject):
         """Stop runtime resources without changing the next-start preference."""
         reconnectOnStartup = AppSettings.isStateON_('Connect')
 
-        if self.state is not ConnectionState.Disconnected:
-            self.startDisconnection()
+        self._shuttingDown = True
 
-        if reconnectOnStartup:
-            AppSettings.turnON_('Connect')
+        try:
+            if (
+                self.state is not ConnectionState.Disconnected
+                or self._startAdmissionPending
+            ):
+                self.startDisconnection()
+
+                if not self._runtimeCleanupFailed:
+                    return
+
+            # Disconnect keeps the UI usable on failure. Final shutdown must
+            # surface retained resources before its owner is destroyed.
+            self._coreManager.stopAll()
+            self._runtimeCleanupFailed = False
+        finally:
+            if reconnectOnStartup:
+                AppSettings.turnON_('Connect')
 
     def toggle(self) -> bool:
         """Perform the operation represented by the current stable state."""

Furious/Core/MultiprocessingRuntime.py

@@ -134,7 +134,8 @@ class MultiprocessingRuntime(CoreRuntime):
         if self.isRunning():
             raise RuntimeStartError('Runtime is already running')
 
-        self._closeProcess()
+        if not self._closeProcess():
+            raise RuntimeStartError('Previous runtime handle could not be closed')
 
         self._stopRequested = False
         self._exitPublished = False
@@ -212,22 +213,28 @@ class MultiprocessingRuntime(CoreRuntime):
         self.publishExit(event)
 
     def _closeProcess(self):
-        """Close and forget the exact inactive multiprocessing handle."""
+        """Release a handle only after close succeeds; keep failures retryable."""
         process = self._process
 
-        if process is not None:
-            try:
-                process.join(0)
-            except (AssertionError, OSError, ValueError):
-                pass
+        if process is None:
+            return True
 
-            try:
-                process.close()
-            except (OSError, ValueError):
-                pass
+        try:
+            process.join(0)
+        except (AssertionError, OSError, ValueError):
+            pass
+
+        try:
+            process.close()
+        except (OSError, ValueError) as ex:
+            logger.error(f'{self.name()} process handle could not be closed: {ex}')
+
+            return False
 
         self._process = None
 
+        return True
+
     def stop(self):
         """Idempotently terminate execution without disposing this object."""
         self._stopRequested = True
@@ -258,8 +265,14 @@ class MultiprocessingRuntime(CoreRuntime):
                 process.kill()
                 process.join(self.StopJoinTimeout)
 
+        if process.is_alive():
+            raise RuntimeError(f'{self.name()} process did not stop after escalation')
+
         event = self._consumeExit()
 
+        if not self._closeProcess():
+            raise RuntimeError(f'{self.name()} process handle could not be closed')
+
         logger.info(f'{self.name()} stopped with exitcode {event.code}')
 
     def dispose(self):

Furious/Service/AGENTS.md

@@ -37,8 +37,10 @@ for execution, and Qt for lifetime primitives. This scope owns multi-stage workf
   replacing the runtime callback. Worker-thread exits are queued to the router's Qt thread, delivered at most once, and
   suppressed after release. Execution liveness and endpoint/TUN readiness remain separate observations. A readiness
   timeout never replaces a typed exit after execution has already stopped, even when that exit is still queued.
-  Lease release currently logs stop/dispose errors and completes logical callback release; this is not evidence that
-  the underlying resource was reaped. Changes to cleanup-failure reporting must cover both runtime and lease owners.
+  Failed stop/dispose keeps the lease in Releasing with terminal delivery suppressed. The manager retains it for
+  retry and refuses new startup while release remains incomplete. A failed attempt transfers unreleased leases
+  back to that durable owner; deleting the attempt must not abandon them. Independent DNS cleanup still runs.
+  Verify runtime liveness and actual handle/thread release separately from the lease's logical state.
 - `HttpGetManager` owns reply/error/timeout cleanup. DNS recursion and external-input caches are bounded. Update,
   connectivity, endpoint, subscription, and asset requests own their exact reply and reject stale generations.
 - Subscription stages remain separate: decoders return neutral items; import constructs profiles/metadata;

Furious/Service/ConnectionManager.py

@@ -99,7 +99,8 @@ class _ConnectionStartAttempt:
             return False
 
         for lease in reversed(self.leases):
-            lease.release()
+            if not lease.release():
+                self.manager._pendingReleases.append(lease)
 
         self.leases.clear()
 
@@ -432,6 +433,9 @@ class ConnectionStartOperation(QtCore.QObject):
 
         self._setStage(ConnectionStartStage.Preparing)
 
+        if not self._isCurrent():
+            return
+
         configcopy = self.attempt.runtimeConfiguration
 
         try:
@@ -461,6 +465,9 @@ class ConnectionStartOperation(QtCore.QObject):
 
         self._setStage(ConnectionStartStage.StartingPrimary)
 
+        if not self._isCurrent():
+            return
+
         router = RuntimeEventRouter()
 
         try:
@@ -491,6 +498,11 @@ class ConnectionStartOperation(QtCore.QObject):
 
         self.attempt.ownRuntime(runtime, router)
 
+        if not self._isCurrent():
+            self.attempt.rollback()
+
+            return
+
         try:
             launch.start()
         except RuntimeStartError as ex:
@@ -520,6 +532,10 @@ class ConnectionStartOperation(QtCore.QObject):
             return
 
         self._setStage(stage)
+
+        if not self._isCurrent():
+            return
+
         self._conditionContinuation = continuation
 
         probe = _RuntimeReadinessProbe(runtime, startup, parent=self)
@@ -594,6 +610,9 @@ class ConnectionStartOperation(QtCore.QObject):
 
         self._setStage(ConnectionStartStage.PreparingTUN)
 
+        if not self._isCurrent():
+            return
+
         configcopy = self.attempt.runtimeConfiguration
 
         if PLATFORM == 'Windows':
@@ -676,6 +695,9 @@ class ConnectionStartOperation(QtCore.QObject):
         if PLATFORM != 'Linux':
             self._setStage(ConnectionStartStage.StartingTUNRuntime)
 
+            if not self._isCurrent():
+                return
+
             try:
                 self._startTUN()
             except RuntimeStartError as ex:
@@ -796,6 +818,10 @@ class ConnectionStartOperation(QtCore.QObject):
             return
 
         self._setStage(ConnectionStartStage.WaitingTUNDevice)
+
+        if not self._isCurrent():
+            return
+
         self._conditionContinuation = continuation
 
         probe = _ConditionProbe(
@@ -1000,6 +1026,9 @@ class ConnectionStartOperation(QtCore.QObject):
 
         self._setStage(ConnectionStartStage.StartingTUNRuntime)
 
+        if not self._isCurrent():
+            return
+
         try:
             self._startTUN()
         except RuntimeStartError as ex:
@@ -1021,6 +1050,9 @@ class ConnectionStartOperation(QtCore.QObject):
 
         self._setStage(ConnectionStartStage.Committing)
 
+        if not self._isCurrent():
+            return
+
         self.attempt.commit(self.exitCallback)
         self._terminal = True
 
@@ -1099,6 +1131,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
 
         self.uniqueCleanup = False
         self._leases = list()
+        self._pendingReleases = []
         self._lastStartError = ''
         self._startGeneration = 0
         self._activeStartOperation = None
@@ -1245,6 +1278,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
         third-party plugins and non-interactive callers that rely on historical
         synchronous launch semantics.
         """
+        self._retryPendingReleases()
         self._lastStartError = ''
 
         attempt = _ConnectionStartAttempt(
@@ -1285,6 +1319,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
         if self._activeStartOperation is not None:
             self._activeStartOperation.cancel()
 
+        self._retryPendingReleases()
         self._lastStartError = ''
         self._startGeneration += 1
 
@@ -1757,7 +1792,8 @@ class ConnectionManager(Mixins.CleanupOnExit):
         if lease is None:
             return
 
-        lease.release()
+        if not lease.release():
+            self._pendingReleases.append(lease)
 
         self._leases.remove(lease)
 
@@ -1765,20 +1801,29 @@ class ConnectionManager(Mixins.CleanupOnExit):
         """Stop every managed proxy-core and TUN runtime."""
         self.cancelStart()
 
-        for lease in reversed(self._leases):
-            lease.release()
-
+        self._pendingReleases.extend(reversed(self._leases))
         self._leases.clear()
+        self._retryPendingReleases()
+
+    def _retryPendingReleases(self):
+        """Keep exact failed resources and refuse replacement until release succeeds."""
+        self._pendingReleases[:] = [
+            lease for lease in self._pendingReleases if not lease.release()
+        ]
+
+        if self._pendingReleases:
+            raise RuntimeError('Connection runtime cleanup is incomplete')
 
     def cleanup(self):
         """Release resources owned by the core manager."""
-        self.cancelStart()
-        self.stopAll()
+        try:
+            self.stopAll()
+        finally:
+            # A failed runtime release must not strand independent DNS replies.
+            if self._dnsResolver is not None:
+                dispose = getattr(self._dnsResolver, 'dispose', None)
 
-        if self._dnsResolver is not None:
-            dispose = getattr(self._dnsResolver, 'dispose', None)
+                if callable(dispose):
+                    dispose()
 
-            if callable(dispose):
-                dispose()
-
-            self._dnsResolver = None
+                self._dnsResolver = None

Furious/Service/RuntimeLease.py

@@ -171,6 +171,7 @@ class RuntimeLease:
 
         self.runtime = runtime
         self.router = router
+        self._releaseInProgress = False
 
     @property
     def state(self):
@@ -182,23 +183,55 @@ class RuntimeLease:
         self.router.commit(callback)
 
     def release(self):
-        """Idempotently stop and dispose the exact owned runtime."""
-        if not self.router.beginRelease():
-            return
+        """Release execution, retaining failed cleanup for the owner to retry."""
+        if self.state is RuntimeLeaseState.Released:
+            return True
+
+        if self._releaseInProgress:
+            return False
+
+        self.router.beginRelease()
+        self._releaseInProgress = True
+
+        stopped, disposed = True, True
 
         try:
-            self.runtime.stop()
-        except Exception as ex:
-            # Any non-exit exceptions
+            try:
+                self.runtime.stop()
+            except Exception as ex:
+                # Any non-exit exceptions
+
+                stopped = False
+
+                logger.error(f'error stopping core runtime: {ex}')
 
-            logger.error(f'error stopping core runtime: {ex}')
-        finally:
             try:
                 self.runtime.dispose()
             except Exception as ex:
                 # Any non-exit exceptions
 
+                disposed = False
+
                 logger.error(f'error disposing core runtime: {ex}')
 
+            if not stopped or not disposed:
+                return False
+
+            try:
+                if self.runtime.isRunning():
+                    logger.error('core runtime remains alive after disposal')
+
+                    return False
+            except Exception as ex:
+                # Any non-exit exceptions
+
+                logger.error(f'could not verify core runtime shutdown: {ex}')
+
+                return False
+
             self.router.finishRelease()
             self.router.deleteLater()
+
+            return True
+        finally:
+            self._releaseInProgress = False

tests/README.md

@@ -89,9 +89,9 @@ worker. Choose tests by the changed contract rather than by filename alone.
 | [test_application_process.py](test_application_process.py) | Exact application-child ownership, shared crash flag, exception/signal handling, temporary crash logs, command-line dispatch. |
 | [test_architecture_refactors.py](test_architecture_refactors.py) | Startup acquisition/rollback including partial controller construction and cleanup retry, singleton election and real isolated IPC race, tray/exit policy, host integration ownership, bounded core-log transport, connection transactions, stylesheet composition. |
 | [test_connection_startup_async.py](test_connection_startup_async.py) | Real local-listener readiness, timeout/cancel/replacement, semantic exits, DNS reply lifetime, mocked platform-specific TUN sequencing. |
-| [test_controllers.py](test_controllers.py) | Connection state/error/reconnect transitions, startup restoration, shared settings, routing fallback persistence and tray/selector agreement after custom-routing disable/re-enable. |
-| [test_runtime_lifecycle.py](test_runtime_lifecycle.py) | Qt-thread exit dispatch, commit/exit races, duplicate and late exits, idempotent release, spawn failure, queue/timer disposal on preparation failure. |
-| [test_external_core.py](test_external_core.py) | Harmless real process launch/output/shutdown, partial thread-start rollback, non-finite timeout rejection, readiness/TUN metadata, Windows paths with spaces, subscription rejection of executable profiles, bounded DNS references. |
+| [test_controllers.py](test_controllers.py) | Connection state/error/reconnect transitions, reentrant cancellation/replacement before launch and during completion, startup restoration, shared settings, routing fallback persistence and tray/selector agreement after custom-routing disable/re-enable. |
+| [test_runtime_lifecycle.py](test_runtime_lifecycle.py) | Qt-thread exit dispatch, commit/exit races, duplicate and late exits, idempotent release and retained failure/retry ownership, failed reap/handle close, spawn failure, queue/timer disposal on preparation failure. |
+| [test_external_core.py](test_external_core.py) | Harmless real process launch/output/shutdown, partial thread-start rollback, non-finite timeout rejection, failed reap/thread-join retry, readiness/TUN metadata, Windows paths with spaces, subscription rejection of executable profiles, bounded DNS references. |
 | [test_frozenlib.py](test_frozenlib.py) | Nested state guards, cleanup isolation, bounded caches/throttling, dual-stack probe selection, mocked proxy/DNS/routes/startup/session boundaries and failure handling. |
 | [test_native_tun_semantics.py](test_native_tun_semantics.py) | Xray/Hysteria2 runtime-copy TUN preservation/replacement, managed-TUN failures, download-test stripping, prevention of a second tun2socks owner. |
 | [test_subscription_sync.py](test_subscription_sync.py) | Group-local preparation/commit, stable duplicate identity, atomic failure, preservation of newer local metadata, rejection of changed source state. |

tests/test_connection_startup_async.py

@@ -158,6 +158,43 @@ class _ResolverFixture(QtCore.QObject):
 class ConnectionStartupAsyncTest(TestCase):
     """Verify readiness, cancellation, rollback, and compatibility."""
 
+    def testStageListenerCancellationStopsNextAcquisitionOrCommit(self):
+        """Cancel through real Qt stage signals before the next ownership boundary."""
+        for stage in (
+            ConnectionStartStage.Preparing,
+            ConnectionStartStage.StartingPrimary,
+            ConnectionStartStage.Committing,
+        ):
+            with self.subTest(stage=stage):
+                manager = ConnectionManager()
+                self.managers.append(manager)
+                runtime = _Runtime()
+                registry = _Registry([PreparedRuntime(runtime)])
+                with mock.patch.object(
+                    manager, '_prepareTUNPolicy', return_value=(False, False)
+                ) as prepare, mock.patch(
+                    'Furious.Service.ConnectionManager.getPluginRegistry',
+                    return_value=registry,
+                ), mock.patch(
+                    'sys.excepthook'
+                ) as qtErrors:
+                    operation = manager.startAsync(_Configuration(), '', deepcopy=False)
+                    operation.stageChanged.connect(
+                        lambda current: operation.cancel() if current is stage else None
+                    )
+                    operation.start()
+                    self.assertEqual(manager.runtimes, [])
+                    self.assertIsNone(manager._activeStartOperation)
+                    if stage is ConnectionStartStage.Preparing:
+                        prepare.assert_not_called()
+                    if stage is not ConnectionStartStage.Committing:
+                        self.assertEqual(runtime.startOptions, [])
+                    else:
+                        self.assertEqual(runtime.stopCount, 1)
+                        self.assertEqual(runtime.disposeCount, 1)
+                    processQtEvents()
+                    qtErrors.assert_not_called()
+
     def setUp(self):
         """Ensure a Qt application exists for real timer/socket delivery."""
         self.app = application()

tests/test_controllers.py

@@ -177,6 +177,111 @@ class ConnectionControllerTest(unittest.TestCase):
 
         processQtEvents()
 
+    def testReentrantDisconnectPreventsStartupAndCompletion(self):
+        """Real signal callbacks cancel admission and later host/publication work."""
+        for phase in (
+            'profile',
+            'connecting',
+            'progress',
+            'runtimes',
+            'finish',
+            'connected',
+            'notification',
+        ):
+            with self.subTest(phase=phase), isolatedSettings(), mock.patch(
+                'sys.excepthook'
+            ) as qtErrors:
+                core = FixtureAsyncCoreManager()
+                controller = ConnectionController(
+                    coreManager=core, updatesManager=FixtureUpdatesManager()
+                )
+                if phase == 'profile':
+                    controller.activeProfileChanged.connect(
+                        lambda profile: (
+                            controller.startDisconnection()
+                            if profile is self.profile
+                            else None
+                        )
+                    )
+                elif phase in ('connecting', 'connected'):
+                    target = (
+                        ConnectionState.Connecting
+                        if phase == 'connecting'
+                        else ConnectionState.Connected
+                    )
+                    controller.stateChanged.connect(
+                        lambda state: (
+                            controller.startDisconnection() if state is target else None
+                        )
+                    )
+                else:
+                    signal = {
+                        'progress': controller.progressStarted,
+                        'runtimes': controller.runtimesChanged,
+                        'finish': controller.progressFinished,
+                        'notification': controller.notificationRequested,
+                    }[phase]
+                    signal.connect(lambda *_: controller.startDisconnection())
+
+                with mock.patch(
+                    'Furious.Controllers.ConnectionController.SystemProxy.set'
+                ) as proxySet, mock.patch(
+                    'Furious.Controllers.ConnectionController.SystemProxy.off'
+                ), mock.patch.object(
+                    controller, '_runPostConnectTasksOnce'
+                ) as postConnect:
+                    admitted = controller.startConnection(self.profile)
+                    if phase in ('profile', 'connecting', 'progress'):
+                        self.assertFalse(admitted)
+                        self.assertEqual(core.operations, [])
+                    else:
+                        self.assertTrue(admitted)
+                        core.operations[0][0].succeed()
+                    self.assertEqual(controller.state, ConnectionState.Disconnected)
+                    self.assertIsNone(controller.activeProfile)
+                    self.assertFalse(controller._actionTimer.isActive())
+                    self.assertEqual(AppSettings.get('Connect'), AppBinarySettings.OFF)
+                    postConnect.assert_not_called()
+                    if phase in ('profile', 'connecting', 'progress', 'runtimes'):
+                        proxySet.assert_not_called()
+                controller.deleteLater()
+                processQtEvents()
+                qtErrors.assert_not_called()
+
+    def testReentrantReplacementOfSameProfileKeepsNewStartPending(self):
+        """State/profile equality must not let an old success finish a new generation."""
+        with isolatedSettings():
+            core = FixtureAsyncCoreManager()
+            controller = ConnectionController(
+                coreManager=core, updatesManager=FixtureUpdatesManager()
+            )
+            replaced = []
+
+            def replace(*_):
+                if replaced:
+                    return
+                replaced.append(True)
+                controller.startDisconnection()
+                controller.startConnection(self.profile)
+
+            with mock.patch(
+                'Furious.Controllers.ConnectionController.SystemProxy.set'
+            ) as proxySet, mock.patch(
+                'Furious.Controllers.ConnectionController.SystemProxy.off'
+            ), mock.patch.object(
+                controller, '_runPostConnectTasksOnce'
+            ):
+                controller.startConnection(self.profile)
+                controller.runtimesChanged.connect(replace)
+                core.operations[0][0].succeed()
+                self.assertEqual(len(core.operations), 2)
+                self.assertIs(controller._startOperation, core.operations[1][0])
+                self.assertTrue(controller.isConnecting())
+                proxySet.assert_not_called()
+                controller.shutdown()
+            controller.deleteLater()
+            processQtEvents()
+
     def testSuccessfulConnectionAndDisconnectionStateMachine(self):
         """Publish stable states while using only injected runtime resources."""
         with isolatedSettings():

tests/test_external_core.py

@@ -117,6 +117,53 @@ class ExternalCoreProcessTest(unittest.TestCase):
                     spawn.assert_not_called()
                 runtime.dispose()
 
+    def testFailedReapRetainsChildUntilRetrySucceeds(self):
+        """Keep independently observed live execution after every escalation fails."""
+        runtime = ExternalCoreProcess(
+            self.configuration(['-c', 'pass'], str(Path.cwd()))
+        )
+        child = mock.Mock(pid=1234)
+        child.poll.return_value = None
+        runtime._process = child
+        runtime.setState(RuntimeState.Alive)
+        with mock.patch.object(runtime, '_requestStop'), mock.patch.object(
+            runtime, '_terminate'
+        ), mock.patch.object(runtime, '_kill'), mock.patch.object(
+            runtime, '_waitForExit', return_value=False
+        ):
+            with self.assertRaisesRegex(RuntimeError, 'could not be reaped'):
+                runtime.dispose()
+        self.assertIsNone(child.poll())
+        self.assertIs(runtime.process, child)
+        self.assertIs(runtime.state, RuntimeState.Stopping)
+        child.poll.return_value = 0
+        runtime.dispose()
+        self.assertIsNone(runtime.process)
+        self.assertIs(runtime.state, RuntimeState.Disposed)
+
+    def testFailedThreadJoinRetainsReaderAndWatcherUntilRetry(self):
+        """A reaped child does not prove its pipe readers or watcher have exited."""
+        runtime = ExternalCoreProcess(
+            self.configuration(['-c', 'pass'], str(Path.cwd()))
+        )
+        child = mock.Mock()
+        child.poll.return_value = 0
+        reader, watcher = mock.Mock(), mock.Mock()
+        reader.is_alive.return_value = watcher.is_alive.return_value = True
+        runtime._process = child
+        runtime._readerThreads = [reader]
+        runtime._watcherThread = watcher
+        with self.assertRaisesRegex(RuntimeError, 'reader or watcher'):
+            runtime.dispose()
+        self.assertIs(runtime.process, child)
+        self.assertEqual(runtime._readerThreads, [reader])
+        self.assertIs(runtime._watcherThread, watcher)
+        reader.is_alive.return_value = watcher.is_alive.return_value = False
+        runtime.dispose()
+        self.assertEqual(runtime._readerThreads, [])
+        self.assertIsNone(runtime._watcherThread)
+        self.assertIsNone(runtime.process)
+
     def testDisposedRuntimeCannotAcquireAnotherProcess(self):
         """Disposal is terminal even when the stored launch specification is valid."""
         runtime = ExternalCoreProcess(
@@ -471,6 +518,7 @@ class ExternalCoreProcessTest(unittest.TestCase):
             dnsResolver.resolve.return_value = (True, [])
 
             tunRuntime = mock.Mock(spec=CoreRuntime)
+            tunRuntime.isRunning.return_value = False
 
             manager = NoCoreRuntimeConnectionManager(dnsResolver=dnsResolver)
 

tests/test_runtime_lifecycle.py

@@ -24,7 +24,12 @@ from Furious.Interface import (
     RuntimeStartError,
     RuntimeState,
 )
-from Furious.Service.RuntimeLease import RuntimeEventRouter, RuntimeLease
+from Furious.Service.RuntimeLease import (
+    RuntimeEventRouter,
+    RuntimeLease,
+    RuntimeLeaseState,
+)
+from Furious.Service.ConnectionManager import ConnectionManager, _ConnectionStartAttempt
 
 from PySide6 import QtCore
 
@@ -99,6 +104,98 @@ class RuntimeLifecycleTest(TestCase):
     def setUpClass(cls):
         cls.app = application()
 
+    def testFailedLeaseReleaseRemainsOwnedUntilSuccessfulRetry(self):
+        """Rollback retains failed resources and blocks replacement admission."""
+        manager = ConnectionManager()
+        attempt = _ConnectionStartAttempt(manager, None)
+        owner = _AttemptOwner()
+        runtime = _Runtime()
+        router = RuntimeEventRouter()
+        router.attach(runtime, owner)
+        lease = attempt.ownRuntime(runtime, router)
+        with mock.patch.object(
+            runtime, 'dispose', side_effect=RuntimeError('still owned')
+        ):
+            attempt.rollback()
+            self.assertEqual(manager._pendingReleases, [lease])
+            self.assertIs(lease.state, RuntimeLeaseState.Releasing)
+            runtime.publishCode(0)
+            processQtEvents()
+            self.assertEqual(owner.events, [])
+            for start in (manager.start, manager.startAsync):
+                with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'):
+                    start(None, '')
+            with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'):
+                manager.stopAll()
+            self.assertEqual(manager._pendingReleases, [lease])
+            resolver = mock.Mock()
+            manager._dnsResolver = resolver
+            with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'):
+                manager.cleanup()
+            resolver.dispose.assert_called_once_with()
+            self.assertIsNone(manager._dnsResolver)
+            self.assertEqual(manager._pendingReleases, [lease])
+        manager.stopAll()
+        self.assertEqual(manager._pendingReleases, [])
+        self.assertIs(lease.state, RuntimeLeaseState.Released)
+        manager.cleanup()
+        owner.deleteLater()
+        processQtEvents()
+
+    def testFailedCommittedReleaseStillCleansIndependentRuntimes(self):
+        """One refusal must not prevent reverse release of other committed leases."""
+        manager = ConnectionManager()
+        first, second = _Runtime(), _Runtime()
+        leases = [
+            RuntimeLease(runtime, RuntimeEventRouter()) for runtime in (first, second)
+        ]
+        manager._leases.extend(leases)
+        with mock.patch.object(second, 'dispose', side_effect=RuntimeError('retry')):
+            with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'):
+                manager.stopAll()
+            self.assertIs(first.state, RuntimeState.Disposed)
+            self.assertEqual(manager._pendingReleases, [leases[1]])
+        manager.stopAll()
+        manager.cleanup()
+        processQtEvents()
+
+    def testMultiprocessingFailedEscalationRetainsLiveChild(self):
+        """Do not manufacture an exit or lose the handle of a surviving child."""
+        runtime = _ProcessRuntime()
+        child = mock.Mock()
+        child.is_alive.return_value = True
+        child.exitcode = None
+        runtime._process = child
+        runtime.setState(RuntimeState.Alive)
+        with self.assertRaisesRegex(RuntimeError, 'did not stop'):
+            runtime.dispose()
+        self.assertTrue(child.is_alive())
+        self.assertIs(runtime.process, child)
+        self.assertIsNone(runtime.lastExit)
+        child.close.assert_not_called()
+        child.is_alive.return_value = False
+        child.exitcode = 0
+        runtime.dispose()
+        self.assertIsNone(runtime.process)
+        self.assertIs(runtime.state, RuntimeState.Disposed)
+
+    def testMultiprocessingCloseFailureRetainsHandleForRetry(self):
+        """Physical exit and handle release need independent evidence."""
+        runtime = _ProcessRuntime()
+        child = mock.Mock()
+        child.is_alive.return_value = False
+        child.exitcode = 0
+        child.close.side_effect = OSError('handle still owned')
+        runtime._process = child
+        with self.assertRaisesRegex(RuntimeError, 'could not be closed'):
+            runtime.dispose()
+        self.assertIs(runtime.process, child)
+        self.assertIsNot(runtime.state, RuntimeState.Disposed)
+        child.close.side_effect = None
+        runtime.dispose()
+        self.assertIsNone(runtime.process)
+        self.assertIs(runtime.state, RuntimeState.Disposed)
+
     def testWatcherThreadExitIsConsumedOnRouterQtThread(self):
         """Queue a worker-thread publication onto the Qt owner's thread."""
         owner = _AttemptOwner()