ssh-tmux: deliver OSC 777/9 notifications from mirrored panes (#833)

Sources/RemoteTmuxControlConnection+CommandResults.swift

@@ -239,8 +239,10 @@ extension RemoteTmuxControlConnection {
                 // (see ``PostAttachAction``).
                 switch pendingPostAttachAction {
                 case .reseed:
+                    pushMirrorSessionEnvironment()
                     reseedAfterReconnect()
                 case .applyClientSize:
+                    pushMirrorSessionEnvironment()
                     // A surface that hasn't computed a grid yet is covered by the
                     // debounced `setClientSize` instead.
                     if let size = lastClientSize {

Sources/RemoteTmuxControlConnection+Commands.swift

@@ -12,6 +12,62 @@ extension RemoteTmuxControlConnection {
         sendInternal(command, kind: .other)
     }
 
+    // MARK: - Mirror session environment (issue #833)
+
+    /// Marker signalling to remote shell integration that this tmux session is
+    /// mirrored by a local cmux over ssh-tmux (`tmux -CC`, no relay socket).
+    static let mirrorMarkerEnvironmentKey = "CMUX_REMOTE_TMUX_MIRROR"
+
+    /// Pushes the mirror marker + identity pairs into the remote tmux SESSION
+    /// environment. Called from the first post-attach `list-windows` result, so
+    /// both paths — first connect and every reconnect — refresh values that
+    /// would otherwise go permanently stale after an app relaunch (issue #833:
+    /// the `tmux -CC` attach has no cmux wrapper shell outside tmux on the
+    /// remote, so nobody else ever re-publishes them).
+    ///
+    /// Deliberately NOT pushed: `CMUX_SOCKET_PATH`. The ssh-tmux transport has
+    /// no relay/reverse forward, so the local Mac socket path is meaningless on
+    /// the remote host; publishing it would point remote `cmux` CLI invocations
+    /// at a dead socket. Notification delivery instead rides the OSC 777/9
+    /// intercept in ``RemoteTmuxSessionMirror`` (see
+    /// ``RemoteTmuxNotificationOSCFilter``).
+    ///
+    /// Session scope (`-t`, not `-g`): the shell-integration refresh path runs
+    /// a session-scoped `show-environment`, which does not surface `-g` values.
+    func pushMirrorSessionEnvironment() {
+        // Target by the stable session id when known so the push can't race a
+        // rename (same convention as `rename-session`).
+        guard let target = sessionId.map({ "$\($0)" })
+            ?? RemoteTmuxHost.controlModeLineSafeName(sessionName)
+                .map(RemoteTmuxHost.shellSingleQuoted)
+        else { return }
+        var pairs = mirrorEnvironment
+        pairs[Self.mirrorMarkerEnvironmentKey] = "1"
+        let commands = Self.mirrorEnvironmentCommands(target: target, pairs: pairs)
+        guard !commands.isEmpty else { return }
+        _ = sendBatchInternal(commands, kinds: commands.map { _ in .other })
+    }
+
+    /// Builds the `set-environment -t <target> KEY VALUE` command lines for a
+    /// push, dropping any pair that could break the line-oriented control
+    /// stream. Pure (and deterministic — sorted by key) so tests can pin the
+    /// exact wire format.
+    static func mirrorEnvironmentCommands(
+        target: String,
+        pairs: [String: String]
+    ) -> [String] {
+        pairs.sorted { $0.key < $1.key }.compactMap { key, value in
+            // Keys are ours (static identifiers), but guard anyway: one CR/LF
+            // would terminate the command line before tmux parses the quotes.
+            guard RemoteTmuxHost.controlModeLineSafeName(key) != nil,
+                  !key.contains(" "),
+                  RemoteTmuxHost.controlModeLineSafeName(value) != nil
+            else { return nil }
+            return "set-environment -t \(target) \(key) "
+                + RemoteTmuxHost.shellSingleQuoted(value)
+        }
+    }
+
     /// Sends a command and reports how its `%begin`/`%end` block resolved:
     /// `true` on `%end`, `false` on `%error` — or `false` if the stream resets
     /// before the block arrives, since a fresh control stream can never answer

Sources/RemoteTmuxControlConnection.swift

@@ -195,6 +195,25 @@ final class RemoteTmuxControlConnection {
     var lastSizingSendAt: ContinuousClock.Instant?
     var pendingPostAttachAction: PostAttachAction?
 
+    /// Session-scoped environment pairs identifying the local mirror, pushed to
+    /// the remote tmux session on every attach AND reconnect (issue #833).
+    /// Session scope (`set-environment -t`) is deliberate: the shell-integration
+    /// pull path runs a session-scoped `show-environment`, which does not see
+    /// global (`-g`) values. Set by the controller when the mirror workspace is
+    /// created; empty until then (and for non-mirror consumers).
+    private(set) var mirrorEnvironment: [String: String] = [:]
+
+    /// Replaces the identity pairs pushed by ``pushMirrorSessionEnvironment()``.
+    /// A connection that already passed its post-attach point (a reused,
+    /// still-connected connection) pushes the fresh pairs immediately;
+    /// otherwise the pending post-attach drain pushes them.
+    func setMirrorEnvironment(_ pairs: [String: String]) {
+        mirrorEnvironment = pairs
+        if connectionState == .connected, attachBlockDrained, pendingPostAttachAction == nil {
+            pushMirrorSessionEnvironment()
+        }
+    }
+
     /// Trailing-edge debounce for `refresh-client -C`. SwiftUI layout settle makes the
     /// rendered grid oscillate (e.g. cols 154→155→156→161→…, ~15 distinct grids in
     /// ~1.3s), and each previously sent its own `refresh-client -C` → ~15 SIGWINCH /

Sources/RemoteTmuxController.swift

@@ -307,6 +307,16 @@ final class RemoteTmuxController {
             applyCreationTitleAsCustomTitle: false
         )
         workspace.isRemoteTmuxMirror = true
+        // Identity pairs the connection pushes into the remote SESSION
+        // environment on attach and every reconnect (issue #833). Workspace id
+        // is published under both keys, matching the SSH-workspace bootstrap
+        // convention (`CMUX_TAB_ID` is the legacy alias). No socket path: the
+        // ssh-tmux transport has no relay, so a local path would be dead on the
+        // remote — see ``RemoteTmuxControlConnection/pushMirrorSessionEnvironment()``.
+        connection.setMirrorEnvironment([
+            "CMUX_WORKSPACE_ID": workspace.id.uuidString,
+            "CMUX_TAB_ID": workspace.id.uuidString,
+        ])
         workspace.remoteTmuxWindowOrderSync = { [weak self, weak workspace] orderedPanelIds, verification in
             guard let self, let workspace else { return false }
             return self.handleMirrorWindowsReordered(

Sources/RemoteTmuxNotificationOSCFilter.swift

@@ -0,0 +1,201 @@
+import Foundation
+
+/// Intercepts OSC desktop-notification escapes in a mirrored pane's output
+/// stream (issue #833).
+///
+/// A remote process inside an ssh-tmux mirror emits notifications with the
+/// xterm/urxvt OSC sequences:
+/// - `ESC ] 777;notify;<title>;<body> BEL|ST` (rxvt-unicode / wezterm style)
+/// - `ESC ] 9;<body> BEL|ST` (iTerm2 growl style; no title)
+///
+/// `%output` is the raw pty copy, so tmux forwards these bytes verbatim to the
+/// mirror. The mirror's Ghostty surface would parse them, but its
+/// `desktop_notification` callback attributes by the surface's local process
+/// TTY — which a manual-mirror surface does not have — and mirror workspaces
+/// are excluded from agent TTY delivery (`AgentDeliveryTargetResolution`). So
+/// the mirror layer intercepts the sequence here, strips it from the stream,
+/// and reports `(title, body)` so the session mirror can attribute the
+/// notification to the pane's surface + workspace itself.
+///
+/// Any other OSC sequence (titles, hyperlinks, clipboard, non-`notify`
+/// OSC 777 subcommands) passes through byte-identical.
+///
+/// Stateful across calls: a `%output` chunk can split the sequence at any
+/// byte. While a sequence still *might* be a notification its bytes are
+/// buffered; the moment the payload diverges from both notification prefixes
+/// the buffer is flushed verbatim and the rest of the sequence streams
+/// through unbuffered. An unfinished candidate that exceeds
+/// ``maxBufferedBytes`` is flushed verbatim too, so a hostile or corrupt
+/// stream can never pin memory or swallow output.
+struct RemoteTmuxNotificationOSCFilter {
+    private enum State {
+        case text        // normal passthrough
+        case esc         // saw ESC, holding it until we know if it's `ESC ]`
+        case collect     // inside an OSC that may still be a notification; buffering
+        case collectEsc  // in collect, saw ESC; maybe the `ESC \` terminator
+        case passOsc     // inside a non-notification OSC; streaming through
+        case passOscEsc  // in passOsc, saw ESC; maybe the `ESC \` terminator
+    }
+
+    /// Ceiling on bytes buffered for an unfinished candidate sequence. An
+    /// overflowing sequence is passed through verbatim instead of stripped.
+    static let maxBufferedBytes = 4096
+
+    private static let notifyPrefix = Array("777;notify;".utf8)
+    private static let osc9Prefix = Array("9;".utf8)
+
+    private var state: State = .text
+    /// Original bytes of the candidate sequence (`ESC ]` + payload so far),
+    /// replayed verbatim when the sequence turns out not to be a notification.
+    private var raw: [UInt8] = []
+    /// Payload bytes only (after `ESC ]`), matched against the prefixes and
+    /// decoded into `(title, body)` on a hit.
+    private var payload: [UInt8] = []
+
+    /// Creates a filter with no buffered escape-sequence state.
+    init() {}
+
+    /// Returns `data` with any complete notification sequences removed,
+    /// invoking `onNotification(title, body)` once per hit in stream order.
+    /// OSC 9 hits report an empty title.
+    mutating func filter(
+        _ data: Data,
+        onNotification: (_ title: String, _ body: String) -> Void
+    ) -> Data {
+        // Hot path: not mid-sequence and no ESC in the chunk — nothing to do.
+        if state == .text, !data.contains(0x1b) { return data }
+        var out = [UInt8]()
+        out.reserveCapacity(data.count)
+        for byte in data {
+            switch state {
+            case .text:
+                if byte == 0x1b {
+                    state = .esc            // hold the ESC; emit it only if it isn't `ESC ]`
+                } else {
+                    out.append(byte)
+                }
+            case .esc:
+                if byte == UInt8(ascii: "]") {
+                    raw = [0x1b, byte]
+                    payload.removeAll(keepingCapacity: true)
+                    state = .collect
+                } else if byte == 0x1b {
+                    out.append(0x1b)        // emit the held ESC, keep holding the new one
+                } else {
+                    out.append(0x1b)
+                    out.append(byte)
+                    state = .text
+                }
+            case .collect:
+                if byte == 0x07 {           // BEL terminator
+                    finishCandidate(terminator: [0x07], into: &out, onNotification)
+                } else if byte == 0x1b {
+                    state = .collectEsc
+                } else {
+                    raw.append(byte)
+                    payload.append(byte)
+                    reclassifyCandidate(into: &out)
+                }
+            case .collectEsc:
+                if byte == 0x5c {           // `ESC \` (ST) terminator
+                    finishCandidate(terminator: [0x1b, 0x5c], into: &out, onNotification)
+                } else {
+                    // An OSC payload cannot legally contain a bare ESC; treat the
+                    // sequence as malformed and pass everything through verbatim.
+                    out.append(contentsOf: raw)
+                    out.append(0x1b)
+                    raw.removeAll(keepingCapacity: true)
+                    payload.removeAll(keepingCapacity: true)
+                    if byte == 0x1b {
+                        state = .esc        // the new ESC may start a fresh sequence
+                    } else {
+                        out.append(byte)
+                        state = .text
+                    }
+                }
+            case .passOsc:
+                if byte == 0x07 {
+                    out.append(byte)
+                    state = .text
+                } else if byte == 0x1b {
+                    state = .passOscEsc
+                } else {
+                    out.append(byte)
+                }
+            case .passOscEsc:
+                if byte == 0x5c {
+                    out.append(0x1b)
+                    out.append(0x5c)
+                    state = .text
+                } else if byte == 0x1b {
+                    out.append(0x1b)        // emit the held ESC, keep holding the new one
+                } else {
+                    out.append(0x1b)
+                    out.append(byte)
+                    state = .passOsc
+                }
+            }
+        }
+        return Data(out)
+    }
+
+    /// While collecting, checks the payload against both notification prefixes
+    /// and the buffer ceiling; a sequence that can no longer be a notification
+    /// (or grew too large unfinished) is flushed verbatim and the remainder
+    /// streams through as a plain OSC.
+    private mutating func reclassifyCandidate(into out: inout [UInt8]) {
+        if raw.count <= Self.maxBufferedBytes,
+           Self.couldMatch(payload, prefix: Self.notifyPrefix)
+            || Self.couldMatch(payload, prefix: Self.osc9Prefix) {
+            return
+        }
+        out.append(contentsOf: raw)
+        raw.removeAll(keepingCapacity: true)
+        payload.removeAll(keepingCapacity: true)
+        state = .passOsc
+    }
+
+    /// Whether `payload` is still compatible with `prefix` (either side is a
+    /// prefix of the other).
+    private static func couldMatch(_ payload: [UInt8], prefix: [UInt8]) -> Bool {
+        payload.count >= prefix.count
+            ? payload.starts(with: prefix)
+            : prefix.starts(with: payload)
+    }
+
+    /// A candidate sequence reached its terminator: strip + report a
+    /// notification hit, or replay the original bytes for anything else
+    /// (including a too-short candidate like `ESC ] 9 BEL`).
+    private mutating func finishCandidate(
+        terminator: [UInt8],
+        into out: inout [UInt8],
+        _ onNotification: (_ title: String, _ body: String) -> Void
+    ) {
+        defer {
+            raw.removeAll(keepingCapacity: true)
+            payload.removeAll(keepingCapacity: true)
+            state = .text
+        }
+        if payload.starts(with: Self.notifyPrefix) {
+            let rest = payload.dropFirst(Self.notifyPrefix.count)
+            if let separator = rest.firstIndex(of: UInt8(ascii: ";")) {
+                onNotification(
+                    Self.decode(rest[..<separator]),
+                    Self.decode(rest[rest.index(after: separator)...])
+                )
+            } else {
+                // `777;notify;<title>` without a body separator: title only.
+                onNotification(Self.decode(rest), "")
+            }
+        } else if payload.starts(with: Self.osc9Prefix) {
+            onNotification("", Self.decode(payload.dropFirst(Self.osc9Prefix.count)))
+        } else {
+            out.append(contentsOf: raw)
+            out.append(contentsOf: terminator)
+        }
+    }
+
+    private static func decode(_ bytes: ArraySlice<UInt8>) -> String {
+        String(decoding: bytes, as: UTF8.self)
+    }
+}