feat(terminal): add Mosh and Eternal Terminal transports

Add SSH-authenticated transports with independent settings, protocol-aware menus and snippets, and tmux session persistence. Restore eligible iOS workspaces from encrypted transport checkpoints without adding a macOS recovery entry. Include schema migrations and regression coverage.
2026-10-16 08:36:56 +08:00
1 parent 5e19dcc219
commit 9738605dd3
@@ -0,0 +1,494 @@
+import ETBootstrap
+import ETSession
+import Foundation
+import MoshBootstrap
+import MoshCore
+import Network
+import TermindApplication
+import TermindDomain
+
+public actor ResumableTerminalClient: ResumableTerminalSession {
+    public typealias RouteFactory = @Sendable () async throws -> SSHRoutedSocket
+    public typealias StateHandler = @Sendable (ResumableTerminalState) -> Void
+
+    private let transport: HostTerminalProtocol
+    private let output: SSHPTYChannel.OutputHandler
+    private let stateChanged: StateHandler
+    private let integrationChanged: SSHPTYChannel.ShellIntegrationHandler
+    private var mosh: MoshClientSession?
+    private var eternal: ETTerminalSession?
+    private var bridge: TerminalTCPBridge?
+    private var reader: Task<Void, Never>?
+    private var observer: Task<Void, Never>?
+    private var maintenance: Task<Void, Never>?
+    private var pathMonitor: NWPathMonitor?
+    private var closed = false
+    private var frozen = false
+    private var state: ResumableTerminalState = .connecting
+    private var credentials: ETCredentials?
+    private var integration: RemoteShellIntegrationBootstrap?
+    private var commandFilter: RemoteCommandEventFilter?
+    private var integratedShell: IntegratedShell?
+    private var pendingCheckpoint: TerminalTransportCheckpoint?
+    private var pendingIntegrationCommand: Data?
+    private var sendsInFlight = 0
+    private var sendWaiters: [CheckedContinuation<Void, Never>] = []
+    private var viewport: (columns: UInt16, rows: UInt16) = (80, 24)
+
+    private struct EternalPayload: Codable {
+        let clientID: String
+        let passkey: Data
+        let checkpoint: ETSessionCheckpoint
+        let integration: RemoteShellIntegrationBootstrap?
+        let commandFilter: RemoteCommandEventFilter?
+        let integratedShell: IntegratedShell?
+        let pendingIntegrationCommand: Data?
+    }
+
+    private struct IntegratedShell: Codable {
+        let shell: String
+        let systemInfo: RemoteSystemInfo?
+        let homeDirectory: String?
+    }
+
+    public init(
+        transport: HostTerminalProtocol, checkpoint: TerminalTransportCheckpoint? = nil,
+        output: @escaping SSHPTYChannel.OutputHandler,
+        stateChanged: @escaping StateHandler,
+        shellIntegration: @escaping SSHPTYChannel.ShellIntegrationHandler
+    ) {
+        self.transport = transport
+        self.pendingCheckpoint = checkpoint
+        self.output = output
+        self.stateChanged = stateChanged
+        self.integrationChanged = shellIntegration
+    }
+
+    public func connect(
+        request: SSHConnectionRequest, options: HostTerminalTransports,
+        environment: [HostEnvironmentVariable], shellIntegrationEnabled: Bool,
+        columns: UInt16, rows: UInt16, route: @escaping RouteFactory,
+        progress: SSHRouteProgressHandler?,
+        startupPreparation: SSHPTYChannel.StartupPreparation? = nil
+    ) async throws {
+        guard !closed, options.isValid, transport.isResumable else { throw CancellationError() }
+        viewport = (columns, rows)
+        guard options.available(isTailscaleManaged: request.configuration.network != nil).contains(transport) else {
+            throw TerminalTransportError.unsupportedRoute
+        }
+        guard environment.allSatisfy(\.isValid) else { throw TerminalTransportError.bootstrapFailed }
+        if let pendingCheckpoint,
+            pendingCheckpoint.version != 1 || pendingCheckpoint.transport != transport
+        {
+            throw TerminalTransportError.invalidCheckpoint
+        }
+        publish(.connecting)
+        sendsInFlight += 1
+        defer { finishSend() }
+        let bootstrap = SSHConnection()
+        var rollbackMoshPID: Int32?
+        do {
+            if pendingCheckpoint == nil {
+                try await bootstrap.connect(using: request, progress: progress, connectionFailure: nil)
+                try checkOpen()
+            }
+            let startup = try await ResumableTerminalStartup.prepare(
+                restoring: pendingCheckpoint != nil, preparation: startupPreparation,
+                execute: { command in try await TerminalTransportRoute.execute(command, on: bootstrap) })
+            try checkOpen()
+            if let checkpoint = pendingCheckpoint {
+                try await restore(checkpoint, route: route)
+            } else {
+                if transport == .mosh {
+                    let range = options.moshUDPPortRange.replacingOccurrences(of: "-", with: ":")
+                    var command = "mosh-server new -s -c 256"
+                    if !range.isEmpty { command += " -p \(range)" }
+                    command += try startup.moshArguments
+                    let variables = environment.map {
+                        "'\($0.variable)=\($0.value.replacingOccurrences(of: "'", with: "'\\''"))'"
+                    }
+                    command = (["env"] + variables + ["LANG=C.UTF-8", "LC_ALL=C.UTF-8", command])
+                        .joined(separator: " ")
+                    let result = try await bootstrap.execute(
+                        command, workingDirectory: nil, stdin: nil, timeout: 20, outputLimit: 65_536, onOutput: nil)
+                    let info: MoshServerConnectInfo
+                    do {
+                        info = try MoshServerOutputParser.parse(result.stdout + result.stderr)
+                    } catch {
+                        throw TerminalTransportError.bootstrapFailed
+                    }
+                    rollbackMoshPID = info.serverPID
+                    mosh = MoshClientSession(
+                        endpoint: .init(
+                            host: request.configuration.host, port: info.port, keyBase64_22: info.key))
+                } else {
+                    let bridge = try TerminalTCPBridge(open: route, initial: await route())
+                    self.bridge = bridge
+                    let executor = BootstrapExecutor(connection: bootstrap)
+                    do {
+                        credentials = try await ETBootstrap().run(using: executor)
+                    } catch {
+                        if let original = await executor.failure { throw original }
+                        throw TerminalTransportError.bootstrapFailed
+                    }
+                    try checkOpen()
+                    guard let credentials else { throw TerminalTransportError.bootstrapFailed }
+                    eternal = try ETTerminalSession(
+                        host: "127.0.0.1", port: bridge.port,
+                        clientID: credentials.clientID, passkey: credentials.passkey,
+                        environmentVariables: Dictionary(
+                            environment.map { ($0.variable, $0.value) },
+                            uniquingKeysWith: { _, last in last }))
+                    if shellIntegrationEnabled && startup.plan == nil {
+                        let integration = RemoteShellIntegrationBootstrap()
+                        self.integration = integration
+                        commandFilter = RemoteCommandEventFilter(token: integration.token)
+                    }
+                }
+            }
+            try checkOpen()
+            if let mosh {
+                try await mosh.start()
+                try await mosh.enqueue(.resize(cols: Int32(columns), rows: Int32(rows)))
+                // An open UDP socket is not proof that the server is reachable.
+                let deadline = ContinuousClock.now.advanced(by: .seconds(20))
+                while true {
+                    try checkOpen()
+                    let operations = await mosh.drainHostOps()
+                    if !operations.isEmpty {
+                        deliver(operations)
+                        break
+                    }
+                    if case .failed = await mosh.state { throw TerminalTransportError.sessionFailed }
+                    guard ContinuousClock.now < deadline else { throw TerminalTransportError.unreachable }
+                    try await Task.sleep(for: .milliseconds(20))
+                }
+                rollbackMoshPID = nil
+                publish(.connected)
+                startMoshReader(mosh)
+            } else if let eternal {
+                startEternalReaders(eternal)
+                try await eternal.connect(initialDimensions: .init(rows: Int(rows), columns: Int(columns)))
+                try checkOpen()
+                if let input = try startup.eternalInput {
+                    try await eternal.send(input)
+                }
+                if pendingCheckpoint == nil, let integration { try await eternal.send(integration.startCommand) }
+                if let command = pendingIntegrationCommand {
+                    pendingIntegrationCommand = nil
+                    try await eternal.send(command)
+                }
+            }
+            await bootstrap.disconnect()
+            pendingCheckpoint = nil
+            startMaintenance()
+            startNetworkMonitoring()
+        } catch {
+            if let pid = rollbackMoshPID, pid > 1 {
+                // Only undo the server this attempt just launched; never kill a user's other sessions.
+                await Task.detached {
+                    _ = try? await bootstrap.execute(
+                        "[ \"$(ps -p \(pid) -o comm=)\" = mosh-server ] && kill -TERM \(pid)",
+                        workingDirectory: nil, stdin: nil, timeout: 3, outputLimit: 1024, onOutput: nil)
+                }.value
+            }
+            await bootstrap.disconnect()
+            await close()
+            throw error
+        }
+    }
+
+    private func restore(_ checkpoint: TerminalTransportCheckpoint, route: @escaping RouteFactory) async throws {
+        if transport == .mosh {
+            let snapshot = try JSONDecoder().decode(MoshSnapshot.self, from: checkpoint.payload)
+            mosh = try await MoshClientSession.restore(from: snapshot)
+        } else {
+            let payload = try JSONDecoder().decode(EternalPayload.self, from: checkpoint.payload)
+            credentials = ETCredentials(clientID: payload.clientID, passkey: payload.passkey)
+            let bridge = try TerminalTCPBridge(open: route, initial: await route())
+            self.bridge = bridge
+            eternal = try ETTerminalSession(
+                host: "127.0.0.1", port: bridge.port, clientID: payload.clientID,
+                passkey: payload.passkey, checkpoint: payload.checkpoint)
+            integration = payload.integration
+            integration?.resetDeadlinesAfterRestore()
+            commandFilter = payload.commandFilter
+            integratedShell = payload.integratedShell
+            pendingIntegrationCommand = payload.pendingIntegrationCommand
+            if let shell = integratedShell {
+                integrationChanged(
+                    .available(shell: shell.shell, systemInfo: shell.systemInfo, homeDirectory: shell.homeDirectory))
+            }
+        }
+    }
+
+    public func write(_ data: Data) async throws {
+        try checkOpen()
+        guard !frozen else { throw TerminalTransportError.suspended }
+        integration?.noteUserInput()
+        sendsInFlight += 1
+        defer { finishSend() }
+        if let mosh { try await mosh.enqueue(.keystrokes(data)) }
+        if let eternal { try await eternal.send(data) }
+    }
+
+    public func resize(columns: UInt16, rows: UInt16) async throws {
+        try checkOpen()
+        viewport = (columns, rows)
+        guard !frozen else { return }
+        sendsInFlight += 1
+        defer { finishSend() }
+        if let mosh { try await mosh.enqueue(.resize(cols: Int32(columns), rows: Int32(rows))) }
+        if let eternal { try await eternal.resize(rows: Int(rows), cols: Int(columns)) }
+    }
+
+    public func suspend() async throws -> TerminalTransportCheckpoint {
+        try checkOpen()
+        guard state == .connected || state == .recovering else { throw TerminalTransportError.sessionFailed }
+        frozen = true
+        if sendsInFlight > 0 {
+            await withCheckedContinuation { sendWaiters.append($0) }
+        }
+        try checkOpen()
+        let payload: Data
+        if let mosh {
+            payload = try JSONEncoder().encode(await mosh.prepareForApplicationBackground())
+        } else if let eternal, let credentials {
+            let checkpoint = try await eternal.prepareForApplicationBackground()
+            payload = try JSONEncoder().encode(
+                EternalPayload(
+                    clientID: credentials.clientID, passkey: credentials.passkey,
+                    checkpoint: checkpoint, integration: integration, commandFilter: commandFilter,
+                    integratedShell: integratedShell, pendingIntegrationCommand: pendingIntegrationCommand))
+        } else {
+            throw TerminalTransportError.invalidCheckpoint
+        }
+        try checkOpen()
+        publish(.suspended)
+        return TerminalTransportCheckpoint(transport: transport, payload: payload)
+    }
+
+    public func resume() async throws {
+        try checkOpen()
+        frozen = false
+        publish(.recovering)
+        if let mosh { try await mosh.resumeFromApplicationBackground() }
+        if let eternal { await eternal.resumeFromApplicationBackground() }
+        try checkOpen()
+        integration?.resetDeadlinesAfterRestore()
+        try await resize(columns: viewport.columns, rows: viewport.rows)
+        if let command = pendingIntegrationCommand {
+            pendingIntegrationCommand = nil
+            sendsInFlight += 1
+            defer { finishSend() }
+            try await eternal?.send(command)
+        }
+    }
+
+    public func retryNow() async throws {
+        try checkOpen()
+        await bridge?.retry()
+        if frozen {
+            try await resume()
+            return
+        }
+        if let eternal { await eternal.notifyNetworkPathChanged() }
+        if let mosh {
+            _ = try await mosh.prepareForApplicationBackground()
+            try await mosh.resumeFromApplicationBackground()
+        }
+    }
+
+    public func close() async {
+        guard !closed else { return }
+        closed = true
+        reader?.cancel()
+        observer?.cancel()
+        maintenance?.cancel()
+        pathMonitor?.cancel()
+        pathMonitor = nil
+        if let mosh { await mosh.stop() }
+        if let eternal { await eternal.close() }
+        await bridge?.close()
+        mosh = nil
+        eternal = nil
+        bridge = nil
+        credentials = nil
+        pendingCheckpoint = nil
+        publish(.closed)
+    }
+
+    private func checkOpen() throws {
+        try Task.checkCancellation()
+        guard !closed else { throw CancellationError() }
+    }
+
+    private func finishSend() {
+        sendsInFlight -= 1
+        if sendsInFlight == 0 {
+            let waiters = sendWaiters
+            sendWaiters.removeAll()
+            for waiter in waiters { waiter.resume() }
+        }
+    }
+
+    private func publish(_ next: ResumableTerminalState) {
+        guard !closed || next == .closed, state != next else { return }
+        state = next
+        stateChanged(next)
+    }
+
+    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())
+                try? await Task.sleep(for: .milliseconds(16))
+            }
+        }
+    }
+
+    private func deliver(_ operations: [MoshHostOp]) {
+        guard !closed else { return }
+        if !operations.isEmpty, !frozen { publish(.connected) }
+        for operation in operations {
+            if case .hostBytes(let data) = operation { output(data) }
+        }
+    }
+
+    private func startEternalReaders(_ session: ETTerminalSession) {
+        reader = Task { [weak self] in
+            for await data in session.output {
+                guard let self, !Task.isCancelled else { return }
+                await self.receive(data)
+            }
+        }
+        observer = Task { [weak self] in
+            for await state in session.stateChanges {
+                guard let self, !Task.isCancelled else { return }
+                await self.receiveEternalState(state)
+            }
+        }
+    }
+
+    private func receiveEternalState(_ state: ETConnectionState) {
+        guard !closed else { return }
+        switch state {
+        case .connected:
+            if !frozen { publish(.connected) }
+        case .reconnecting, .disconnected:
+            if !frozen { publish(.recovering) }
+        case .failed: publish(.failed)
+        case .sessionEnded: publish(.ended)
+        case .closed, .idle, .bootstrapping, .connecting: break
+        }
+    }
+
+    private func receive(_ data: Data) async {
+        guard !closed else { return }
+        if var integration {
+            let update = integration.receive(data)
+            self.integration = update.isComplete ? nil : integration
+            await apply(update)
+        } else {
+            deliverEternalOutput(data)
+        }
+    }
+
+    private func apply(_ update: RemoteShellIntegrationBootstrap.Update) async {
+        if let command = update.command {
+            if frozen {
+                pendingIntegrationCommand = command
+            } else {
+                sendsInFlight += 1
+                do { try await eternal?.send(command) } catch {
+                    if !closed { publish(.failed) }
+                }
+                finishSend()
+            }
+        }
+        guard !closed else { return }
+        if update.integrationInstalled {
+            let shell = IntegratedShell(
+                shell: update.shell ?? "unknown", systemInfo: update.systemInfo, homeDirectory: update.homeDirectory)
+            integratedShell = shell
+            integrationChanged(
+                .available(shell: shell.shell, systemInfo: shell.systemInfo, homeDirectory: shell.homeDirectory))
+        }
+        if update.homeResolutionUnavailable { integrationChanged(.homeResolutionUnavailable) }
+        deliverEternalOutput(update.visibleOutput ?? Data())
+    }
+
+    private func deliverEternalOutput(_ data: Data) {
+        guard !closed else { return }
+        var visible = data
+        if var commandFilter {
+            let update = commandFilter.receive(visible)
+            self.commandFilter = commandFilter
+            visible = update.visibleOutput
+            for event in update.events {
+                switch event {
+                case .commandStarted(let command): integrationChanged(.commandStarted(command))
+                case .commandFinished(let code, let duration):
+                    integrationChanged(.commandFinished(code, duration: duration))
+                case .promptReady: integrationChanged(.promptReady)
+                case .draft(let draft): integrationChanged(.draft(draft))
+                }
+            }
+        }
+        if !visible.isEmpty { output(visible) }
+    }
+
+    private func startMaintenance() {
+        maintenance = Task { [weak self] in
+            while !Task.isCancelled {
+                do { try await Task.sleep(for: .milliseconds(100)) } catch { return }
+                await self?.pollIntegration()
+            }
+        }
+    }
+
+    private func pollIntegration() async {
+        guard !closed, !frozen, var integration else { return }
+        let update = integration.poll()
+        self.integration = update.isComplete ? nil : integration
+        await apply(update)
+    }
+
+    private func startNetworkMonitoring() {
+        let monitor = NWPathMonitor()
+        monitor.pathUpdateHandler = { [weak self] path in
+            let available = path.status == .satisfied
+            Task { await self?.networkPathChanged(available: available) }
+        }
+        monitor.start(queue: DispatchQueue(label: "com.akinokaede.termind.terminal-network"))
+        pathMonitor = monitor
+    }
+
+    private func networkPathChanged(available: Bool) async {
+        guard !closed, !frozen, state == .connected || state == .recovering else { return }
+        if !available { publish(.recovering) }
+        // Never bootstrap another shell: both engines retain their own protocol state.
+        if available { await eternal?.notifyNetworkPathChanged() }
+    }
+}
+
+private actor BootstrapExecutor: ETBootstrapExecutor {
+    let connection: SSHConnection
+    private(set) var failure: (any Error)?
+    init(connection: SSHConnection) { self.connection = connection }
+    func run(command: String) async throws -> String {
+        do {
+            let result = try await connection.execute(
+                command, workingDirectory: nil, stdin: nil, timeout: 20, outputLimit: 65_536, onOutput: nil)
+            return result.stdout + result.stderr
+        } catch {
+            failure = error
+            throw error
+        }
+    }
+}
@@ -0,0 +1,29 @@
+import Foundation
+import TermindApplication
+
+nonisolated struct ResumableTerminalStartup {
+    let plan: SSHPTYStartupPlan?
+
+    static func prepare(
+        restoring: Bool, preparation: SSHPTYChannel.StartupPreparation?,
+        execute: @escaping RemoteSessionCommandExecutor
+    ) async throws -> Self {
+        // A restored transport already owns its remote PTY, including any attached tmux client.
+        guard !restoring else { return Self(plan: nil) }
+        return Self(plan: try await preparation?(execute))
+    }
+
+    var moshArguments: String {
+        get throws {
+            guard let plan else { return "" }
+            return " -- /bin/sh -c \(try RemoteTmuxCommands.quote(plan.command))"
+        }
+    }
+
+    var eternalInput: Data? {
+        get throws {
+            guard let plan else { return nil }
+            return Data((" exec /bin/sh -c \(try RemoteTmuxCommands.quote(plan.command))\r").utf8)
+        }
+    }
+}
@@ -10,7 +10,7 @@ import struct TermindDomain.Host
 /// Installs a small cwd-reporting hook before remote output reaches Ghostty.
 /// The probe, its echo, and the installation command stay inside this state
 /// machine; only the login banner and the final integrated prompt are emitted.
