diff --git a/FrameworkTests/ResumableTerminalExitTests.swift b/FrameworkTests/ResumableTerminalExitTests.swift new file mode 100644 index 00000000..a4706d60 --- /dev/null +++ b/FrameworkTests/ResumableTerminalExitTests.swift @@ -0,0 +1,107 @@ +import ETSession +import Foundation +import MoshCore +import TermindApplication +import TermindDomain +import Testing + +@testable import TermindRemoteInfrastructure + +@Suite +struct ResumableTerminalExitTests { + @Test func moshStoppedDrainsFinalBytesAndEndsOnlyOnce() async throws { + let events = Events() + let client = events.client(.mosh) + let tail = Data("logout\r\n".utf8) + + #expect(await !client.receiveMoshPoll(state: .stopped, operations: [.hostBytes(tail)])) + #expect(events.values == [.state(.connected), .output(tail), .state(.ended)]) + #expect(await !client.receiveMoshPoll(state: .stopped, operations: [.hostBytes(tail)])) + await client.close() + #expect(events.values == [.state(.connected), .output(tail), .state(.ended)]) + await #expect(throws: CancellationError.self) { try await client.write(Data("pwd\n".utf8)) } + await #expect(throws: CancellationError.self) { try await client.suspend() } + } + + @Test(arguments: [MoshSessionState.running, .suspended, .failed(.timeout(timeoutMs: 1000))]) + func moshSilenceSuspensionAndFailureAreNotRemoteExit(_ state: MoshSessionState) async { + let events = Events() + let client = events.client(.mosh) + _ = await client.receiveMoshPoll(state: state, operations: []) + #expect(!events.values.contains(.state(.ended))) + await client.close() + _ = await client.receiveMoshPoll(state: .stopped, operations: []) + #expect(!events.values.contains(.state(.ended))) + } + + @Test func eternalEndWaitsForTheOutputConsumerBeforeClosing() async { + let events = Events() + let client = events.client(.eternalTerminal) + let output = AsyncStream.makeStream() + let states = AsyncStream.makeStream() + await client.startEternalReaders(output: output.stream, states: states.stream) + states.continuation.yield(.connected) + #expect(await eventually { events.values.contains(.state(.connected)) }) + + states.continuation.yield(.sessionEnded) + let tail = Data("last terminal output\r\n".utf8) + output.continuation.yield(tail) + #expect(await eventually { events.values.contains(.output(tail)) }) + #expect(!events.values.contains(.state(.ended))) + output.continuation.finish() + + #expect(await eventually { events.values.contains(.state(.ended)) }) + #expect(events.values == [.state(.connected), .output(tail), .state(.ended)]) + states.continuation.yield(.connected) + states.continuation.yield(.sessionEnded) + states.continuation.finish() + await client.close() + #expect(events.values == [.state(.connected), .output(tail), .state(.ended)]) + } + + @Test func eternalDisconnectAndOutputEOFAloneDoNotCloseTheSession() async { + let events = Events() + let client = events.client(.eternalTerminal) + let output = AsyncStream.makeStream() + let states = AsyncStream.makeStream() + await client.startEternalReaders(output: output.stream, states: states.stream) + states.continuation.yield(.disconnected) + states.continuation.yield(.reconnecting) + output.continuation.finish() + #expect(await eventually { events.values.contains(.state(.recovering)) }) + states.continuation.yield(.connected) + #expect(await eventually { events.values.contains(.state(.connected)) }) + #expect(!events.values.contains(.state(.ended))) + await client.close() + states.continuation.yield(.sessionEnded) + states.continuation.finish() + #expect(!events.values.contains(.state(.ended))) + } + + private func eventually(_ condition: () -> Bool) async -> Bool { + for _ in 0..<200 { + if condition() { return true } + try? await Task.sleep(for: .milliseconds(10)) + } + return condition() + } +} + +private nonisolated final class Events: @unchecked Sendable { + enum Event: Equatable { + case output(Data) + case state(ResumableTerminalState) + } + + private let lock = NSLock() + private var recorded: [Event] = [] + var values: [Event] { lock.withLock { recorded } } + + func client(_ transport: HostTerminalProtocol) -> ResumableTerminalClient { + ResumableTerminalClient( + transport: transport, + output: { [self] data in lock.withLock { recorded.append(.output(data)) } }, + stateChanged: { [self] state in lock.withLock { recorded.append(.state(state)) } }, + shellIntegration: { _ in }) + } +} diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/ResumableTerminalClient.swift b/Frameworks/TermindRemoteInfrastructure/SSH/ResumableTerminalClient.swift index 894e050f..8a102cfb 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/ResumableTerminalClient.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/ResumableTerminalClient.swift @@ -159,7 +159,14 @@ public actor ResumableTerminalClient: ResumableTerminalSession { deliver(operations) break } - if case .failed = await mosh.state { throw TerminalTransportError.sessionFailed } + let state = await mosh.state + if case .failed = state { throw TerminalTransportError.sessionFailed } + if state == .stopped { + rollbackMoshPID = nil + _ = await receiveMoshPoll(state: state, operations: mosh.drainHostOps()) + await bootstrap.disconnect() + return + } guard ContinuousClock.now < deadline else { throw TerminalTransportError.unreachable } try await Task.sleep(for: .milliseconds(20)) } @@ -181,9 +188,14 @@ public actor ResumableTerminalClient: ResumableTerminalSession { } await bootstrap.disconnect() pendingCheckpoint = nil + guard !closed else { return } startMaintenance() startNetworkMonitoring() } catch { + if state == .ended { + await bootstrap.disconnect() + return + } if let pid = rollbackMoshPID, pid > 1 { // Only undo the server this attempt just launched; never kill a user's other sessions. await Task.detached { @@ -315,7 +327,7 @@ public actor ResumableTerminalClient: ResumableTerminalSession { bridge = nil credentials = nil pendingCheckpoint = nil - publish(.closed) + if state != .ended { publish(.closed) } } private func checkOpen() throws { @@ -333,25 +345,45 @@ public actor ResumableTerminalClient: ResumableTerminalSession { } private func publish(_ next: ResumableTerminalState) { - guard !closed || next == .closed, state != next else { return } + guard !closed || next == .closed, state != .ended, state != next else { return } state = next stateChanged(next) } + private func finishRemoteExit() async { + guard !closed, state != .ended else { return } + publish(.ended) + await close() + } + private func startMoshReader(_ session: MoshClientSession) { reader = Task { [weak self] in while !Task.isCancelled { guard let self else { return } - if case .failed = await session.state { - await self.publish(.failed) - return - } - await self.deliver(session.drainHostOps()) + // Read terminal state before draining, so the final poll cannot miss trailing output. + let state = await session.state + guard await self.receiveMoshPoll(state: state, operations: session.drainHostOps()) else { return } try? await Task.sleep(for: .milliseconds(16)) } } } + func receiveMoshPoll(state: MoshSessionState, operations: [MoshHostOp]) async -> Bool { + guard !closed else { return false } + deliver(operations) + switch state { + case .stopped: + // Remote shutdown reaches this state only with swift-mosh's shutdown-marker fix (PR #2). + await finishRemoteExit() + return false + case .failed: + publish(.failed) + return false + case .idle, .starting, .running, .suspending, .suspended, .stopping: + return true + } + } + private func deliver(_ operations: [MoshHostOp]) { guard !closed else { return } if !operations.isEmpty, !frozen { publish(.connected) } @@ -361,21 +393,25 @@ public actor ResumableTerminalClient: ResumableTerminalSession { } private func startEternalReaders(_ session: ETTerminalSession) { + startEternalReaders(output: session.output, states: session.stateChanges) + } + + func startEternalReaders(output: AsyncStream, states: AsyncStream) { reader = Task { [weak self] in - for await data in session.output { + for await data in output { guard let self, !Task.isCancelled else { return } await self.receive(data) } } observer = Task { [weak self] in - for await state in session.stateChanges { + for await state in states { guard let self, !Task.isCancelled else { return } await self.receiveEternalState(state) } } } - private func receiveEternalState(_ state: ETConnectionState) { + private func receiveEternalState(_ state: ETConnectionState) async { guard !closed else { return } switch state { case .connected: @@ -383,7 +419,10 @@ public actor ResumableTerminalClient: ResumableTerminalSession { case .reconnecting, .disconnected: if !frozen { publish(.recovering) } case .failed: publish(.failed) - case .sessionEnded: publish(.ended) + case .sessionEnded: + // The vendor finishes output before emitting this state, but our consumer may still be draining it. + await reader?.value + await finishRemoteExit() case .closed, .idle, .bootstrapping, .connecting: break } } diff --git a/TFI/Features/Workspaces/MobileWorkspace.swift b/TFI/Features/Workspaces/MobileWorkspace.swift index 64145b82..d2cd2bfa 100644 --- a/TFI/Features/Workspaces/MobileWorkspace.swift +++ b/TFI/Features/Workspaces/MobileWorkspace.swift @@ -867,6 +867,7 @@ final class MobileWorkspace: Identifiable { let reconnecting = connection.isRecovering let token = UUID() generation = token + transportState = nil commandMonitor.beginConnection() if descriptor.kind.isSSHTerminal { commandMonitor.requireScopedEvents() } activeHistoryID = nil @@ -1178,7 +1179,7 @@ final class MobileWorkspace: Identifiable { pendingInitialSFTPPath = nil journey.receive(.connected(host.id)) } - guard generation == token, !Task.isCancelled else { return } + guard generation == token, !Task.isCancelled, transportState != .ended else { return } let additionalSnapshot = hosts.additionalConnectionSnapshot(from: config) transferSnapshot = additionalSnapshot if let sftp { @@ -1543,7 +1544,8 @@ final class MobileWorkspace: Identifiable { } } - private func receiveTransportState(_ state: ResumableTerminalState) { + func receiveTransportState(_ state: ResumableTerminalState) { + guard transportState != .ended else { return } transportState = state switch state { case .connected: @@ -1554,7 +1556,17 @@ final class MobileWorkspace: Identifiable { case .recovering, .suspended: connected = false status = String(localized: AppStrings.Connections.reconnecting) - case .failed, .ended: + case .ended: + connected = false + busy = false + connectionResumeIntent = .none + recoveryLease = nil + transportSuspension?.cancel() + transportSuspension = nil + let recoveryID = recoveryID + Task { [terminalRecoveryStore] in try? await terminalRecoveryStore.discard(recoveryID) } + handleRemoteExit(.success(0)) + case .failed: connected = false busy = false connection.resetRecovery() diff --git a/TFITests/MobileTerminalRestorationTests.swift b/TFITests/MobileTerminalRestorationTests.swift index 7a8490e5..a798133e 100644 --- a/TFITests/MobileTerminalRestorationTests.swift +++ b/TFITests/MobileTerminalRestorationTests.swift @@ -8,6 +8,28 @@ import Testing @Suite @MainActor struct MobileTerminalRestorationTests { + @Test(arguments: [HostTerminalProtocol.mosh, .eternalTerminal]) + func remoteEndClosesWorkspaceOnceButRecoveryAndFailureDoNot(_ transport: HostTerminalProtocol) { + let host = Host(name: "Server", host: "example.invalid") + let workspace = MobileWorkspace( + descriptor: .init(hostID: host.id, title: host.name, kind: .terminal(transport)), + hosts: HostLibraryModel(hosts: [host])) + defer { workspace.disconnect() } + var closed: [UUID] = [] + workspace.didRequestClose = { closed.append($0) } + + workspace.receiveTransportState(.recovering) + workspace.receiveTransportState(.failed) + #expect(closed.isEmpty) + workspace.receiveTransportState(.ended) + #expect(closed == [workspace.id]) + workspace.receiveTransportState(.connected) + workspace.receiveTransportState(.ended) + #expect(closed == [workspace.id]) + #expect(!workspace.connected) + #expect(!workspace.busy) + } + @Test(arguments: [HostTerminalProtocol.mosh, .eternalTerminal]) func savedTmuxReferenceSurvivesTransportRecoveryAndMissingCheckpoint(_ transport: HostTerminalProtocol) async throws { diff --git a/TFM/Features/Terminal/SSHSessionModel.swift b/TFM/Features/Terminal/SSHSessionModel.swift index 749474fe..22045a52 100644 --- a/TFM/Features/Terminal/SSHSessionModel.swift +++ b/TFM/Features/Terminal/SSHSessionModel.swift @@ -449,6 +449,7 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable { statusStore = store let attemptID = UUID() connectionAttemptID = attemptID + transportState = nil if let ssh { Task { await ssh.disconnect() } } ssh = nil guard let host = connectionHost(hostID, using: store) else { @@ -960,6 +961,7 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable { ) try Task.checkCancellation() guard connectionAttemptID == attemptID else { throw CancellationError() } + guard transportState != .ended else { return } transferSnapshot = statusStore?.additionalConnectionSnapshot(from: configuration) ?? SSHAdditionalConnectionSnapshot(configuration: configuration) @@ -996,7 +998,8 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable { } } - private func receiveTransportState(_ state: ResumableTerminalState) { + func receiveTransportState(_ state: ResumableTerminalState) { + guard transportState != .ended else { return } transportState = state switch state { case .connected: @@ -1004,7 +1007,10 @@ final class SSHSessionModel: RemotePathDisplaying, Identifiable { connectionJourney.hide() case .recovering: status = .reconnecting - case .failed, .ended: + case .ended: + // These protocols report session end, not the remote process's numeric exit status. + handleRemoteExit(.success(0)) + case .failed: connection.resetRecovery() status = .failed(TerminalTransportError.sessionFailed.localizedDescription) case .suspended: status = .reconnecting diff --git a/TFMTests/WindowStateTests.swift b/TFMTests/WindowStateTests.swift index 87b00d38..8757c607 100644 --- a/TFMTests/WindowStateTests.swift +++ b/TFMTests/WindowStateTests.swift @@ -1520,6 +1520,31 @@ struct WindowStateTests { #expect(app.closedWorkspaceRecords.isEmpty) } + @Test + func resumableRemoteExitClosesOnlyItsSplitAndDoesNotTreatRecoveryAsExit() throws { + for transport in [HostTerminalProtocol.mosh, .eternalTerminal] { + let host = Host(name: "Server", host: "server.example") + let pane = Pane.sshSession(host: host, transport: transport) + let session = try #require(pane.sshSession) + let workspace = Workspace(pane: pane) + workspace.splitFocusedPane(axis: .horizontal, with: .sshSession(host: host)) + let neighbor = workspace.focusedPane + let window = WindowModel(workspaces: [workspace]) + + session.receiveTransportState(.recovering) + session.receiveTransportState(.failed) + #expect(workspace.panes.count == 2) + + session.receiveTransportState(.ended) + #expect(window.workspaces.count == 1) + #expect(workspace.panes.map(\.id) == [neighbor.id]) + session.receiveTransportState(.connected) + session.receiveTransportState(.ended) + #expect(session.status == .disconnected) + #expect(workspace.panes.map(\.id) == [neighbor.id]) + } + } + @Test func normalRemoteExitClosesLastPaneAndSelectsNeighboringWorkspace() { let host = Host(name: "Host", host: "host.example")