diff --git a/Packages/macOS/CmuxControlSocket/Sources/CmuxControlSocket/Wire/ControlCommandExecutionPolicy.swift b/Packages/macOS/CmuxControlSocket/Sources/CmuxControlSocket/Wire/ControlCommandExecutionPolicy.swift index 30407f5f1..b8cca1935 100644 --- a/Packages/macOS/CmuxControlSocket/Sources/CmuxControlSocket/Wire/ControlCommandExecutionPolicy.swift +++ b/Packages/macOS/CmuxControlSocket/Sources/CmuxControlSocket/Wire/ControlCommandExecutionPolicy.swift @@ -148,6 +148,9 @@ public enum ControlCommandExecutionPolicy: Sendable, Equatable { // finishes — that wait must happen on a socket worker, never on the // main actor (same shape as the remote.tmux.* methods above). "debug.amux.mirror_local", + // debug.amux.mirror_ssh has the same v2VmCall shape (and an SSH + // round-trip on top), so it parks on the worker lane too. + "debug.amux.mirror_ssh", // debug.amux.parse_choices captures a pane via the local transport // (an await), so it parks on v2VmCall's semaphore — worker lane. "debug.amux.parse_choices", diff --git a/Resources/Localizable.xcstrings b/Resources/Localizable.xcstrings index 806c2b047..20e4ec20a 100644 --- a/Resources/Localizable.xcstrings +++ b/Resources/Localizable.xcstrings @@ -17505,6 +17505,363 @@ } } }, + "amux.alarm.error.title": { + "extractionState": "manual", + "localizations": { + "ar": { + "stringUnit": { + "state": "translated", + "value": "واجه %@ خطأ" + } + }, + "bs": { + "stringUnit": { + "state": "translated", + "value": "%@ je naišao na grešku" + } + }, + "da": { + "stringUnit": { + "state": "translated", + "value": "%@ stødte på en fejl" + } + }, + "de": { + "stringUnit": { + "state": "translated", + "value": "Bei %@ ist ein Fehler aufgetreten" + } + }, + "en": { + "stringUnit": { + "state": "translated", + "value": "%@ hit an error" + } + }, + "es": { + "stringUnit": { + "state": "translated", + "value": "%@ encontró un error" + } + }, + "fr": { + "stringUnit": { + "state": "translated", + "value": "%@ a rencontré une erreur" + } + }, + "it": { + "stringUnit": { + "state": "translated", + "value": "%@ ha riscontrato un errore" + } + }, + "ja": { + "stringUnit": { + "state": "translated", + "value": "%@でエラーが発生しました" + } + }, + "ko": { + "stringUnit": { + "state": "translated", + "value": "%@에서 오류가 발생했습니다" + } + }, + "nb": { + "stringUnit": { + "state": "translated", + "value": "%@ støtte på en feil" + } + }, + "pl": { + "stringUnit": { + "state": "translated", + "value": "%@ napotkał błąd" + } + }, + "pt-BR": { + "stringUnit": { + "state": "translated", + "value": "%@ encontrou um erro" + } + }, + "ru": { + "stringUnit": { + "state": "translated", + "value": "%@ столкнулся с ошибкой" + } + }, + "th": { + "stringUnit": { + "state": "translated", + "value": "%@ พบข้อผิดพลาด" + } + }, + "tr": { + "stringUnit": { + "state": "translated", + "value": "%@ bir hatayla karşılaştı" + } + }, + "uk": { + "stringUnit": { + "state": "translated", + "value": "%@ зіткнувся з помилкою" + } + }, + "zh-Hans": { + "stringUnit": { + "state": "translated", + "value": "%@ 遇到错误" + } + }, + "zh-Hant": { + "stringUnit": { + "state": "translated", + "value": "%@ 發生錯誤" + } + } + } + }, + "amux.alarm.finished.title": { + "extractionState": "manual", + "localizations": { + "ar": { + "stringUnit": { + "state": "translated", + "value": "أنهى %@ العمل" + } + }, + "bs": { + "stringUnit": { + "state": "translated", + "value": "%@ je završio" + } + }, + "da": { + "stringUnit": { + "state": "translated", + "value": "%@ er færdig" + } + }, + "de": { + "stringUnit": { + "state": "translated", + "value": "%@ ist fertig" + } + }, + "en": { + "stringUnit": { + "state": "translated", + "value": "%@ finished" + } + }, + "es": { + "stringUnit": { + "state": "translated", + "value": "%@ terminó" + } + }, + "fr": { + "stringUnit": { + "state": "translated", + "value": "%@ a terminé" + } + }, + "it": { + "stringUnit": { + "state": "translated", + "value": "%@ ha terminato" + } + }, + "ja": { + "stringUnit": { + "state": "translated", + "value": "%@が完了しました" + } + }, + "ko": { + "stringUnit": { + "state": "translated", + "value": "%@ 작업이 완료되었습니다" + } + }, + "nb": { + "stringUnit": { + "state": "translated", + "value": "%@ er ferdig" + } + }, + "pl": { + "stringUnit": { + "state": "translated", + "value": "%@ zakończył pracę" + } + }, + "pt-BR": { + "stringUnit": { + "state": "translated", + "value": "%@ concluiu" + } + }, + "ru": { + "stringUnit": { + "state": "translated", + "value": "%@ завершил работу" + } + }, + "th": { + "stringUnit": { + "state": "translated", + "value": "%@ ทำงานเสร็จแล้ว" + } + }, + "tr": { + "stringUnit": { + "state": "translated", + "value": "%@ tamamladı" + } + }, + "uk": { + "stringUnit": { + "state": "translated", + "value": "%@ завершив роботу" + } + }, + "zh-Hans": { + "stringUnit": { + "state": "translated", + "value": "%@ 已完成" + } + }, + "zh-Hant": { + "stringUnit": { + "state": "translated", + "value": "%@ 已完成" + } + } + } + }, + "amux.alarm.waiting.title": { + "extractionState": "manual", + "localizations": { + "ar": { + "stringUnit": { + "state": "translated", + "value": "%@ بحاجة إلى إدخالك" + } + }, + "bs": { + "stringUnit": { + "state": "translated", + "value": "%@ treba vaš unos" + } + }, + "da": { + "stringUnit": { + "state": "translated", + "value": "%@ har brug for dit input" + } + }, + "de": { + "stringUnit": { + "state": "translated", + "value": "%@ benötigt Ihre Eingabe" + } + }, + "en": { + "stringUnit": { + "state": "translated", + "value": "%@ needs your input" + } + }, + "es": { + "stringUnit": { + "state": "translated", + "value": "%@ necesita tu respuesta" + } + }, + "fr": { + "stringUnit": { + "state": "translated", + "value": "%@ attend votre réponse" + } + }, + "it": { + "stringUnit": { + "state": "translated", + "value": "%@ è in attesa del tuo input" + } + }, + "ja": { + "stringUnit": { + "state": "translated", + "value": "%@が入力を待っています" + } + }, + "ko": { + "stringUnit": { + "state": "translated", + "value": "%@이(가) 입력을 기다리고 있습니다" + } + }, + "nb": { + "stringUnit": { + "state": "translated", + "value": "%@ trenger innspill fra deg" + } + }, + "pl": { + "stringUnit": { + "state": "translated", + "value": "%@ czeka na Twoją odpowiedź" + } + }, + "pt-BR": { + "stringUnit": { + "state": "translated", + "value": "%@ precisa da sua resposta" + } + }, + "ru": { + "stringUnit": { + "state": "translated", + "value": "%@ ожидает вашего ввода" + } + }, + "th": { + "stringUnit": { + "state": "translated", + "value": "%@ ต้องการข้อมูลจากคุณ" + } + }, + "tr": { + "stringUnit": { + "state": "translated", + "value": "%@ girişinizi bekliyor" + } + }, + "uk": { + "stringUnit": { + "state": "translated", + "value": "%@ очікує на ваше введення" + } + }, + "zh-Hans": { + "stringUnit": { + "state": "translated", + "value": "%@ 需要您的输入" + } + }, + "zh-Hant": { + "stringUnit": { + "state": "translated", + "value": "%@ 需要您的輸入" + } + } + } + }, "amux.choice.cancel": { "extractionState": "manual", "localizations": { @@ -232132,4 +232489,4 @@ } }, "version": "1.0" -} \ No newline at end of file +} diff --git a/Sources/AmuxAgentAlarmPolicy.swift b/Sources/AmuxAgentAlarmPolicy.swift new file mode 100644 index 000000000..120ec1635 --- /dev/null +++ b/Sources/AmuxAgentAlarmPolicy.swift @@ -0,0 +1,93 @@ +import CmuxMuxa +import Foundation + +/// One alarm the agent-observation layer should surface for a state +/// transition — the payload ``AmuxAgentAlarmPolicy`` produces and the +/// notification pipeline renders. +struct AmuxAgentAlarm: Equatable { + /// Notification title (localized, carries the outcome). + let title: String + /// Notification body (the agent's last notification/prompt when known). + let body: String + /// Debounce key so one agent flapping through the same state doesn't + /// spam (scoped per agent session + entered state). + let cooldownKey: String + /// Debounce window in seconds. + let cooldownInterval: TimeInterval +} + +/// Maps muxad state transitions to user-facing alarms (plan §4.3: desktop +/// notifications for needs-input / needs-choice / error, plus the +/// work-finished edge) — pure and synchronous so the policy is unit-testable +/// apart from the notification pipeline. +/// +/// Works identically for the local daemon and forwarded remote daemons: the +/// transition's origin is irrelevant, only the edge matters. +struct AmuxAgentAlarmPolicy { + /// Seconds within which a repeat of the same agent+state edge is dropped. + static let cooldownInterval: TimeInterval = 20 + + /// The alarm `transition` warrants, or `nil` for quiet edges. + /// + /// Alarming edges: entering `waiting_input`/`waiting_choice` (blocked on + /// the user), entering `error`, and `working` → `idle` (the turn + /// finished). Everything else — starting, refreshes into the same state, + /// stops — is badge-only. + static func alarm(for transition: MuxaTransition) -> AmuxAgentAlarm? { + guard transition.from != transition.to else { return nil } + let agent = transition.agent + let title: String + let body: String + switch transition.to { + case .waitingInput, .waitingChoice: + title = String( + format: String( + localized: "amux.alarm.waiting.title", + defaultValue: "%@ needs your input" + ), + displayName(for: agent.kind) + ) + body = agent.lastNotification ?? agent.lastPrompt ?? "" + case .error: + title = String( + format: String( + localized: "amux.alarm.error.title", + defaultValue: "%@ hit an error" + ), + displayName(for: agent.kind) + ) + body = agent.lastNotification ?? agent.lastPrompt ?? "" + case .idle where transition.from == .working: + title = String( + format: String( + localized: "amux.alarm.finished.title", + defaultValue: "%@ finished" + ), + displayName(for: agent.kind) + ) + // A finished turn is described by what was asked, not by a stale + // needs-input notification from earlier in the turn. + body = agent.lastPrompt ?? "" + default: + return nil + } + return AmuxAgentAlarm( + title: title, + body: body, + cooldownKey: "amux.agent.\(agent.sessionId).\(transition.to.rawValue)", + cooldownInterval: Self.cooldownInterval + ) + } + + /// Human-readable product name for an agent kind (an unknown kind shows + /// its raw wire string — never localized, it's a product name). + static func displayName(for kind: MuxaAgentKind) -> String { + switch kind { + case .claudeCode: "Claude Code" + case .codex: "Codex" + case .geminiCli: "Gemini CLI" + case .opencode: "opencode" + case .unknown(let raw): raw + } + } +} diff --git a/Sources/AmuxAgentObservationHub.swift b/Sources/AmuxAgentObservationHub.swift new file mode 100644 index 000000000..561c16119 --- /dev/null +++ b/Sources/AmuxAgentObservationHub.swift @@ -0,0 +1,115 @@ +import Foundation + +/// Composes agent observation across daemons: the local muxad plus one +/// forwarded muxad per SSH host that currently has a live mirror. +/// +/// The hub owns lifecycle only — each ``AmuxAgentStatusService`` still does +/// its own snapshot/subscribe/join — and answers the cross-daemon questions +/// (attend target, a workspace's agent pane) by asking every service. +/// ``reconcile()`` runs on every mirror-topology change: a host gaining its +/// first mirror gets a forwarder + service, a host losing its last one gets +/// both torn down. +@MainActor +final class AmuxAgentObservationHub { + /// One remote host's observation pair: the socket forward and the + /// status service reading through it. + struct RemoteObserver { + let forwarder: AmuxRemoteMuxaForwarder + let service: AmuxAgentStatusService + } + + private let localService: AmuxAgentStatusService + /// The SSH hosts that should be observed right now (those with ≥1 live + /// mirror) — injected so tests drive reconcile without a controller. + private let sshHosts: @MainActor () -> [RemoteTmuxHost] + /// Builds a host's forwarder + service pair (composition-root injected: + /// the service needs host-scoped join closures and the alarm sink). + private let makeRemoteObserver: @MainActor (RemoteTmuxHost) -> RemoteObserver + + private var remoteObservers: [String: RemoteObserver] = [:] + private var started = false + + init( + localService: AmuxAgentStatusService, + sshHosts: @escaping @MainActor () -> [RemoteTmuxHost], + makeRemoteObserver: @escaping @MainActor (RemoteTmuxHost) -> RemoteObserver + ) { + self.localService = localService + self.sshHosts = sshHosts + self.makeRemoteObserver = makeRemoteObserver + } + + /// Starts local observation and reconciles remote observers (idempotent). + func start() { + guard !started else { return } + started = true + localService.start() + reconcile() + } + + /// Stops every service and forward. + func stop() { + started = false + localService.stop() + for (_, observer) in remoteObservers { + teardown(observer) + } + remoteObservers = [:] + } + + /// Diffs the wanted host set against running observers: starts a + /// forwarder + service for newly mirrored hosts, tears down observers + /// whose host has no mirror left. Safe to call spuriously. + func reconcile() { + guard started else { return } + let wanted = Dictionary( + sshHosts().map { ($0.id, $0) }, + uniquingKeysWith: { first, _ in first } + ) + for (hostId, observer) in remoteObservers where wanted[hostId] == nil { + teardown(observer) + remoteObservers[hostId] = nil + } + for (hostId, host) in wanted where remoteObservers[hostId] == nil { + let observer = makeRemoteObserver(host) + remoteObservers[hostId] = observer + observer.service.start() + } + } + + /// The number of remote hosts currently observed (diagnostics/tests). + var remoteObserverCount: Int { remoteObservers.count } + + /// The workspace (and tmux pane) of the agent blocked on the user the + /// longest across every observed daemon — the attend jump target. + func attendTarget() -> (workspace: Workspace, tmuxPane: Int?)? { + let candidates = ([localService] + remoteObservers.values.map(\.service)) + .compactMap { $0.attendCandidate() } + return candidates + .min { $0.blockedSince < $1.blockedSince } + .map { ($0.workspace, $0.tmuxPane) } + } + + /// The tmux pane of `workspaceId`'s most relevant agent, asking each + /// daemon in turn (a workspace mirrors exactly one host, so at most one + /// service resolves it). + func agentPane(inWorkspace workspaceId: UUID) -> Int? { + if let pane = localService.agentPane(inWorkspace: workspaceId) { + return pane + } + for observer in remoteObservers.values { + if let pane = observer.service.agentPane(inWorkspace: workspaceId) { + return pane + } + } + return nil + } + + private func teardown(_ observer: RemoteObserver) { + observer.service.stop() + // Synchronous (`stop()` is nonisolated) so the app-termination path + // tears the `ssh -N` down for certain — a scheduled actor hop might + // never run once the app is exiting. + observer.forwarder.stop() + } +} diff --git a/Sources/AmuxAgentStatusService.swift b/Sources/AmuxAgentStatusService.swift index 4159e40c2..e96c6f603 100644 --- a/Sources/AmuxAgentStatusService.swift +++ b/Sources/AmuxAgentStatusService.swift @@ -23,6 +23,13 @@ final class AmuxAgentStatusService { /// the fallback join while muxad rows carry no session name (v3 dropped /// `tmux_session` from the wire). private let workspaceForPane: @MainActor (Int) -> Workspace? + /// Runs before every connect attempt — the remote-host seam where the + /// hub re-establishes the SSH socket forward. `nil` for the local daemon. + private let prepareConnection: (@Sendable () async throws -> Void)? + /// Receives every live state transition together with the workspace it + /// resolved to (`nil` when the agent isn't in a mirrored session) — the + /// alarm/notification seam. Snapshot rows never fire this. + private let onTransition: (@MainActor (MuxaTransition, Workspace?) -> Void)? /// Live agent rows by muxad session id (agent-CLI session, not tmux). private var agentsBySessionId: [String: MuxaAgent] = [:] /// Weakly holds a workspace we wrote a row into. @@ -38,11 +45,15 @@ final class AmuxAgentStatusService { init( client: MuxaClient = MuxaClient(), workspaceForSession: @escaping @MainActor (String) -> Workspace?, - workspaceForPane: @escaping @MainActor (Int) -> Workspace? + workspaceForPane: @escaping @MainActor (Int) -> Workspace?, + prepareConnection: (@Sendable () async throws -> Void)? = nil, + onTransition: (@MainActor (MuxaTransition, Workspace?) -> Void)? = nil ) { self.client = client self.workspaceForSession = workspaceForSession self.workspaceForPane = workspaceForPane + self.prepareConnection = prepareConnection + self.onTransition = onTransition } /// Starts the snapshot+subscribe loop (idempotent). @@ -61,6 +72,7 @@ final class AmuxAgentStatusService { private func run() async { while !Task.isCancelled { do { + try await prepareConnection?() let agents = try await client.snapshot() agentsBySessionId = Dictionary( agents.map { ($0.sessionId, $0) }, @@ -70,6 +82,7 @@ final class AmuxAgentStatusService { for try await transition in try await client.transitions() { upsert(transition.agent) applyToWorkspaces() + onTransition?(transition, workspace(for: transition.agent)) } } catch { // muxad not running (or protocol failure): degrade silently. @@ -90,10 +103,29 @@ final class AmuxAgentStatusService { } } + /// The mirror workspace `agent` resolves to: session-name join first + /// (correct across servers), pane-id join as fallback. + private func workspace(for agent: MuxaAgent) -> Workspace? { + if let session = agent.tmuxSession { + return workspaceForSession(session) + } + if let pane = Self.paneNumber(agent.pane) { + return workspaceForPane(pane) + } + return nil + } + /// The workspace (and tmux pane, when known) of the agent that has been /// blocked on the user the longest — the attend jump target. `nil` when /// no tracked agent needs attention or none resolves to a workspace. func attendTarget() -> (workspace: Workspace, tmuxPane: Int?)? { + attendCandidate().map { ($0.workspace, $0.tmuxPane) } + } + + /// Like ``attendTarget()`` but carries *when* the winning agent got + /// blocked, so the observation hub can pick the longest-blocked agent + /// across several daemons (local + one per remote host). + func attendCandidate() -> (workspace: Workspace, tmuxPane: Int?, blockedSince: Date)? { let blocked = agentsBySessionId.values .filter { $0.state.needsAttention } .sorted { lhs, rhs in @@ -102,17 +134,12 @@ final class AmuxAgentStatusService { return l < r } for agent in blocked { - let pane = Self.paneNumber(agent.pane) - let workspace: Workspace? - if let session = agent.tmuxSession { - workspace = workspaceForSession(session) - } else if let pane { - workspace = workspaceForPane(pane) - } else { - workspace = nil - } - if let workspace { - return (workspace, pane) + if let workspace = workspace(for: agent) { + return ( + workspace, + Self.paneNumber(agent.pane), + agent.stateEnteredDate ?? agent.lastActivityDate ?? .distantPast + ) } } return nil @@ -125,13 +152,7 @@ final class AmuxAgentStatusService { func agentPane(inWorkspace workspaceId: UUID) -> Int? { let candidates = agentsBySessionId.values.compactMap { agent -> (agent: MuxaAgent, pane: Int)? in guard let pane = Self.paneNumber(agent.pane) else { return nil } - let workspace: Workspace? - if let session = agent.tmuxSession { - workspace = workspaceForSession(session) - } else { - workspace = workspaceForPane(pane) - } - guard workspace?.id == workspaceId else { return nil } + guard workspace(for: agent)?.id == workspaceId else { return nil } return (agent, pane) } let best = candidates.sorted { lhs, rhs in @@ -164,15 +185,7 @@ final class AmuxAgentStatusService { private func applyToWorkspaces() { var byWorkspace: [UUID: (workspace: Workspace, agents: [MuxaAgent])] = [:] for agent in agentsBySessionId.values { - let workspace: Workspace? - if let session = agent.tmuxSession { - workspace = workspaceForSession(session) - } else if let pane = Self.paneNumber(agent.pane) { - workspace = workspaceForPane(pane) - } else { - workspace = nil - } - guard let workspace else { continue } + guard let workspace = workspace(for: agent) else { continue } byWorkspace[workspace.id, default: (workspace, [])].agents.append(agent) } diff --git a/Sources/AmuxRemoteMuxaForwarder.swift b/Sources/AmuxRemoteMuxaForwarder.swift new file mode 100644 index 000000000..70fdcf2a2 --- /dev/null +++ b/Sources/AmuxRemoteMuxaForwarder.swift @@ -0,0 +1,143 @@ +import Darwin +import Foundation +import os + +/// Holds one SSH host's muxad socket forward open. +/// +/// Resolves the remote daemon socket path once (a one-shot probe over the +/// host's shared ControlMaster), then establishes an `ssh -N -L` unix-socket +/// forward so a local `MuxaClient` at ``localSocketPath`` reaches the remote +/// muxad. Because the forward multiplexes over the existing master, the +/// *master* ends up owning the listener: the `-N` mux client registers the +/// forwarding and exits immediately (observed OpenSSH mux behavior), so +/// liveness is judged by connecting to the local socket, never by the spawn +/// process. Observe-only and crash-tolerant: the owner's snapshot/subscribe +/// loop calls ``ensureForward()`` before every connect attempt, so a dead +/// forward (master closed, network drop, remote reboot) is re-established on +/// the loop's existing retry cadence. +actor AmuxRemoteMuxaForwarder { + /// The SSH host whose muxad socket this forward carries. + let host: RemoteTmuxHost + + /// The local unix-socket path clients connect to (the `-L` bind). + nonisolated var localSocketPath: String { host.muxaForwardSocketPath } + + /// Lock carve-out: an in-flight spawn must be terminable *synchronously* + /// from the app-termination path (an actor hop scheduled in + /// `applicationWillTerminate` may never run). `uncheckedState` because + /// `Process` isn't `Sendable`; the handle never escapes the lock's short + /// critical sections. + private nonisolated let processBox = OSAllocatedUnfairLock(uncheckedState: nil) + private var remoteSocketPath: String? + + /// Creates a forwarder for `host` (kind ``RemoteTmuxHostKind/ssh``). + init(host: RemoteTmuxHost) { + self.host = host + } + + /// Ensures a live forward, establishing one when needed (idempotent). + /// + /// A connectable local socket means the master already serves the + /// forward — re-requesting it would just stack another listener on the + /// master. Otherwise: probes the remote socket path on first use, + /// unlinks any stale local socket file, spawns the multiplexed + /// `ssh -N -L`, and waits briefly for the local end to accept + /// connections. Throws when the probe cannot resolve a remote path or + /// the process cannot launch; a forward that is slow to come up is not + /// an error (the caller's connect simply retries). + func ensureForward() async throws { + if Self.canConnect(unixSocketPath: localSocketPath) { return } + processBox.withLockUnchecked { $0 = nil } + + try host.ensureControlSocketDirectory() + let remotePath: String + if let remoteSocketPath { + remotePath = remoteSocketPath + } else { + let probe = try await RemoteTmuxSSHTransport.runProcess( + executable: "/usr/bin/ssh", + arguments: host.sshControlArguments(controlPersistSeconds: 180, batchMode: true) + + ["--", host.destination] + + [RemoteTmuxHost.remoteMuxaSocketProbeArgv + .map(RemoteTmuxHost.shellSingleQuoted) + .joined(separator: " ")] + ) + guard probe.exitCode == 0, + let resolved = RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: probe.stdout) + else { + throw RemoteTmuxError.commandFailed(exitCode: probe.exitCode, stderr: probe.stderr) + } + remoteSocketPath = resolved + remotePath = resolved + } + + // The previous forward's bind lingers as a dead socket file; ssh + // refuses to bind over it, so clear our own path before spawning. + try? FileManager.default.removeItem(atPath: localSocketPath) + + let argv = host.muxaForwardInvocation(remoteSocketPath: remotePath) + let forward = Process() + forward.executableURL = URL(fileURLWithPath: argv[0]) + forward.arguments = Array(argv.dropFirst()) + forward.standardInput = FileHandle.nullDevice + forward.standardOutput = FileHandle.nullDevice + forward.standardError = FileHandle.nullDevice + try forward.run() + processBox.withLockUnchecked { $0 = forward } + + // Bounded, cancellable readiness wait: the mux client registers the + // forward with the master and exits, offering no readiness signal on + // the success path (only ExitOnForwardFailure on failure), so briefly + // confirm connectability — same documented pattern as the status + // service's reconnect poll. Never an error: the caller's connect + // retries. Process exit is NOT a failure here (see the type doc). + for _ in 0..<10 { + if Self.canConnect(unixSocketPath: localSocketPath) { break } + try? await Task.sleep(for: .milliseconds(100)) + if Task.isCancelled { break } + } + } + + /// Tears the forward down: terminates an in-flight spawn and unlinks the + /// local socket file, which unreaches the master's listener (new + /// connections need the path; the orphaned listener fd dies with the + /// master when its ControlPersist window closes after the host's last + /// mirror detaches). + /// + /// `nonisolated` (synchronous, via the process-handle lock) so the + /// app-termination path and the hub's teardown can both call it without + /// an actor hop that might never be scheduled. + nonisolated func stop() { + let process = processBox.withLockUnchecked { handle -> Process? in + defer { handle = nil } + return handle + } + if let process, process.isRunning { + process.terminate() + } + try? FileManager.default.removeItem(atPath: localSocketPath) + } + + /// Whether a unix socket at `path` currently accepts a connection — the + /// forward-readiness check (a bind exists *and* ssh is serving it). + static func canConnect(unixSocketPath path: String) -> Bool { + let fd = socket(AF_UNIX, SOCK_STREAM, 0) + guard fd >= 0 else { return false } + defer { close(fd) } + var addr = sockaddr_un() + addr.sun_family = sa_family_t(AF_UNIX) + let pathBytes = Array(path.utf8) + let capacity = MemoryLayout.size(ofValue: addr.sun_path) - 1 + guard pathBytes.count <= capacity else { return false } + withUnsafeMutableBytes(of: &addr.sun_path) { raw in + raw.copyBytes(from: pathBytes) + } + let size = socklen_t(MemoryLayout.size) + let result = withUnsafePointer(to: &addr) { pointer in + pointer.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in + connect(fd, sa, size) + } + } + return result == 0 + } +} diff --git a/Sources/AppDelegate+AmuxLocalTmux.swift b/Sources/AppDelegate+AmuxLocalTmux.swift index 0a9c56492..cf58b3260 100644 --- a/Sources/AppDelegate+AmuxLocalTmux.swift +++ b/Sources/AppDelegate+AmuxLocalTmux.swift @@ -2,20 +2,69 @@ import AppKit import CmuxMuxa extension AppDelegate { - /// Builds the muxa agent-status service, joining muxad rows to amux - /// local-engine mirror workspaces (composition root wiring; the service - /// itself takes closures so it stays testable without an AppDelegate). - func makeAmuxAgentStatusService() -> AmuxAgentStatusService { - AmuxAgentStatusService( + /// Builds the agent-observation hub: the local muxad status service plus + /// a factory for per-SSH-host observers (socket forwarder + status + /// service joined against that host's mirrors). Composition-root wiring; + /// the hub and services take closures so they stay testable without an + /// AppDelegate. + func makeAmuxAgentObservationHub() -> AmuxAgentObservationHub { + let localService = AmuxAgentStatusService( workspaceForSession: { [weak self] sessionName in self?.remoteTmuxController.localMirrorWorkspace(sessionName: sessionName) }, workspaceForPane: { [weak self] paneId in self?.remoteTmuxController.localMirrorWorkspace(containingPane: paneId) + }, + onTransition: { [weak self] transition, workspace in + self?.amuxHandleAgentTransition(transition, workspace: workspace) + } + ) + return AmuxAgentObservationHub( + localService: localService, + sshHosts: { [weak self] in + self?.remoteTmuxController.activeSSHMirrorHosts() ?? [] + }, + makeRemoteObserver: { [weak self] host in + let forwarder = AmuxRemoteMuxaForwarder(host: host) + let hostId = host.id + let service = AmuxAgentStatusService( + client: MuxaClient(address: MuxaSocketAddress(path: forwarder.localSocketPath)), + workspaceForSession: { [weak self] sessionName in + self?.remoteTmuxController.mirrorWorkspace(hostId: hostId, sessionName: sessionName) + }, + workspaceForPane: { [weak self] paneId in + self?.remoteTmuxController.mirrorWorkspace(hostId: hostId, containingPane: paneId) + }, + prepareConnection: { try await forwarder.ensureForward() }, + onTransition: { [weak self] transition, workspace in + self?.amuxHandleAgentTransition(transition, workspace: workspace) + } + ) + return AmuxAgentObservationHub.RemoteObserver(forwarder: forwarder, service: service) } ) } + /// Routes an agent state transition into the notification pipeline: + /// alarming edges (needs input / choice, error, work finished — see + /// ``AmuxAgentAlarmPolicy``) become workspace-scoped notifications, so + /// remote agents alert on this Mac instead of only on the remote host. + /// Transitions that don't resolve to a mirrored workspace are dropped. + func amuxHandleAgentTransition(_ transition: MuxaTransition, workspace: Workspace?) { + guard let workspace, + let alarm = AmuxAgentAlarmPolicy.alarm(for: transition), + let notificationStore else { return } + notificationStore.addNotification( + tabId: workspace.id, + surfaceId: nil, + title: alarm.title, + subtitle: workspace.customTitle ?? "", + body: alarm.body, + cooldownKey: alarm.cooldownKey, + cooldownInterval: alarm.cooldownInterval + ) + } + /// Launch-time reconcile for the amux local engine: any session left on /// the dedicated local server (detach-by-default means the server /// outlives the app) is re-mirrored as a workspace, so a restart brings @@ -200,7 +249,7 @@ extension AppDelegate { func amuxSendPrompt(_ text: String, to workspace: Workspace) -> Bool { let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines) guard !trimmed.isEmpty else { return false } - let agentPane = amuxAgentStatusService.agentPane(inWorkspace: workspace.id) + let agentPane = amuxAgentObservation.agentPane(inWorkspace: workspace.id) return remoteTmuxController.sendPromptToMirror( workspaceId: workspace.id, tmuxPane: agentPane, @@ -243,7 +292,7 @@ extension AppDelegate { /// pane. The `workspace` should be the agent's mirror workspace. @MainActor func amuxPresentChoiceSheet(for workspace: Workspace) { - let agentPane = amuxAgentStatusService.agentPane(inWorkspace: workspace.id) + let agentPane = amuxAgentObservation.agentPane(inWorkspace: workspace.id) Task { @MainActor in let text = await remoteTmuxController.captureMirrorPaneText( workspaceId: workspace.id, @@ -299,7 +348,7 @@ extension AppDelegate { @discardableResult func amuxAttend() -> Bool { - guard let target = amuxAgentStatusService.attendTarget() else { return false } + guard let target = amuxAgentObservation.attendTarget() else { return false } guard let (manager, _) = amuxWorkspace(withId: target.workspace.id) else { return false } manager.selectWorkspace(target.workspace) if let pane = target.tmuxPane { diff --git a/Sources/AppDelegate.swift b/Sources/AppDelegate.swift index ec12c9d28..b48c1dd4e 100644 --- a/Sources/AppDelegate.swift +++ b/Sources/AppDelegate.swift @@ -511,9 +511,10 @@ final class AppDelegate: NSObject, NSApplicationDelegate, UNUserNotificationCent var aboutTitlebarDebugStore: AboutTitlebarDebugStore { debugWindowsCoordinator.aboutTitlebarStore } /// Coordinates remote tmux (`ssh … tmux -CC`) mirroring; composition-root owned. let remoteTmuxController = RemoteTmuxController() - /// Streams muxad agent state into mirror workspaces' sidebar rows; built - /// lazily so its closure can capture the fully-initialized delegate. - lazy var amuxAgentStatusService: AmuxAgentStatusService = makeAmuxAgentStatusService() + /// Streams muxad agent state (local daemon + one forwarded daemon per + /// mirrored SSH host) into mirror workspaces' sidebar rows; built lazily + /// so its closures can capture the fully-initialized delegate. + lazy var amuxAgentObservation: AmuxAgentObservationHub = makeAmuxAgentObservationHub() private static let reloadConfigurationMenuItemIdentifier = NSUserInterfaceItemIdentifier("com.cmux.reloadConfiguration") private static let cachedIsRunningUnderXCTest = detectRunningUnderXCTest(ProcessInfo.processInfo.environment) @@ -1547,7 +1548,10 @@ final class AppDelegate: NSObject, NSApplicationDelegate, UNUserNotificationCent } #endif reconcileLocalAmuxSessionsAtLaunch() - amuxAgentStatusService.start() + remoteTmuxController.mirrorTopologyDidChange = { [weak self] in + self?.amuxAgentObservation.reconcile() + } + amuxAgentObservation.start() // First-run onboarding (consent-gated; only offered once). Deferred a // beat so the main window is up before the modal appears. DispatchQueue.main.async { @@ -2010,6 +2014,9 @@ final class AppDelegate: NSObject, NSApplicationDelegate, UNUserNotificationCent sentryStopMemoryContextRefresh() // Plain quit detaches local ssh clients; explicit close already killed marked sessions. remoteTmuxController.detachAll() + // Tear down remote muxad observers: unlink their forward sockets and + // terminate any in-flight forward spawn so nothing outlives the app. + amuxAgentObservation.stop() // Best-effort presence goodbye; unclean exits are covered by the // service's missed-heartbeat timeout. PresenceHeartbeatClient.shared.appWillTerminate() diff --git a/Sources/RemoteTmuxController.swift b/Sources/RemoteTmuxController.swift index 68ba507a4..1aba845ac 100644 --- a/Sources/RemoteTmuxController.swift +++ b/Sources/RemoteTmuxController.swift @@ -258,7 +258,18 @@ final class RemoteTmuxController { /// Active session→workspace mirrors keyed `connectionHash\u{1}session` /// (see ``connectionKey(host:sessionName:)``). - private var sessionMirrors: [String: RemoteTmuxSessionMirror] = [:] + /// + /// The `didSet` fires on every mutation (add, remove, rekey) and pings + /// ``mirrorTopologyDidChange`` so the agent-observation layer can + /// reconcile which hosts it watches; observers must tolerate spurious + /// pings (the value may be unchanged in content). + private var sessionMirrors: [String: RemoteTmuxSessionMirror] = [:] { + didSet { mirrorTopologyDidChange?() } + } + + /// Invoked after any mirror add/remove/rekey — the hook the + /// agent-observation hub uses to start/stop per-host muxad observers. + var mirrorTopologyDidChange: (() -> Void)? /// Dedicated-window bindings (host↔window) and the in-flight-attach guard for /// the "one cmux window per remote endpoint" mirror mode (Option 1), owned by @@ -565,11 +576,41 @@ final class RemoteTmuxController { /// any — the join key between muxad agent rows (which carry the tmux /// session name) and sidebar workspaces. func localMirrorWorkspace(sessionName: String) -> Workspace? { + mirrorWorkspace(hostId: RemoteTmuxHost.amuxLocal().id, sessionName: sessionName) + } + + /// The workspace mirroring `sessionName` on the host identified by + /// `hostId` (``RemoteTmuxHost/id``) — the session-name join between a + /// host's muxad agent rows and its mirror workspaces, host-scoped so two + /// hosts with same-named sessions never cross-match. + func mirrorWorkspace(hostId: String, sessionName: String) -> Workspace? { sessionMirrors.values - .first { $0.host.kind == .localAmux && $0.sessionName == sessionName }? + .first { $0.host.id == hostId && $0.sessionName == sessionName }? .workspace } + /// The workspace whose mirrored session on host `hostId` currently + /// contains tmux pane `%paneId` — the pane-id join fallback, host-scoped + /// (pane ids are only unique per tmux server). + func mirrorWorkspace(hostId: String, containingPane paneId: Int) -> Workspace? { + sessionMirrors.values + .first { $0.host.id == hostId && $0.containsPane(paneId) }? + .workspace + } + + /// The distinct SSH hosts that currently have at least one live mirror — + /// the set of hosts whose remote muxad the observation hub should watch. + func activeSSHMirrorHosts() -> [RemoteTmuxHost] { + var seen = Set() + var hosts: [RemoteTmuxHost] = [] + for mirror in sessionMirrors.values where mirror.host.kind == .ssh { + if seen.insert(mirror.host.id).inserted { + hosts.append(mirror.host) + } + } + return hosts + } + /// Captures the visible text of `workspaceId`'s mirrored agent pane via /// `tmux capture-pane -p` (a one-shot read against the same server the /// control stream is attached to). Targets `tmuxPane` when given and @@ -613,9 +654,7 @@ final class RemoteTmuxController { /// another server sharing the id can mismatch — acceptable for a status /// badge; revisit when muxa regains session names on the wire. func localMirrorWorkspace(containingPane paneId: Int) -> Workspace? { - sessionMirrors.values - .first { $0.host.kind == .localAmux && $0.containsPane(paneId) }? - .workspace + mirrorWorkspace(hostId: RemoteTmuxHost.amuxLocal().id, containingPane: paneId) } /// The panel socket send/read should target inside `workspaceId` when diff --git a/Sources/RemoteTmuxHost+MuxaForward.swift b/Sources/RemoteTmuxHost+MuxaForward.swift new file mode 100644 index 000000000..f92ec1c43 --- /dev/null +++ b/Sources/RemoteTmuxHost+MuxaForward.swift @@ -0,0 +1,79 @@ +import Foundation + +/// muxad socket forwarding for SSH hosts: the local forward socket path and +/// the remote-side probe that locates (or predicts) the remote daemon socket. +/// +/// The agent-observation layer joins muxad rows to mirror workspaces, but +/// muxad listens on a unix socket only. For a remote host the daemon runs on +/// the remote machine, so amux forwards its socket over the host's existing +/// SSH ControlMaster (`ssh -N -L :`) and points a `MuxaClient` +/// at the local end. +extension RemoteTmuxHost { + /// The local unix-socket path amux binds as the forwarded end of this + /// host's remote muxad socket. + /// + /// Lives next to the ControlMaster socket under `~/.cmux/ssh/` and uses + /// the same slug + ``connectionHash`` scheme, so two distinct endpoints + /// never collide on one forward. The slug is trimmed to the AF_UNIX path + /// budget exactly like ``controlSocketPath`` (keeping the OpenSSH + /// transient-suffix reserve for headroom); the hash is never trimmed. + var muxaForwardSocketPath: String { + let home = FileManager.default.homeDirectoryForCurrentUser.path + let prefix = "\(home)/.cmux/ssh/muxa-" + let suffix = "-\(connectionHash).sock" + let fixedBytes = prefix.utf8.count + suffix.utf8.count + Self.opensshTransientSuffixLength + let slugBudget = max(0, Self.maxUnixSocketPathLength - fixedBytes) + return "\(prefix)\(Self.trimmedToUTF8ByteBudget(slug, slugBudget))\(suffix)" + } + + /// The remote argv that prints the muxad socket path on the remote host. + /// + /// Mirrors muxad's own bind rule (`$XDG_RUNTIME_DIR/muxa.sock`, else + /// `/tmp/muxa-.sock`): prefers a path where a socket actually exists + /// right now, and otherwise falls back to the path a default-configured + /// muxad *would* bind — so the forward can be established before the + /// daemon starts and begin working the moment it does. + static var remoteMuxaSocketProbeArgv: [String] { + let script = + "if [ -n \"${XDG_RUNTIME_DIR:-}\" ] && [ -S \"$XDG_RUNTIME_DIR/muxa.sock\" ]; then " + + "printf '%s\\n' \"$XDG_RUNTIME_DIR/muxa.sock\"; " + + "elif [ -S \"/tmp/muxa-$(id -u).sock\" ]; then " + + "printf '%s\\n' \"/tmp/muxa-$(id -u).sock\"; " + + "elif [ -n \"${XDG_RUNTIME_DIR:-}\" ]; then " + + "printf '%s\\n' \"$XDG_RUNTIME_DIR/muxa.sock\"; " + + "else printf '%s\\n' \"/tmp/muxa-$(id -u).sock\"; fi" + return ["/bin/sh", "-c", script] + } + + /// Parses the probe's stdout into the remote muxad socket path. + /// + /// The remote login shell may prepend banner/profile noise, so this takes + /// the *last* line that looks like an absolute `muxa` socket path rather + /// than trusting the whole output. `nil` when no line qualifies. + static func remoteMuxaSocketPath(fromProbeOutput output: String) -> String? { + output + .split(whereSeparator: \.isNewline) + .map { $0.trimmingCharacters(in: .whitespaces) } + .last { $0.hasPrefix("/") && $0.hasSuffix(".sock") && $0.contains("muxa") } + } + + /// The `ssh` argv (executable first) that holds this host's muxad socket + /// forward open: `-N` (no remote command) multiplexed over the shared + /// ControlMaster, `-L :` unix-socket forwarding, and + /// `ExitOnForwardFailure` so a failed bind surfaces as process exit + /// instead of a silent no-op forward. + func muxaForwardInvocation( + remoteSocketPath: String, + sshExecutablePath: String = "/usr/bin/ssh", + controlPersistSeconds: Int = 180 + ) -> [String] { + [sshExecutablePath] + + sshControlArguments(controlPersistSeconds: controlPersistSeconds, batchMode: true) + + [ + "-N", + "-o", "ExitOnForwardFailure=yes", + "-L", "\(muxaForwardSocketPath):\(remoteSocketPath)", + "--", destination, + ] + } +} diff --git a/Sources/RemoteTmuxHost.swift b/Sources/RemoteTmuxHost.swift index 054808113..4b2dada76 100644 --- a/Sources/RemoteTmuxHost.swift +++ b/Sources/RemoteTmuxHost.swift @@ -129,12 +129,12 @@ struct RemoteTmuxHost: Sendable, Equatable, Identifiable { /// macOS caps an AF_UNIX `sun_path` at 104 bytes (including the NUL /// terminator), so the usable path length is 103 bytes. - private static let maxUnixSocketPathLength = 103 + static let maxUnixSocketPathLength = 103 /// Bytes OpenSSH appends to `ControlPath` for its transient pre-rename bind /// socket: a `.` plus 16 random characters (see `mux.c`). The bound path must /// fit the AF_UNIX limit, not just the final renamed `ControlPath`. - private static let opensshTransientSuffixLength = 17 + static let opensshTransientSuffixLength = 17 /// Whether the path OpenSSH would actually bind for `controlPath` — i.e. /// `controlPath` plus its 17-byte transient suffix — fits the AF_UNIX limit. @@ -149,7 +149,7 @@ struct RemoteTmuxHost: Sendable, Equatable, Identifiable { /// encoding fits `byteBudget`. Trims on Character (not byte) boundaries so a /// multi-byte scalar is never split, and counts bytes (not characters) /// because the AF_UNIX limit is measured in bytes. - private static func trimmedToUTF8ByteBudget(_ value: String, _ byteBudget: Int) -> String { + static func trimmedToUTF8ByteBudget(_ value: String, _ byteBudget: Int) -> String { guard value.utf8.count > byteBudget else { return value } var result = "" var used = 0 diff --git a/Sources/RemoteTmuxSSHTransport.swift b/Sources/RemoteTmuxSSHTransport.swift index f76e60850..c7820524d 100644 --- a/Sources/RemoteTmuxSSHTransport.swift +++ b/Sources/RemoteTmuxSSHTransport.swift @@ -399,7 +399,7 @@ actor RemoteTmuxSSHTransport { /// We capture only the raw fds (`Int32`, `Sendable`) across the task /// boundary — never the non-`Sendable` `FileHandle` — and the `Pipe`s stay /// alive because `process` retains them until this function returns. - private static func runProcess( + static func runProcess( executable: String, arguments: [String] ) async throws -> RemoteTmuxCommandResult { diff --git a/Sources/TerminalController+RemoteTmux.swift b/Sources/TerminalController+RemoteTmux.swift index 12a1fc24f..45a038185 100644 --- a/Sources/TerminalController+RemoteTmux.swift +++ b/Sources/TerminalController+RemoteTmux.swift @@ -495,6 +495,39 @@ extension TerminalController { } } + /// `debug.amux.mirror_ssh` — DEBUG-only verification entry for the remote + /// agent-observation slice: mirror ONE named session on an SSH host as a + /// workspace (production `remote.tmux.mirror` mirrors every session on + /// the host, which an E2E against a shared machine must not do). Params: + /// `host` (required), `session` (required), optional `create` (Bool). + /// + /// Same focus posture as `debug.amux.mirror_local`: creates UI state on + /// the main actor, selects nothing, raises no window. + nonisolated func v2AmuxMirrorSSH(id: Any?, params: [String: Any]) -> String { + guard let host = Self.remoteTmuxHost(from: params) else { + return v2Error(id: id, code: "invalid_params", message: "host is required") + } + guard let session = (params["session"] as? String), !session.isEmpty else { + return v2Error(id: id, code: "invalid_params", message: "session is required") + } + let createIfMissing = (params["create"] as? Bool) ?? false + return v2VmCall(id: id, timeoutSeconds: 60) { + try await MainActor.run { + guard let appDelegate = AppDelegate.shared, + let manager = appDelegate.tabManager else { + throw RemoteTmuxError.unreachable("app not ready") + } + let mirrored = try appDelegate.remoteTmuxController.mirrorSession( + host: host, + sessionName: session, + createIfMissing: createIfMissing, + into: manager + ) + return ["host": host.destination, "session": session, "mirrored": mirrored] + } + } + } + /// `debug.amux.parse_choices` — DEBUG-only verification for the /// waiting_choice sheet's capture→parse chain (the sheet itself is a /// modal NSAlert that can't be driven headlessly). Captures the @@ -510,7 +543,7 @@ extension TerminalController { } return v2VmCall(id: id, timeoutSeconds: 15) { let agentPane = await MainActor.run { - AppDelegate.shared?.amuxAgentStatusService.agentPane(inWorkspace: workspaceId) + AppDelegate.shared?.amuxAgentObservation.agentPane(inWorkspace: workspaceId) } guard let controller = await MainActor.run(body: { AppDelegate.shared?.remoteTmuxController }) else { throw RemoteTmuxError.unreachable("app not ready") diff --git a/Sources/TerminalController.swift b/Sources/TerminalController.swift index a2f507f6d..6ea3e90cf 100644 --- a/Sources/TerminalController.swift +++ b/Sources/TerminalController.swift @@ -1406,6 +1406,8 @@ class TerminalController { return v2Result(id: request.id, v2DebugSidebarSimulateDrag(params: request.params)) case "debug.amux.mirror_local": return v2AmuxMirrorLocal(id: request.id, params: request.params) + case "debug.amux.mirror_ssh": + return v2AmuxMirrorSSH(id: request.id, params: request.params) case "debug.amux.parse_choices": return v2AmuxParseChoices(id: request.id, params: request.params) #endif diff --git a/cmux.xcodeproj/project.pbxproj b/cmux.xcodeproj/project.pbxproj index fa206a750..66ac4a5ab 100644 --- a/cmux.xcodeproj/project.pbxproj +++ b/cmux.xcodeproj/project.pbxproj @@ -72,12 +72,17 @@ A9F200000000000000000014 /* AgentSessionWebView.swift in Sources */ = {isa = PBXBuildFile; fileRef = A9F100000000000000000014 /* AgentSessionWebView.swift */; }; REE0CA0000000000000000F2 /* AIAccountCredentialSources.swift in Sources */ = {isa = PBXBuildFile; fileRef = REE0CA0000000000000000F1 /* AIAccountCredentialSources.swift */; }; REE0CA0000000000000000E2 /* AIAccountsClient.swift in Sources */ = {isa = PBXBuildFile; fileRef = REE0CA0000000000000000E1 /* AIAccountsClient.swift */; }; + 52AD40A1F31F245A72A92F9E /* AmuxAgentAlarmPolicy.swift in Sources */ = {isa = PBXBuildFile; fileRef = B2680ACD6796886000A02DBD /* AmuxAgentAlarmPolicy.swift */; }; + 22281B0DF8F14335041509F8 /* AmuxAgentAlarmPolicyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7AD19910C49AA84AD7A2EA6B /* AmuxAgentAlarmPolicyTests.swift */; }; + 27B00B5B8F6B98F75A211AB9 /* AmuxAgentObservationHub.swift in Sources */ = {isa = PBXBuildFile; fileRef = 6B043F5076DAEA5D7D7A2EFC /* AmuxAgentObservationHub.swift */; }; 1501F27F8F13705D6A45FE8B /* AmuxAgentStatusService.swift in Sources */ = {isa = PBXBuildFile; fileRef = FCBCA2F8FD0979D0A0220ACE /* AmuxAgentStatusService.swift */; }; CBE9ED114E0851AC45E68AA2 /* AmuxChoiceParser.swift in Sources */ = {isa = PBXBuildFile; fileRef = 633B3B33234501085E3589FE /* AmuxChoiceParser.swift */; }; 810D92864A73A66F234DEE62 /* AmuxChoiceParserTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 30B2B74D3A88D2117ABDF858 /* AmuxChoiceParserTests.swift */; }; F1D7EC2E03E780781BFDF619 /* AmuxMuxadLaunchAgent.swift in Sources */ = {isa = PBXBuildFile; fileRef = 406DA564A5BDD51A652E12C8 /* AmuxMuxadLaunchAgent.swift */; }; 7CF5DA6BDEBA29C857C55524 /* AmuxMuxadLaunchAgentTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = F763B6F21E006430645D4A60 /* AmuxMuxadLaunchAgentTests.swift */; }; + 20B556DFFD31982C1FBAD2AD /* AmuxMuxaForwardTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 19DF33891C136F2E0B0D44E0 /* AmuxMuxaForwardTests.swift */; }; F0D3873B704AB7DC7AF68D0A /* AmuxOnboarding.swift in Sources */ = {isa = PBXBuildFile; fileRef = DE47828B1FB148362FE46E65 /* AmuxOnboarding.swift */; }; + 6698709B89C493E14DE21A25 /* AmuxRemoteMuxaForwarder.swift in Sources */ = {isa = PBXBuildFile; fileRef = 65A547D2196B0A5ED47E1592 /* AmuxRemoteMuxaForwarder.swift */; }; F4350A110000000000000001 /* AppBundleIconPersistencePolicy.swift in Sources */ = {isa = PBXBuildFile; fileRef = F4350A130000000000000001 /* AppBundleIconPersistencePolicy.swift */; }; F4350A120000000000000001 /* AppBundleIconPersistencePolicy.swift in Sources */ = {isa = PBXBuildFile; fileRef = F4350A130000000000000001 /* AppBundleIconPersistencePolicy.swift */; }; 6F4D7FB8E98E231A6DA8F7A6 /* AppDelegate+AmuxLocalTmux.swift in Sources */ = {isa = PBXBuildFile; fileRef = E761B3D025E9AF34360B7369 /* AppDelegate+AmuxLocalTmux.swift */; }; @@ -915,6 +920,7 @@ E1B13CF7FE1A77324A142521 /* RemoteTmuxControlStreamParser.swift in Sources */ = {isa = PBXBuildFile; fileRef = FD5030BD9629431ACE9A4A74 /* RemoteTmuxControlStreamParser.swift */; }; B0555301B0555301B0555301 /* RemoteTmuxControlStreamParserBudgetTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = B0555302B0555302B0555302 /* RemoteTmuxControlStreamParserBudgetTests.swift */; }; 2FAA10BC0D7B50DE57F507A4 /* RemoteTmuxError.swift in Sources */ = {isa = PBXBuildFile; fileRef = AF44822DC080B2110CFECF88 /* RemoteTmuxError.swift */; }; + 11E3075A63892F2440FB9353 /* RemoteTmuxHost+MuxaForward.swift in Sources */ = {isa = PBXBuildFile; fileRef = 3364ABCFB2FA3644ADF93AB7 /* RemoteTmuxHost+MuxaForward.swift */; }; 240EDE51707EB251790C8985 /* RemoteTmuxHost.swift in Sources */ = {isa = PBXBuildFile; fileRef = 02B8A9858C9804AC9C37D7B4 /* RemoteTmuxHost.swift */; }; C57A929DB467A3245427BA83 /* RemoteTmuxHostKind.swift in Sources */ = {isa = PBXBuildFile; fileRef = 8D416EC6BF52FD9B9B9AF146 /* RemoteTmuxHostKind.swift */; }; F6724602A1B2C3D4E5F67246 /* RemoteTmuxHostRemoteCommandOverrideTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = F6724603A1B2C3D4E5F67246 /* RemoteTmuxHostRemoteCommandOverrideTests.swift */; }; @@ -1572,12 +1578,17 @@ A9F100000000000000000014 /* AgentSessionWebView.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Panels/AgentSessionWebView.swift; sourceTree = ""; }; REE0CA0000000000000000F1 /* AIAccountCredentialSources.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = AIAccountCredentialSources.swift; sourceTree = ""; }; REE0CA0000000000000000E1 /* AIAccountsClient.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = AIAccountsClient.swift; sourceTree = ""; }; + B2680ACD6796886000A02DBD /* AmuxAgentAlarmPolicy.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxAgentAlarmPolicy.swift"; sourceTree = ""; }; + 7AD19910C49AA84AD7A2EA6B /* AmuxAgentAlarmPolicyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxAgentAlarmPolicyTests.swift"; sourceTree = ""; }; + 6B043F5076DAEA5D7D7A2EFC /* AmuxAgentObservationHub.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxAgentObservationHub.swift"; sourceTree = ""; }; FCBCA2F8FD0979D0A0220ACE /* AmuxAgentStatusService.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxAgentStatusService.swift"; sourceTree = ""; }; 633B3B33234501085E3589FE /* AmuxChoiceParser.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxChoiceParser.swift"; sourceTree = ""; }; 30B2B74D3A88D2117ABDF858 /* AmuxChoiceParserTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxChoiceParserTests.swift"; sourceTree = ""; }; 406DA564A5BDD51A652E12C8 /* AmuxMuxadLaunchAgent.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxMuxadLaunchAgent.swift"; sourceTree = ""; }; F763B6F21E006430645D4A60 /* AmuxMuxadLaunchAgentTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxMuxadLaunchAgentTests.swift"; sourceTree = ""; }; + 19DF33891C136F2E0B0D44E0 /* AmuxMuxaForwardTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxMuxaForwardTests.swift"; sourceTree = ""; }; DE47828B1FB148362FE46E65 /* AmuxOnboarding.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxOnboarding.swift"; sourceTree = ""; }; + 65A547D2196B0A5ED47E1592 /* AmuxRemoteMuxaForwarder.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AmuxRemoteMuxaForwarder.swift"; sourceTree = ""; }; F4350A130000000000000001 /* AppBundleIconPersistencePolicy.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = App/AppBundleIconPersistencePolicy.swift; sourceTree = ""; }; E761B3D025E9AF34360B7369 /* AppDelegate+AmuxLocalTmux.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AppDelegate+AmuxLocalTmux.swift"; sourceTree = ""; }; C4A570010000000000000002 /* AppDelegate+CanvasShortcutRouting.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "AppDelegate+CanvasShortcutRouting.swift"; sourceTree = ""; }; @@ -2353,6 +2364,7 @@ FD5030BD9629431ACE9A4A74 /* RemoteTmuxControlStreamParser.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RemoteTmuxControlStreamParser.swift; sourceTree = ""; }; B0555302B0555302B0555302 /* RemoteTmuxControlStreamParserBudgetTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RemoteTmuxControlStreamParserBudgetTests.swift; sourceTree = ""; }; AF44822DC080B2110CFECF88 /* RemoteTmuxError.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RemoteTmuxError.swift; sourceTree = ""; }; + 3364ABCFB2FA3644ADF93AB7 /* RemoteTmuxHost+MuxaForward.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "RemoteTmuxHost+MuxaForward.swift"; sourceTree = ""; }; 02B8A9858C9804AC9C37D7B4 /* RemoteTmuxHost.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RemoteTmuxHost.swift; sourceTree = ""; }; 8D416EC6BF52FD9B9B9AF146 /* RemoteTmuxHostKind.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "RemoteTmuxHostKind.swift"; sourceTree = ""; }; F6724603A1B2C3D4E5F67246 /* RemoteTmuxHostRemoteCommandOverrideTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RemoteTmuxHostRemoteCommandOverrideTests.swift; sourceTree = ""; }; @@ -3837,6 +3849,10 @@ 406DA564A5BDD51A652E12C8 /* AmuxMuxadLaunchAgent.swift */, 633B3B33234501085E3589FE /* AmuxChoiceParser.swift */, FCBCA2F8FD0979D0A0220ACE /* AmuxAgentStatusService.swift */, + B2680ACD6796886000A02DBD /* AmuxAgentAlarmPolicy.swift */, + 6B043F5076DAEA5D7D7A2EFC /* AmuxAgentObservationHub.swift */, + 65A547D2196B0A5ED47E1592 /* AmuxRemoteMuxaForwarder.swift */, + 3364ABCFB2FA3644ADF93AB7 /* RemoteTmuxHost+MuxaForward.swift */, E1C0FE8EE0A38F9F2FE7FF17 /* RemoteTmuxLocalTransport.swift */, E0FACC363EAAC64617102784 /* RemoteTmuxTransport.swift */, E761B3D025E9AF34360B7369 /* AppDelegate+AmuxLocalTmux.swift */, @@ -4301,6 +4317,8 @@ 42092CDB2109E250F7F2A76E /* TabManagerUnitTests.swift */, F763B6F21E006430645D4A60 /* AmuxMuxadLaunchAgentTests.swift */, 30B2B74D3A88D2117ABDF858 /* AmuxChoiceParserTests.swift */, + 7AD19910C49AA84AD7A2EA6B /* AmuxAgentAlarmPolicyTests.swift */, + 19DF33891C136F2E0B0D44E0 /* AmuxMuxaForwardTests.swift */, C0DE1A080000000000000002 /* SavedLayoutDefinitionTests.swift */, C0DE1A070000000000000002 /* SavedLayoutStoreTests.swift */, 5830CA11BAC0000000000001 /* SocketCallbackAwaiterMainThreadTests.swift */, @@ -4894,10 +4912,13 @@ A9F200000000000000000014 /* AgentSessionWebView.swift in Sources */, REE0CA0000000000000000F2 /* AIAccountCredentialSources.swift in Sources */, REE0CA0000000000000000E2 /* AIAccountsClient.swift in Sources */, + 52AD40A1F31F245A72A92F9E /* AmuxAgentAlarmPolicy.swift in Sources */, + 27B00B5B8F6B98F75A211AB9 /* AmuxAgentObservationHub.swift in Sources */, 1501F27F8F13705D6A45FE8B /* AmuxAgentStatusService.swift in Sources */, CBE9ED114E0851AC45E68AA2 /* AmuxChoiceParser.swift in Sources */, F1D7EC2E03E780781BFDF619 /* AmuxMuxadLaunchAgent.swift in Sources */, F0D3873B704AB7DC7AF68D0A /* AmuxOnboarding.swift in Sources */, + 6698709B89C493E14DE21A25 /* AmuxRemoteMuxaForwarder.swift in Sources */, F4350A110000000000000001 /* AppBundleIconPersistencePolicy.swift in Sources */, 6F4D7FB8E98E231A6DA8F7A6 /* AppDelegate+AmuxLocalTmux.swift in Sources */, C4A570010000000000000001 /* AppDelegate+CanvasShortcutRouting.swift in Sources */, @@ -5369,6 +5390,7 @@ AFEC676A8FB901E750BA5641 /* RemoteTmuxControlPipeWriter.swift in Sources */, E1B13CF7FE1A77324A142521 /* RemoteTmuxControlStreamParser.swift in Sources */, 2FAA10BC0D7B50DE57F507A4 /* RemoteTmuxError.swift in Sources */, + 11E3075A63892F2440FB9353 /* RemoteTmuxHost+MuxaForward.swift in Sources */, 240EDE51707EB251790C8985 /* RemoteTmuxHost.swift in Sources */, C57A929DB467A3245427BA83 /* RemoteTmuxHostKind.swift in Sources */, 00B223273DE0DBED98437C7A /* RemoteTmuxLayoutContainer.swift in Sources */, @@ -5859,8 +5881,10 @@ D3610B020000000000000001 /* AgentSessionAutoResumeSwiftTests.swift in Sources */, A9E030000000000000000002 /* AgentSessionSocketSurfaceTests.swift in Sources */, A9E050000000000000000002 /* AgentSessionWebRendererTests.swift in Sources */, + 22281B0DF8F14335041509F8 /* AmuxAgentAlarmPolicyTests.swift in Sources */, 810D92864A73A66F234DEE62 /* AmuxChoiceParserTests.swift in Sources */, 7CF5DA6BDEBA29C857C55524 /* AmuxMuxadLaunchAgentTests.swift in Sources */, + 20B556DFFD31982C1FBAD2AD /* AmuxMuxaForwardTests.swift in Sources */, 725746692D9647948561044D /* AppDelegateBareSpaceShortcutRoutingTests.swift in Sources */, E3309A09 /* AppDelegateEqualizeSplitsShortcutTests.swift in Sources */, F5996001A1B2C3D4E5F60001 /* AppDelegateFileExplorerShortcutRoutingTests.swift in Sources */, diff --git a/cmuxTests/AmuxAgentAlarmPolicyTests.swift b/cmuxTests/AmuxAgentAlarmPolicyTests.swift new file mode 100644 index 000000000..5e2f27cb5 --- /dev/null +++ b/cmuxTests/AmuxAgentAlarmPolicyTests.swift @@ -0,0 +1,109 @@ +import CmuxMuxa +import XCTest + +#if canImport(cmux_DEV) +@testable import cmux_DEV +#elseif canImport(cmux) +@testable import cmux +#endif + +final class AmuxAgentAlarmPolicyTests: XCTestCase { + private func agent( + state: CmuxMuxa.MuxaAgentState, + kind: CmuxMuxa.MuxaAgentKind = .claudeCode, + lastNotification: String? = nil, + lastPrompt: String? = nil + ) -> CmuxMuxa.MuxaAgent { + CmuxMuxa.MuxaAgent( + kind: kind, + sessionId: "session-1", + state: state, + lastPrompt: lastPrompt, + lastNotification: lastNotification + ) + } + + func testEnteringWaitingInputAlarms() { + let alarm = AmuxAgentAlarmPolicy.alarm(for: .init( + from: .working, + to: .waitingInput, + agent: agent(state: .waitingInput, lastNotification: "Permission needed") + )) + XCTAssertNotNil(alarm) + XCTAssertTrue(alarm?.title.contains("Claude Code") == true) + XCTAssertEqual(alarm?.body, "Permission needed") + XCTAssertEqual(alarm?.cooldownKey, "amux.agent.session-1.waiting_input") + } + + func testEnteringWaitingChoiceAlarms() { + XCTAssertNotNil(AmuxAgentAlarmPolicy.alarm(for: .init( + from: .working, + to: .waitingChoice, + agent: agent(state: .waitingChoice) + ))) + } + + func testEnteringErrorAlarms() { + XCTAssertNotNil(AmuxAgentAlarmPolicy.alarm(for: .init( + from: .working, + to: .error, + agent: agent(state: .error) + ))) + } + + func testWorkingToIdleAlarmsAsFinished() { + let alarm = AmuxAgentAlarmPolicy.alarm(for: .init( + from: .working, + to: .idle, + agent: agent(state: .idle, lastPrompt: "refactor the parser") + )) + XCTAssertNotNil(alarm) + XCTAssertEqual(alarm?.body, "refactor the parser") + } + + func testFinishedBodyPrefersPromptOverStaleNotification() { + let alarm = AmuxAgentAlarmPolicy.alarm(for: .init( + from: .working, + to: .idle, + agent: agent( + state: .idle, + lastNotification: "Claude needs your permission", + lastPrompt: "refactor the parser" + ) + )) + XCTAssertEqual(alarm?.body, "refactor the parser") + } + + func testStartingToIdleIsQuiet() { + XCTAssertNil(AmuxAgentAlarmPolicy.alarm(for: .init( + from: .starting, + to: .idle, + agent: agent(state: .idle) + ))) + } + + func testSameStateRefreshIsQuiet() { + XCTAssertNil(AmuxAgentAlarmPolicy.alarm(for: .init( + from: .waitingInput, + to: .waitingInput, + agent: agent(state: .waitingInput) + ))) + } + + func testWorkingAndStoppedAreQuiet() { + XCTAssertNil(AmuxAgentAlarmPolicy.alarm(for: .init( + from: .idle, + to: .working, + agent: agent(state: .working) + ))) + XCTAssertNil(AmuxAgentAlarmPolicy.alarm(for: .init( + from: .working, + to: .stopped, + agent: agent(state: .stopped) + ))) + } + + func testUnknownAgentKindUsesRawWireName() { + XCTAssertEqual(AmuxAgentAlarmPolicy.displayName(for: .unknown("aider")), "aider") + } +} diff --git a/cmuxTests/AmuxMuxaForwardTests.swift b/cmuxTests/AmuxMuxaForwardTests.swift new file mode 100644 index 000000000..9a7b1484b --- /dev/null +++ b/cmuxTests/AmuxMuxaForwardTests.swift @@ -0,0 +1,94 @@ +import XCTest + +#if canImport(cmux_DEV) +@testable import cmux_DEV +#elseif canImport(cmux) +@testable import cmux +#endif + +final class AmuxMuxaForwardTests: XCTestCase { + func testForwardSocketPathIsHostUniqueAndBounded() { + let a = RemoteTmuxHost(destination: "alice@gpu-box") + let b = RemoteTmuxHost(destination: "alice@other-box") + XCTAssertNotEqual(a.muxaForwardSocketPath, b.muxaForwardSocketPath) + XCTAssertTrue(a.muxaForwardSocketPath.contains("/.cmux/ssh/muxa-")) + XCTAssertTrue(a.muxaForwardSocketPath.hasSuffix("-\(a.connectionHash).sock")) + XCTAssertTrue( + RemoteTmuxHost.controlSocketPathFitsUnixLimit(a.muxaForwardSocketPath), + "forward socket path must fit the AF_UNIX limit" + ) + } + + func testForwardSocketPathBoundedForLongDestination() { + let long = RemoteTmuxHost( + destination: "some-extremely-long-user-name@a-very-long-hostname.internal.example.company.network.example.com" + ) + XCTAssertTrue(RemoteTmuxHost.controlSocketPathFitsUnixLimit(long.muxaForwardSocketPath)) + XCTAssertTrue(long.muxaForwardSocketPath.hasSuffix("-\(long.connectionHash).sock")) + } + + func testForwardSocketPathDiffersFromControlSocketPath() { + let host = RemoteTmuxHost(destination: "alice@gpu-box") + XCTAssertNotEqual(host.muxaForwardSocketPath, host.controlSocketPath) + } + + func testProbeOutputParserPicksSocketPathLine() { + let output = """ + Welcome to gpu-box! Message of the day. + /run/user/1000/muxa.sock + """ + XCTAssertEqual( + RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: output), + "/run/user/1000/muxa.sock" + ) + } + + func testProbeOutputParserAcceptsTmpFallbackPath() { + XCTAssertEqual( + RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: "/tmp/muxa-501.sock\n"), + "/tmp/muxa-501.sock" + ) + } + + func testProbeOutputParserRejectsNoise() { + XCTAssertNil(RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: "")) + XCTAssertNil(RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: "bash: no such file\n")) + XCTAssertNil(RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: "muxa.sock")) + XCTAssertNil(RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: "/etc/passwd")) + } + + func testProbeOutputParserTakesLastCandidate() { + let output = """ + /tmp/muxa-0.sock + /run/user/1000/muxa.sock + """ + XCTAssertEqual( + RemoteTmuxHost.remoteMuxaSocketPath(fromProbeOutput: output), + "/run/user/1000/muxa.sock" + ) + } + + func testForwardInvocationShape() { + let host = RemoteTmuxHost(destination: "alice@gpu-box") + let argv = host.muxaForwardInvocation(remoteSocketPath: "/run/user/1000/muxa.sock") + XCTAssertEqual(argv.first, "/usr/bin/ssh") + XCTAssertTrue(argv.contains("-N")) + XCTAssertTrue(argv.contains("ExitOnForwardFailure=yes")) + guard let lIndex = argv.firstIndex(of: "-L") else { + return XCTFail("missing -L") + } + XCTAssertEqual(argv[lIndex + 1], "\(host.muxaForwardSocketPath):/run/user/1000/muxa.sock") + // `--` must guard the destination so a dash-prefixed destination can + // never be parsed as an ssh option. + XCTAssertEqual(Array(argv.suffix(2)), ["--", "alice@gpu-box"]) + XCTAssertTrue(argv.contains("BatchMode=yes")) + } + + func testCanConnectIsFalseForMissingSocket() { + XCTAssertFalse( + AmuxRemoteMuxaForwarder.canConnect( + unixSocketPath: NSTemporaryDirectory() + "cmux-test-nonexistent-\(UUID().uuidString).sock" + ) + ) + } +}