-public nonisolated struct RemoteShellIntegrationBootstrap: Sendable {
+public nonisolated struct RemoteShellIntegrationBootstrap: Codable, Sendable {
    enum Diagnostic: Sendable, Equatable {
        enum TimeoutStage: String, Sendable {
            case lateProbe = "late_probe"
@@ -41,7 +41,7 @@ public nonisolated struct RemoteShellIntegrationBootstrap: Sendable {
        var diagnostic: Diagnostic?
    }

-    private enum Phase: Sendable {
+    private enum Phase: Codable, Sendable {
        case probing
        case scanningForLateProbe
        case suppressingLateProbe
@@ -103,6 +103,12 @@ public nonisolated struct RemoteShellIntegrationBootstrap: Sendable {
        userDidWrite = true
    }

+    /// Monotonic deadlines from another process cannot be reused after restoration.
+    mutating func resetDeadlinesAfterRestore(now: UInt64 = DispatchTime.now().uptimeNanoseconds) {
+        deadline = now + Self.responseTimeout
+        if lastActivity != nil { lastActivity = now }
+    }
+
    public mutating func receive(
        _ data: Data, now: UInt64 = DispatchTime.now().uptimeNanoseconds
    ) -> Update {
@@ -558,7 +564,7 @@ public nonisolated struct RemoteShellIntegrationBootstrap: Sendable {
 /// Removes authenticated command-report OSC sequences before output reaches
 /// Ghostty and emits their decoded payloads separately. The session token makes
 /// ordinary OSC output in a remote program ineligible to forge local history.
-public nonisolated struct RemoteCommandEventFilter: Sendable {
+public nonisolated struct RemoteCommandEventFilter: Codable, Sendable {
    public static let maximumPayloadBytes = 64 * 1024

    public nonisolated enum Event: Sendable, Equatable {
@@ -731,6 +737,15 @@ public actor SSHPTYChannel {
    private var environmentVariables: [HostEnvironmentVariable] = []
    private var environmentRejection: EnvironmentRejectionHandler = { _ in }
    private var startupPreparation: StartupPreparation?
+    private let transportOutput: OutputHandler
+    private let transportIntegration: ShellIntegrationHandler
+    private var resumable: ResumableTerminalClient?
+    private var transportOptions = HostTerminalTransports()
+    private var transportKind: HostTerminalProtocol = .ssh
+    private var transportCheckpoint: TerminalTransportCheckpoint?
+    private var transportRoute: ResumableTerminalClient.RouteFactory?
+    private var transportControl: RemoteSessionCommandExecutor?
+    private var transportState: ResumableTerminalClient.StateHandler = { _ in }

    public init(
        output: @escaping OutputHandler, exit: @escaping ExitHandler,
@@ -740,6 +755,8 @@ public actor SSHPTYChannel {
        logger: Logging.Logger = ApplicationLogger.disabled("termind.ssh-pty-channel"),
        maxSessionsPolicy: SSHMaxSessionsPolicy = SSHMaxSessionsPolicy()
    ) {
+        transportOutput = output
+        transportIntegration = shellIntegration
        self.keyboardInteractive = keyboardInteractive
        self.agentForwardingFactory = agentForwardingFactory
        worker = SSHPTYChannelWorker(
@@ -756,6 +773,21 @@ public actor SSHPTYChannel {
    ) async throws {
        defer { request.configuration.clearSecrets() }
        var request = request.withKeyboardInteractiveFallback(keyboardInteractive)
+        if transportKind.isResumable {
+            request.agentForwarding = nil
+            request.permitsAgentForwarding = false
+            guard let transportRoute else { throw TerminalTransportError.unsupportedRoute }
+            let client = ResumableTerminalClient(
+                transport: transportKind, checkpoint: transportCheckpoint, output: transportOutput,
+                stateChanged: transportState, shellIntegration: transportIntegration)
+            resumable = client
+            try await client.connect(
+                request: request, options: transportOptions, environment: environmentVariables,
+                shellIntegrationEnabled: shellIntegrationEnabled, columns: columns, rows: rows,
+                route: transportRoute, progress: progress, startupPreparation: startupPreparation)
+            transportCheckpoint = nil
+            return
+        }
        request.agentForwarding = try await agentForwardingFactory?(request)
        try await worker.connect(
            using: request,
@@ -779,8 +811,42 @@ public actor SSHPTYChannel {
        startupPreparation = preparation
    }

+    public func configureTransport(
+        _ kind: HostTerminalProtocol, options: HostTerminalTransports,
+        checkpoint: TerminalTransportCheckpoint? = nil,
+        route: @escaping ResumableTerminalClient.RouteFactory,
+        control: RemoteSessionCommandExecutor? = nil,
+        stateChanged: @escaping ResumableTerminalClient.StateHandler
+    ) {
+        transportKind = kind
+        transportOptions = options
+        transportCheckpoint = checkpoint
+        transportRoute = route
+        transportControl = control
+        transportState = stateChanged
+    }
+
+    public func suspendTransport() async throws -> TerminalTransportCheckpoint {
+        guard let resumable else { throw TerminalTransportError.invalidCheckpoint }
+        return try await resumable.suspend()
+    }
+
+    public func resumeTransport() async throws {
+        guard let resumable else { throw TerminalTransportError.invalidCheckpoint }
+        try await resumable.resume()
+    }
+
+    public func retryTransport() async throws {
+        guard let resumable else { throw TerminalTransportError.invalidCheckpoint }
+        try await resumable.retryNow()
+    }
+
    public func sessionCommandExecutor() async throws -> RemoteSessionCommandExecutor {
-        try await worker.sessionCommandExecutor()
+        if transportKind.isResumable {
+            guard let transportControl else { throw TermindSSHError.disconnected }
+            return transportControl
+        }
+        return try await worker.sessionCommandExecutor()
    }

    public func runtimeCommandExecutor() async throws -> RemoteSessionCommandExecutor {
@@ -796,10 +862,25 @@ public actor SSHPTYChannel {
    }

    public func releaseTerminalForControl() async throws {
+        if let resumable {
+            await resumable.close()
+            return
+        }
        try await worker.releaseTerminalForControl()
    }

-    public func write(_ data: Data) {
+    public func write(_ data: Data) async {
+        if let resumable {
+            do {
+                try await resumable.write(data)
+            } catch TerminalTransportError.suspended {
+                // Background checkpoints must remain frozen until eligibility is invalidated.
+            } catch is CancellationError {
+            } catch {
+                transportState(.failed)
+            }
+            return
+        }
        worker.write(data)
    }

@@ -823,14 +904,25 @@ public actor SSHPTYChannel {
        try await worker.historySource()
    }

-    public func resize(columns: UInt16, rows: UInt16, widthPixels: UInt32, heightPixels: UInt32) {
+    public func resize(columns: UInt16, rows: UInt16, widthPixels: UInt32, heightPixels: UInt32) async {
+        if let resumable {
+            do { try await resumable.resize(columns: columns, rows: rows) } catch { transportState(.failed) }
+            return
+        }
        worker.resize(
            columns: columns, rows: rows,
            widthPixels: widthPixels, heightPixels: heightPixels
        )
    }

-    public func disconnect() async { await worker.disconnect() }
+    public func disconnect() async {
+        await resumable?.close()
+        resumable = nil
+        transportCheckpoint = nil
+        transportRoute = nil
+        transportControl = nil
+        await worker.disconnect()
+    }
 }

 nonisolated final class SSHPTYChannelWorker: @unchecked Sendable {
@@ -0,0 +1,191 @@
+import Darwin
+import Foundation
+import TermindApplication
+
+/// Adapts the ET library's TCP-only endpoint API to our routed socket owner.
+/// Socket ownership and bounded queues stay on one serial queue, including cancellation.
+nonisolated final class TerminalTCPBridge: @unchecked Sendable {
+    typealias Open = @Sendable () async throws -> SSHRoutedSocket
+    let port: UInt16
+    private let queue = DispatchQueue(label: "com.akinokaede.termind.terminal-tcp-bridge")
+    private let open: Open
+    private var listener: Int32
+    private var local: Int32 = -1
+    private var remote: SSHRoutedSocket?
+    private var timer: (any DispatchSourceTimer)?
+    private var opening: Task<Void, Never>?
+    private var generation = UUID()
+    private var toLocal = Data()
+    private var toRemote = Data()
+    private var localEOF = false
+    private var remoteEOF = false
+    private var closed = false
+    private var openingBlocked = false
+
+    init(open: @escaping Open, initial: SSHRoutedSocket? = nil) throws {
+        self.open = open
+        self.remote = initial
+        if let initial { Self.configure(initial.socket) }
+        listener = Darwin.socket(AF_INET, SOCK_STREAM, 0)
+        guard listener >= 0 else { throw POSIXError(.ENOBUFS) }
+        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")
+        let fd = listener
+        let bound = withUnsafePointer(to: &address) {
+            $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {
+                Darwin.bind(fd, $0, socklen_t(MemoryLayout<sockaddr_in>.size))
+            }
+        }
+        var length = socklen_t(MemoryLayout<sockaddr_in>.size)
+        let named = withUnsafeMutablePointer(to: &address) {
+            $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { getsockname(fd, $0, &length) }
+        }
+        guard bound == 0, named == 0, listen(fd, 1) == 0 else {
+            Darwin.close(fd)
+            throw POSIXError(.EADDRNOTAVAIL)
+        }
+        port = UInt16(bigEndian: address.sin_port)
+        Self.configure(fd)
+        let timer = DispatchSource.makeTimerSource(queue: queue)
+        self.timer = timer
+        timer.schedule(deadline: .now(), repeating: .milliseconds(20))
+        timer.setEventHandler { [weak self] in self?.pump() }
+        timer.resume()
+    }
+
+    deinit {
+        timer?.cancel()
+        opening?.cancel()
+        if listener >= 0 { Darwin.close(listener) }
+        if local >= 0 { Darwin.close(local) }
+        remote?.close()
+    }
+
+    func close() async {
+        await withCheckedContinuation { continuation in
+            queue.async { [self] in
+                closed = true
+                timer?.cancel()
+                timer = nil
+                retire()
+                if listener >= 0 {
+                    Darwin.close(listener)
+                    listener = -1
+                }
+                continuation.resume()
+            }
+        }
+    }
+
+    func retry() async {
+        await withCheckedContinuation { continuation in
+            queue.async {
+                self.openingBlocked = false
+                continuation.resume()
+            }
+        }
+    }
+
+    private static func configure(_ fd: Int32) {
+        _ = fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK)
+        var one: Int32 = 1
+        _ = setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, socklen_t(MemoryLayout<Int32>.size))
+    }
+
+    private func retire() {
+        generation = UUID()
+        opening?.cancel()
+        opening = nil
+        if local >= 0 {
+            Darwin.close(local)
+            local = -1
+        }
+        remote?.close()
+        remote = nil
+        toLocal.removeAll()
+        toRemote.removeAll()
+        localEOF = false
+        remoteEOF = false
+    }
+
+    private func pump() {
+        guard !closed else { return }
+        let accepted = Darwin.accept(listener, nil, nil)
+        if accepted >= 0 {
+            if openingBlocked {
+                Darwin.close(accepted)
+                return
+            }
+            if local >= 0, remote != nil { retire() }
+            if local >= 0 { Darwin.close(local) }
+            local = accepted
+            Self.configure(accepted)
+            toLocal.removeAll()
+            toRemote.removeAll()
+            if remote == nil, opening == nil { startOpening() }
+        }
+        guard local >= 0, let remote else { return }
+        guard Self.move(from: local, to: remote.socket, buffer: &toRemote, eof: &localEOF),
+            Self.move(from: remote.socket, to: local, buffer: &toLocal, eof: &remoteEOF)
+        else {
+            retire()
+            return
+        }
+        if localEOF && remoteEOF { retire() }
+    }
+
+    private func startOpening() {
+        let token = generation
+        opening = Task { [weak self, open] in
+            do {
+                let socket = try await open()
+                guard let self else {
+                    socket.close()
+                    return
+                }
+                queue.async {
+                    guard !self.closed, self.generation == token else {
+                        socket.close()
+                        return
+                    }
+                    Self.configure(socket.socket)
+                    self.remote = socket
+                    self.opening = nil
+                }
+            } catch {
+                guard let self else { return }
+                let sshError = error as? TermindSSHError
+                let blocksRetry =
+                    error is CancellationError || error is SSHInteractionRequired
+                    || sshError?.isLocalAuthenticationRequirement == true
+                    || sshError?.shouldRequestAuthentication == true
+                    || sshError?.shouldRequestHostKey == true || sshError?.isHostKeyChanged == true
+                queue.async {
+                    guard self.generation == token else { return }
+                    self.openingBlocked = blocksRetry
+                    self.retire()
+                }
+            }
+        }
+    }
+
+    private static func move(from input: Int32, to output: Int32, buffer: inout Data, eof: inout Bool) -> Bool {
+        guard !eof else { return true }
+        if buffer.isEmpty {
+            var bytes = [UInt8](repeating: 0, count: 32_768)
+            let count = Darwin.recv(input, &bytes, bytes.count, 0)
+            if count == 0 {
+                eof = true
+                _ = Darwin.shutdown(output, SHUT_WR)
+                return true
+            }
+            if count < 0 { return errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR }
+            buffer.append(contentsOf: bytes.prefix(count))
+        }
+        let sent = buffer.withUnsafeBytes { Darwin.send(output, $0.baseAddress, $0.count, 0) }
+        if sent > 0 { buffer.removeFirst(sent) }
+        return sent >= 0 || errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR
+    }
+}
@@ -0,0 +1,39 @@
+import Foundation
+import TermindApplication
+import TermindDomain
+
+public nonisolated enum TerminalTransportRoute {
+    public static func execute(_ command: String, using request: SSHConnectionRequest) async throws
+        -> RemoteSessionCommandResult
+    {
+        defer { request.configuration.clearSecrets() }
+        let connection = SSHConnection()
+        do {
+            try await connection.connect(using: request)
+            let result = try await execute(command, on: connection)
+            await connection.disconnect()
+            return result
+        } catch {
+            await connection.disconnect()
+            throw error
+        }
+    }
+
+    static func execute(_ command: String, on connection: SSHConnection) async throws -> RemoteSessionCommandResult {
+        let result = try await connection.execute(
+            command, workingDirectory: nil, stdin: nil, timeout: 20, outputLimit: 262_144, onOutput: nil)
+        return RemoteSessionCommandResult(
+            stdout: result.stdout, stderr: result.stderr, exitCode: result.exitCode,
+            truncated: result.stdoutTruncated || result.stderrTruncated)
+    }
+
+    public static func open(using request: SSHConnectionRequest, port: Int) async throws -> SSHRoutedSocket {
+        defer { request.configuration.clearSecrets() }
+        let config = request.configuration
+        return try await SSHRoutedSocket.open(
+            host: config.host, port: port, proxy: config.proxy, jumps: config.jumps, timeout: 20,
+            authorizeHostKey: request.authorizeHostKey, network: config.network,
+            keyboardInteractive: request.keyboardInteractive, securityKeySigning: request.securityKeySigning,
+            interactionPolicy: request.interactionPolicy, hostKeyConstraints: request.hostKeyConstraints)
+    }
+}
@@ -0,0 +1,141 @@
+import CryptoKit
+import Foundation
+import Security
+import TermindApplication
+import TermindDomain
+
+import struct TermindDomain.Host
+
+/// One device-local store, independent of SwiftData/Vault/CloudKit. Leases fence saves racing an explicit close.
+public actor TerminalRecoveryStore {
+    public static let shared = TerminalRecoveryStore()
+    private let directory: URL
+    private let testingKey: Data?
+    private var owners: [UUID: UUID] = [:]
+
+    private struct Envelope: Codable {
+        let metadata: Data
+        let ciphertext: Data
+    }
+
+    public init(directory: URL? = nil, testingKey: Data? = nil) {
+        self.directory =
+            directory
+            ?? URL.applicationSupportDirectory
+            .appendingPathComponent("Termind/TerminalRecovery", isDirectory: true)
+        self.testingKey = testingKey
+    }
+
+    public func register(_ id: UUID) -> UUID {
+        let lease = UUID()
+        owners[id] = lease
+        return lease
+    }
+
+    public func record(_ id: UUID) throws -> TerminalRecoveryRecord {
+        let envelope = try JSONDecoder().decode(Envelope.self, from: Data(contentsOf: location(id)))
+        _ = try AES.GCM.open(
+            AES.GCM.SealedBox(combined: envelope.ciphertext), using: key(), authenticating: envelope.metadata)
+        let record = try JSONDecoder().decode(TerminalRecoveryRecord.self, from: envelope.metadata)
+        guard record.id == id else { throw TerminalTransportError.invalidCheckpoint }
+        return record
+    }
+
+    public func records() throws -> [TerminalRecoveryRecord] {
+        guard FileManager.default.fileExists(atPath: directory.path) else { return [] }
+        return try FileManager.default.contentsOfDirectory(
+            at: directory, includingPropertiesForKeys: nil
+        )
+        .filter { $0.pathExtension == "terminal" }
+        .map {
+            let envelope = try JSONDecoder().decode(Envelope.self, from: Data(contentsOf: $0))
+            return try JSONDecoder().decode(TerminalRecoveryRecord.self, from: envelope.metadata)
+        }
+        .sorted { $0.savedAt > $1.savedAt }
+    }
+
+    public func save(
+        _ record: TerminalRecoveryRecord, checkpoint: TerminalTransportCheckpoint, lease: UUID
+    ) throws {
+        guard owners[record.id] == lease, checkpoint.transport == record.transport else {
+            throw CancellationError()
+        }
+        let metadata = try JSONEncoder().encode(record)
+        let payload = try JSONEncoder().encode(checkpoint)
+        let ciphertext = try AES.GCM.seal(payload, using: key(), authenticating: metadata).combined!
+        try FileManager.default.createDirectory(
+            at: directory, withIntermediateDirectories: true, attributes: [.posixPermissions: 0o700])
+        var excluded = directory
+        var resources = URLResourceValues()
+        resources.isExcludedFromBackup = true
+        try excluded.setResourceValues(resources)
+        try JSONEncoder().encode(Envelope(metadata: metadata, ciphertext: ciphertext))
+            .write(to: location(record.id), options: [.atomic, .completeFileProtectionUntilFirstUserAuthentication])
+        try FileManager.default.setAttributes([.posixPermissions: 0o600], ofItemAtPath: location(record.id).path)
+    }
+
+    /// Removes eligibility before returning protocol counters to a caller that can transmit.
+    public func consume(_ id: UUID, host: Host) throws -> TerminalTransportCheckpoint {
+        let envelope = try JSONDecoder().decode(Envelope.self, from: Data(contentsOf: location(id)))
+        let payload = try AES.GCM.open(
+            AES.GCM.SealedBox(combined: envelope.ciphertext), using: key(), authenticating: envelope.metadata)
+        let record = try JSONDecoder().decode(TerminalRecoveryRecord.self, from: envelope.metadata)
+        guard record.id == id, record.matches(host) else { throw TerminalTransportError.invalidCheckpoint }
+        let checkpoint = try JSONDecoder().decode(TerminalTransportCheckpoint.self, from: payload)
+        guard checkpoint.version == 1, checkpoint.transport == record.transport else {
+            throw TerminalTransportError.invalidCheckpoint
+        }
+        try remove(id)
+        return checkpoint
+    }
+
+    public func invalidate(_ id: UUID) throws { try remove(id) }
+
+    public func discard(_ id: UUID) throws {
+        owners.removeValue(forKey: id)
+        try remove(id)
+    }
+
+    private func remove(_ id: UUID) throws {
+        let path = location(id)
+        if FileManager.default.fileExists(atPath: path.path) { try FileManager.default.removeItem(at: path) }
+    }
+
+    private func location(_ id: UUID) -> URL {
+        directory.appendingPathComponent(id.uuidString.lowercased()).appendingPathExtension("terminal")
+    }
+
+    private func key() throws -> SymmetricKey {
+        if let testingKey { return SymmetricKey(data: testingKey) }
+        let match: [String: Any] = [
+            kSecClass as String: kSecClassGenericPassword,
+            kSecAttrService as String: "com.akinokaede.termind.terminal-recovery",
+            kSecAttrAccount as String: "device-key-v1",
+            kSecAttrSynchronizable as String: false,
+        ]
+        var query = match
+        query[kSecReturnData as String] = true
+        query[kSecMatchLimit as String] = kSecMatchLimitOne
+        var result: CFTypeRef?
+        let status = SecItemCopyMatching(query as CFDictionary, &result)
+        if status == errSecSuccess, let data = result as? Data { return SymmetricKey(data: data) }
+        guard status == errSecItemNotFound else {
+            throw NSError(domain: NSOSStatusErrorDomain, code: Int(status))
+        }
+        // A missing key must not silently turn existing checkpoints into unrecoverable ciphertext.
+        if FileManager.default.fileExists(atPath: directory.path),
+            try FileManager.default.contentsOfDirectory(atPath: directory.path).contains(where: {
+                $0.hasSuffix(".terminal")
+            })
+        {
+            throw TerminalTransportError.invalidCheckpoint
+        }
+        let key = SymmetricKey(size: .bits256)
+        var add = match
+        add[kSecValueData as String] = key.withUnsafeBytes { Data($0) }
+        add[kSecAttrAccessible as String] = kSecAttrAccessibleAfterFirstUnlockThisDeviceOnly
+        let inserted = SecItemAdd(add as CFDictionary, nil)
+        guard inserted == errSecSuccess else { throw NSError(domain: NSOSStatusErrorDomain, code: Int(inserted)) }
+        return key
+    }
+}