cmux NEXT Cloud: a first chat an older daemon refused updates the machine and starts once more

Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/DaemonUpdatePolicy.swift

@@ -106,6 +106,9 @@ public actor DaemonUpdatePolicy {
     private var queue: [Pending] = []
     /// The id of the target whose update runs now.
     private var running: String?
+    /// The id of the target inside its `update(to:whileBusy:)` call now: a turn ``updateNow``
+    /// joins without marking it, since that update answers the refusal (review of #29497).
+    private var updating: String?
     /// The loop that drains ``queue``; nil while it is empty.
     private var drain: Task<Void, Never>?
     /// The drain's backoff wait; ``offer(_:)`` cuts it short.
@@ -122,6 +125,9 @@ public actor DaemonUpdatePolicy {
     /// 6 hours, or disconnected while offline): their next offer clears that line when they need
     /// no update.
     private var offlineShown: Set<String> = []
+    /// Callers of ``updateNow(_:)`` waiting for a turn of their target to end, by target id: each
+    /// stream is finished when that turn ends (any outcome) or the queue drops the target.
+    private var turnWaiters: [String: [UUID: AsyncStream<Void>.Continuation]] = [:]
 
     public init(_ dependencies: Dependencies) {
         self.dependencies = dependencies
@@ -167,6 +173,68 @@ public actor DaemonUpdatePolicy {
         return true
     }
 
+    /// A chat start that the machine's older daemon refused (a fresh Cloud machine whose image
+    /// baked a cmux-tui from before cmux#29107): `target` is updated at the front of the queue now,
+    /// as the person's Update click would be (neither the ledger nor this run's memory holds it
+    /// back), and this returns once that turn ended, whatever its outcome, so the caller can start
+    /// again on the daemon it left. A turn of `target` already running is joined, never run twice.
+    /// Returns at once while `agent.updatePeers` is off, and when the calling task is cancelled.
+    public func updateNow(_ target: any DaemonUpdateTarget) async {
+        guard dependencies.enabled() else { return }
+        let id = target.id
+        // Its update already runs: that update answers this refusal, so only wait for it. Marking
+        // it would queue a second update once the turn ends.
+        if updating == id {
+            await turnEnded(id)
+            return
+        }
+        // As the person's click: neither the ledger nor this run's memory holds the next turn back.
+        // A turn that already reached its update takes this mark (``attempt``); one still reading
+        // the machine (its `cannotUpdate` check) leaves it, and ``drainQueue()`` queues it again.
+        retried.insert(id)
+        if running == id {
+            offered[id] = target
+        } else {
+            enqueue(target, urgent: true)
+            // Due now, also an urgent entry waiting out a busy backoff (review of #29497).
+            if let index = queue.firstIndex(where: { $0.target.id == id }), queue[index].due > elapsed() {
+                queue[index].due = elapsed()
+                sleeper?.cancel()
+            }
+        }
+        await turnEnded(id)
+        // The joined turn ended before its update ran: the mark queued its next turn, which runs
+        // the update; wait for that one too.
+        if retried.contains(id) { await turnEnded(id) }
+    }
+
+    /// Whether an ``updateNow`` waits on a turn of target `id` (tests order a join against it).
+    func awaitsTurn(_ id: String) -> Bool { turnWaiters[id] != nil }
+
+    /// Waits until a turn of target `id` ends, or the queue drops it, or the task is cancelled.
+    private func turnEnded(_ id: String) async {
+        guard running == id || queue.contains(where: { $0.target.id == id }) else { return }
+        // One element at most: the stream only ever finishes.
+        let (stream, continuation) = AsyncStream<Void>.makeStream(bufferingPolicy: .bufferingNewest(1))
+        let key = UUID()
+        turnWaiters[id, default: [:]][key] = continuation
+        await withTaskCancellationHandler {
+            for await _ in stream { break }
+        } onCancel: {
+            continuation.finish()
+        }
+        turnWaiters[id]?[key] = nil
+        if turnWaiters[id]?.isEmpty == true { turnWaiters[id] = nil }
+    }
+
+    /// Ends the waits of ``updateNow(_:)`` on target `id` (nil: on every target).
+    private func endTurnWaits(_ id: String?) {
+        let ended = id.map { turnWaiters.removeValue(forKey: $0).map { Array($0.values) } ?? [] }
+            ?? turnWaiters.values.flatMap(\.values)
+        if id == nil { turnWaiters.removeAll() }
+        for continuation in ended { continuation.finish() }
+    }
+
     /// Offers every target of ``Dependencies/targets`` and returns when the queue is empty: each
     /// target done, left for the next launch, or the toggle turned off.
     public func run() async {
@@ -223,6 +291,7 @@ public actor DaemonUpdatePolicy {
             Self.logger.info("daemon update skipped: off, or the bundled build is unknown or has local changes")
             queue.removeAll()
             drain = nil
+            endTurnWaits(nil)
             return
         }
         var ledger = DaemonUpdateLedger.read(deps.directory, build: local.hash)
@@ -236,6 +305,7 @@ public actor DaemonUpdatePolicy {
                 Self.logger.info("daemon update stopped: agent.updatePeers turned off")
                 let retried = queue.filter { $0.firstBusy != nil }
                 queue.removeAll()
+                endTurnWaits(nil)
                 for each in retried {
                     await deps.report(DaemonUpdateEvent(each.target, .cleared))
                 }
@@ -249,6 +319,7 @@ public actor DaemonUpdatePolicy {
                 queue.append(again)
             }
             running = nil
+            endTurnWaits(pending.target.id)
             // The person clicked Update while this turn read the machine, before any update ran.
             if retried.contains(pending.target.id), let target = offered[pending.target.id] {
                 enqueue(target, urgent: false)
@@ -350,7 +421,9 @@ public actor DaemonUpdatePolicy {
         await deps.report(DaemonUpdateEvent(target, .updating))
         // A click until now is answered by this update.
         retried.remove(target.id)
+        updating = target.id
         let outcome = await target.update(to: local, whileBusy: whileBusy)
+        updating = nil
         Self.logger.info("\(Self.logLine(target.displayName, to: local.hash, whileBusy: whileBusy, outcome), privacy: .public)")
         if outcome == .busy || outcome == .unreachable {
             var again = pending

Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentTabRemoteHost.swift

@@ -33,6 +33,14 @@ nonisolated struct AgentTabRemoteHost: AgentPaneHostProviding {
     /// A Cloud machine's link owner answer (``CloudMachineLinks/ownerLink``), read at each trust call
     /// and never kept. Nil for a machine that is not a Cloud machine.
     var ownerLink: AgentSessionAttachClient.OwnerLink?
+    /// Brings the machine's daemon to this app's build now and answers whether an update turn ran
+    /// (``DaemonUpdatePolicy/updateNow(_:)``); nil where the app cannot. A start that the machine's
+    /// older daemon refused (``olderDaemonRefusals``) runs it, then tries once more: a fresh Cloud
+    /// machine boots its image's cmux-tui, which may predate what this app's start needs (a team VM
+    /// that takes its creator's chat, cmux#29107), and the app's automatic update runs in the
+    /// background, so a first chat right after the machine came up would otherwise reach the old
+    /// daemon (Leo: "whenever i start a chat it should be able to send me to cloud automatically").
+    var updateMachine: (@Sendable () async -> Bool)?
     /// The chat runs on the person's own Cloud VM (``RemoteAcpmuxWire/ownerCloud(_:served:)``).
     var ownerCloud: RemoteAcpmuxWire.OwnerCloud?
 
@@ -80,11 +88,12 @@ nonisolated struct AgentTabRemoteHost: AgentPaneHostProviding {
             // A new chat: nothing runs on that machine until the page's first send (cx-k21v). The
             // page shows the New Tab page or an empty chat; its `session/new` starts the chat.
             route = AgentPaneRemoteRoute(machine: machine, hostID: hostID, approvals: approvals) { [endpoint, reconnect] in
-                RemoteAcpmuxWire(start: { [approvals = approvals, servesKeyedReason] cwd, harness, model in
+                RemoteAcpmuxWire(start: { [approvals = approvals, servesKeyedReason, updateMachine] cwd, harness, model in
                     let started: AgentTabRemoteStart.Started
                     do {
                         started = try await Self.keyedStart(cwd, harness, model, approvals: approvals,
-                                                            servesKeyedReason: servesKeyedReason, proceed: Self.proceedAlways, start: start)
+                                                            servesKeyedReason: servesKeyedReason, updateMachine: updateMachine,
+                                                            proceed: Self.proceedAlways, start: start)
                     } catch let refusal as AgentRemoteStartRefusal {
                         throw AgentSessionRemoteError(code: refusal.code, message: servesTrust ? refusal.userMessage : refusal.withoutTrust.userMessage, folder: refusal.folder)
                     }
@@ -116,22 +125,38 @@ nonisolated struct AgentTabRemoteHost: AgentPaneHostProviding {
     /// try): its throw stops the start there, and what it threw is what this throws.
     func startNow(cwd: String?, harness: String, model: String? = nil, proceed: @escaping @Sendable () async throws -> Void) async throws -> AgentTabRemoteStart.Started {
         guard let start else { throw AgentRemoteStartRefusal.startUnsupported(machine: machine) }
-        return try await Self.keyedStart(cwd, harness, model, approvals: approvals, servesKeyedReason: servesKeyedReason, proceed: proceed, start: start)
+        return try await Self.keyedStart(cwd, harness, model, approvals: approvals, servesKeyedReason: servesKeyedReason,
+                                         updateMachine: updateMachine, proceed: proceed, start: start)
     }
 
+    /// The refusals of a start that an older daemon on the machine gives where its current build
+    /// would start the chat: a team VM's (`team_vm_blocked`, before cmux#29107) and an owner check
+    /// that daemon cannot take (`host_unverified`, a daemon without the caller's `vm_kind` word).
+    /// ``updateMachine`` runs only while the link's owner answer says the machine is the caller's,
+    /// so a teammate's machine is never retried (the app refuses that start before it is sent).
+    static let olderDaemonRefusals: Set<String> = [AgentSessionAttachError.teamMachine, "agent_session.host_unverified"]
+
     /// The chat's Allow needs this app's key on that machine's agent daemon (cx-oa6ah): an idle
     /// daemon gets it before the start; a busy one waits for the person's restart click on the
     /// Allow notice. That machine's cmux-tui decides whether a restart loses nothing (cx-0w0jj):
     /// the start asks for this app's key, and its `needs_keyed_start` answer says whether to
     /// restart; after the restart (or without one) the chat starts as it is.
     private static func keyedStart(
         _ cwd: String?, _ harness: String?, _ model: String?, approvals: AgentPaneRemoteApprovals?, servesKeyedReason: Bool,
+        updateMachine: (@Sendable () async -> Bool)? = nil,
         proceed: @escaping @Sendable () async throws -> Void,
         start unchecked: @escaping @Sendable (String?, String?, String?, AgentSessionAttachClient.PersonKey?) async throws -> AgentTabRemoteStart.Started
     ) async throws -> AgentTabRemoteStart.Started {
         let start = { @Sendable (cwd: String?, harness: String?, model: String?, key: AgentSessionAttachClient.PersonKey?) in
             try await proceed()
-            return try await unchecked(cwd, harness, model, key)
+            do {
+                return try await unchecked(cwd, harness, model, key)
+            } catch let refusal as AgentRemoteStartRefusal where olderDaemonRefusals.contains(refusal.code) {
+                // The machine's daemon may be older than this app's: update it now, then once more.
+                guard let updateMachine, await updateMachine() else { throw refusal }
+                try await proceed()
+                return try await unchecked(cwd, harness, model, key)
+            }
         }
         guard let approvals, let prepare = approvals.prepare else { return try await start(cwd, harness, model, nil) }
         guard servesKeyedReason else {

Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentTabs+Wiring.swift

@@ -365,6 +365,14 @@ extension AgentTabStore {
             // it is sent, through any daemon (the legacy link's or the app link's own answer).
             let links = services.machines.session(daemon.machineID).map { CloudMachineLinks(appLink: $0.appLink, link: $0.link) }
             host.ownerLink = links?.ownerLink
+            // A start its older daemon refused updates the machine to this app's build and tries
+            // once more, only on the caller's own machine (the link's owner answer, read then).
+            if let cloudMachineID = host.cloudMachineID, let ownerLink = links?.ownerLink {
+                host.updateMachine = { [weak services] in
+                    guard await ownerLink() != nil, let services else { return false }
+                    return await CloudMachineUpdates.updateNow(cloudMachineID, services: services)
+                }
+            }
             host.ownerCloud = RemoteAcpmuxWire.ownerCloud(links?.ownerLink, served: daemon.supports(RemoteAcpmuxWire.ownerCloudCapability))
             if started {
                 // The tab may still be provisional here: each handshake reads its store surface.

Packages/macOS/CmuxNext/Sources/CmuxNextApp/Cloud/CloudMachineUpdates.swift

@@ -25,7 +25,7 @@ final class CloudMachineUpdates {
     private weak var services: AppServices?
     /// Which handshakes are still to offer (``HandshakeFollower``).
     private var follower = HandshakeFollower()
-    private let logger = Logger(subsystem: "com.cmuxterm.app.next", category: "cloud.machine-update")
+    private nonisolated static let logger = Logger(subsystem: "com.cmuxterm.app.next", category: "cloud.machine-update")
 
     private init(services: AppServices) {
         self.services = services
@@ -80,17 +80,35 @@ final class CloudMachineUpdates {
     /// no `cmux` client; false when it made no offer. The policy keeps one entry per machine, so an
     /// offer per handshake is safe.
     private func offer(_ machineID: String) -> Bool {
-        guard CloudTunables.updatesCloudMachines, let services,
-              let session = services.machines.session(machineID), let binary = services.cloud.move.cli else { return false }
-        let cli = CloudMoveCLI(binary: binary, timeout: services.cloud.move.timeout)
-        let target = CloudMachineUpdateTarget(machineID: machineID, displayName: session.machine.title,
-                                              source: Self.source(session, machines: services.machines, cli: cli, api: services.cloud.machineAPI,
-                                                                  logger: logger))
+        guard let services, let target = Self.target(machineID, services: services) else { return false }
         // task-owner: returns once the policy queued the machine; the policy's drain owns the update
         Task { [policy = services.daemonUpdates.policy] in await policy.offer(target) }
         return true
     }
 
+    /// A chat start on machine `machineID` that its older daemon refused (``AgentTabRemoteHost/updateMachine``):
+    /// the machine is updated to the app's build now, at the front of the shared policy's queue (joining
+    /// an update of it that already runs), and this returns once that turn ended. False when no turn ran:
+    /// the toggles are off, the app has no `cmux` client, or the machine is gone. Same guarantees as the
+    /// automatic update: `--to-bundled --if-older`, never a move back, terminals and chats keep running.
+    static func updateNow(_ machineID: String, services: AppServices) async -> Bool {
+        guard let target = target(machineID, services: services) else { return false }
+        logger.info("cloud machine \(machineID, privacy: .public): a start was refused by its daemon; updating it before starting again")
+        await services.daemonUpdates.policy.updateNow(target)
+        return true
+    }
+
+    /// Machine `machineID` as the shared policy's target; nil while the toggles are off, the app has no
+    /// `cmux` client, or the machine is gone.
+    private static func target(_ machineID: String, services: AppServices) -> CloudMachineUpdateTarget? {
+        guard CloudTunables.updatesCloudMachines, let session = services.machines.session(machineID),
+              let binary = services.cloud.move.cli else { return nil }
+        let cli = CloudMoveCLI(binary: binary, timeout: services.cloud.move.timeout)
+        return CloudMachineUpdateTarget(machineID: machineID, displayName: session.machine.title,
+                                        source: source(session, machines: services.machines, cli: cli, api: services.cloud.machineAPI,
+                                                       logger: logger))
+    }
+
     // MARK: The machine's side
 
     /// The live ``CloudMachineUpdateTarget/Source`` of `session`. Each step reads the session on