remote-tmux: name the transport seam so a persistent-session transport can be added

CLI/cmux.swift

@@ -11450,6 +11450,8 @@ struct CMUXCLI {
         var identityFile: String?
         var noFocus = false
         var newWindow = false
+        var transport: String?
+        var transportPort: Int?
 
         // Intentional subset of parseSSHCommandOptions: ssh-tmux has no relay,
         // passthrough, --ssh-option, --name, or --window support.
@@ -11472,6 +11474,25 @@ struct CMUXCLI {
                 }
                 identityFile = commandArgs[index + 1]
                 index += 2
+            case "--transport":
+                guard index + 1 < commandArgs.count else {
+                    throw CLIError(message: "ssh-tmux: --transport requires a value (ssh or et)")
+                }
+                let raw = commandArgs[index + 1].lowercased()
+                guard raw == "ssh" || raw == "et" else {
+                    throw CLIError(message: "ssh-tmux: --transport must be ssh or et")
+                }
+                transport = raw
+                index += 2
+            case "--transport-port":
+                guard index + 1 < commandArgs.count else {
+                    throw CLIError(message: "ssh-tmux: --transport-port requires a value")
+                }
+                guard let parsed = Int(commandArgs[index + 1]), parsed > 0, parsed <= 65535 else {
+                    throw CLIError(message: "ssh-tmux: --transport-port must be 1-65535")
+                }
+                transportPort = parsed
+                index += 2
             case "--no-focus":
                 noFocus = true
                 index += 1
@@ -11500,6 +11521,8 @@ struct CMUXCLI {
         var params: [String: Any] = ["host": destination]
         if let port { params["port"] = port }
         if let identityFile, !identityFile.isEmpty { params["identity_file"] = identityFile }
+        if let transport { params["transport"] = transport }
+        if let transportPort { params["transport_port"] = transportPort }
         params["activate"] = !noFocus
         if !newWindow {
             try applyWindowOrCallerContext(to: &params, client: client, windowRaw: nil)
@@ -19276,7 +19299,8 @@ struct CMUXCLI {
             return Self.moshTmuxCommandUsage
         case "ssh-tmux":
             let help = String(localized: "cli.help.ssh-tmux", defaultValue: """
-            Usage: cmux ssh-tmux <destination> [--port <n>] [--identity <path>] [--no-focus]
+            Usage: cmux ssh-tmux <destination> [--port <n>] [--identity <path>]
+                                 [--transport ssh|et] [--transport-port <n>] [--no-focus]
 
             Mirror a remote host's tmux sessions into the current window's sidebar over
             SSH tmux control mode (tmux -CC). Each session becomes a workspace, each
@@ -19290,13 +19314,19 @@ struct CMUXCLI {
             with no prompt. ~/.ssh/config aliases and their IdentityFile/ProxyJump/Port settings are honored.
 
             Flags:
-              --port <n>          SSH port
-              --identity <path>   SSH identity file path
-              --no-focus          Do not select the mirror workspace or focus its window
+              --port <n>            SSH port
+              --identity <path>     SSH identity file path
+              --transport <name>    ssh (default) or et. et carries the control stream over
+                                    EternalTerminal, which reconnects on its own, so the
+                                    mirror survives a network change instead of respawning.
+              --transport-port <n>  Port the transport connects to: sshd's for ssh, etserver's
+                                    for et (default 22 for ssh, 2022 for et)
+              --no-focus            Do not select the mirror workspace or focus its window
 
             Example:
               cmux ssh-tmux dev@my-host
               cmux ssh-tmux dev@my-host --port 2222 --identity ~/.ssh/id_ed25519
+              cmux ssh-tmux dev@my-host --transport et --transport-port 8080
             """)
             let newWindowHelp = String(
                 localized: "cli.help.ssh-tmux.newWindow",

Sources/RemoteTmuxControlConnection.swift

@@ -123,7 +123,10 @@ final class RemoteTmuxControlConnection {
     private var stderrTask: Task<Void, Never>?
     private var parser = RemoteTmuxControlStreamParser()
     private var ingestTask: Task<Void, Never>?
-    private var processGeneration: UInt64 = 0
+    /// Bumped on every spawn. Readable across the type's extensions so a completion that
+    /// outlived its process — a liveness probe answered after a respawn, say — can tell that its
+    /// answer describes a stream that no longer exists. Writable only here.
+    private(set) var processGeneration: UInt64 = 0
     var pendingCommands: [CommandKind] = []
     var windowListRequestInFlight = false
     var windowListRequestDirty = false
@@ -329,12 +332,19 @@ final class RemoteTmuxControlConnection {
     static let altScreenEnterSequence = Data("\u{1b}[?1049h".utf8)
     static let altScreenExitSequence = Data("\u{1b}[?1049l".utf8)
 
+    /// How this connection is carried, derived from the host's transport unless a caller
+    /// overrides it (which tests do, to assert argv without spawning anything).
+    let transportProfile: RemoteTmuxTransportProfile
+
     init(
         host: RemoteTmuxHost,
         sessionName: String,
         createIfMissing: Bool = false,
-        pendingPaneSeedByteLimit: Int = RemoteTmuxControlConnection.maximumPendingPaneSeedBytes
+        pendingPaneSeedByteLimit: Int = RemoteTmuxControlConnection.maximumPendingPaneSeedBytes,
+        transportProfile: RemoteTmuxTransportProfile? = nil
     ) {
+        self.transportProfile = transportProfile
+            ?? host.transport.profile(port: host.transportPort, terminalPath: host.transportTerminalPath)
         self.host = host
         self.sessionName = sessionName
         self.createIfMissing = createIfMissing
@@ -423,11 +433,25 @@ final class RemoteTmuxControlConnection {
         enterReceived = false
 
         let proc = Process()
-        proc.executableURL = URL(fileURLWithPath: RemoteTmuxHost.defaultSSHExecutablePath())
-        proc.arguments = host.controlModeArguments(
+        let transportExecutable = transportProfile.executablePath()
+        let transportArgv = transportProfile.controlStreamArgv(
+            host: host,
             sessionName: sessionName,
             createIfMissing: createIfMissing
         )
+        if transportProfile.requiresPseudoTerminal {
+            // A terminal client will not talk over pipes: measured against et 6.2.11, it
+            // emits nothing at all and aborts at session end, which reads as "the host
+            // produced no output" rather than "this was spawned without a tty". Give it one.
+            record("transport-pty")
+            proc.executableURL = URL(fileURLWithPath: RemoteTmuxPseudoTerminal.allocatorPath)
+            proc.arguments = RemoteTmuxPseudoTerminal.wrap(
+                executable: transportExecutable, arguments: transportArgv
+            )
+        } else {
+            proc.executableURL = URL(fileURLWithPath: transportExecutable)
+            proc.arguments = transportArgv
+        }
         let inPipe = Pipe(), outPipe = Pipe(), errPipe = Pipe()
         proc.standardInput = inPipe
         proc.standardOutput = outPipe
@@ -585,10 +609,78 @@ final class RemoteTmuxControlConnection {
         stderrPipeReader = nil
         stdinWriter?.close()
         stdinWriter = nil
-        process?.terminate()
+        terminateProcessTree(process)
         process = nil
     }
 
+    /// Ends a spawned transport and everything it started.
+    ///
+    /// `Process.terminate()` signals one pid, and a transport is rarely one process: cmux may
+    /// launch a pty allocator that execs a broker that finally execs the client. Signalling only
+    /// the allocator leaves the broker and client running, and because the client holds the
+    /// remote end open they keep their session too — two such trees were found alive hours after
+    /// their respawns, each still holding a control client on the remote server.
+    ///
+    /// Signalling the allocator's process GROUP does not fix it either: `/usr/bin/script` puts
+    /// its command in a group of its own (measured — the allocator and its payload had different
+    /// pgids, and the payload survived a group kill). So the tree is walked instead, children
+    /// before parents, and each process that leads its own group takes that group with it.
+    ///
+    /// SIGTERM first because these clients close their remote end on it; anything still alive a
+    /// moment later is sent SIGKILL, so a client that ignores the polite signal cannot outlive
+    /// the stream that owns it.
+    private func terminateProcessTree(_ proc: Process?) {
+        guard let proc, proc.processIdentifier > 0 else { return }
+        let root = proc.processIdentifier
+        let tree = Self.processTree(root: root)
+        for pid in tree.reversed() { Self.signalProcess(pid, SIGTERM) }
+        proc.terminate()
+        Task.detached {
+            try? await Task.sleep(nanoseconds: 2_000_000_000)
+            for pid in tree.reversed() where Darwin.kill(pid, 0) == 0 {
+                Self.signalProcess(pid, SIGKILL)
+            }
+        }
+    }
+
+    /// `root` and its descendants, parents before children, bounded in depth so a pathological
+    /// tree cannot make teardown expensive.
+    nonisolated static func processTree(root: pid_t, childrenOf: (pid_t) -> [pid_t] = childPIDs) -> [pid_t] {
+        var out: [pid_t] = [root]
+        var frontier = [root]
+        for _ in 0..<4 {
+            let next = frontier.flatMap(childrenOf).filter { !out.contains($0) }
+            if next.isEmpty { break }
+            out.append(contentsOf: next)
+            frontier = next
+        }
+        return out
+    }
+
+    /// Sends `signal` to `pid`, and to its process group when `pid` leads one. A leader's group
+    /// holds the processes it started that the walk cannot see (anything spawned between the
+    /// listing and the signal).
+    nonisolated private static func signalProcess(_ pid: pid_t, _ signal: Int32) {
+        guard pid > 1 else { return }
+        if getpgid(pid) == pid { _ = Darwin.kill(-pid, signal) }
+        _ = Darwin.kill(pid, signal)
+    }
+
+    /// Direct children of `pid`, via the kernel process table (no subprocess, so teardown does
+    /// not spawn anything while it is tearing down).
+    nonisolated static let childPIDs: (pid_t) -> [pid_t] = { parent in
+        var name: [Int32] = [CTL_KERN, KERN_PROC, KERN_PROC_ALL, 0]
+        var length = 0
+        guard sysctl(&name, 4, nil, &length, nil, 0) == 0, length > 0 else { return [] }
+        let count = length / MemoryLayout<kinfo_proc>.stride
+        var procs = [kinfo_proc](repeating: kinfo_proc(), count: count)
+        guard sysctl(&name, 4, &procs, &length, nil, 0) == 0 else { return [] }
+        let actual = length / MemoryLayout<kinfo_proc>.stride
+        return procs[0..<min(actual, count)].compactMap { entry in
+            entry.kp_eproc.e_ppid == parent ? entry.kp_proc.p_pid : nil
+        }
+    }
+
     // MARK: - Internals
 
     @discardableResult
@@ -699,9 +791,34 @@ final class RemoteTmuxControlConnection {
         case .ended:
             return
         case .connecting, .connected:
-            // The control stream died without `%exit` — a transport loss. Keep the
-            // mirror frozen and reconnect.
-            beginReconnecting()
+            // The control stream died without `%exit`. What that means depends on who owns
+            // reconnection: for ssh it is a transport loss cmux recovers from, but a
+            // transport that reconnects internally does not end for a network drop, so its
+            // exit is the session genuinely ending.
+            // A transport that could not start will not start on the next try either, and
+            // retrying hides the reason: end-of-stream no longer implies the session is over, so
+            // without this the mirror waits out the attach timeout with nothing to explain it.
+            if RemoteTmuxSSHTransport.indicatesUnrecoverableTransportFailure(stderrBuffer) {
+                record("stream-end-unrecoverable")
+                connectionState = .ended
+                cancelScheduledWork()
+                teardownProcessHandles()
+                observers.notifyExit()
+                return
+            }
+            switch RemoteTmuxStreamEndDisposition.forStreamEnd(hasReachedControlMode: enterReceived) {
+            case .reconnect:
+                // Keep the mirror frozen and reconnect.
+                beginReconnecting()
+            case .sessionOver:
+                // Either the session ended, or the transport never started — both are terminal, and
+                // both must report rather than retry.
+                record(enterReceived ? "stream-end-session-over" : "stream-end-before-connect")
+                connectionState = .ended
+                cancelScheduledWork()
+                teardownProcessHandles()
+                observers.notifyExit()
+            }
         case .reconnecting:
             // A reconnect attempt's process exited before reaching control mode
             // (a successful attach would have moved us to `.connected` via `.enter`).
@@ -717,8 +834,12 @@ final class RemoteTmuxControlConnection {
             // (host unreachable, refused) is transient — keep retrying with backoff.
             let sessionGone = decoding.stderrIndicatesSessionGone(stderrBuffer)
                 || decoding.controlOutputIndicatesSessionGone(preControlOutputBuffer)
+            // Same rule on the retry path: a reconnect that failed because the transport cannot run
+            // is not transient, and looping on it burns the backoff forever.
+            let unrecoverable = RemoteTmuxSSHTransport.indicatesUnrecoverableTransportFailure(stderrBuffer)
             teardownProcessHandles()
-            if sessionGone {
+            if sessionGone || unrecoverable {
+                if unrecoverable { record("reconnect-unrecoverable") }
                 record("reconnect-session-gone")
                 connectionState = .ended
                 reconnectTask?.cancel()

Sources/RemoteTmuxControlStreamParser.swift

@@ -21,6 +21,10 @@ struct RemoteTmuxControlStreamParser {
     private let maxCommandBlockBytes: Int
     private var buffer: [UInt8] = []
     private var inBlock = false
+    /// Whether control mode has been entered. Entering happens once per stream, so the scan for
+    /// the enter DCS stops after it — searching later lines could match a DCS that is genuinely
+    /// part of a pane's own bytes.
+    private var sawEnter = false
     private var blockNumber = 0
     private var blockLines: [String] = []
     private var blockBufferedBytes = 0
@@ -36,6 +40,22 @@ struct RemoteTmuxControlStreamParser {
     /// The DCS sequence tmux emits to enter control mode: `ESC P 1000 p`.
     private static let enterSequence: [UInt8] = [0x1b, 0x50, 0x31, 0x30, 0x30, 0x30, 0x70]
 
+    /// Index range of the first occurrence of `needle` in `haystack`, or nil.
+    private static func firstRange(of needle: [UInt8], in haystack: [UInt8]) -> Range<Int>? {
+        guard !needle.isEmpty, haystack.count >= needle.count else { return nil }
+        let last = haystack.count - needle.count
+        var start = 0
+        while start <= last {
+            if haystack[start] == needle[0] {
+                var offset = 1
+                while offset < needle.count, haystack[start + offset] == needle[offset] { offset += 1 }
+                if offset == needle.count { return start..<(start + needle.count) }
+            }
+            start += 1
+        }
+        return nil
+    }
+
     /// ASCII bytes of the `%output ` notification prefix (used to detect and parse
     /// `%output` lines from raw bytes, before any String decode).
     private static let outputPrefix: [UInt8] = Array("%output ".utf8)
@@ -67,10 +87,23 @@ struct RemoteTmuxControlStreamParser {
         var bytes = rawBytes
         var prefixMessages: [RemoteTmuxControlMessage] = []
 
-        // Strip a leading enter DCS (it is prepended to the first %begin line).
-        if bytes.starts(with: Self.enterSequence) {
+        // Strip the enter DCS. It usually opens the line, but it does not have to: a transport
+        // that types the command into a login shell (et) leaves the shell's echo and its OSC
+        // title sequences on the same line, with the DCS arriving partway through and no
+        // newline before it. Requiring it at offset 0 means `.enter` is never emitted there,
+        // and since commands are withheld until `.enter`, the mirror waits forever while the
+        // notifications after it parse normally.
+        //
+        // Everything before the DCS is pre-control-mode shell noise by definition — tmux emits
+        // this sequence when it takes over the terminal — so dropping that prefix is right
+        // rather than merely convenient.
+        // Scoped to before control mode is entered and outside a command block: block content is
+        // raw pane bytes (`capture-pane -e`) that can legitimately contain this DCS, and matching
+        // it there would cut the painted pane apart — the same hazard the ST strip below avoids.
+        if !sawEnter, !inBlock, let dcs = Self.firstRange(of: Self.enterSequence, in: bytes) {
+            sawEnter = true
             prefixMessages.append(.enter)
-            bytes.removeFirst(Self.enterSequence.count)
+            bytes.removeFirst(dcs.upperBound)
         }
         // Drop ST (ESC \) DCS-teardown framing — but ONLY on notification lines.
         // Command-block content (e.g. `capture-pane -e` output) is raw terminal