From 73a6fa44948894bd9f3dc065ddddfcd0cf49e0ce Mon Sep 17 00:00:00 2001 From: Jiun Bae Date: Thu, 9 Jul 2026 10:37:25 +0900 Subject: [PATCH 1/4] =?UTF-8?q?feat(amux):=20observe=20remote=20muxad=20ov?= =?UTF-8?q?er=20SSH=20=E2=80=94=20per-host=20badges=20+=20agent=20alarms?= =?UTF-8?q?=20(Phase=205)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mirroring half already handled remote hosts, but agent observation was local-only: every muxad join was scoped to the .localAmux engine, so an ssh->tmux->claude workflow showed the terminal without any agent state. - RemoteTmuxHost+MuxaForward: local forward socket path (AF_UNIX budget, slug+connectionHash scheme), remote muxad socket probe (mirrors muxad's XDG/tmp bind rule) + stdout parser, ssh -N -L invocation builder. - AmuxRemoteMuxaForwarder: actor holding one host's muxad socket forward open over the existing SSH ControlMaster; respawned lazily by the status service's reconnect loop via prepareConnection. - AmuxAgentObservationHub: composes the local status service with one forwarder+service pair per SSH host that has a live mirror; reconciles on RemoteTmuxController.mirrorTopologyDidChange (sessionMirrors didSet) and aggregates attendTarget/agentPane across daemons. - RemoteTmuxController: host-scoped joins mirrorWorkspace(hostId:...) that the localAmux joins now delegate to, plus activeSSHMirrorHosts(). - AmuxAgentAlarmPolicy: waiting_input/waiting_choice/error entry and working->idle (finished) transitions become workspace-scoped cmux notifications with per-agent-state cooldown — remote agents now alert on this Mac instead of only on the remote host; local agents gain the same completion alarm. - debug.amux.mirror_ssh: DEBUG-only single-session SSH mirror RPC so E2E against a shared host doesn't mirror every session like remote.tmux.mirror. - Localization: 3 new alarm title keys x 19 locales. - Tests: AmuxMuxaForwardTests (9) + AmuxAgentAlarmPolicyTests (8), wired into cmux.xcodeproj, all passing via cmux-unit. Co-Authored-By: Claude Fable 5 --- .../Wire/ControlCommandExecutionPolicy.swift | 3 + Resources/Localizable.xcstrings | 359 +++++++++++++++++- Sources/AmuxAgentAlarmPolicy.swift | 87 +++++ Sources/AmuxAgentObservationHub.swift | 113 ++++++ Sources/AmuxAgentStatusService.swift | 69 ++-- Sources/AmuxRemoteMuxaForwarder.swift | 118 ++++++ Sources/AppDelegate+AmuxLocalTmux.swift | 65 +++- Sources/AppDelegate.swift | 12 +- Sources/RemoteTmuxController.swift | 49 ++- Sources/RemoteTmuxHost+MuxaForward.swift | 79 ++++ Sources/RemoteTmuxHost.swift | 6 +- Sources/RemoteTmuxSSHTransport.swift | 2 +- Sources/TerminalController+RemoteTmux.swift | 35 +- Sources/TerminalController.swift | 2 + cmux.xcodeproj/project.pbxproj | 24 ++ cmuxTests/AmuxAgentAlarmPolicyTests.swift | 96 +++++ cmuxTests/AmuxMuxaForwardTests.swift | 94 +++++ 17 files changed, 1162 insertions(+), 51 deletions(-) create mode 100644 Sources/AmuxAgentAlarmPolicy.swift create mode 100644 Sources/AmuxAgentObservationHub.swift create mode 100644 Sources/AmuxRemoteMuxaForwarder.swift create mode 100644 Sources/RemoteTmuxHost+MuxaForward.swift create mode 100644 cmuxTests/AmuxAgentAlarmPolicyTests.swift create mode 100644 cmuxTests/AmuxMuxaForwardTests.swift 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..2254e84a5 --- /dev/null +++ b/Sources/AmuxAgentAlarmPolicy.swift @@ -0,0 +1,87 @@ +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 + switch transition.to { + case .waitingInput, .waitingChoice: + title = String( + format: String( + localized: "amux.alarm.waiting.title", + defaultValue: "%@ needs your input" + ), + displayName(for: agent.kind) + ) + case .error: + title = String( + format: String( + localized: "amux.alarm.error.title", + defaultValue: "%@ hit an error" + ), + displayName(for: agent.kind) + ) + case .idle where transition.from == .working: + title = String( + format: String( + localized: "amux.alarm.finished.title", + defaultValue: "%@ finished" + ), + displayName(for: agent.kind) + ) + default: + return nil + } + return AmuxAgentAlarm( + title: title, + body: agent.lastNotification ?? agent.lastPrompt ?? "", + 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..e593e4a8d --- /dev/null +++ b/Sources/AmuxAgentObservationHub.swift @@ -0,0 +1,113 @@ +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() + let forwarder = observer.forwarder + Task { await 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..982499f4f --- /dev/null +++ b/Sources/AmuxRemoteMuxaForwarder.swift @@ -0,0 +1,118 @@ +import Darwin +import Foundation + +/// 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 keeps an `ssh -N -L` unix-socket +/// forward process alive so a local `MuxaClient` at ``localSocketPath`` +/// reaches the remote muxad. 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 respawned 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 } + + private var process: Process? + private var remoteSocketPath: String? + + /// Creates a forwarder for `host` (kind ``RemoteTmuxHostKind/ssh``). + init(host: RemoteTmuxHost) { + self.host = host + } + + /// Ensures a live forward process, spawning one when needed (idempotent). + /// + /// 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 let process, process.isRunning { return } + process = 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() + process = forward + + // Bounded, cancellable readiness wait: ssh binds the local socket a + // few ms after launch and offers 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. + for _ in 0..<10 { + if !forward.isRunning { break } + if Self.canConnect(unixSocketPath: localSocketPath) { break } + try? await Task.sleep(for: .milliseconds(100)) + if Task.isCancelled { break } + } + } + + /// Terminates the forward process and removes the local socket file. + func stop() { + process?.terminationHandler = nil + process?.terminate() + process = nil + 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..5c94d61a3 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 { 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..35ef29e70 --- /dev/null +++ b/cmuxTests/AmuxAgentAlarmPolicyTests.swift @@ -0,0 +1,96 @@ +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 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" + ) + ) + } +} From 951b51452a09c3488554096d65d909afddad7b08 Mon Sep 17 00:00:00 2001 From: Jiun Bae Date: Thu, 9 Jul 2026 10:58:02 +0900 Subject: [PATCH 2/4] fix(amux): finished alarm body prefers the prompt over a stale needs-input notification Verified live in the ssh-localhost E2E: the working->idle alarm was showing the earlier permission-request message instead of what the turn was about. Co-Authored-By: Claude Fable 5 --- Sources/AmuxAgentAlarmPolicy.swift | 8 +++++++- cmuxTests/AmuxAgentAlarmPolicyTests.swift | 13 +++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/Sources/AmuxAgentAlarmPolicy.swift b/Sources/AmuxAgentAlarmPolicy.swift index 2254e84a5..120ec1635 100644 --- a/Sources/AmuxAgentAlarmPolicy.swift +++ b/Sources/AmuxAgentAlarmPolicy.swift @@ -37,6 +37,7 @@ struct AmuxAgentAlarmPolicy { 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( @@ -46,6 +47,7 @@ struct AmuxAgentAlarmPolicy { ), displayName(for: agent.kind) ) + body = agent.lastNotification ?? agent.lastPrompt ?? "" case .error: title = String( format: String( @@ -54,6 +56,7 @@ struct AmuxAgentAlarmPolicy { ), displayName(for: agent.kind) ) + body = agent.lastNotification ?? agent.lastPrompt ?? "" case .idle where transition.from == .working: title = String( format: String( @@ -62,12 +65,15 @@ struct AmuxAgentAlarmPolicy { ), 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: agent.lastNotification ?? agent.lastPrompt ?? "", + body: body, cooldownKey: "amux.agent.\(agent.sessionId).\(transition.to.rawValue)", cooldownInterval: Self.cooldownInterval ) diff --git a/cmuxTests/AmuxAgentAlarmPolicyTests.swift b/cmuxTests/AmuxAgentAlarmPolicyTests.swift index 35ef29e70..5e2f27cb5 100644 --- a/cmuxTests/AmuxAgentAlarmPolicyTests.swift +++ b/cmuxTests/AmuxAgentAlarmPolicyTests.swift @@ -61,6 +61,19 @@ final class AmuxAgentAlarmPolicyTests: XCTestCase { 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, From 18dc725723132ca1b749b195c2e85353747e1e0d Mon Sep 17 00:00:00 2001 From: Jiun Bae Date: Thu, 9 Jul 2026 11:02:25 +0900 Subject: [PATCH 3/4] fix(amux): tear the muxad forward down synchronously on app termination An ssh -N mux client left behind by a quit pins the host's ControlMaster past its ControlPersist window (an attached session stops the countdown), so applicationWillTerminate now stops the observation hub, and the forwarder's stop() is nonisolated over an OSAllocatedUnfairLock-held process handle (lock carve-out: an actor hop scheduled during termination may never run). Co-Authored-By: Claude Fable 5 --- Sources/AmuxAgentObservationHub.swift | 6 ++++-- Sources/AmuxRemoteMuxaForwarder.swift | 31 ++++++++++++++++++++------- Sources/AppDelegate.swift | 4 ++++ 3 files changed, 31 insertions(+), 10 deletions(-) diff --git a/Sources/AmuxAgentObservationHub.swift b/Sources/AmuxAgentObservationHub.swift index e593e4a8d..561c16119 100644 --- a/Sources/AmuxAgentObservationHub.swift +++ b/Sources/AmuxAgentObservationHub.swift @@ -107,7 +107,9 @@ final class AmuxAgentObservationHub { private func teardown(_ observer: RemoteObserver) { observer.service.stop() - let forwarder = observer.forwarder - Task { await forwarder.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/AmuxRemoteMuxaForwarder.swift b/Sources/AmuxRemoteMuxaForwarder.swift index 982499f4f..929259564 100644 --- a/Sources/AmuxRemoteMuxaForwarder.swift +++ b/Sources/AmuxRemoteMuxaForwarder.swift @@ -1,5 +1,6 @@ import Darwin import Foundation +import os /// Holds one SSH host's muxad socket forward open. /// @@ -17,7 +18,13 @@ actor AmuxRemoteMuxaForwarder { /// The local unix-socket path clients connect to (the `-L` bind). nonisolated var localSocketPath: String { host.muxaForwardSocketPath } - private var process: Process? + /// Lock carve-out: the forward process handle must be terminable + /// *synchronously* from the app-termination path (an actor hop scheduled + /// in `applicationWillTerminate` may never run, orphaning an `ssh -N` + /// that pins the ControlMaster past its ControlPersist window). + /// `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``). @@ -33,8 +40,8 @@ actor AmuxRemoteMuxaForwarder { /// 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 let process, process.isRunning { return } - process = nil + if processBox.withLockUnchecked({ $0?.isRunning == true }) { return } + processBox.withLockUnchecked { $0 = nil } try host.ensureControlSocketDirectory() let remotePath: String @@ -70,7 +77,7 @@ actor AmuxRemoteMuxaForwarder { forward.standardOutput = FileHandle.nullDevice forward.standardError = FileHandle.nullDevice try forward.run() - process = forward + processBox.withLockUnchecked { $0 = forward } // Bounded, cancellable readiness wait: ssh binds the local socket a // few ms after launch and offers no readiness signal on the success @@ -86,10 +93,18 @@ actor AmuxRemoteMuxaForwarder { } /// Terminates the forward process and removes the local socket file. - func stop() { - process?.terminationHandler = nil - process?.terminate() - process = nil + /// + /// `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) } diff --git a/Sources/AppDelegate.swift b/Sources/AppDelegate.swift index 5c94d61a3..40c0bed6f 100644 --- a/Sources/AppDelegate.swift +++ b/Sources/AppDelegate.swift @@ -2014,6 +2014,10 @@ 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 so their `ssh -N` forward clients + // don't outlive the app (an attached mux client pins the ControlMaster + // past its ControlPersist window). + amuxAgentObservation.stop() // Best-effort presence goodbye; unclean exits are covered by the // service's missed-heartbeat timeout. PresenceHeartbeatClient.shared.appWillTerminate() From 9c9247c6e3a8ccc4700c8ac0efd044cf0cd35669 Mon Sep 17 00:00:00 2001 From: Jiun Bae Date: Thu, 9 Jul 2026 15:33:06 +0900 Subject: [PATCH 4/4] fix(amux): judge forward liveness by the socket, not the mux client process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Observed against OpenSSH: a multiplexed `ssh -N -L` registers the forward with the ControlMaster and exits — the master owns the listener. So ensureForward() now treats a connectable local socket as 'already forwarded' (re-requesting would stack listeners on the master), the readiness wait no longer reads process exit as failure, and stop()/the app-termination hook are documented as unlink-the-path teardown (the master's orphaned listener fd dies with the master's ControlPersist window). Co-Authored-By: Claude Fable 5 --- Sources/AmuxRemoteMuxaForwarder.swift | 62 ++++++++++++++++----------- Sources/AppDelegate.swift | 5 +-- 2 files changed, 38 insertions(+), 29 deletions(-) diff --git a/Sources/AmuxRemoteMuxaForwarder.swift b/Sources/AmuxRemoteMuxaForwarder.swift index 929259564..70fdcf2a2 100644 --- a/Sources/AmuxRemoteMuxaForwarder.swift +++ b/Sources/AmuxRemoteMuxaForwarder.swift @@ -5,12 +5,16 @@ 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 keeps an `ssh -N -L` unix-socket -/// forward process alive so a local `MuxaClient` at ``localSocketPath`` -/// reaches the remote muxad. 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 respawned on the loop's existing retry cadence. +/// 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 @@ -18,12 +22,11 @@ actor AmuxRemoteMuxaForwarder { /// The local unix-socket path clients connect to (the `-L` bind). nonisolated var localSocketPath: String { host.muxaForwardSocketPath } - /// Lock carve-out: the forward process handle must be terminable - /// *synchronously* from the app-termination path (an actor hop scheduled - /// in `applicationWillTerminate` may never run, orphaning an `ssh -N` - /// that pins the ControlMaster past its ControlPersist window). - /// `uncheckedState` because `Process` isn't `Sendable`; the handle never - /// escapes the lock's short critical sections. + /// 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? @@ -32,15 +35,18 @@ actor AmuxRemoteMuxaForwarder { self.host = host } - /// Ensures a live forward process, spawning one when needed (idempotent). + /// Ensures a live forward, establishing one when needed (idempotent). /// - /// 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). + /// 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 processBox.withLockUnchecked({ $0?.isRunning == true }) { return } + if Self.canConnect(unixSocketPath: localSocketPath) { return } processBox.withLockUnchecked { $0 = nil } try host.ensureControlSocketDirectory() @@ -79,20 +85,24 @@ actor AmuxRemoteMuxaForwarder { try forward.run() processBox.withLockUnchecked { $0 = forward } - // Bounded, cancellable readiness wait: ssh binds the local socket a - // few ms after launch and offers 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. + // 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 !forward.isRunning { break } if Self.canConnect(unixSocketPath: localSocketPath) { break } try? await Task.sleep(for: .milliseconds(100)) if Task.isCancelled { break } } } - /// Terminates the forward process and removes the local socket file. + /// 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 diff --git a/Sources/AppDelegate.swift b/Sources/AppDelegate.swift index 40c0bed6f..b48c1dd4e 100644 --- a/Sources/AppDelegate.swift +++ b/Sources/AppDelegate.swift @@ -2014,9 +2014,8 @@ 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 so their `ssh -N` forward clients - // don't outlive the app (an attached mux client pins the ControlMaster - // past its ControlPersist window). + // 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.