Support reconnect in port forwarding

Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift

@@ -10,6 +10,7 @@ public nonisolated enum PortForwardingRuntimeState: Hashable, Sendable {
     case awaitingTrust
     case awaitingAuthentication
     case connecting
+    case reconnecting
     case starting
     case running(activeConnections: Int)
     case failed(String)
@@ -33,6 +34,7 @@ public nonisolated enum PortForwardingRuntimeState: Hashable, Sendable {
         case .awaitingTrust: String(localized: "Awaiting trust")
         case .awaitingAuthentication: String(localized: "Awaiting authentication")
         case .connecting: String(localized: "Connecting")
+        case .reconnecting: String(localized: "Reconnecting")
         case .starting: String(localized: "Starting")
         case .running(let count):
             count == 0
@@ -49,7 +51,7 @@ public nonisolated enum PortForwardingRuntimeState: Hashable, Sendable {
 @Observable
 public final class PortForwardingConnection {
     public let hostID: Host.ID
-    public let connection = RemoteSessionConnectionController()
+    public let connection: RemoteSessionConnectionController
     public var journey: ConnectionJourney { connection.journey }
     public private(set) var authenticationPrompt: SSHAuthenticationPrompt?
     public fileprivate(set) var hostKeyRequestID: UUID?
@@ -59,17 +61,21 @@ public final class PortForwardingConnection {
     fileprivate var session: (any SSHForwardingSessionProtocol)?
     fileprivate var task: Task<Void, Never>?
     fileprivate var connected = false
+    fileprivate var ready = false
+    fileprivate var silentAttempt = false
+    fileprivate var configuration: SSHConnectionConfiguration?
     fileprivate var connectionCounts: [PortForwardingRule.ID: Int] = [:]
 
-    fileprivate init(host: Host, route: [Host]) {
+    fileprivate init(host: Host, route: [Host], reconnect: SessionReconnectController) {
         hostID = host.id
         self.route = route
+        connection = RemoteSessionConnectionController(reconnect: reconnect)
     }
 
     fileprivate func setPrompt(_ prompt: SSHAuthenticationPrompt?) { authenticationPrompt = prompt }
 }
 
-/// Runtime intent exists only between explicit Start and Stop/failure. No defaults or retry timers.
+/// Explicit Start owns runtime intent. Established connections recover silently until Stop or a terminal failure.
 @Observable
 public final class PortForwardingController {
     public typealias SessionFactory =
@@ -79,20 +85,26 @@ public final class PortForwardingController {
 
     private let hosts: HostLibraryModel
     private let makeSession: SessionFactory
+    private let makeReconnectController: @MainActor () -> SessionReconnectController
     public private(set) var states: [PortForwardingRule.ID: PortForwardingRuntimeState] = [:]
     private var connections: [Host.ID: PortForwardingConnection] = [:]
     private var presentations: [PortForwardingRule.ID: PortForwardingConnection] = [:]
     private var generations: [PortForwardingRule.ID: UUID] = [:]
     private var ruleTasks: [PortForwardingRule.ID: Task<Void, Never>] = [:]
+    private var cleanupTasks: [Host.ID: Task<Void, Never>] = [:]
     private var savingAuthentication: PortForwardingConnection?
 
     public init(
         hosts: HostLibraryModel,
         makeSession: @escaping SessionFactory = {
             SSHForwardingSession(connectionHandler: $0, failureHandler: $1)
+        },
+        makeReconnectController: @escaping @MainActor () -> SessionReconnectController = {
+            SessionReconnectController()
         }
     ) {
         self.hosts = hosts
+        self.makeReconnectController = makeReconnectController
         self.makeSession = makeSession
     }
 
@@ -107,7 +119,8 @@ public final class PortForwardingController {
         if let existing = connections[host.id], !activeIDs(existing).isEmpty {
             context = existing
         } else {
-            context = PortForwardingConnection(host: host, route: hosts.route(for: host))
+            context = PortForwardingConnection(
+                host: host, route: hosts.route(for: host), reconnect: makeReconnectController())
             connections[host.id] = context
         }
         generations[id] = UUID()
@@ -115,6 +128,8 @@ public final class PortForwardingController {
         states[id] = .preparing
         if context.connected {
             startRule(rule, in: context)
+        } else if context.connection.isRecovering {
+            states[id] = .reconnecting
         } else if context.task == nil && context.authenticationPrompt == nil {
             open(context)
         } else {
@@ -133,8 +148,8 @@ public final class PortForwardingController {
         let session = context.session
         let previous = ruleTasks[id]
         ruleTasks[id] = Task {
-            if let previous { await previous.value }
             if let session { await session.stop(id) }
+            if let previous { await previous.value }
         }
         if activeIDs(context).isEmpty { retire(context) }
     }
@@ -237,18 +252,24 @@ public final class PortForwardingController {
     }
 
     private func open(_ context: PortForwardingConnection) {
-        guard let host = hosts.host(context.hostID), !activeIDs(context).isEmpty else { return }
+        guard let host = hosts.host(context.hostID), connections[host.id] === context,
+            !activeIDs(context).isEmpty, context.task == nil, !context.connected
+        else { return }
+        let silently = context.connection.isRecovering
         let attempt = UUID()
         context.attemptID = attempt
-        context.journey.begin(
-            host: host, route: hosts.route(for: host), purpose: .portForwarding,
-            username: context.submissions[host.id]?.username ?? hosts.identity(for: host)?.username)
-        update(context, state: .preparing)
+        context.silentAttempt = silently
+        if !silently {
+            context.journey.begin(
+                host: host, route: hosts.route(for: host), purpose: .portForwarding,
+                username: context.submissions[host.id]?.username ?? hosts.identity(for: host)?.username)
+            update(context, state: .preparing)
+        }
         let session = makeSession(
-            { [weak self, weak context] id, count in
+            { [weak self, weak context] id, generation, count in
                 Task { @MainActor in
                     guard let self, let context, self.isCurrent(context, attempt: attempt),
-                        self.generations[id] != nil, self.presentations[id] === context
+                        self.generations[id] == generation, self.presentations[id] === context
                     else { return }
                     context.connectionCounts[id] = count
                     if case .running = self.states[id], self.states[id] != .running(activeConnections: count) {
@@ -259,21 +280,34 @@ public final class PortForwardingController {
             { [weak self, weak context] message in
                 Task { @MainActor in
                     guard let self, let context, self.isCurrent(context, attempt: attempt) else { return }
-                    self.fail(context, message: message)
+                    self.recover(context, message: message)
                 }
             })
         context.session = session
+        let cleanup = cleanupTasks[context.hostID]
         context.task = Task { [weak self, weak context] in
             guard let self, let context else { return }
             do {
-                let preparation = try hosts.prepareSSHConnection(
-                    for: host, submissions: context.submissions,
-                    temporaryHostKeys: context.connection.temporaryHostKeys)
-                let configuration = try await preparation.resolve(trustPolicy: .allowChallenge)
+                await cleanup?.value
+                try Task.checkCancellation()
+                let configuration: SSHConnectionConfiguration
+                if silently {
+                    guard let snapshot = context.configuration else { throw TermindSSHError.missingCredential }
+                    configuration = snapshot
+                } else {
+                    let preparation = try hosts.prepareSSHConnection(
+                        for: host, submissions: context.submissions,
+                        temporaryHostKeys: context.connection.temporaryHostKeys)
+                    configuration = try await preparation.resolve(trustPolicy: .allowChallenge)
+                }
                 guard isCurrent(context, attempt: attempt) else { throw CancellationError() }
-                update(context, state: .connecting)
-                try await session.connect(
-                    SSHConnectionRequest(
+                if silently {
+                    try await session.connect(
+                        SSHConnectionRequest(configuration: configuration, interactionPolicy: .forbidden)
+                            .enforcingInteractionPolicy(), progress: nil)
+                } else {
+                    update(context, state: .connecting)
+                    let request = SSHConnectionRequest(
                         configuration: configuration,
                         authorizeHostKey: { [weak self, weak context] challenge in
                             guard let self, let context else { throw CancellationError() }
@@ -282,17 +316,23 @@ public final class PortForwardingController {
                         keyboardInteractive: { [weak self, weak context] challenge in
                             guard let self, let context else { throw CancellationError() }
                             return try await self.answer(challenge, in: context, attempt: attempt)
-                        }),
-                    progress: { [weak self, weak context] event in
-                        Task { @MainActor in
-                            guard let self, let context, self.isCurrent(context, attempt: attempt) else { return }
-                            context.journey.receive(event)
-                        }
-                    })
+                        })
+                    try await session.connect(
+                        request,
+                        progress: { [weak self, weak context] event in
+                            Task { @MainActor in
+                                guard let self, let context, self.isCurrent(context, attempt: attempt) else { return }
+                                context.journey.receive(event)
+                            }
+                        })
+                }
                 try Task.checkCancellation()
                 guard isCurrent(context, attempt: attempt) else { throw CancellationError() }
                 context.task = nil
                 context.connected = true
+                if !silently {
+                    context.configuration = hosts.configurationForAdditionalConnection(from: configuration)
+                }
                 context.submissions.removeAll()
                 context.connection.temporaryHostKeys.removeAll()
                 for id in activeIDs(context) {
@@ -304,7 +344,13 @@ public final class PortForwardingController {
                 context.task = nil
                 context.attemptID = nil
                 context.session = nil
-                if let ssh = error as? TermindSSHError,
+                if silently {
+                    if Self.requiresManualRetry(error) {
+                        fail(context, message: error.localizedDescription, presentJourney: false)
+                    } else {
+                        recover(context, message: error.localizedDescription)
+                    }
+                } else if let ssh = error as? TermindSSHError,
                     ssh.isLocalAuthenticationRequirement || ssh.shouldRequestAuthentication
                 {
                     requestAuthentication(context, error: ssh)
@@ -368,51 +414,112 @@ public final class PortForwardingController {
     }
 
     private func startRule(_ rule: PortForwardingRule, in context: PortForwardingConnection) {
-        guard let token = generations[rule.id], let session = context.session else { return }
+        guard let token = generations[rule.id], let session = context.session, let attempt = context.attemptID else {
+            return
+        }
+        let silently = context.silentAttempt
         context.connectionCounts[rule.id] = 0
-        states[rule.id] = .starting
+        states[rule.id] = silently ? .reconnecting : .starting
         let previous = ruleTasks[rule.id]
         ruleTasks[rule.id] = Task { [weak self] in
             if let previous { await previous.value }
-            guard let self, generations[rule.id] == token else { return }
+            guard let self, generations[rule.id] == token, isCurrent(context, attempt: attempt) else { return }
             do {
-                try await session.start(rule)
-                guard generations[rule.id] == token else { return }
+                try await session.start(rule, generation: token)
+                guard generations[rule.id] == token, isCurrent(context, attempt: attempt) else {
+                    await session.stop(rule.id)
+                    return
+                }
                 states[rule.id] = .running(activeConnections: context.connectionCounts[rule.id] ?? 0)
-                context.journey.succeed()
+                if !context.ready {
+                    context.ready = true
+                    context.connection.markConnected()
+                }
+                if !silently { context.journey.succeed() }
             } catch {
-                guard generations[rule.id] == token else { return }
+                guard generations[rule.id] == token, isCurrent(context, attempt: attempt) else { return }
+                // Transport cleanup can finish a pending remote listen before its failure callback arrives.
+                if error is CancellationError
+                    || (error as? TermindSSHError).map({
+                        if case .disconnected = $0 { return true }
+                        return false
+                    }) == true
+                {
+                    recover(context, message: error.localizedDescription)
+                    return
+                }
                 generations[rule.id] = nil
                 states[rule.id] = .failed(error.localizedDescription)
                 if activeIDs(context).isEmpty {
                     retire(context)
-                    context.journey.fail(error.localizedDescription)
+                    if !silently { context.journey.fail(error.localizedDescription) }
                 }
             }
         }
     }
 
-    private func fail(_ context: PortForwardingConnection, message: String) {
+    private static func requiresManualRetry(_ error: any Error) -> Bool {
+        if error is SSHInteractionRequired || error is CancellationError { return true }
+        guard let ssh = error as? TermindSSHError else { return false }
+        if ssh.isLocalAuthenticationRequirement || ssh.shouldRequestAuthentication || ssh.shouldRequestHostKey {
+            return true
+        }
+        switch ssh {
+        case .untrustedHost, .mismatchedHostKey, .hostKeyChanged: return true
+        default: return false
+        }
+    }
+
+    private func recover(_ context: PortForwardingConnection, message: String) {
+        guard context.connection.hasConnectedOnce, context.configuration != nil else {
+            fail(context, message: message)
+            return
+        }
+        closeTransport(context)
+        context.journey.hide()
+        update(context, state: .reconnecting)
+        context.connection.scheduleRecovery(after: message) { [weak self, weak context] in
+            guard let self, let context, self.connections[context.hostID] === context,
+                context.connection.isRecovering
+            else { return }
+            self.open(context)
+        }
+    }
+
+    private func fail(_ context: PortForwardingConnection, message: String, presentJourney: Bool = true) {
         for id in activeIDs(context) {
             generations[id] = nil
             states[id] = .failed(message)
         }
         retire(context)
-        context.journey.fail(message)
+        if presentJourney { context.journey.fail(message) }
     }
 
-    private func retire(_ context: PortForwardingConnection) {
+    private func closeTransport(_ context: PortForwardingConnection) {
         context.attemptID = nil
+        let opening = context.task
         context.task?.cancel()
         context.task = nil
         context.connected = false
+        context.ready = false
+        context.connectionCounts.removeAll()
+        let session = context.session
+        context.session = nil
+        let previous = cleanupTasks[context.hostID]
+        cleanupTasks[context.hostID] = Task {
+            await previous?.value
+            await opening?.value
+            await session?.disconnect()
+        }
+    }
+
+    private func retire(_ context: PortForwardingConnection) {
+        closeTransport(context)
         context.setPrompt(nil)
         context.hostKeyRequestID = nil
         context.submissions.removeAll()
+        context.configuration = nil
         context.connection.disconnect()
-        let session = context.session
-        context.session = nil
-        if let session { Task { await session.disconnect() } }
         if connections[context.hostID] === context { connections[context.hostID] = nil }
     }
 }

Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingSession.swift

@@ -15,13 +15,13 @@ nonisolated struct SSHForwardingStatistics: Sendable {
 
 public protocol SSHForwardingSessionProtocol: Actor {
     func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws
-    func start(_ rule: PortForwardingRule) async throws
+    func start(_ rule: PortForwardingRule, generation: UUID) async throws
     func stop(_ id: PortForwardingRule.ID) async
     func disconnect() async
 }
 
 public actor SSHForwardingSession: SSHForwardingSessionProtocol {
-    public typealias ConnectionHandler = @Sendable (PortForwardingRule.ID, Int) -> Void
+    public typealias ConnectionHandler = @Sendable (PortForwardingRule.ID, UUID, Int) -> Void
     public typealias FailureHandler = @Sendable (String) -> Void
 
     private let worker: SSHForwardingWorker
@@ -39,8 +39,8 @@ public actor SSHForwardingSession: SSHForwardingSessionProtocol {
         try await worker.connect(request, progress: progress)
     }
 
-    public func start(_ rule: PortForwardingRule) async throws {
-        try await worker.start(rule)
+    public func start(_ rule: PortForwardingRule, generation: UUID = UUID()) async throws {
+        try await worker.start(rule, generation: generation)
     }
 
     public func stop(_ id: PortForwardingRule.ID) async {
@@ -61,6 +61,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
 
     private final class RuleContext: @unchecked Sendable {
         let rule: PortForwardingRule
+        let generation: UUID
         var socket: Int32 = -1
         var events: SSHSocketEvents?
         var remoteListener: OpaquePointer?
@@ -70,7 +71,10 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         var stop: (@Sendable () -> Void)?
         var reportedCount = -1
 
-        init(rule: PortForwardingRule) { self.rule = rule }
+        init(rule: PortForwardingRule, generation: UUID) {
+            self.rule = rule
+            self.generation = generation
+        }
     }
 
     private final class Flow: @unchecked Sendable {
@@ -103,6 +107,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
     }
 
     private let queue = DispatchQueue(label: "com.akinokaede.termind.ssh.forwarding")
+    private let closingSockets = DispatchGroup()
     private let connectionHandler: SSHForwardingSession.ConnectionHandler
     private let failureHandler: SSHForwardingSession.FailureHandler
     private var socket: Int32 = -1
@@ -140,7 +145,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
     deinit { cleanup() }
 
     func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws {
-        await queue.perform { self.cleanup() }
+        await disconnect()
         let opened = try await SSHSessionWorker.openAuthenticatedSession(
             using: request, timeout: 20, progress: progress)
         do {
@@ -156,12 +161,12 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
             }
             try Task.checkCancellation()
         } catch {
-            await queue.perform { self.cleanup() }
+            await disconnect()
             throw error
         }
     }
 
-    func start(_ rule: PortForwardingRule) async throws {
+    func start(_ rule: PortForwardingRule, generation: UUID) async throws {
         try await withCheckedThrowingContinuation { continuation in
             queue.async { [self] in
                 do {
@@ -170,7 +175,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
                         continuation.resume()
                         return
                     }
-                    let context = RuleContext(rule: rule)
+                    let context = RuleContext(rule: rule, generation: generation)
                     switch rule.kind {
                     case .local, .dynamic:
                         do { context.socket = try makeListener(host: rule.bindHost, port: rule.bindPort) } catch {
@@ -212,7 +217,14 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         }
     }
 
-    func disconnect() async { await queue.perform { self.cleanup() } }
+    func disconnect() async {
+        await withCheckedContinuation { continuation in
+            queue.async { [self] in
+                cleanup()
+                closingSockets.notify(queue: queue) { continuation.resume() }
+            }
+        }
+    }
 
     func statistics() async -> SSHForwardingStatistics {
         await withCheckedContinuation { continuation in
@@ -387,15 +399,25 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
             }
             return true
         }
-        if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN {
+        let code = libssh2_session_last_errno(session)
+        if code == LIBSSH2_ERROR_EAGAIN {
             blocked(session)
             return false
         }
         listening = nil
         if rules[context.rule.id] === context { rules[context.rule.id] = nil }
-        context.start?.resume(
-            throwing: SSHSessionWorker.sessionError(
-                session, prefix: String(localized: "Unable to start remote forwarding")))
+        let error: TermindSSHError
+        switch code {
+        case LIBSSH2_ERROR_SOCKET_NONE, LIBSSH2_ERROR_SOCKET_SEND, LIBSSH2_ERROR_SOCKET_RECV,
+            LIBSSH2_ERROR_SOCKET_DISCONNECT, LIBSSH2_ERROR_SOCKET_TIMEOUT, LIBSSH2_ERROR_BAD_SOCKET,
+            LIBSSH2_ERROR_TIMEOUT:
+            // A lost transport must not turn a pending listener into a rule-local failure.
+            error = .disconnected
+        default:
+            error = SSHSessionWorker.sessionError(
+                session, prefix: String(localized: "Unable to start remote forwarding"))
+        }
+        context.start?.resume(throwing: error)
         context.start = nil
         context.stop?()
         context.stop = nil
@@ -571,7 +593,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
     private func report(_ context: RuleContext) {
         guard !context.stopped, context.start == nil, context.reportedCount != context.flows.count else { return }
         context.reportedCount = context.flows.count
-        connectionHandler(context.rule.id, context.reportedCount)
+        connectionHandler(context.rule.id, context.generation, context.reportedCount)
     }
 
     private func close(_ flow: Flow) {
@@ -587,7 +609,11 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         let descriptor = flow.stream.socket
         flow.stream.socket = -1
         if let events = flow.events {
-            events.cancel { Darwin.close(descriptor) }
+            closingSockets.enter()
+            events.cancel { [closingSockets] in
+                Darwin.close(descriptor)
+                closingSockets.leave()
+            }
             flow.events = nil
         } else if descriptor >= 0 {
             Darwin.close(descriptor)
@@ -613,9 +639,11 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
             let descriptor = context.socket
             context.socket = -1
             context.events = nil
-            events.cancel {
+            closingSockets.enter()
+            events.cancel { [closingSockets] in
                 Darwin.close(descriptor)
                 completion()
+                closingSockets.leave()
             }
         } else {
             completion()
@@ -685,7 +713,15 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         ready = SSHReadyQueue()
         pendingOpens = SSHReadyQueue()
         if let transport = routeTransport {
-            if let events = transportEvents { events.cancel { transport.close() } } else { transport.close() }
+            if let events = transportEvents {
+                closingSockets.enter()
+                events.cancel { [closingSockets] in
+                    transport.close()
+                    closingSockets.leave()
+                }
+            } else {
+                transport.close()
+            }
         }
         transportEvents = nil
         routeTransport = nil
@@ -782,7 +818,10 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
             if candidate >= 0 {
                 var yes: Int32 = 1
                 setsockopt(candidate, SOL_SOCKET, SO_NOSIGPIPE, &yes, socklen_t(MemoryLayout<Int32>.size))
-                if Darwin.bind(candidate, info.ai_addr, info.ai_addrlen) == 0,
+                // Closed streams can remain in TIME_WAIT after the listener is released.
+                // REUSEADDR permits rebinding then without sharing an active listener.
+                if setsockopt(candidate, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout<Int32>.size)) == 0,
+                    Darwin.bind(candidate, info.ai_addr, info.ai_addrlen) == 0,
                     Darwin.listen(candidate, 128) == 0
                 {
                     makeNonblocking(candidate)

Memory/ARCHITECTURE.md

@@ -197,13 +197,23 @@ window-owned multiple/marquee selection with Stop (disabled without active rules
 Delete. Start/retry always target one rule; neither platform has bulk start or Start All/Stop All.
 Mobile pushes detail with trailing Start/Stop/Retry.
 
-Only explicit actions start/retry rules: launch, foreground, saved/synced edits and failures never do;
-legacy enabled defaults are unused. Background expiration/termination stop resources; navigation owns
-neither transport nor prompts. Shared host sessions share authentication; new connections require fresh
-challenges. Stopping one rule preserves peers; stopping the last cancels opening/prompts. Listener failure
-is rule-local; transport failure fails all without restart. Host/jump-route edits and rule replacement
-stop affected runs; authentication Continue/Save may continue its requested opening. Connection/rule
-generations reject stale results, and stop/start serializes listener cleanup.
+Only explicit actions create rule-running intent; launch, foreground and saved/synced edits never do.
+After a listener has run, transport loss preserves that intent and silently recovers all active rules on
+one shared Host connection. Retry delays are 1/2/5/10/30 seconds, capped at 30, reset after 60 stable seconds.
+Recovery reuses an in-memory authenticated configuration with durable trust refreshed after initial login;
+once-only trust and keyboard-interactive answers never authorize another connection. Like SFTP workers,
+the complete route forbids app authentication/trust forms, permits system key authorization and does not
+gate eligibility on saved credentials. Authentication/trust/interaction failures end recovery with an inline
+error; only explicit Retry opens Journey again. Recovery shows Reconnecting and Stop, never Journey or alerts.
+Initial failures still need manual Retry. Listener failure is rule-local; old TCP streams are not replayed.
+
+Legacy enabled defaults are unused. Background expiration/termination stop resources and retries; navigation
+owns neither transport nor prompts. Stopping one rule preserves peers; stopping the last cancels opening,
+prompts and retries and releases credential snapshots. Host/jump-route edits and rule replacement stop affected
+runs; authentication Continue/Save may continue its requested opening. Connection/rule generations reject stale
+results, including pending listener completions, and replacement sessions await old transport/listener cleanup.
+Disconnect returns only after socket cancellation handlers close descriptors. Local/SOCKS listeners use
+SO_REUSEADDR to permit rebinding after closed streams enter TIME_WAIT, never SO_REUSEPORT or shared live binds.
 
 Forwarding/jump bridges use DispatchSource readiness and exactly one serial queue per libssh2 session.
 Stream pumps retain write chunks across EAGAIN, use a 256 KiB per-direction quantum and 256 KiB/session-pass

TFI/App/MobileAppDependencies.swift

@@ -1,6 +1,7 @@
 import Foundation
 import SwiftData
 import TermindApplication
+import TermindRemoteInfrastructure
 
 @MainActor
 struct MobileAppDependencies {
@@ -14,6 +15,7 @@ struct MobileAppDependencies {
     var container: ModelContainer? = nil
     var workspaceStore: MobileWorkspaceStore? = nil
     var liveActivities: MobileLiveActivityController? = nil
+    var makeForwardings: (HostLibraryModel) -> PortForwardingController = { PortForwardingController(hosts: $0) }
 
     static func live() -> MobileAppDependencies {
         MobileAppDependencies(

TFI/App/MobileAppModel.swift

@@ -189,7 +189,7 @@ final class MobileAppModel {
         snippets = SnippetLibraryModel(library: library)
         knownHosts = KnownHostsLibraryModel(library: library)
         forwardingLibrary = ForwardingLibraryModel(library: library)
-        forwardings = PortForwardingController(hosts: library)
+        forwardings = dependencies.makeForwardings(library)
         transfers = SFTPTransferCoordinator(
             store: library, clientFactory: { SFTPClient() },
             globalLimit: sftpSettings.preferences.globalLimit, perHostLimit: sftpSettings.preferences.perHostLimit)

TFITests/MobileForwardingLifecycleTests.swift

@@ -1,12 +1,60 @@
 import Foundation
 import TermindApplication
 import TermindDomain
+import TermindRemoteInfrastructure
 import Testing
 
 @testable import TermindIOS
 
 @Suite @MainActor
 struct MobileForwardingLifecycleTests {
+    @Test func backgroundExpirationStopsEstablishedRecoveryAndForegroundCannotReviveIt() async throws {
+        let background = ForwardingBackgroundTask()
+        var dependencies = MobileAppDependencies.testing(background: background)
+        var sessions: [MobileRecoverySession] = []
+        dependencies.makeForwardings = { library in
+            PortForwardingController(
+                hosts: library,
+                makeSession: { _, failure in
+                    let session = MobileRecoverySession(failure: failure)
+                    sessions.append(session)
+                    return session
+                },
+                makeReconnectController: {
+                    SessionReconnectController(sleep: { _ in try await Task.sleep(for: .seconds(3600)) })
+                })
+        }
+        let app = try MobileAppModel(dependencies: dependencies)
+        let identity = Identity(name: "User", username: "user")
+        let host = TermindDomain.Host(name: "Server", host: "server.invalid", identityID: identity.id)
+        app.library.upsert(identity)
+        app.library.upsert(host)
+        let rule = PortForwardingRule(hostID: host.id)
+        try app.forwardingLibrary.save(rule)
+        let scene = UUID()
+        defer {
+            app.forwardings.stopAll()
+            app.removeScene(scene)
+            app.outbox.stop()
+        }
+        app.sceneChanged(scene, phase: .active)
+        app.forwardings.start(rule.id)
+        try await wait { app.forwardings.state(for: rule.id) == .running(activeConnections: 0) }
+        let context = try #require(app.forwardings.connection(for: rule.id))
+        await sessions[0].fail()
+        try await wait { app.forwardings.state(for: rule.id) == .reconnecting }
+        #expect(!context.journey.isPresented)
+        app.sceneChanged(scene, phase: .background)
+        let expire = try #require(background.expiration)
+        expire()
+        app.sceneChanged(scene, phase: .active)
+        context.connection.reconnect.retryNow()
+        #expect(app.forwardings.state(for: rule.id) == .stopped)
+        #expect(app.forwardings.connection(for: rule.id) == nil)
+        #expect(context.connection.reconnectState == nil)
+        #expect(sessions.count == 1)
+    }
+
     @Test func backgroundExpirationCancelsSharedAuthenticationAndForegroundNeverRestartsIt() async throws {
         let background = ForwardingBackgroundTask()
         let app = try MobileAppModel(inMemory: true, background: background)
@@ -82,6 +130,16 @@ struct MobileForwardingLifecycleTests {
     }
 }
 
+private actor MobileRecoverySession: SSHForwardingSessionProtocol {
+    let failure: SSHForwardingSession.FailureHandler
+    init(failure: @escaping SSHForwardingSession.FailureHandler) { self.failure = failure }
+    func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) {}
+    func start(_ rule: PortForwardingRule, generation: UUID) {}
+    func stop(_ id: UUID) {}
+    func disconnect() {}
+    func fail() { failure("Connection lost") }
+}
+
 @MainActor private final class ForwardingBackgroundTask: MobileBackgroundTaskManaging {
     var expiration: (@MainActor () -> Void)?
     func begin(expiration: @escaping @MainActor () -> Void) { self.expiration = expiration }

TFMTests/ArchitectureBoundaryTests.swift

@@ -9,9 +9,12 @@ struct ArchitectureBoundaryTests {
             contentsOf: sourceRoot.appending(
                 path: "../Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift"),
             encoding: .utf8)
-        for symbol in ["UserDefaults", "SettingsKey", "SessionReconnectController", "ExternalAuthentication"] {
+        for symbol in ["UserDefaults", "SettingsKey", "ExternalAuthentication"] {
             #expect(!containsToken(symbol, in: controller))
         }
+        #expect(containsToken("SessionReconnectController", in: controller))
+        #expect(controller.contains("interactionPolicy: .forbidden"))
+        #expect(containsToken("enforcingInteractionPolicy", in: controller))
         for path in ["../TFI/App/MobileAppModel.swift", "../TFI/App/MobileRootView.swift"] {
             let source = try String(contentsOf: sourceRoot.appending(path: path), encoding: .utf8)
             #expect(!containsToken("SSHKeyboardInteractiveQueue", in: source))

TFMTests/PortForwardingJourneyTests.swift

@@ -42,7 +42,7 @@ struct PortForwardingJourneyTests {
         #expect(app.forwarding.connection(for: rule.id) == nil)
     }
 
-    @Test func savedRulesNeverStartUntilRequestedAndFailureNeverRetries() async throws {
+    @Test func savedRulesNeverStartUntilRequestedAndInitialFailureNeverRetries() async throws {
         let fixture = Fixture(mode: .failure)
         let rule = fixture.rules[0]
         #expect(fixture.sessions.isEmpty)
@@ -226,7 +226,7 @@ struct PortForwardingJourneyTests {
         #expect(fixture.sessions.count == 1)
     }
 
-    @Test func listenerFailureDoesNotDisconnectAnExistingRuleAndDisconnectNeverRestarts() async throws {
+    @Test func listenerFailureDoesNotDisconnectAnExistingRuleAndOnlyItsPeerRecovers() async throws {
         let fixture = Fixture(mode: .listenerFailure)
         defer { fixture.controller.stopAll() }
         let first = fixture.rules[0]
@@ -243,12 +243,14 @@ struct PortForwardingJourneyTests {
         let session = try #require(fixture.sessions.first)
         #expect(await session.disconnects == 0)
         await session.fail()
-        try await wait {
-            if case .failed = fixture.controller.state(for: first.id) { return true }
-            return false
-        }
+        try await wait { fixture.controller.state(for: first.id) == .reconnecting }
         fixture.controller.reconcileConfiguration()
         #expect(fixture.sessions.count == 1)
+        fixture.controller.connection(for: first.id)?.connection.reconnect.retryNow()
+        try await wait { fixture.controller.state(for: first.id) == .running(activeConnections: 0) }
+        #expect(fixture.sessions.count == 2)
+        #expect(await fixture.sessions[1].starts == [first.id])
+        #expect(fixture.controller.state(for: second.id).isFailed)
     }
 
     @Test func rapidStopAndStartCannotLeaveTheNewListenerStopped() async throws {
@@ -362,10 +364,10 @@ private actor ForwardingTestSession: SSHForwardingSessionProtocol {
         try Task.checkCancellation()
         progress?(.connected(configuration.hostID))
     }
-    func start(_ rule: PortForwardingRule) async throws {
+    func start(_ rule: PortForwardingRule, generation: UUID) async throws {
         if mode == .listenerFailure && rule.bindPort == 9000 { throw TermindSSHError.route("Port in use") }
         if mode == .countDuringStart {
-            count(rule.id, 3)
+            count(rule.id, generation, 3)
             try await Task.sleep(for: .milliseconds(20))
         }
         starts.append(rule.id)

TFMTests/PortForwardingReconnectTests.swift

@@ -0,0 +1,582 @@
+import Foundation
+import TermindApplication
+import TermindDomain
+import TermindRemoteInfrastructure
+import Testing
+
+import struct TermindDomain.Host
+
+@Suite @MainActor
+struct PortForwardingReconnectTests {
+    @Test func sharedHostRestoresEveryKindAndJoinsNewRulesWithoutPresentingJourney() async throws {
+        let fixture = ReconnectFixture()
+        defer { fixture.stop() }
+        try await fixture.start([0, 1, 2])
+        let context = try #require(fixture.controller.connection(for: fixture.rules[0].id))
+        let old = fixture.sessions[0]
+        await old.fail()
+        try await fixture.waitForRetry()
+        fixture.controller.start(fixture.rules[3].id)
+        #expect(fixture.sessions.count == 1)
+        #expect(fixture.rules.allSatisfy { fixture.controller.state(for: $0.id) == .reconnecting })
+        #expect(PortForwardingRuntimeState.reconnecting.isActive)
+        #expect(!PortForwardingRuntimeState.reconnecting.isFailed)
+        #expect(PortForwardingRuntimeState.reconnecting.label == String(localized: "Reconnecting"))
+        #expect(!context.journey.isPresented)
+        await fixture.gate.release()
+        try await fixture.waitForRunning([0, 1, 2, 3])
+        let replacement = fixture.sessions[1]
+        #expect(await replacement.starts == Set(fixture.rules.map(\.id)))
+        #expect(await replacement.requests.first?.interactionPolicy == .forbidden)
+        #expect(await replacement.hasProgress == false)
+        #expect(await replacement.requests.first?.configuration.credentials.isEmpty == true)
+        #expect(context.connection.reconnectState == nil)
+        #expect(context.authenticationPrompt == nil)
+        #expect(context.connection.pendingHostKey == nil)
+        #expect(context.connection.keyboardAuthentication.pending.isEmpty)
+        #expect(!context.journey.isPresented)
+        await old.report(fixture.rules[0].id, count: 17)
+        await old.fail()
+        await fixture.settle()
+        #expect(fixture.controller.state(for: fixture.rules[0].id) == .running(activeConnections: 0))
+        #expect(fixture.sessions.count == 2)
+    }
+
+    @Test func networkFailuresUseCappedBackoffUntilTheConnectionRecovers() async throws {
+        let fixture = ReconnectFixture(modes: [.success] + Array(repeating: .offline, count: 5) + [.success])
+        defer { fixture.stop() }
+        try await fixture.start([0])
+        await fixture.sessions[0].fail()
+        for (index, delay) in [1, 2, 5, 10, 30, 30].enumerated() {
+            try await fixture.waitForRetry()
+            let state = try #require(fixture.context?.connection.reconnectState)
+            #expect(state.attempt == index + 1)
+            #expect(await fixture.gate.nextDelay == .seconds(delay))
+            #expect(!fixture.context!.journey.isPresented)
+            await fixture.gate.release()
+            try await eventually { fixture.sessions.count == index + 2 }
+        }
+        try await fixture.waitForRunning([0])
+        #expect(fixture.context?.connection.reconnectState == nil)
+    }
+
+    @Test func restartingOneRuleOnTheSameTransportRejectsItsOldCounts() async throws {
+        let fixture = ReconnectFixture()
+        defer { fixture.stop() }
+        try await fixture.start([0, 1])
+        let session = fixture.sessions[0]
+        let id = fixture.rules[0].id
+        let oldGeneration = try #require(await session.generations[id])
+        fixture.controller.stop(id)
+        fixture.controller.start(id)
+        try await fixture.waitForRunning([0, 1])
+        let newGeneration = try #require(await session.generations[id])
+        #expect(newGeneration != oldGeneration)
+        #expect(fixture.sessions.count == 1)
+        await session.report(id, generation: newGeneration, count: 2)
+        try await eventually { fixture.controller.state(for: id) == .running(activeConnections: 2) }
+        await session.report(id, generation: oldGeneration, count: 17)
+        await fixture.settle()
+        #expect(fixture.controller.state(for: id) == .running(activeConnections: 2))
+    }
+
+    @Test(arguments: [ReconnectSession.Mode.keyboard, .trust], [false, true])
+    fileprivate func interactionOnDestinationOrJumpEndsRecoveryWithoutOpeningForms(
+        _ mode: ReconnectSession.Mode, _ jump: Bool
+    ) async throws {
+        let fixture = ReconnectFixture(modes: [.success, mode, mode], usesJump: jump, trusted: mode != .trust)
+        defer { fixture.stop() }
+        try await fixture.start([0])
+        let context = try #require(fixture.context)
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        try await eventually { fixture.controller.state(for: fixture.rules[0].id).isFailed }
+        #expect(
+            fixture.controller.state(for: fixture.rules[0].id) == .failed(SSHInteractionRequired().localizedDescription)
+        )
+        #expect(context.connection.reconnectState == nil)
+        #expect(context.connection.keyboardAuthentication.pending.isEmpty)
+        #expect(context.connection.pendingHostKey == nil)
+        #expect(context.authenticationPrompt == nil)
+        #expect(!context.journey.isPresented)
+        fixture.controller.retry(fixture.rules[0].id)
+        try await eventually {
+            fixture.context?.connection.keyboardAuthentication.current != nil
+                || fixture.context?.connection.pendingHostKey != nil
+        }
+        #expect(fixture.sessions.count == 3)
+        #expect(fixture.context !== context)
+        #expect(fixture.context?.journey.isPresented == true)
+        #expect(await fixture.sessions[2].requests.first?.interactionPolicy == .allowed)
+    }
+
+    @Test(arguments: [
+        TermindSSHError.authenticationFailed("Rejected"), .missingCredential, .keyPassphraseRequired,
+        .missingUsername, .untrustedHost, .mismatchedHostKey, .hostKeyChanged(expected: ["old"], received: "new"),
+        .routeHost(hostID: UUID(), authentication: true, serverChallenge: true, hostKey: false, message: "Jump auth"),
+        .routeHost(hostID: UUID(), authentication: false, serverChallenge: false, hostKey: true, message: "Jump trust"),
+    ])
+    func authenticationAndTrustErrorsRequireManualRetry(_ error: TermindSSHError) async throws {
+        let fixture = ReconnectFixture()
+        defer { fixture.stop() }
+        try await fixture.start([0])
+        fixture.nextError = error
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        try await eventually { fixture.controller.state(for: fixture.rules[0].id).isFailed }
+        #expect(fixture.controller.state(for: fixture.rules[0].id) == .failed(error.localizedDescription))
+        #expect(fixture.context?.journey.isPresented == false)
+        #expect(fixture.context?.authenticationPrompt == nil)
+        #expect(fixture.context?.connection.reconnectState == nil)
+    }
+
+    @Test(arguments: [false, true])
+    func onlyRememberedHostTrustCanAuthorizeRecovery(_ remember: Bool) async throws {
+        let fixture = ReconnectFixture(modes: [.trust, .trust], trusted: false)
+        defer { fixture.stop() }
+        fixture.controller.start(fixture.rules[0].id)
+        try await eventually { fixture.context?.connection.pendingHostKey != nil }
+        let context = try #require(fixture.context)
+        fixture.controller.acceptHostKey(
+            for: fixture.rules[0].id, requestID: try #require(context.hostKeyRequestID),
+            observation: try #require(context.connection.pendingHostKey?.observation), remember: remember)
+        try await fixture.waitForRunning([0])
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        if remember {
+            try await fixture.waitForRunning([0])
+        } else {
+            try await eventually { fixture.controller.state(for: fixture.rules[0].id).isFailed }
+        }
+        let request = try #require(await fixture.sessions[1].requests.first)
+        #expect(request.configuration.expectedHostKeys.isEmpty == !remember)
+        #expect(fixture.library.knownHosts.isEmpty == !remember)
+        #expect(!context.journey.isPresented)
+    }
+
+    @Test func temporaryPasswordAndKeyPassphraseSurviveOnlyInTheConnectionSnapshot() async throws {
+        let fixture = ReconnectFixture()
+        defer { fixture.stop() }
+        let material = try SSHKeyMaterial.generate(.ed25519, comment: "recovery", passphrase: "temporary-passphrase")
+        let key = SSHKeyRef(
+            label: "Encrypted", algorithm: .ed25519, publicKey: material.publicKey, fingerprint: material.fingerprint,
+            credentialRef: "recovery-key", storage: .thisDevice, hasPassphrase: true)
+        try fixture.library.upsert(key, credential: .software(SSHPrivateKeySecret(privateKey: material.privateKey)))
+        var identity = fixture.identity
+        identity.keyID = key.id
+        fixture.library.upsert(identity)
+        fixture.controller.start(fixture.rules[0].id)
+        try await eventually { fixture.context?.authenticationPrompt != nil }
+        let prompt = try #require(fixture.context?.authenticationPrompt)
+        fixture.controller.submitAuthentication(
+            for: fixture.rules[0].id, promptID: prompt.id, identityID: identity.id, username: identity.username,
+            password: "temporary-password", passphrase: "temporary-passphrase", keyID: key.id, save: false)
+        try await fixture.waitForRunning([0])
+        await fixture.sessions.last!.fail()
+        try await fixture.retry()
+        try await fixture.waitForRunning([0])
+        let request = try #require(await fixture.sessions.last!.requests.first)
+        #expect(request.interactionPolicy == .forbidden)
+        #expect(request.configuration.credentials.count == 2)
+        guard case .privateKey(_, _, let passphrase) = request.configuration.credentials[0],
+            case .password(let password) = request.configuration.credentials[1]
+        else {
+            Issue.record("Expected the authenticated credential snapshot")
+            return
+        }
+        #expect(passphrase == "temporary-passphrase")
+        #expect(password == "temporary-password")
+        #expect(fixture.library.identity(identity.id)?.usesPassword == false)
+        #expect(fixture.context?.journey.plainTextLog.contains("temporary-") == false)
+    }
+
+    @Test func stoppingOneRuleDuringBackoffPreservesOnlyItsPeer() async throws {
+        let fixture = ReconnectFixture()
+        defer { fixture.stop() }
+        try await fixture.start([0, 1])
+        await fixture.sessions[0].fail()
+        try await fixture.waitForRetry()
+        fixture.controller.stop(fixture.rules[0].id)
+        await fixture.gate.release()
+        try await fixture.waitForRunning([1])
+        #expect(await fixture.sessions[1].starts == [fixture.rules[1].id])
+        #expect(fixture.controller.state(for: fixture.rules[0].id) == .stopped)
+    }
+
+    @Test(arguments: ["stop", "remove", "host", "jump", "rule", "all"])
+    func retiringIntentRejectsEvenAnUncooperativeRetryTimer(_ action: String) async throws {
+        let fixture = ReconnectFixture(usesJump: true)
+        defer { fixture.stop() }
+        try await fixture.start([0])
+        let context = try #require(fixture.context)
+        await fixture.sessions[0].fail()
+        try await fixture.waitForRetry()
+        fixture.retire(action, index: 0)
+        await fixture.gate.release()
+        await fixture.settle()
+        #expect(fixture.sessions.count == 1)
+        #expect(fixture.controller.state(for: fixture.rules[0].id) == .stopped)
+        #expect(fixture.context == nil)
+        #expect(context.connection.reconnectState == nil)
+    }
+
+    @Test(arguments: ["stop", "remove", "host", "jump", "rule", "all"], [false, true])
+    func retiringDuringRecoveryRejectsLateOpeningAndListenerSuccess(_ action: String, _ listening: Bool) async throws {
+        let fixture = ReconnectFixture(modes: [.success, listening ? .holdLateStart : .holdConnect], usesJump: true)
+        defer { fixture.stop() }
+        try await fixture.start([1])
+        let context = try #require(fixture.controller.connection(for: fixture.rules[1].id))
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        let session = fixture.sessions[1]
+        try await eventually { await (listening ? session.hasPendingStart : session.waitingToConnect) }
+        fixture.retire(action, index: 1)
+        if listening { await session.releaseStart() } else { await session.releaseConnect() }
+        try await eventually { await session.disconnects > 0 }
+        await fixture.settle()
+        #expect(fixture.controller.state(for: fixture.rules[1].id) == .stopped)
+        #expect(fixture.controller.connection(for: fixture.rules[1].id) == nil)
+        #expect(context.connection.reconnectState == nil)
+        #expect(!context.journey.isPresented)
+        #expect(await session.listening.isEmpty)
+        #expect(fixture.sessions.count == 2)
+    }
+
+    @Test(arguments: [false, true])
+    func stopAndStartOnASharedTransportRejectsLateListenerResults(_ fails: Bool) async throws {
+        let fixture = ReconnectFixture(modes: [.success, .holdLateStart])
+        defer { fixture.stop() }
+        try await fixture.start([0, 1])
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        try await fixture.waitForRunning([0])
+        let session = fixture.sessions[1]
+        try await eventually { await session.hasPendingStart }
+        let id = fixture.rules[1].id
+        let oldGeneration = try #require(await session.generations[id])
+        fixture.controller.stop(id)
+        fixture.controller.start(id)
+        await session.releaseStart(failing: fails)
+        try await eventually {
+            guard await session.generations[id] != oldGeneration else { return false }
+            return await session.hasPendingStart
+        }
+        await session.releaseStart()
+        try await fixture.waitForRunning([0, 1])
+        #expect(fixture.sessions.count == 2)
+        #expect(!fixture.context!.journey.isPresented)
+        #expect(await session.listening == [fixture.rules[0].id, id])
+    }
+
+    @Test func stopAndStartDuringOpeningRejectsItsLateSuccessAndWaitsForCleanup() async throws {
+        let fixture = ReconnectFixture(modes: [.success, .holdConnect, .success])
+        defer { fixture.stop() }
+        try await fixture.start([0])
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        try await eventually { await fixture.sessions[1].waitingToConnect }
+        let old = fixture.sessions[1]
+        fixture.controller.stop(fixture.rules[0].id)
+        fixture.controller.start(fixture.rules[0].id)
+        try await eventually { fixture.sessions.count == 3 }
+        #expect(await fixture.sessions[2].requests.isEmpty)
+        await old.releaseConnect()
+        try await fixture.waitForRunning([0])
+        #expect(await old.starts.isEmpty)
+        #expect(await old.disconnects > 0)
+        #expect(await fixture.sessions[2].requests.first?.interactionPolicy == .allowed)
+        await old.fail()
+        await fixture.settle()
+        #expect(fixture.controller.state(for: fixture.rules[0].id) == .running(activeConnections: 0))
+    }
+
+    @Test func recoveryWaitsForTheOldListenerToFinishClosingBeforeOpeningItsReplacement() async throws {
+        let fixture = ReconnectFixture(modes: [.holdDisconnect, .success])
+        defer { fixture.stop() }
+        try await fixture.start([0])
+        let old = fixture.sessions[0]
+        await old.fail()
+        try await fixture.retry()
+        try await eventually {
+            guard fixture.sessions.count == 2 else { return false }
+            return await old.waitingToDisconnect
+        }
+        #expect(await fixture.sessions[1].requests.isEmpty)
+        #expect(fixture.controller.state(for: fixture.rules[0].id) == .reconnecting)
+        await old.releaseDisconnect()
+        try await fixture.waitForRunning([0])
+        #expect(await fixture.sessions[1].starts == [fixture.rules[0].id])
+    }
+
+    @Test func pendingListenerCancellationCannotEraseRecoveryAndStopDoesNotWaitForItsStart() async throws {
+        let fixture = ReconnectFixture(modes: [.holdRemoteStart, .holdRemoteStart])
+        defer { fixture.stop() }
+        try await fixture.start([0])
+        fixture.controller.start(fixture.rules[1].id)
+        try await eventually { await fixture.sessions[0].hasPendingStart }
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        try await fixture.waitForRunning([0])
+        try await eventually { await fixture.sessions[1].hasPendingStart }
+        #expect(fixture.controller.state(for: fixture.rules[1].id) == .reconnecting)
+        fixture.controller.stop(fixture.rules[1].id)
+        try await eventually { await !fixture.sessions[1].hasPendingStart }
+        #expect(fixture.controller.state(for: fixture.rules[0].id) == .running(activeConnections: 0))
+        #expect(fixture.controller.state(for: fixture.rules[1].id) == .stopped)
+    }
+
+    @Test func listenerFailureDuringRecoveryIsLocalAndStaysOutOfJourney() async throws {
+        let fixture = ReconnectFixture(modes: [.success, .remoteListenerFailure])
+        defer { fixture.stop() }
+        try await fixture.start([0, 1])
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        try await fixture.waitForRunning([0])
+        try await eventually { fixture.controller.state(for: fixture.rules[1].id).isFailed }
+        #expect(!fixture.context!.journey.isPresented)
+        await fixture.sessions[1].fail()
+        try await fixture.retry()
+        try await fixture.waitForRunning([0])
+        #expect(await fixture.sessions[2].starts == [fixture.rules[0].id])
+    }
+
+    @Test func transportLossWhileStartingAListenerKeepsTheRuleInRecovery() async throws {
+        let fixture = ReconnectFixture(modes: [.success, .disconnectedListener, .success])
+        defer { fixture.stop() }
+        try await fixture.start([1])
+        await fixture.sessions[0].fail()
+        try await fixture.retry()
+        try await fixture.waitForRetry()
+        #expect(fixture.controller.state(for: fixture.rules[1].id) == .reconnecting)
+        #expect(await fixture.gate.nextDelay == .seconds(2))
+        await fixture.gate.release()
+        try await fixture.waitForRunning([1])
+        #expect(await fixture.sessions[2].starts == [fixture.rules[1].id])
+        #expect(fixture.controller.connection(for: fixture.rules[1].id)?.journey.isPresented == false)
+    }
+}
+
+@MainActor
+private final class ReconnectFixture {
+    let identity = Identity(name: "User", username: "user")
+    let jump: Host
+    let host: Host
+    let rules: [PortForwardingRule]
+    let library: HostLibraryModel
+    let gate = ForwardingRetryGate()
+    var sessions: [ReconnectSession] = []
+    var nextError: TermindSSHError?
+    let modes: [ReconnectSession.Mode]
+    lazy var controller = makeController()
+
+    private func makeController() -> PortForwardingController {
+        let makeSession: PortForwardingController.SessionFactory = { [unowned self] count, failure in
+            let mode = sessions.count < modes.count ? modes[sessions.count] : .success
+            let session = ReconnectSession(mode: mode, error: nextError, count: count, failure: failure)
+            nextError = nil
+            sessions.append(session)
+            return session
+        }
+        return PortForwardingController(
+            hosts: library, makeSession: makeSession,
+            makeReconnectController: { [gate] in
+                SessionReconnectController(policy: .sessionDefault, sleep: { await gate.sleep($0) }, now: Date.init)
+            })
+    }
+    var context: PortForwardingConnection? { controller.connection(for: rules[0].id) }
+
+    init(modes: [ReconnectSession.Mode] = [], usesJump: Bool = false, trusted: Bool = true) {
+        self.modes = modes
+        jump = Host(name: "Jump", host: "jump.invalid", identityID: identity.id)
+        host = Host(name: "Host", host: "host.invalid", identityID: identity.id, jumpHostID: usesJump ? jump.id : nil)
+        rules = [
+            PortForwardingRule(hostID: host.id, kind: .local, bindPort: 8000),
+            PortForwardingRule(hostID: host.id, kind: .remote, bindPort: 9000),
+            PortForwardingRule(hostID: host.id, kind: .dynamic, bindPort: 1080),
+            PortForwardingRule(hostID: host.id, kind: .local, bindPort: 8001),
+        ]
+        library = HostLibraryModel(
+            hosts: usesJump ? [jump, host] : [host], identities: [identity],
+            portForwardings: rules,
+            knownHosts: trusted
+                ? [jump, host].map {
+                    KnownHostEntry(hostname: $0.host, port: 22, algorithm: .ed25519, fingerprint: "trusted")
+                } : [])
+        library.willDeletePortForwardings = { [weak self] in self?.controller.stop($0) }
+    }
+
+    func start(_ indices: [Int]) async throws {
+        for index in indices { controller.start(rules[index].id) }
+        try await waitForRunning(indices)
+    }
+    func waitForRunning(_ indices: [Int]) async throws {
+        try await eventually {
+            indices.allSatisfy { self.controller.state(for: self.rules[$0].id) == .running(activeConnections: 0) }
+        }
+    }
+    func waitForRetry() async throws {
+        try await eventually { await self.gate.nextDelay != nil }
+    }
+    func retry() async throws {
+        try await waitForRetry()
+        let count = sessions.count
+        await gate.release()
+        try await eventually { self.sessions.count == count + 1 }
+    }
+    func settle() async { for _ in 0..<30 { await Task.yield() } }
+    func retire(_ action: String, index: Int) {
+        switch action {
+        case "stop": controller.stop(rules[index].id)
+        case "remove": controller.remove([rules[index].id])
+        case "host": controller.hostConfigurationChanged(host.id)
+        case "jump": controller.hostConfigurationChanged(jump.id)
+        case "rule":
+            var rule = rules[index]
+            rule.bindPort += 10
+            library.upsert(rule)
+        default: controller.stopAll()
+        }
+    }
+    func stop() {
+        controller.stopAll()
+        Task { await gate.finish() }
+    }
+}
+
+private actor ForwardingRetryGate {
+    private var pending: [(Duration, CheckedContinuation<Void, Never>)] = []
+    private var finished = false
+    var nextDelay: Duration? { pending.first?.0 }
+    func sleep(_ duration: Duration) async {
+        guard !finished else { return }
+        await withCheckedContinuation { pending.append((duration, $0)) }
+    }
+    func release() { if !pending.isEmpty { pending.removeFirst().1.resume() } }
+    func finish() {
+        finished = true
+        while !pending.isEmpty { release() }
+    }
+}
+
+private actor ReconnectSession: SSHForwardingSessionProtocol {
+    enum Mode: Sendable {
+        case success, offline, keyboard, trust, holdConnect, holdDisconnect, holdRemoteStart, holdLateStart
+        case remoteListenerFailure, disconnectedListener
+    }
+    let mode: Mode
+    let error: TermindSSHError?
+    let count: SSHForwardingSession.ConnectionHandler
+    let failure: SSHForwardingSession.FailureHandler
+    var requests: [SSHConnectionRequest] = []
+    var starts: Set<UUID> = []
+    var listening: Set<UUID> = []
+    var generations: [UUID: UUID] = [:]
+    var hasProgress = false
+    var disconnects = 0
+    private var connectGate: CheckedContinuation<Void, Never>?
+    private var disconnectGate: CheckedContinuation<Void, Never>?
+    private var pendingStarts: [UUID: CheckedContinuation<Void, any Error>] = [:]
+    var waitingToConnect: Bool { connectGate != nil }
+    var waitingToDisconnect: Bool { disconnectGate != nil }
+    var hasPendingStart: Bool { !pendingStarts.isEmpty }
+
+    init(
+        mode: Mode, error: TermindSSHError?, count: @escaping SSHForwardingSession.ConnectionHandler,
+        failure: @escaping SSHForwardingSession.FailureHandler
+    ) {
+        self.mode = mode
+        self.error = error
+        self.count = count
+        self.failure = failure
+    }
+    func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws {
+        requests.append(request)
+        hasProgress = progress != nil
+        if let error { throw error }
+        if mode == .offline { throw TermindSSHError.socket("Offline") }
+        if mode == .holdConnect { await withCheckedContinuation { connectGate = $0 } }
+        let configuration = request.configuration
+        let hop = configuration.jumps.first
+        if mode == .keyboard {
+            _ = try await request.keyboardInteractive?(
+                SSHKeyboardInteractiveChallenge(
+                    connectionID: UUID(), hostID: hop?.id ?? configuration.hostID,
+                    hostname: hop?.host ?? configuration.host, port: 22, username: "user", round: 1,
+                    name: "OTP", instruction: "", prompts: [.init(text: "Code", echo: false)]))
+        }
+        if mode == .trust {
+            let observation = SSHHostKeyObservation(
+                hostname: hop?.host ?? configuration.host, port: 22, algorithm: .ed25519, fingerprint: "trusted")
+            let keys = hop?.expectedHostKeys ?? configuration.expectedHostKeys
+            if !keys.contains(observation.identity) {
+                _ = try await request.authorizeHostKey?(
+                    SSHHostKeyChallenge(
+                        hostID: hop?.id ?? configuration.hostID, observation: observation, expectedHostKeys: keys))
+            }
+        }
+        progress?(.connected(configuration.hostID))
+    }
+    func start(_ rule: PortForwardingRule, generation: UUID) async throws {
+        generations[rule.id] = generation
+        if rule.kind == .remote {
+            if mode == .remoteListenerFailure { throw TermindSSHError.route("Port in use") }
+            if mode == .disconnectedListener { throw TermindSSHError.disconnected }
+            if mode == .holdRemoteStart || mode == .holdLateStart {
+                try await withCheckedThrowingContinuation { pendingStarts[rule.id] = $0 }
+            }
+        }
+        starts.insert(rule.id)
+        listening.insert(rule.id)
+    }
+    func stop(_ id: UUID) {
+        listening.remove(id)
+        if mode != .holdLateStart { pendingStarts.removeValue(forKey: id)?.resume(throwing: CancellationError()) }
+    }
+    func disconnect() async {
+        disconnects += 1
+        if mode == .holdDisconnect { await withCheckedContinuation { disconnectGate = $0 } }
+        listening.removeAll()
+        if mode != .holdLateStart { cancelStarts() }
+    }
+    func releaseDisconnect() {
+        disconnectGate?.resume()
+        disconnectGate = nil
+    }
+    func releaseConnect() {
+        connectGate?.resume()
+        connectGate = nil
+    }
+    func releaseStart(failing: Bool = false) {
+        let pending = pendingStarts.values
+        pendingStarts.removeAll()
+        for continuation in pending {
+            if failing {
+                continuation.resume(throwing: TermindSSHError.route("Late listen failure"))
+            } else {
+                continuation.resume()
+            }
+        }
+    }
+    func fail() {
+        cancelStarts()
+        failure("Connection lost")
+    }
+    func report(_ id: UUID, generation: UUID? = nil, count value: Int) {
+        guard let generation = generation ?? generations[id] else { return }
+        count(id, generation, value)
+    }
+    private func cancelStarts() {
+        let pending = pendingStarts.values
+        pendingStarts.removeAll()
+        for continuation in pending { continuation.resume(throwing: CancellationError()) }
+    }
+}
+
+@MainActor
+private func eventually(_ predicate: () async -> Bool) async throws {
+    for _ in 0..<2000 {
+        if await predicate() { return }
+        try await Task.sleep(for: .milliseconds(1))
+    }
+    throw CocoaError(.coderInvalidValue)
+}

TFMTests/SSHForwardingWireTests.swift

@@ -9,6 +9,55 @@ import Testing
 /// Uses the loopback Paramiko or forwarding-only OpenSSH fixture in Scripts/.
 @Suite(.enabled(if: ProcessInfo.processInfo.environment["TERMIND_SSH_FIXTURE"] != nil), .serialized)
 nonisolated struct SSHForwardingWireTests {
+    @Test(arguments: [PortForwardingRule.Kind.local, .dynamic])
+    func disconnectReleasesThePortBeforeReturningEvenWithEstablishedStreams(_ kind: PortForwardingRule.Kind)
+        async throws
+    {
+        let fixture = try fixture()
+        let rule = try rule(kind, fixture: fixture)
+        for _ in 0..<5 {
+            let count = ForwardingWireConnectionCount()
+            let session = SSHForwardingSession(
+                connectionHandler: { _, _, value in count.set(value) }, failureHandler: { _ in })
+            try await session.connect(request(fixture, jumps: 0))
+            do {
+                try await session.start(rule)
+                var duplicate = rule
+                duplicate.id = UUID()
+                await #expect(throws: TermindSSHError.self) { try await session.start(duplicate) }
+                let client = try SSHSocketConnector.open(host: "127.0.0.1", port: rule.bindPort, timeout: 5)
+                defer { Darwin.close(client) }
+                for _ in 0..<1000 where count.value == 0 { try await Task.sleep(for: .milliseconds(1)) }
+                try #require(count.value == 1)
+                // Close the server side first so its accepted stream can enter TIME_WAIT.
+                await session.disconnect()
+                _ = Darwin.shutdown(client, SHUT_RDWR)
+
+                let candidate = Darwin.socket(AF_INET, SOCK_STREAM, 0)
+                try #require(candidate >= 0)
+                defer { Darwin.close(candidate) }
+                var yes: Int32 = 1
+                try #require(
+                    setsockopt(candidate, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout<Int32>.size)) == 0)
+                var address = sockaddr_in()
+                address.sin_len = UInt8(MemoryLayout<sockaddr_in>.size)
+                address.sin_family = sa_family_t(AF_INET)
+                address.sin_addr.s_addr = inet_addr("127.0.0.1")
+                address.sin_port = UInt16(rule.bindPort).bigEndian
+                let bound = withUnsafePointer(to: &address) {
+                    $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {
+                        Darwin.bind(candidate, $0, socklen_t(MemoryLayout<sockaddr_in>.size))
+                    }
+                }
+                try #require(bound == 0, "The previous forwarding listener must have released its port")
+                try #require(Darwin.listen(candidate, 1) == 0)
+            } catch {
+                await session.disconnect()
+                throw error
+            }
+        }
+    }
+
     @Test(arguments: [PortForwardingRule.Kind.local, .remote, .dynamic], [1, 16, 64])
     func concurrentStreamsKeepTheirPayloadsAndHalfClose(_ kind: PortForwardingRule.Kind, _ concurrency: Int)
         async throws
@@ -24,7 +73,7 @@ nonisolated struct SSHForwardingWireTests {
     @Test func stoppingOneRulePreservesItsPeerAndReleasesItsListener() async throws {
         let fixture = try fixture()
         let session = SSHForwardingSession(
-            connectionHandler: { _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) })
+            connectionHandler: { _, _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) })
         try await session.connect(request(fixture, jumps: 0))
         do {
             let first = try rule(.local, fixture: fixture)
@@ -58,7 +107,7 @@ nonisolated struct SSHForwardingWireTests {
 
     @Test func slowConsumerPreservesPayloadWhileItsPeerProgresses() async throws {
         let fixture = try fixture()
-        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in })
         try await session.connect(request(fixture, jumps: 2))
         do {
             let rule = try rule(.local, fixture: fixture)
@@ -78,7 +127,7 @@ nonisolated struct SSHForwardingWireTests {
 
     @Test func disconnectUnderBackpressureReleasesTheListenerAcrossRepeatedStarts() async throws {
         let fixture = try fixture()
-        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in })
         let rule = try rule(.local, fixture: fixture)
         for _ in 0..<5 {
             try await session.connect(request(fixture, jumps: 1))
@@ -113,7 +162,7 @@ nonisolated struct SSHForwardingWireTests {
         let fixture = try fixture()
         // Only Paramiko provides the deliberately delayed listener reply.
         guard fixture.server != "openssh" else { return }
-        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in })
         try await session.connect(request(fixture, jumps: 0, username: "none-slow-forward"))
         do {
             let rule = try rule(.remote, fixture: fixture)
@@ -135,7 +184,7 @@ nonisolated struct SSHForwardingWireTests {
 
     @Test func multipleRemoteListenersCanStopAndRestartIndependently() async throws {
         let fixture = try fixture()
-        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in })
         try await session.connect(request(fixture, jumps: 1))
         do {
             let first = try rule(.remote, fixture: fixture)
@@ -158,7 +207,7 @@ nonisolated struct SSHForwardingWireTests {
     @Test(.enabled(if: ProcessInfo.processInfo.environment["TERMIND_FORWARDING_BENCHMARK"] == "1"))
     func idleForwardingCPU() async throws {
         let fixture = try fixture()
-        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in })
         try await session.connect(request(fixture, jumps: 2))
         let rule = try rule(.local, fixture: fixture)
         try await session.start(rule)
@@ -182,7 +231,7 @@ nonisolated struct SSHForwardingWireTests {
     ) async throws {
         let fixture = try fixture()
         let session = SSHForwardingSession(
-            connectionHandler: { _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) })
+            connectionHandler: { _, _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) })
         try await session.connect(request(fixture, jumps: jumps))
         do {
             let rule = try rule(kind, fixture: fixture)
@@ -421,3 +470,10 @@ nonisolated struct SSHForwardingWireTests {
         return data
     }
 }
+
+private nonisolated final class ForwardingWireConnectionCount: @unchecked Sendable {
+    private let lock = NSLock()
+    private var count = 0
+    var value: Int { lock.withLock { count } }
+    func set(_ value: Int) { lock.withLock { count = value } }
+}

TFMTests/SSHKeyboardInteractiveTests.swift

@@ -69,7 +69,7 @@ struct SSHKeyboardInteractiveWireTests {
     @Test func forwardingAuthenticatesAndReconnectionRequestsFreshAnswers() async throws {
         let fixture = try fixture()
         let recorder = WireChallengeRecorder()
-        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in })
         let request = request(mode: "kbd", fixture: fixture, handler: recorder.handler)
         try await session.connect(request)
         await session.disconnect()