From a50755a1cd56faaaa099ec94969c878139f87cb9 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 08:21:59 +0000 Subject: [PATCH 1/5] fix(ios): bound link requests and keep terminal input flowing Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-557bb07834475125d7ba --- .../CmuxMobileLive/AcpmuxConnection.swift | 54 ++++-- .../Sources/CmuxMobileLive/AcpmuxHub.swift | 65 +++++-- .../Sources/CmuxMobileLive/AcpmuxReplay.swift | 18 ++ .../DaemonTerminalSession.swift | 28 +-- .../Sources/CmuxMobileLive/LinkPairing.swift | 61 +++++- .../LiveCoordinator+Links.swift | 17 +- .../BoundedEventStream.swift | 142 ++++++++++++++ .../CmuxMobileTransport/DaemonClient.swift | 173 ++++++++++++------ .../CmuxMobileTransport/LinkDeadline.swift | 88 +++++++++ .../AcpmuxReplayTests.swift | 85 +++++++++ .../DaemonTerminalSessionTests.swift | 56 ++++-- .../LinkRequestDeadlineTests.swift | 160 ++++++++++++++++ .../TerminalInputThroughputTests.swift | 98 ++++++++++ .../DaemonBackpressureTests.swift | 60 ++++++ .../DaemonClientTests.swift | 99 ++++++++++ .../LinkDeadlineTests.swift | 98 ++++++++++ 16 files changed, 1169 insertions(+), 133 deletions(-) create mode 100644 cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxReplay.swift create mode 100644 cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/BoundedEventStream.swift create mode 100644 cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/LinkDeadline.swift create mode 100644 cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift create mode 100644 cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift create mode 100644 cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/TerminalInputThroughputTests.swift create mode 100644 cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonBackpressureTests.swift create mode 100644 cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift index 4be1a18f5e33..045c8cadc682 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift @@ -25,6 +25,8 @@ public actor AcpmuxConnection { private var nextID = 1 private var pending: [Int: CheckedContinuation] = [:] private var readTask: Task? + private var sends: [Int: Task] = [:] + private var ended: (any Error)? public nonisolated let notifications: AsyncStream private let continuation: AsyncStream.Continuation @@ -38,34 +40,49 @@ public actor AcpmuxConnection { readTask = Task { await self.readLoop() } } + /// Prompts resolve when the turn ends, which can include a person answering an + /// approval. Other requests are short reads or mutations; both have finite bounds. + static func replyTimeout(for method: String) -> Duration { + method == "session/prompt" ? .seconds(24 * 60 * 60) : .seconds(30) + } + public func request(_ method: String, _ params: JSONValue = .object([:])) async throws -> JSONValue { + try await request(method, params, timeout: Self.replyTimeout(for: method)) + } + + /// A late reply is ignored; cancellation and expiry release the request and + /// cancel its write without closing unrelated requests on the same channel. + public func request(_ method: String, _ params: JSONValue, timeout: Duration) async throws -> JSONValue { + try Task.checkCancellation() + if let ended { throw ended } start() let id = nextID nextID += 1 let message: JSONValue = .object(["jsonrpc": .string("2.0"), "id": .number(Double(id)), "method": .string(method), "params": params]) let data = try JSONEncoder().encode(message) rpcLog.debug("-> \(id) \(method, privacy: .public) \(data.count) bytes") + return try await LinkDeadline.run( + timeout: timeout, + operation: { try await self.sendAndWait(id: id, data: data) }, + onCancel: { await self.fail(id, CancellationError()) } + ) + } + + private func sendAndWait(id: Int, data: Data) async throws -> JSONValue { + try Task.checkCancellation() + if let ended { throw ended } return try await withCheckedThrowingContinuation { continuation in pending[id] = continuation - Task { - do { try await channel.send(data) } catch { self.fail(id, error) } + sends[id] = Task { + do { + try Task.checkCancellation() + try await channel.send(data) + sends[id] = nil + } catch { fail(id, error) } } } } - /// `request` that gives up after `timeout` (a read the person waits on: the - /// agent catalog). A late reply is dropped. - public func request(_ method: String, _ params: JSONValue, timeout: Duration) async throws -> JSONValue { - let id = nextID - let timer = Task { [weak self] in - try? await Task.sleep(for: timeout) - guard !Task.isCancelled else { return } - await self?.fail(id, LinkError(code: "timeout", message: "The Mac didn't answer (\(method))")) - } - defer { timer.cancel() } - return try await request(method, params) - } - /// A JSON-RPC notification (no id, no reply), e.g. `session/cancel`. public func notify(_ method: String, _ params: JSONValue) async throws { start() @@ -75,11 +92,12 @@ public actor AcpmuxConnection { public func close() async { readTask?.cancel() - await channel.close(code: "eof", reason: "client closed") finish(LinkError(code: "closed", message: "acpmux channel closed")) + await channel.close(code: "eof", reason: "client closed") } private func fail(_ id: Int, _ error: any Error) { + sends.removeValue(forKey: id)?.cancel() pending.removeValue(forKey: id)?.resume(throwing: error) } @@ -116,6 +134,7 @@ public actor AcpmuxConnection { rpcLog.debug("<- \(Int(id)) with no request waiting") return } + sends.removeValue(forKey: Int(id))?.cancel() rpcLog.debug("<- \(Int(id)) \(value["error"] == nil ? "result" : "error", privacy: .public)") if let error = value["error"] { continuation.resume(throwing: RPCError( @@ -172,6 +191,9 @@ public actor AcpmuxConnection { } private func finish(_ error: any Error) { + ended = error + for task in sends.values { task.cancel() } + sends.removeAll() for (_, continuation) in pending { continuation.resume(throwing: error) } pending.removeAll() continuation.finish() diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift index 82170d74df93..f8e8b8b01095 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift @@ -26,7 +26,7 @@ public final class AcpmuxHub { /// Callers of `waitUntilReady` (a Feed answer whose Mac's link is opening). @ObservationIgnored private var readyWaiters: [UUID: CheckedContinuation] = [:] @ObservationIgnored private var pumpTask: Task? - @ObservationIgnored private var sources: [String: WeakSource] = [:] + @ObservationIgnored private var sources: [String: [WeakSource]] = [:] /// Watch-level events for the Feed: `(sessionId, notification)`. @ObservationIgnored public var onWatchEvent: (@MainActor (AcpmuxHub.WatchEvent) -> Void)? /// Signs an Allow or an answer with the phone's approval key after Face ID @@ -85,6 +85,11 @@ public final class AcpmuxHub { } private struct WeakSource { weak var source: AcpmuxSessionSource? } + /// A chat with an expired replay cache may still have an open reader. Keep routing + /// to it while a new reader loads its own complete snapshot from the Mac. + private var liveSources: [AcpmuxSessionSource] { + sources.values.flatMap { $0.compactMap(\.source) } + } public init() {} @@ -95,7 +100,7 @@ public final class AcpmuxHub { connection = nil isReady = false fullAccess = false - for entry in sources.values { entry.source?.linkDown() } + for source in liveSources { source.linkDown() } guard let channel else { return } let connection = AcpmuxConnection(channel: channel) self.connection = connection @@ -126,7 +131,7 @@ public final class AcpmuxHub { readyWaiters = [:] for waiter in waiters.values { waiter.resume(returning: true) } log.info("acpmux ready, \(next.count, privacy: .public) sessions") - for entry in sources.values { entry.source?.linkUp() } + for source in liveSources { source.linkUp() } onWatchEvent?(.sessionsLoaded(Array(next.values))) await seedPending(connection) } catch { @@ -207,11 +212,13 @@ public final class AcpmuxHub { try await connection.notify(method, params) } - /// The chat source for an acpmux session (one per session). + /// Reuse a complete replay; after its budget expires, a new reader gets a fresh + /// attach instead of a suffix whose first deltas have no message to append to. public func source(sessionID: String, summary: AgentSessionSummary) -> AcpmuxSessionSource { - if let existing = sources[sessionID]?.source { return existing } + let existing = (sources[sessionID] ?? []).compactMap(\.source) + if let source = existing.last(where: { !$0.replay.requiresResync }) { return source } let source = AcpmuxSessionSource(hub: self, sessionID: sessionID, summary: summary) - sources[sessionID] = WeakSource(source: source) + sources[sessionID] = (existing + [source]).map { WeakSource(source: $0) } return source } @@ -313,7 +320,7 @@ public final class AcpmuxHub { /// Full access mode again. func fullAccessRefused() { fullAccess = false - for entry in sources.values { entry.source?.fullAccessChanged(false) } + for source in liveSources { source.fullAccessChanged(false) } } private func handle(_ notification: AcpmuxConnection.Notification) { @@ -321,7 +328,9 @@ public final class AcpmuxHub { let sessionID = params["sessionId"]?.stringValue switch notification.method { case "_acpmux/event": - if let sessionID { sources[sessionID]?.source?.receive(record: params) } + if let sessionID { + for entry in sources[sessionID] ?? [] { entry.source?.receive(record: params) } + } case "_acpmux/session_changed": guard let value = params["session"], let session = AcpmuxSession(value) else { return } sessions[session.id] = session @@ -338,7 +347,7 @@ public final class AcpmuxHub { Task { [weak self] in guard let self, let connection = self.connection else { return } _ = try? await connection.request("_acpmux/watch", .object(["enabled": .bool(true)])) - for entry in self.sources.values { entry.source?.linkUp() } + for source in self.liveSources { source.linkUp() } } default: break @@ -354,8 +363,8 @@ public final class AcpmuxSessionSource: AgentSessionSource { let sessionID: String private weak var hub: AcpmuxHub? private let records: AcpmuxRecords - private var continuations: [AsyncStream.Continuation] = [] - private var replay: [AgentSessionEvent] = [] + private var continuations: [UUID: AsyncStream.Continuation] = [:] + fileprivate var replay = AcpmuxReplay(limit: AcpmuxSessionSource.maxHistory) private var lastSeq = 0 private var attached = false private var attaching = false @@ -382,9 +391,31 @@ public final class AcpmuxSessionSource: AgentSessionSource { public func events() -> AsyncStream { let (stream, continuation) = AsyncStream.makeStream(bufferingPolicy: .unbounded) + if replay.requiresResync { + guard let hub else { + continuation.yield(.clientError("Reconnect to load this chat.")) + continuation.finish() + return stream + } + let source = hub.source(sessionID: sessionID, summary: summary) + let upstream = source.events() + let task = Task { + // The proxy reader owns its fresh source until cancellation; the hub's + // registry remains weak, so closing a chat releases its replay too. + defer { withExtendedLifetime(source) {} } + for await event in upstream { continuation.yield(event) } + continuation.finish() + } + continuation.onTermination = { _ in task.cancel() } + return stream + } // What this source already has is history to a new reader: one change, drawn in place. - if !replay.isEmpty { continuation.yield(.batch(replay)) } - continuations.append(continuation) + if !replay.events.isEmpty { continuation.yield(.batch(replay.events)) } + let id = UUID() + continuations[id] = continuation + continuation.onTermination = { [weak self] _ in + Task { @MainActor in self?.continuations.removeValue(forKey: id) } + } return stream } @@ -853,8 +884,8 @@ public final class AcpmuxSessionSource: AgentSessionSource { guard let events = history else { return } history = nil guard !events.isEmpty else { return } - replay.append(contentsOf: events) - for continuation in continuations { continuation.yield(.batch(events)) } + replay.append(events) + for continuation in continuations.values { continuation.yield(.batch(events)) } } private func emit(_ event: AgentSessionEvent) { @@ -866,8 +897,8 @@ public final class AcpmuxSessionSource: AgentSessionSource { history?.append(event) return } - replay.append(event) + replay.append([event]) } - for continuation in continuations { continuation.yield(event) } + for continuation in continuations.values { continuation.yield(event) } } } diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxReplay.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxReplay.swift new file mode 100644 index 000000000000..894c93fc317e --- /dev/null +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxReplay.swift @@ -0,0 +1,18 @@ +/// A replay is useful only while it is complete. Once its budget is exceeded, +/// release it and require a fresh snapshot; keeping only a tail would apply +/// deltas without their turn, permission request or earlier message chunks. +struct AcpmuxReplay { + let limit: Int + private(set) var events: [Event] = [] + private(set) var requiresResync = false + + mutating func append(_ batch: [Event]) { + guard !requiresResync else { return } + guard batch.count <= limit - events.count else { + events = [] + requiresResync = true + return + } + events.append(contentsOf: batch) + } +} diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/DaemonTerminalSession.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/DaemonTerminalSession.swift index faa225d1a1df..1e18a067af00 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/DaemonTerminalSession.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/DaemonTerminalSession.swift @@ -125,19 +125,14 @@ public final class DaemonTerminalSession: TerminalSession { /// a lost reply does not stop input (later sends keep both directions /// moving, and the loss is repaired by duplicate ACKs). static let maxInputInFlight = 64 + /// A missing input reply frees its slot and reports uncertain delivery. + var inputRequestDeadline: Duration = .seconds(10) /// How long a `send` may go unanswered before ``armTailProbe()`` writes /// `ping`s behind it: many round trips, well under the 1 s RTO. var tailProbeDelay: Duration = .milliseconds(150) /// Writes and answers so far: a probe fires only if none came since it was armed. private var inputProgress = 0 private var tailProbe: Task? - /// When each unanswered send went out, by batch number. - private var inputWrittenAt: [Int: TimeInterval] = [:] - /// No new send while the oldest unanswered one is older than this - /// (seconds): a stalled link holds typed keys in `input`, where the hold - /// rules apply, instead of writing up to `maxInputInFlight` of them to - /// arrive late. Scrolling keeps many in flight; a stall does not. - var inputStallLimit: TimeInterval = 0.5 /// Sends written and not yet answered (tests). var inputsInFlight: Int { inputInFlight } /// A tail probe is armed (tests). @@ -832,6 +827,9 @@ public final class DaemonTerminalSession: TerminalSession { /// else on the attach channel. Keys that waited longer than /// `inputHoldLimit` are dropped first. /// + /// Reply age does not stop the pipeline: a high-latency link or one lost + /// reply must not turn typing into stop-and-wait. The bounded in-flight + /// window and each request's deadline bound a channel that truly stalls. /// The next batch goes out once this one is written, not when its reply /// comes (TUI fast scroll), up to `maxInputInFlight` unanswered. One /// small request at a time stalled a flick for the link's whole @@ -850,24 +848,23 @@ public final class DaemonTerminalSession: TerminalSession { expireInput() guard !input.isEmpty, let target = inputClient ?? client else { return } if inputInFlight > 0, inputRecovering || inputTarget !== target || inputInFlight >= Self.maxInputInFlight { return } - if let oldest = inputWrittenAt.values.min(), now() - oldest > inputStallLimit { return } inputWriting = true inputInFlight += 1 inputTarget = target let number = nextInputBatch nextInputBatch += 1 - inputWrittenAt[number] = now() let typedAt = input[0].at let batch = input.reduce(into: Data()) { $0.append($1.bytes) } input.removeAll() let surface = surface let onLane = target !== client + let deadline = inputRequestDeadline Task { [weak self] in guard let self else { return } var written = false do { let submit: @Sendable () async throws -> String = { - try await target.submit("send", params: ["surface": surface, "bytes": batch.base64EncodedString()]) + try await target.submit("send", params: ["surface": surface, "bytes": batch.base64EncodedString()], deadline: deadline) } // A lane request gets written and answered within seconds or // the lane is given up. @@ -904,7 +901,7 @@ public final class DaemonTerminalSession: TerminalSession { if !written { inputWriting = false } lastError = LinkController.reason(error) inputRecovering = true - if written || target === abandonedInputLane { + if written || target === abandonedInputLane || Self.inputDeliveryUncertain(error) { inputFailed.append((number, batch, typedAt, written: true)) } else { // Never written: safe to send again (on the attach channel @@ -923,11 +920,16 @@ public final class DaemonTerminalSession: TerminalSession { handler?(.connection(connectionState)) } } - inputWrittenAt[number] = nil inputSettled() } } + /// A write deadline/cancellation can race a partial or completed write. + /// Treat it like a lost reply: the Mac may have applied the keystrokes. + static func inputDeliveryUncertain(_ error: any Error) -> Bool { + error is CancellationError || (error as? LinkError)?.code == "write_timeout" + } + /// The input lane gave no answer in time (`wasWriting`: the batch's /// write itself hung): given up, and the batch is not sent again (the /// Mac may already have typed it). @@ -937,7 +939,6 @@ public final class DaemonTerminalSession: TerminalSession { inputRecovering = true inputFailed.append((number, Data(count: bytes), now(), written: true)) if wasWriting { inputWriting = false } - inputWrittenAt[number] = nil if inputClient === lane { dropInputLane() } inputSettled() } @@ -955,7 +956,6 @@ public final class DaemonTerminalSession: TerminalSession { } tailProbe?.cancel() tailProbe = nil - inputWrittenAt.removeAll() inputTarget = nil abandonedInputLane = nil if inputRecovering { diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift index 5eb25373ba95..7dd4268d630f 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift @@ -26,6 +26,10 @@ public struct LinkPairing: Sendable { /// A scanned pairing code waiting for its Mac's challenge (``PairInvite``). public let invites: PairInvites + // Network exchange and the Mac's Allow prompt have different budgets. + var handshakeTimeout: Duration = .seconds(15) + var approvalTimeout: Duration = .seconds(180) + public init(keys: PhoneKeys, phoneID: String, name: String, pins: MacPins = .standard, push: PushKeys? = nil, invites: PairInvites = PairInvites()) { self.keys = keys @@ -112,7 +116,12 @@ public struct LinkPairing: Sendable { ] let pushArgs = macID.flatMap { pushKeyArgs(mac: $0) } if let pushArgs { args.merge(pushArgs) { a, _ in a } } - let channel = try await link.openChannel(service: "auth", args: args) + let channelArgs = args + let channel = try await LinkDeadline.run( + timeout: handshakeTimeout, + operation: { try await link.openChannel(service: "auth", args: channelArgs) }, + onDiscard: { await $0.close(code: "eof", reason: "pairing expired") } + ) var opened = Opened(channel: channel, identity: identity, approval: approval, macNonce: macNonce) opened.sentPushKey = pushArgs != nil return .success(opened) @@ -171,9 +180,12 @@ public struct LinkPairing: Sendable { /// The box key that key vouched for (sealed replies go to it), and its signature. var boxKey: String? var boxSignature: String? + var waitingForApproval = false + var deadline = ContinuousClock.now.advanced(by: handshakeTimeout) do { while true { - guard let frame = try await channel.receive() else { + guard let frame = try await exchange(channel, until: deadline, approval: waitingForApproval, + operation: { try await channel.receive() }) else { let closed = channel.closedBy ?? LinkError(code: "closed", message: "The Mac closed pairing") // A Mac with no `auth` cannot prove a pinned key either. if closed.code == "unknown_service", pinned != nil { return .refused(Self.changed) } @@ -192,7 +204,12 @@ public struct LinkPairing: Sendable { return .paired(macDevice: macID) } if value["pending"] == .bool(true) { - await onWaiting() + // Repeated pending frames never extend the approval deadline. + if !waitingForApproval { + waitingForApproval = true + deadline = ContinuousClock.now.advanced(by: approvalTimeout) + await onWaiting() + } continue } guard let nonce = value["challenge"]?.stringValue, let mac = value["mac"]?.stringValue else { @@ -220,7 +237,9 @@ public struct LinkPairing: Sendable { // A scanned code stands in for the Mac's Allow prompt, only on the Mac that just // proved the key the code names (never in `open`'s args, sent before any proof). if let token = invites.token(mac: mac, provenKey: provenNow) { reply["pairToken"] = .string(token) } - try await channel.send(try JSONEncoder().encode(JSONValue.object(reply))) + let bytes = try JSONEncoder().encode(JSONValue.object(reply)) + try await exchange(channel, until: deadline, approval: waitingForApproval, + operation: { try await channel.send(bytes) }) } } catch let error as LinkError { return .refused(error) @@ -256,19 +275,41 @@ public struct LinkPairing: Sendable { /// "Remove This iPhone from My Macs": asks the Mac to forget this phone (best effort, over a live link). public func unpair(link: any MacLink, macID: String) async { + let deadline = ContinuousClock.now.advanced(by: handshakeTimeout) guard let (identity, approval) = try? keys.publicKeys(), - let channel = try? await link.openChannel(service: "auth", args: [ - "identityKey": identity, "approvalKey": approval, "name": "", "action": "unpair", - ]) else { return } + let channel = try? await LinkDeadline.run( + timeout: handshakeTimeout, + operation: { try await link.openChannel(service: "auth", args: [ + "identityKey": identity, "approvalKey": approval, "name": "", "action": "unpair", + ]) }, + onDiscard: { await $0.close(code: "eof", reason: "unpair expired") } + ) else { return } defer { Task { await channel.close(code: "eof", reason: "") } } - guard let frame = try? await channel.receive(), let value = try? JSONDecoder().decode(JSONValue.self, from: frame), + guard let frame = try? await exchange(channel, until: deadline, operation: { try await channel.receive() }), let value = try? JSONDecoder().decode(JSONValue.self, from: frame), let nonce = value["challenge"]?.stringValue, value["mac"]?.stringValue == macID else { return } let payload = LinkAuthPayload(mac: macID, phone: phoneID, nonce: nonce, identityKey: identity, approvalKey: approval, action: "unpair") guard let signature = try? keys.signIdentity(payload.message), let reply = try? JSONEncoder().encode(JSONValue.object(["signature": .string(ApprovalPayload.hex(signature))])) else { return } - try? await channel.send(reply) - _ = try? await channel.receive() + do { + try await exchange(channel, until: deadline, operation: { try await channel.send(reply) }) + _ = try await exchange(channel, until: deadline, operation: { try await channel.receive() }) + } catch { return } + } + + /// Deadline/cancellation closes this auth channel, unblocking its reader. + /// Network silence must not be mistaken for an unanswered human prompt. + private func exchange( + _ channel: any LinkChannel, until deadline: ContinuousClock.Instant, approval: Bool = false, + operation: @escaping @Sendable () async throws -> Value + ) async throws -> Value { + try await LinkDeadline.run( + timeout: ContinuousClock.now.duration(to: deadline), + timeoutError: LinkError(code: approval ? "timeout" : "reply_timeout", + message: "The Mac didn't answer. Retry."), + operation: operation, + onCancel: { await channel.close(code: "eof", reason: "pairing ended") } + ) } } diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift index f9d05f6997d9..0b0699fd71a3 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift @@ -349,9 +349,20 @@ extension LiveCoordinator { /// ahead of data, no reader stalls) and drops links under load. private func checkVersion(_ session: MacSession, _ link: any MacLink) async { let macID = session.macID - guard let channel = try? await link.openChannel(service: "version", args: [:]) else { return } - let reply = try? await channel.receive() - await channel.close(code: "eof", reason: "") + guard let channel = try? await LinkDeadline.run( + timeout: .seconds(15), + operation: { try await link.openChannel(service: "version", args: [:]) }, + onDiscard: { await $0.close(code: "eof", reason: "version expired") } + ) else { return } + defer { Task { await channel.close(code: "eof", reason: "") } } + let reply: Data? + do { + reply = try await LinkDeadline.run( + timeout: .seconds(15), + operation: { try await channel.receive() }, + onCancel: { await channel.close(code: "eof", reason: "version expired") } + ) + } catch { return } // Only a host without the service is old; another refusal (`unpaired` while pairing // fails) says nothing about its protocol. if reply == nil, let closed = channel.closedBy, closed.code != "unknown_service" { return } diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/BoundedEventStream.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/BoundedEventStream.swift new file mode 100644 index 000000000000..877de3d599b8 --- /dev/null +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/BoundedEventStream.swift @@ -0,0 +1,142 @@ +import Foundation + +/// A lossless async sequence backed by a byte-bounded queue. +/// +/// Daemon channels carry both replies and unsolicited event lines. Keeping the +/// events in an unbounded `AsyncStream` lets the daemon reader consume forever +/// even when the terminal renderer is behind, so transport-level flow control +/// never sees the slow consumer. This queue makes the producer wait once the +/// byte budget is full; the reader therefore stops calling `receive()` until a +/// consumer drains some data. +public struct DaemonEventStream: AsyncSequence, Sendable { + public typealias Element = Data + + public struct AsyncIterator: AsyncIteratorProtocol { + private let buffer: BoundedDaemonEventBuffer + + init(buffer: BoundedDaemonEventBuffer) { + self.buffer = buffer + } + + public mutating func next() async -> Data? { + await buffer.next() + } + } + + let buffer: BoundedDaemonEventBuffer + + init(buffer: BoundedDaemonEventBuffer) { + self.buffer = buffer + } + + public func makeAsyncIterator() -> AsyncIterator { + AsyncIterator(buffer: buffer) + } +} + +/// The bounded queue used by `DaemonClient`'s unsolicited event stream. +/// +/// There is one producer (the daemon reader) and normally one consumer (the +/// terminal/chat event task), but the implementation tolerates a replacement +/// consumer while preserving FIFO order. `finish()` wakes both sides so a +/// cancelled/closed daemon never leaves a task suspended forever. +public actor BoundedDaemonEventBuffer { + public static let defaultCapacity = 4 << 20 + + private let capacity: Int + private var queue: [Data] = [] + private var head = 0 + private var queuedBytes = 0 + private var closed = false + private var consumers: [(id: UUID, continuation: CheckedContinuation)] = [] + private var producer: CheckedContinuation? + + public init(capacity: Int = BoundedDaemonEventBuffer.defaultCapacity) { + self.capacity = max(1, capacity) + } + + /// Enqueues one complete event line. Returns `false` after `finish()`. + /// A single event larger than the nominal capacity is admitted when the + /// queue is empty; dropping it would corrupt the terminal stream. + public func push(_ data: Data) async -> Bool { + guard !data.isEmpty else { return !closed } + while !closed && !Task.isCancelled && (queuedBytes + data.count > capacity) && !queue.isEmpty { + await withTaskCancellationHandler { + await withCheckedContinuation { continuation in + if Task.isCancelled { continuation.resume() } + else { producer = continuation } + } + } onCancel: { + Task { await self.cancelProducer() } + } + } + guard !closed, !Task.isCancelled else { return false } + if !consumers.isEmpty { + consumers.removeFirst().continuation.resume(returning: data) + } else { + queue.append(data) + queuedBytes += data.count + } + return true + } + + /// Returns the oldest event, waiting for the next one when needed. + public func next() async -> Data? { + guard !Task.isCancelled else { return nil } + if head < queue.count { + let data = queue[head] + queue[head] = Data() + head += 1 + queuedBytes -= data.count + if head >= queue.count { + queue.removeAll(keepingCapacity: true) + head = 0 + } else if head >= 64, head * 2 >= queue.count { + queue.removeFirst(head) + head = 0 + } + resumeProducerIfRoom() + return data + } + if closed { return nil } + let id = UUID() + return await withTaskCancellationHandler { + await withCheckedContinuation { continuation in + if Task.isCancelled { continuation.resume(returning: nil) } + else { consumers.append((id, continuation)) } + } + } onCancel: { + Task { await self.cancelConsumer(id) } + } + } + + /// Closes the stream and wakes any blocked producer or consumer. + /// Already queued events remain drainable (the previous AsyncStream EOF + /// behavior); consumed slots release their data immediately. Cancellation + /// only removes that waiter, so a replacement consumer can still drain. + public func finish() { + guard !closed else { return } + closed = true + cancelProducer() + for consumer in consumers { consumer.continuation.resume(returning: nil) } + consumers.removeAll() + } + + public var pendingBytes: Int { queuedBytes } + + private func cancelProducer() { + producer?.resume() + producer = nil + } + + private func cancelConsumer(_ id: UUID) { + guard let index = consumers.firstIndex(where: { $0.id == id }) else { return } + consumers.remove(at: index).continuation.resume(returning: nil) + } + + private func resumeProducerIfRoom() { + guard queuedBytes < capacity, let producer else { return } + self.producer = nil + producer.resume() + } +} diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/DaemonClient.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/DaemonClient.swift index 5960e4fab68e..d4a4be49fa90 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/DaemonClient.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileTransport/DaemonClient.swift @@ -8,20 +8,28 @@ public actor DaemonClient { private var pending: [String: CheckedContinuation] = [:] /// Ids of a batch sent but not yet awaited, and replies that came first. private var expected: Set = [] - private var early: [String: Data] = [:] + private var early: [String: Result] = [:] + private var deadlines: [String: Task] = [:] /// Set once the channel ended; later waits fail with it. private var ended: (any Error)? private var buffer = Data() private var readTask: Task? - public nonisolated let events: AsyncStream - private let eventContinuation: AsyncStream.Continuation + /// Unsolicited daemon events (terminal output, tree notifications, etc.). + /// The sequence is byte-bounded; when its consumer falls behind, the + /// daemon reader waits before reading another channel frame so link-level + /// backpressure reaches the Mac. + public nonisolated let events: DaemonEventStream + private let eventBuffer: BoundedDaemonEventBuffer /// The `LinkError` code of a request whose deadline passed. public static let replyTimeout = "reply_timeout" + public static let defaultDeadline: Duration = .seconds(15) - public init(channel: any LinkChannel) { + public init(channel: any LinkChannel, eventCapacity: Int = BoundedDaemonEventBuffer.defaultCapacity) { self.channel = channel - (events, eventContinuation) = AsyncStream.makeStream() + let eventBuffer = BoundedDaemonEventBuffer(capacity: eventCapacity) + self.eventBuffer = eventBuffer + self.events = DaemonEventStream(buffer: eventBuffer) } public func start() { @@ -44,25 +52,14 @@ public actor DaemonClient { } /// Sends `{"id", "cmd", ...params}` and returns the raw JSON reply line. - /// With `deadline`, a reply that has not come by then fails with + /// The default deadline is 15 seconds. A reply that has not come by then fails with /// `LinkError` `reply_timeout` (a late reply is dropped). public func request(_ cmd: String, params: [String: Any] = [:], deadline: Duration? = nil, clock: any Clock = ContinuousClock()) async throws -> Data { start() let id = nextRequestID() let line = try line(id: id, cmd: cmd, params: params) - if let deadline { - Task { [weak self] in - try? await clock.sleep(for: deadline) - await self?.fail(id: id, error: LinkError(code: Self.replyTimeout, message: "no reply to \(cmd) in time")) - } - } - return try await withCheckedThrowingContinuation { continuation in - pending[id] = continuation - Task { - do { try await channel.send(line) } catch { self.fail(id: id, error: error) } - } - } + return try await perform(id: id, line: line, deadline: deadline ?? Self.defaultDeadline, clock: clock) } /// Closes the channel (the Mac drops its daemon connection) and fails @@ -70,27 +67,21 @@ public actor DaemonClient { public func close() async { readTask?.cancel() await channel.close(code: "eof", reason: "") - finish(LinkError(code: "closed", message: "daemon channel closed")) + await finish(LinkError(code: "closed", message: "daemon channel closed")) } /// Sends a `cmux.protocol/2` request (`machine` and `session` default to /// `current`, as the Mac app's `ResourceRequestEnvelope`) and returns the /// raw response line (`ok`, `result` or `error`). /// - /// With `deadline`, a reply that has not come by then fails with - /// `LinkError` `reply_timeout` (a late reply is dropped as an event), so + /// The default deadline is 15 seconds. A reply that has not come by then fails with + /// `LinkError` `reply_timeout` (a late reply is dropped), so /// a request on a link that went quiet never waits forever. public func resourceRequest(_ operation: String, params: sending [String: Any] = [:], idempotencyKey: String? = nil, deadline: Duration? = nil, clock: any Clock = ContinuousClock()) async throws -> Data { start() let id = "r\(nextID)" nextID += 1 - if let deadline { - Task { [weak self] in - try? await clock.sleep(for: deadline) - await self?.fail(id: id, error: LinkError(code: Self.replyTimeout, message: "no reply to \(operation) in time")) - } - } var fields = params fields["machine"] = fields["machine"] ?? "current" fields["session"] = fields["session"] ?? "current" @@ -98,11 +89,29 @@ public actor DaemonClient { if let idempotencyKey { object["idempotency_key"] = idempotencyKey } var line = try JSONSerialization.data(withJSONObject: object, options: [.sortedKeys, .withoutEscapingSlashes]) line.append(0x0A) - return try await withCheckedThrowingContinuation { continuation in - pending[id] = continuation - Task { - do { try await channel.send(line) } catch { self.fail(id: id, error: error) } - } + return try await perform(id: id, line: line, deadline: deadline ?? Self.defaultDeadline, clock: clock) + } + + private func perform(id: String, line: Data, deadline: Duration, clock: any Clock) async throws -> Data { + try Task.checkCancellation() + if let ended { throw ended } + expected.insert(id) + arm(id: id, deadline: deadline, clock: clock) + let write = Task { + do { + try Task.checkCancellation() + try await channel.send(line) + } catch { self.fail(id: id, error: error) } + } + defer { write.cancel() } + return try await reply(id) + } + + private func arm(id: String, deadline: Duration, clock: any Clock = ContinuousClock()) { + deadlines[id] = Task { [weak self] in + do { try await clock.sleep(for: deadline) } catch { return } + guard !Task.isCancelled else { return } + await self?.fail(id: id, error: LinkError(code: Self.replyTimeout, message: "The Mac didn't answer. Retry.")) } } @@ -110,7 +119,8 @@ public actor DaemonClient { /// in the same order: one round trip instead of one per request (the /// daemon handles a connection's lines in order). A failed request is a /// `.failure` in its slot; a failed write fails them all. - public func requests(_ batch: [(cmd: String, params: [String: Any])]) async -> [Result] { + public func requests(_ batch: [(cmd: String, params: [String: Any])], + deadline: Duration = DaemonClient.defaultDeadline) async -> [Result] { start() var ids: [String] = [] var bytes = Data() @@ -121,14 +131,19 @@ public actor DaemonClient { } ids.append(id) } + guard !Task.isCancelled else { return ids.map { _ in .failure(CancellationError()) } } + if let ended { return ids.map { _ in .failure(ended) } } expected.formUnion(ids) - do { try await channel.send(bytes) } catch { - for id in ids { - expected.remove(id) - early[id] = nil + for id in ids { arm(id: id, deadline: deadline) } + let write = Task { + do { + try Task.checkCancellation() + try await channel.send(bytes) + } catch { + for id in ids { self.fail(id: id, error: error) } } - return ids.map { _ in .failure(error) } } + defer { write.cancel() } var results: [Result] = [] for id in ids { do { results.append(.success(try await reply(id))) } catch { results.append(.failure(error)) } @@ -141,14 +156,23 @@ public actor DaemonClient { /// submit before the next keeps its requests in order on the channel /// while several are unanswered (the daemon handles a connection's lines /// in order). - public func submit(_ cmd: String, params: [String: Any] = [:]) async throws -> String { + public func submit(_ cmd: String, params: [String: Any] = [:], + deadline: Duration = DaemonClient.defaultDeadline) async throws -> String { start() let id = nextRequestID() let line = try line(id: id, cmd: cmd, params: params) + try Task.checkCancellation() + if let ended { throw ended } expected.insert(id) - do { try await channel.send(line) } catch { - expected.remove(id) - early[id] = nil + arm(id: id, deadline: deadline) + do { + let channel = self.channel + try await LinkDeadline.run(timeout: deadline, + timeoutError: LinkError(code: "write_timeout", message: "The Mac didn't answer. Retry."), + operation: { try await channel.send(line) }, + onCancel: { await channel.close(code: "timeout", reason: "write interrupted") }) + } catch { + discard(id) throw error } return id @@ -161,19 +185,41 @@ public actor DaemonClient { /// The reply to batch request `id`, which may already have come. private func reply(_ id: String) async throws -> Data { - if let line = early.removeValue(forKey: id) { return line } - if let ended { - expected.remove(id) - throw ended - } - return try await withCheckedThrowingContinuation { continuation in - expected.remove(id) - pending[id] = continuation + try await withTaskCancellationHandler { + if Task.isCancelled { + discard(id) + throw CancellationError() + } + if let result = early.removeValue(forKey: id) { return try result.get() } + if let ended { + discard(id) + throw ended + } + guard expected.contains(id) else { + throw LinkError(code: "unknown_request", message: "Request is no longer waiting") + } + return try await withCheckedThrowingContinuation { continuation in + expected.remove(id) + pending[id] = continuation + } + } onCancel: { + Task { await self.fail(id: id, error: CancellationError()) } } } + private func discard(_ id: String) { + deadlines.removeValue(forKey: id)?.cancel() + expected.remove(id) + early[id] = nil + } + private func fail(id: String, error: any Error) { - pending.removeValue(forKey: id)?.resume(throwing: error) + deadlines.removeValue(forKey: id)?.cancel() + if let waiter = pending.removeValue(forKey: id) { + waiter.resume(throwing: error) + } else if expected.remove(id) != nil { + early[id] = .failure(error) + } } private func readLoop() async { @@ -183,35 +229,40 @@ public actor DaemonClient { while let newline = buffer.firstIndex(of: 0x0A) { let line = buffer[buffer.startIndex..( + timeout: Duration, + timeoutError: LinkError = LinkError(code: "reply_timeout", message: "The Mac didn't answer. Retry."), + operation: @escaping @Sendable () async throws -> Value, + onCancel: @escaping @Sendable () async -> Void = {}, + onDiscard: @escaping @Sendable (Value) async -> Void = { _ in } + ) async throws -> Value { + // Do not create an unstructured worker for an already-cancelled caller. + // `Task {}` does not inherit cancellation, so relying on the worker's + // first check alone leaves a small window where operation can start. + try Task.checkCancellation() + guard timeout > .zero else { + Task { await onCancel() } + throw timeoutError + } + let race = LinkDeadlineRace() + return try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { continuation in + race.install(continuation) + let work = Task { + do { + guard race.beginOperation() else { return } + try Task.checkCancellation() + let value = try await operation() + if !race.finish(.success(value)) { Task { await onDiscard(value) } } + } catch { race.finish(.failure(error)) } + } + let timer = Task { + do { try await Task.sleep(for: timeout) } catch { return } + if race.finish(.failure(timeoutError)) { Task { await onCancel() } } + } + race.install(work: work, timer: timer) + } + } onCancel: { + if race.finish(.failure(CancellationError())) { Task { await onCancel() } } + } + } +} + +private final class LinkDeadlineRace: @unchecked Sendable { // crash-allow: lock-guarded state + private let lock = NSLock() + private var result: Result? + private var continuation: CheckedContinuation? + private var tasks: [Task] = [] + + func install(_ continuation: CheckedContinuation) { + let result: Result? = lock.withLock { + if let result { return result } + self.continuation = continuation + return nil + } + if let result { continuation.resume(with: result) } + } + + func install(work: Task, timer: Task) { + let finished = lock.withLock { + guard result == nil else { return true } + tasks = [work, timer] + return false + } + if finished { work.cancel(); timer.cancel() } + } + + /// Claims the operation slot only while the race is still active. This + /// closes the gap between creating the worker and installing its handle. + func beginOperation() -> Bool { + lock.withLock { result == nil } + } + + @discardableResult + func finish(_ result: Result) -> Bool { + let settled = lock.withLock { () -> (CheckedContinuation?, [Task])? in + guard self.result == nil else { return nil } + self.result = result + defer { continuation = nil; tasks = [] } + return (continuation, tasks) + } + guard let (continuation, tasks) = settled else { return false } + for task in tasks { task.cancel() } + continuation?.resume(with: result) + return true + } +} diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift new file mode 100644 index 000000000000..efc382e94070 --- /dev/null +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift @@ -0,0 +1,85 @@ +import CmuxMobileAgentChat +import CmuxMobileTransport +import Foundation +import Testing +@testable import CmuxMobileLive + +@Suite struct AcpmuxReplayTests { + @Test func overflowDiscardsTheWholeReplayRatherThanPartialDeltas() { + var replay = AcpmuxReplay(limit: 3) + replay.append(["turn", "first chunk"]) + replay.append(["second chunk"]) + #expect(replay.events == ["turn", "first chunk", "second chunk"]) + #expect(!replay.requiresResync) + replay.append(["third chunk"]) + #expect(replay.events.isEmpty) + #expect(replay.requiresResync) + replay.append(Array(repeating: "later chunk", count: 10_000)) + #expect(replay.events.isEmpty, "an expired replay never grows again") + } + + @Test func anOversizedHistoryBatchAlsoRequiresResync() { + var replay = AcpmuxReplay(limit: 3) + replay.append([1, 2, 3, 4]) + #expect(replay.requiresResync) + #expect(replay.events.isEmpty) + } +} + +@MainActor +@Suite struct AcpmuxReplayResyncTests { + @Test func newReaderResyncsWithoutInterruptingTheExistingReader() async throws { + let attaches = ChatOpenSnapshotLiveTests.Attaches() + let link = MockMacLink { _, data in + guard let object = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let method = object["method"] as? String else { return [] } + let id = object["id"].map { "\($0)" } ?? "null" + func reply(_ result: String) -> Data { Data(#"{"jsonrpc":"2.0","id":\#(id),"result":\#(result)}"#.utf8) } + switch method { + case "initialize": return [reply(#"{"protocolVersion":1}"#)] + case "_acpmux/watch": return [reply(#"{"sessions":[]}"#)] + case "_acpmux/attach": + let params = object["params"] as? [String: Any] + attaches.append((params?["afterSeq"] as? NSNumber)?.intValue ?? 0) + let history = ChatOpenSnapshotLiveTests.turnRecords(0).joined(separator: ",") + return [reply(#"{"session":{"sessionId":"s1","status":"idle","pending":[]},"hasMore":false,"events":[\#(history)]}"#)] + case "_test/live": + return [reply("{}"), Data(#"{"jsonrpc":"2.0","method":"_acpmux/event","params":{"sessionId":"s1","seq":100000,"dir":"mux","kind":"user_message","msg":{"text":"live","promptId":"live","turnId":"live"}}}"#.utf8)] + default: return [reply("{}")] + } + } + let hub = AcpmuxHub() + await hub.bind(try await link.openChannel(service: "acpmux", args: [:])) + let source = hub.source(sessionID: "s1", summary: ChatOpenSnapshotLiveTests.summary) + let (first, firstTask) = ChatOpenSnapshotLiveTests.collect(source.events()) + defer { firstTask.cancel() } + for _ in 0..<2000 where first.events.isEmpty { await Task.yield() } + try #require(!first.events.isEmpty) + + // Stream past the cache budget while the original reader remains attached. + for n in 0..<(AcpmuxSessionSource.maxHistory + 1) { + source.receive(record: .object([ + "seq": .number(Double(n + 10)), "dir": .string("mux"), "kind": .string("turn_started"), + "msg": .object(["turnId": .string("t0")]), + ])) + } + let (second, secondTask) = ChatOpenSnapshotLiveTests.collect(source.events()) + defer { secondTask.cancel() } + for _ in 0..<2000 where second.events.isEmpty { await Task.yield() } + try #require(attaches.all == [0, 0], "the new reader fetches a snapshot, not a partial afterSeq replay") + guard case .batch(let history)? = second.events.first else { + Issue.record("the resynced reader did not receive one complete history batch") + return + } + #expect(ChatOpenSnapshotLiveTests.userMessages(history) == ["t0"]) + #expect(first.events.filter { if case .batch = $0 { true } else { false } }.count == 1, + "resync never repeats history into the existing reader") + + _ = try await hub.request("_test/live", .object([:])) + for _ in 0..<2000 where !ChatOpenSnapshotLiveTests.userMessages(second.events).contains("live") { + await Task.yield() + } + #expect(ChatOpenSnapshotLiveTests.userMessages(first.events).contains("live")) + #expect(ChatOpenSnapshotLiveTests.userMessages(second.events).contains("live")) + } +} diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/DaemonTerminalSessionTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/DaemonTerminalSessionTests.swift index 510edb6f05a1..a4f533151bc1 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/DaemonTerminalSessionTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/DaemonTerminalSessionTests.swift @@ -220,7 +220,7 @@ final class FakeTerminalDaemon: @unchecked Sendable { let session = DaemonTerminalSession(surface: 7) { DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) } - // A frozen clock: a slow host must not age these sends past `inputStallLimit`. + // A fixed clock keeps detached-input expiration out of this test. session.now = { 1000 } session.start { _ in } try await waitUntil { session.isAttached } @@ -251,7 +251,7 @@ final class FakeTerminalDaemon: @unchecked Sendable { return DaemonClient(channel: channel) } var events: [TerminalSessionEvent] = [] - // A frozen clock: a slow host must not age these sends past `inputStallLimit`. + // A fixed clock keeps detached-input expiration out of this test. session.now = { 1000 } session.start { events.append($0) } try await waitUntil { session.isAttached } @@ -307,7 +307,7 @@ final class FakeTerminalDaemon: @unchecked Sendable { DaemonClient(channel: GatedChannel(try await link.openChannel(service: "daemon", args: [:]), gate: gate)) } var events: [TerminalSessionEvent] = [] - // A frozen clock: a slow host must not age these sends past `inputStallLimit`. + // A fixed clock keeps detached-input expiration out of this test. session.now = { 1000 } session.start { events.append($0) } try await waitUntil { session.isAttached } @@ -339,7 +339,7 @@ final class FakeTerminalDaemon: @unchecked Sendable { DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) } session.tailProbeDelay = .milliseconds(30) - // A frozen clock: a slow host must not age these sends past `inputStallLimit`. + // A fixed clock keeps detached-input expiration out of this test. session.now = { 1000 } session.start { _ in } try await waitUntil { session.isAttached } @@ -377,7 +377,7 @@ final class FakeTerminalDaemon: @unchecked Sendable { DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) } session.tailProbeDelay = .seconds(60) - session.now = { 1000 } // a frozen clock: no send ages past the stall limit + session.now = { 1000 } // detached-input expiration is not under test session.start { _ in } try await waitUntil { session.isAttached } let sends = { daemon.recorded.filter { $0["cmd"] as? String == "send" }.count } @@ -393,10 +393,9 @@ final class FakeTerminalDaemon: @unchecked Sendable { #expect(sends() == cap) } - /// Review: a stalled link (the oldest unanswered send older than - /// `inputStallLimit`) holds new keys instead of writing them to arrive - /// late; they go once the stall clears. - @Test func aStalledLinkHoldsNewKeys() async throws { + /// One lost reply must not block later keystrokes after 500 ms. Request + /// deadlines settle it; only the bounded in-flight window applies. + @Test func anOldUnansweredSendDoesNotBlockLaterKeys() async throws { let daemon = FakeTerminalDaemon() daemon.unansweredSends = true let link = MockMacLink { service, line in daemon.respond(service, line) } @@ -411,10 +410,43 @@ final class FakeTerminalDaemon: @unchecked Sendable { let sends = { daemon.recorded.filter { $0["cmd"] as? String == "send" }.count } session.send(Data("a".utf8)) try await waitUntil { sends() == 1 } - clock += session.inputStallLimit + 0.1 + clock += 0.6 session.send(Data("b".utf8)) - for _ in 0..<10 where sends() <= 1 { try await Task.sleep(for: .milliseconds(10)) } - #expect(sends() == 1) + try await waitUntil { sends() == 2 } + #expect(sends() == 2) + session.stop() + } + + @Test func lostFallbackReplyTimesOutAndNewInputWorksAfterReattach() async throws { + let daemon = FakeTerminalDaemon() + daemon.unansweredSends = true + let link = MockMacLink { service, line in daemon.respond(service, line) } + let session = DaemonTerminalSession(surface: 7) { + DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) + } + session.inputRequestDeadline = .milliseconds(50) + var events: [TerminalSessionEvent] = [] + session.start { events.append($0) } + try await waitUntil { session.isAttached } + session.send(Data("a".utf8)) + try await waitUntil { events.contains(.inputNotSent(bytes: 1)) } + #expect(session.inputsInFlight == 0) + daemon.unansweredSends = false + session.reconnect() + try await waitUntil { session.isAttached } + session.send(Data("b".utf8)) + try await waitUntil { + daemon.recorded.last { $0["cmd"] as? String == "send" }?["bytes"] as? String == Data("b".utf8).base64EncodedString() + } + let sent = daemon.recorded.filter { $0["cmd"] as? String == "send" } + #expect(sent.count == 2, "the uncertain first keystroke is never replayed") + session.stop() + } + + @Test func timedOutOrCancelledWritesAreNeverClassifiedAsSafeToReplay() { + #expect(DaemonTerminalSession.inputDeliveryUncertain(LinkError(code: "write_timeout", message: "partial write"))) + #expect(DaemonTerminalSession.inputDeliveryUncertain(CancellationError())) + #expect(!DaemonTerminalSession.inputDeliveryUncertain(LinkError(code: "not_written", message: "not submitted"))) } /// TUI fast scroll: what one main-queue turn queues (a scroll frame's diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift new file mode 100644 index 000000000000..5c3bc6efe77d --- /dev/null +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift @@ -0,0 +1,160 @@ +import CmuxMobileAgentChat +import CmuxMobileTransport +import Foundation +import Testing +@testable import CmuxMobileLive + +@Suite(.timeLimit(.minutes(1))) struct LinkRequestDeadlineTests { + @Test func aDroppedRPCReplyTimesOutAndTheNextRequestStillWorks() async throws { + let channel = DeadlineChannel() + let connection = AcpmuxConnection(channel: channel) + do { + _ = try await connection.request("missing", .null, timeout: .milliseconds(20)) + Issue.record("A dropped reply succeeded") + } catch let error as LinkError { + #expect(error.code == "reply_timeout") + } + // The old response cannot answer the next request. + channel.push(Data(#"{"jsonrpc":"2.0","id":1,"result":"late"}"#.utf8)) + let result = try await connection.request("answer", .null, timeout: .seconds(1)) + #expect(result == .string("ok")) + await connection.close() + } + + @Test func cancellingAnRPCReleasesItWithoutClosingOtherRequests() async throws { + let channel = DeadlineChannel() + let connection = AcpmuxConnection(channel: channel) + let request = Task { try await connection.request("missing") } + var sends = channel.sent.makeAsyncIterator() + _ = await sends.next() + request.cancel() + do { + _ = try await request.value + Issue.record("A cancelled request succeeded") + } catch is CancellationError {} + #expect(channel.closed == false) + #expect(try await connection.request("answer") == .string("ok")) + await connection.close() + do { + _ = try await connection.request("answer") + Issue.record("A closed connection accepted a request") + } catch let error as LinkError { #expect(error.code == "closed") } + } + + @Test func aMissingPairingChallengeTimesOutAndClosesItsChannel() async throws { + var pairing = PairInviteTests.pairing(PairInvites()) + pairing.handshakeTimeout = .milliseconds(20) + let channel = DeadlineChannel() + let result = await pairing.finish(.success(opened(channel)), macID: "test-mac") {} + guard case .refused(let error) = result else { Issue.record("Pairing unexpectedly succeeded"); return } + #expect(error.code == "reply_timeout") + var closes = channel.closes.makeAsyncIterator() + _ = await closes.next() + #expect(channel.closed) + } + + @Test func approvalCanFinishAfterTheHandshakeDeadline() async throws { + var pairing = PairInviteTests.pairing(PairInvites()) + pairing.handshakeTimeout = .milliseconds(10) + pairing.approvalTimeout = .seconds(5) + let channel = DeadlineChannel() + channel.push(Data(#"{"pending":true}"#.utf8)) + let waiting = AsyncStream.makeStream() + let task = Task { + await pairing.finish(.success(opened(channel)), macID: "test-mac") { waiting.continuation.yield(()) } + } + var notices = waiting.stream.makeAsyncIterator() + _ = await notices.next() + try await Task.sleep(for: .milliseconds(30)) + channel.push(Data(#"{"ok":true}"#.utf8)) + #expect(await task.value == .paired(macDevice: "test-mac")) + waiting.continuation.finish() + } + + @Test func aMissingApprovalHasAVisibleTimeout() async throws { + var pairing = PairInviteTests.pairing(PairInvites()) + pairing.approvalTimeout = .milliseconds(20) + let channel = DeadlineChannel() + channel.push(Data(#"{"pending":true}"#.utf8)) + let result = await pairing.finish(.success(opened(channel)), macID: "test-mac") {} + guard case .refused(let error) = result else { Issue.record("Pairing unexpectedly succeeded"); return } + #expect(error.code == "timeout") + } + + @Test func cancellingPairingClosesItsWaitingReceive() async throws { + let pairing = PairInviteTests.pairing(PairInvites()) + let channel = DeadlineChannel() + let task = Task { await pairing.finish(.success(opened(channel)), macID: "test-mac") {} } + var reads = channel.reads.makeAsyncIterator() + _ = await reads.next() + task.cancel() + guard case .refused = await task.value else { Issue.record("Cancelled pairing succeeded"); return } + var closes = channel.closes.makeAsyncIterator() + _ = await closes.next() + #expect(channel.closed) + } + + private func opened(_ channel: DeadlineChannel) -> LinkPairing.Opened { + .init(channel: channel, identity: "04aa", approval: "04bb", macNonce: "test-nonce") + } +} + +/// Its receive deliberately ignores Task cancellation; only close releases it. +private final class DeadlineChannel: LinkChannel, @unchecked Sendable { + let id: UInt32 = 1 + private let lock = NSLock() + private var queue: [Data] = [] + private var waiter: CheckedContinuation? + private var ended = false + private let sendEvents = AsyncStream.makeStream() + private let readEvents = AsyncStream.makeStream() + private let closeEvents = AsyncStream.makeStream() + var sent: AsyncStream { sendEvents.stream } + var reads: AsyncStream { readEvents.stream } + var closes: AsyncStream { closeEvents.stream } + var closed: Bool { lock.withLock { ended } } + + func send(_ data: Data) async throws { + sendEvents.continuation.yield(()) + if let request = try? JSONDecoder().decode(JSONValue.self, from: data), + request["method"] == .string("answer"), let id = request["id"] { + push(try JSONEncoder().encode(JSONValue.object(["jsonrpc": .string("2.0"), "id": id, "result": .string("ok")]))) + } + } + + func receive() async throws -> Data? { + await withCheckedContinuation { continuation in + let immediate: Data?? = lock.withLock { + if !queue.isEmpty { return .some(queue.removeFirst()) } + if ended { return .some(nil) } + waiter = continuation + return nil + } + if let immediate { continuation.resume(returning: immediate) } + readEvents.continuation.yield(()) + } + } + + func push(_ data: Data) { + let continuation = lock.withLock { () -> CheckedContinuation? in + guard !ended else { return nil } + if let waiter { self.waiter = nil; return waiter } + queue.append(data) + return nil + } + continuation?.resume(returning: data) + } + + func close(code: String, reason: String) async { + let continuation = lock.withLock { + ended = true + defer { waiter = nil; queue = [] } + return waiter + } + continuation?.resume(returning: nil) + closeEvents.continuation.yield(()) + closeEvents.continuation.finish() + sendEvents.continuation.finish() + readEvents.continuation.finish() + } +} diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/TerminalInputThroughputTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/TerminalInputThroughputTests.swift new file mode 100644 index 000000000000..5dfd8332b505 --- /dev/null +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/TerminalInputThroughputTests.swift @@ -0,0 +1,98 @@ +import CmuxMobileTransport +import Foundation +import Testing +@testable import CmuxMobileLive + +/// Same harness for base and head: restore only DaemonTerminalSession.swift +/// from the base revision, then run this exact test file again. A synthetic +/// daemon acknowledges sends after 500 ms; both runs type 100 keys at 50/s. +@MainActor @Suite struct TerminalInputThroughputTests { + @Test func keystrokeThroughputAt500msRTT() async throws { + for dropFirstReply in [false, true] { + let channel = LatencyTerminalChannel(dropFirstReply: dropFirstReply) + let session = DaemonTerminalSession(surface: 7) { DaemonClient(channel: channel) } + session.tailProbeDelay = .seconds(60) + session.start { _ in } + let clock = ContinuousClock() + let attachDeadline = clock.now.advanced(by: .seconds(3)) + while !session.isAttached, clock.now < attachDeadline { + try await Task.sleep(for: .milliseconds(5)) + } + try #require(session.isAttached) + let began = clock.now + for _ in 0..<100 { + session.send(Data("x".utf8)) + try await Task.sleep(for: .milliseconds(20)) + } + // Record what reached the daemon during the typing interval, + // rather than counting queued bytes as successful delivery. + let elapsed = began.duration(to: clock.now) + let seconds = Double(elapsed.components.seconds) + Double(elapsed.components.attoseconds) / 1e18 + let delivered = channel.sentBytes + print("TERMINAL_INPUT_PERF rtt_ms=500 first_reply_dropped=\(dropFirstReply) typed=100 delivered=\(delivered) seconds=\(seconds) keys_per_second=\(Double(delivered) / seconds)") + #expect(delivered > 0) + session.stop() + await channel.close(code: "eof", reason: "benchmark complete") + } + } +} + +private final class LatencyTerminalChannel: LinkChannel, @unchecked Sendable { + let id: UInt32 = 1 + private let daemon = FakeTerminalDaemon() + private let lock = NSLock() + private let dropFirstReply: Bool + private var firstSend = true + private var bytes = 0 + private var closed = false + private var timers: [UUID: Task] = [:] + private let continuation: AsyncStream.Continuation + private var iterator: AsyncStream.Iterator + + init(dropFirstReply: Bool) { + self.dropFirstReply = dropFirstReply + let (stream, continuation) = AsyncStream.makeStream() + self.continuation = continuation + iterator = stream.makeAsyncIterator() + } + + var sentBytes: Int { lock.withLock { bytes } } + + func send(_ data: Data) async throws { + let object = try JSONSerialization.jsonObject(with: data.split(separator: 0x0A)[0]) as? [String: Any] + let isInput = object?["cmd"] as? String == "send" + let replies = daemon.respond("daemon", data) + guard isInput else { + for reply in replies { continuation.yield(reply) } + return + } + let shouldDrop = lock.withLock { + bytes += (object?["bytes"] as? String).flatMap { Data(base64Encoded: $0) }?.count ?? 0 + defer { firstSend = false } + return firstSend && dropFirstReply + } + guard !shouldDrop else { return } + let id = UUID() + lock.withLock { + guard !closed else { return } + timers[id] = Task { [weak self] in + do { try await Task.sleep(for: .milliseconds(500)) } catch { return } + guard let self else { return } + let open = self.lock.withLock { self.timers[id] = nil; return !self.closed } + if open { for reply in replies { self.continuation.yield(reply) } } + } + } + } + + func receive() async throws -> Data? { await iterator.next() } + + func close(code: String, reason: String) async { + let timers = lock.withLock { + closed = true + defer { self.timers.removeAll() } + return Array(self.timers.values) + } + for timer in timers { timer.cancel() } + continuation.finish() + } +} diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonBackpressureTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonBackpressureTests.swift new file mode 100644 index 000000000000..60e2c2c2e846 --- /dev/null +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonBackpressureTests.swift @@ -0,0 +1,60 @@ +import Foundation +import Testing +@testable import CmuxMobileTransport + +@Suite(.timeLimit(.minutes(1))) struct DaemonBackpressureTests { + @Test func slowEventReaderStopsReceivingFromTheLink() async throws { + let line = Data(#"{"event":"output"}"#.utf8) + let channel = BackpressureChannel() + for _ in 0..<3 { channel.deliver(line + Data([0x0A])) } + let client = DaemonClient(channel: channel, eventCapacity: line.count) + await client.start() + try await Task.sleep(for: .milliseconds(50)) + // One buffered event and one blocked enqueue; the third frame stays + // at the link until the application consumes the first event. + #expect(channel.readCount == 2) + var events = client.events.makeAsyncIterator() + #expect(await events.next() == line) + try await Task.sleep(for: .milliseconds(50)) + #expect(channel.readCount == 3) + await client.close() + #expect(await events.next() == line) + #expect(await events.next() == nil) + } + + @Test func closeReleasesABlockedProducerAndCancellationReleasesAReader() async throws { + let buffer = BoundedDaemonEventBuffer(capacity: 1) + #expect(await buffer.push(Data([1]))) + let producer = Task { await buffer.push(Data([2])) } + await buffer.finish() + #expect(await producer.value == false) + + let empty = BoundedDaemonEventBuffer(capacity: 1) + let reader = Task { await empty.next() } + reader.cancel() + #expect(await reader.value == nil) + await empty.finish() + } +} + +private final class BackpressureChannel: LinkChannel, @unchecked Sendable { + let id: UInt32 = 1 + private let lock = NSLock() + private var reads = 0 + private let stream: AsyncStream + private let continuation: AsyncStream.Continuation + private var iterator: AsyncStream.Iterator + + init() { + (stream, continuation) = AsyncStream.makeStream() + iterator = stream.makeAsyncIterator() + } + var readCount: Int { lock.withLock { reads } } + func send(_ data: Data) async throws {} + func receive() async throws -> Data? { + lock.withLock { reads += 1 } + return await iterator.next() + } + func deliver(_ data: Data) { continuation.yield(data) } + func close(code: String, reason: String) async { continuation.finish() } +} diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonClientTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonClientTests.swift index a1fed7dc0b87..8200a93884ca 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonClientTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/DaemonClientTests.swift @@ -3,6 +3,21 @@ import Testing @testable import CmuxMobileTransport @Suite struct DaemonClientTests { + @Test func unsolicitedEventsApplyByteBackpressureWithoutDroppingOrder() async throws { + let buffer = BoundedDaemonEventBuffer(capacity: 4) + #expect(await buffer.push(Data("one".utf8))) + let blocked = Task { await buffer.push(Data("two".utf8)) } + try await Task.sleep(for: .milliseconds(20)) + #expect(!blocked.isCancelled) + #expect(await buffer.pendingBytes == 3) + var iterator = DaemonEventStream(buffer: buffer).makeAsyncIterator() + #expect(await iterator.next() == Data("one".utf8)) + #expect(await blocked.value) + #expect(await iterator.next() == Data("two".utf8)) + await buffer.finish() + #expect(await iterator.next() == nil) + } + @Test func repliesRouteByIDAndEventsStream() async throws { let link = MockMacLink { _, request in let object = try! JSONSerialization.jsonObject(with: request.dropLast()) as! [String: Any] @@ -48,6 +63,75 @@ import Testing } #expect(commands == ["identify", "refused", "list-workspaces"]) } + + @Test func missingNormalAndResourceRepliesFailAndLaterRequestsWork() async throws { + let link = MockMacLink { _, bytes in + let object = try! JSONSerialization.jsonObject(with: bytes.dropLast()) as! [String: Any] + guard object["cmd"] as? String == "ping" else { return [] } + return [Data("{\"id\":\"\(object["id"]!)\",\"ok\":true}\n".utf8)] + } + let client = DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) + await #expect(throws: LinkError(code: DaemonClient.replyTimeout, message: "The Mac didn't answer. Retry.")) { + try await client.request("list-workspaces", deadline: .milliseconds(20)) + } + await #expect(throws: LinkError(code: DaemonClient.replyTimeout, message: "The Mac didn't answer. Retry.")) { + try await client.resourceRequest("workspace.list", deadline: .milliseconds(20)) + } + let reply = try await client.request("ping") + #expect(String(decoding: reply, as: UTF8.self).contains("true")) + await client.close() + } + + @Test func oneMissingBatchReplyDoesNotLoseOtherReplies() async throws { + let link = MockMacLink { _, bytes in + bytes.split(separator: 0x0A).compactMap { line in + let object = try! JSONSerialization.jsonObject(with: Data(line)) as! [String: Any] + guard object["cmd"] as? String != "drop" else { return nil } + return Data("{\"id\":\"\(object["id"]!)\",\"ok\":true}\n".utf8) + } + } + let client = DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) + let replies = await client.requests([("drop", [:]), ("ping", [:])], deadline: .milliseconds(20)) + if case .failure(let error) = replies[0] { + #expect((error as? LinkError)?.code == DaemonClient.replyTimeout) + } else { Issue.record("Missing reply succeeded") } + #expect(try replies[1].get().isEmpty == false) + await client.close() + } + + @Test func submittedReplyCanExpireBeforeItIsAwaited() async throws { + let link = MockMacLink { _, _ in [] } + let client = DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) + let id = try await client.submit("send", deadline: .milliseconds(20)) + try await Task.sleep(for: .milliseconds(50)) + await #expect(throws: LinkError(code: DaemonClient.replyTimeout, message: "The Mac didn't answer. Retry.")) { + try await client.reply(to: id) + } + await client.close() + } + + @Test func cancellingReplyDoesNotWaitForDeadline() async throws { + let link = MockMacLink { _, _ in [] } + let client = DaemonClient(channel: try await link.openChannel(service: "daemon", args: [:])) + let id = try await client.submit("send", deadline: .seconds(60)) + let waiter = Task { try await client.reply(to: id) } + waiter.cancel() + await #expect(throws: CancellationError.self) { try await waiter.value } + await client.close() + } + + @Test func lateReplyIsDroppedInsteadOfBecomingAnEvent() async throws { + let channel = DeadlineTestChannel() + let client = DaemonClient(channel: channel) + let id = try await client.submit("send", deadline: .milliseconds(20)) + await #expect(throws: LinkError.self) { try await client.reply(to: id) } + channel.deliver(Data("{\"id\":\"\(id)\",\"ok\":true}\n{\"event\":\"tree.changed\"}\n".utf8)) + var iterator = client.events.makeAsyncIterator() + let event = await iterator.next() + #expect(event.map { String(decoding: $0, as: UTF8.self) } == "{\"event\":\"tree.changed\"}") + await client.close() + } + } final class Counter: @unchecked Sendable { @@ -56,3 +140,18 @@ final class Counter: @unchecked Sendable { func add() { lock.withLock { count += 1 } } var value: Int { lock.withLock { count } } } + +private final class DeadlineTestChannel: LinkChannel, @unchecked Sendable { + let id: UInt32 = 1 + private let stream: AsyncStream + private let continuation: AsyncStream.Continuation + private var iterator: AsyncStream.Iterator + init() { + (stream, continuation) = AsyncStream.makeStream() + iterator = stream.makeAsyncIterator() + } + func send(_ data: Data) async throws {} + func receive() async throws -> Data? { await iterator.next() } + func deliver(_ data: Data) { continuation.yield(data) } + func close(code: String, reason: String) async { continuation.finish() } +} diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift new file mode 100644 index 000000000000..547d802e7114 --- /dev/null +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift @@ -0,0 +1,98 @@ +import Foundation +import Testing +@testable import CmuxMobileTransport + +@Suite(.timeLimit(.minutes(1))) struct LinkDeadlineTests { + @Test func preCancelledRunDoesNotStartOperation() async throws { + let starts = Counter() + let task = Task { + try await LinkDeadline.run(timeout: .seconds(1), operation: { + starts.increment() + return 1 + }) + } + task.cancel() + do { + _ = try await task.value + Issue.record("A pre-cancelled run succeeded") + } catch is CancellationError {} + #expect(starts.value == 0) + } + + @Test func deadlineReturnsDespiteUncancellableIOAndDisposesItsLateResult() async throws { + let gate = DeadlineGate() + let abandoned = AsyncStream.makeStream() + let discarded = AsyncStream.makeStream() + let task = Task { + try await LinkDeadline.run( + timeout: .milliseconds(30), + operation: { await gate.wait() }, + onCancel: { abandoned.continuation.yield(()) }, + onDiscard: { discarded.continuation.yield($0) } + ) + } + var started = gate.started.makeAsyncIterator() + _ = await started.next() + do { + _ = try await task.value + Issue.record("Unanswered I/O succeeded") + } catch let error as LinkError { #expect(error.code == "reply_timeout") } + var cleanup = abandoned.stream.makeAsyncIterator() + _ = await cleanup.next() + gate.release(7) + var late = discarded.stream.makeAsyncIterator() + #expect(await late.next() == 7) + abandoned.continuation.finish() + discarded.continuation.finish() + } + + @Test func cancellationReturnsDespiteUncancellableIO() async throws { + let gate = DeadlineGate() + let abandoned = AsyncStream.makeStream() + let task = Task { + try await LinkDeadline.run(timeout: .seconds(60), operation: { await gate.wait() }, + onCancel: { abandoned.continuation.yield(()) }) + } + var started = gate.started.makeAsyncIterator() + _ = await started.next() + task.cancel() + do { + _ = try await task.value + Issue.record("Cancelled I/O succeeded") + } catch is CancellationError {} + var cleanup = abandoned.stream.makeAsyncIterator() + _ = await cleanup.next() + gate.release(7) + abandoned.continuation.finish() + } +} + +private final class Counter: @unchecked Sendable { + private let lock = NSLock() + private var count = 0 + func increment() { lock.withLock { count += 1 } } + var value: Int { lock.withLock { count } } +} + +private final class DeadlineGate: @unchecked Sendable { + private let lock = NSLock() + private var continuation: CheckedContinuation? + private let entered = AsyncStream.makeStream() + var started: AsyncStream { entered.stream } + + func wait() async -> Int { + await withCheckedContinuation { continuation in + lock.withLock { self.continuation = continuation } + entered.continuation.yield(()) + } + } + + func release(_ value: Int) { + let continuation = lock.withLock { + defer { self.continuation = nil } + return self.continuation + } + continuation?.resume(returning: value) + entered.continuation.finish() + } +} From 9b7735f196e47cf4c10f55b6006857a8a4be52dd Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 08:54:51 +0000 Subject: [PATCH 2/5] fix(ios): preserve the pairing window for eager link setup Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-557bb07834475125d7ba --- .../Sources/CmuxMobileLive/LinkPairing.swift | 8 +++- .../LiveCoordinator+Links.swift | 3 +- .../CmuxMobileLive/WorkspaceSync.swift | 10 ++++- .../LinkRequestDeadlineTests.swift | 42 +++++++++++++++++-- .../LinkDeadlineTests.swift | 4 +- 5 files changed, 57 insertions(+), 10 deletions(-) diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift index 7dd4268d630f..a3557c37abc4 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LinkPairing.swift @@ -27,8 +27,12 @@ public struct LinkPairing: Sendable { public let invites: PairInvites // Network exchange and the Mac's Allow prompt have different budgets. - var handshakeTimeout: Duration = .seconds(15) - var approvalTimeout: Duration = .seconds(180) + static let defaultHandshakeTimeout: Duration = .seconds(15) + static let defaultApprovalTimeout: Duration = .seconds(180) + /// Channels opened behind auth may wait for both the proof and the Allow prompt. + public static let authorizationTimeout = defaultHandshakeTimeout + defaultApprovalTimeout + var handshakeTimeout: Duration = defaultHandshakeTimeout + var approvalTimeout: Duration = defaultApprovalTimeout public init(keys: PhoneKeys, phoneID: String, name: String, pins: MacPins = .standard, push: PushKeys? = nil, invites: PairInvites = PairInvites()) { diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift index 0b0699fd71a3..aa33f697205e 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/LiveCoordinator+Links.swift @@ -358,7 +358,8 @@ extension LiveCoordinator { let reply: Data? do { reply = try await LinkDeadline.run( - timeout: .seconds(15), + // This channel was opened behind auth, before its Allow prompt. + timeout: LinkPairing.authorizationTimeout + DaemonClient.defaultDeadline, operation: { try await channel.receive() }, onCancel: { await channel.close(code: "eof", reason: "version expired") } ) diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/WorkspaceSync.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/WorkspaceSync.swift index 8f396e5215f1..efba75e7b553 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/WorkspaceSync.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/WorkspaceSync.swift @@ -74,7 +74,11 @@ public final class WorkspaceSync { /// agents, list workspaces. The daemon answers a connection's lines in /// order; events after the subscribe wait in the client's stream. Does not /// wait for the answers (a Mac that pairs holds them until `auth` passes). - public nonisolated static func prepare(_ channel: any LinkChannel, deviceID: String, deviceName: String) async -> Prepared { + public nonisolated static func prepare( + _ channel: any LinkChannel, deviceID: String, deviceName: String, + authorizationTimeout: Duration = LinkPairing.authorizationTimeout, + replyTimeout: Duration = DaemonClient.defaultDeadline + ) async -> Prepared { let client = DaemonClient(channel: channel) await client.start() let replies = Task.detached { () -> [Result] in @@ -88,7 +92,9 @@ public final class WorkspaceSync { ("subscribe", ["tree_events": "deltas"]), ("list-agents", [:]), ("list-workspaces", [:]), - ]) + ], deadline: authorizationTimeout + replyTimeout) + // This eager batch can be held behind auth's human approval. Only + // setup gets that grace; subsequent daemon reads keep their normal bound. ConnectTimeline.shared.mark("first_daemon_response") return replies } diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift index 5c3bc6efe77d..433337be9445 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift @@ -94,6 +94,42 @@ import Testing #expect(channel.closed) } + @Test func eagerWorkspaceSetupWaitsForAuthorizationWithoutAnotherWrite() async throws { + let channel = DeadlineChannel() + let prepared = await WorkspaceSync.prepare(channel, deviceID: "phone", deviceName: "iPhone", + authorizationTimeout: .seconds(1), replyTimeout: .milliseconds(10)) + var writes = channel.sent.makeAsyncIterator() + let batch = try #require(await writes.next()) + // The normal response budget passes while the Mac's Allow prompt is open. + try await Task.sleep(for: .milliseconds(30)) + var replies = Data() + for line in batch.split(separator: 0x0A) { + let request = try JSONDecoder().decode(JSONValue.self, from: Data(line)) + let id = try #require(request["id"]) + replies.append(try JSONEncoder().encode(JSONValue.object(["id": id, "ok": .bool(true)]))) + replies.append(0x0A) + } + channel.push(replies) + let results = await prepared.replies.value + #expect(results.count == 5) + for result in results { _ = try result.get() } + await prepared.client.close() + #expect(await writes.next() == nil) + } + + @Test func eagerWorkspaceSetupStillExpiresWhenAuthorizationNeverCompletes() async throws { + let channel = DeadlineChannel() + let prepared = await WorkspaceSync.prepare(channel, deviceID: "phone", deviceName: "iPhone", + authorizationTimeout: .milliseconds(10), replyTimeout: .milliseconds(10)) + let results = await prepared.replies.value + #expect(results.count == 5) + for result in results { + guard case .failure(let error) = result else { Issue.record("Unanswered setup succeeded"); continue } + #expect((error as? LinkError)?.code == "reply_timeout") + } + await prepared.client.close() + } + private func opened(_ channel: DeadlineChannel) -> LinkPairing.Opened { .init(channel: channel, identity: "04aa", approval: "04bb", macNonce: "test-nonce") } @@ -106,16 +142,16 @@ private final class DeadlineChannel: LinkChannel, @unchecked Sendable { private var queue: [Data] = [] private var waiter: CheckedContinuation? private var ended = false - private let sendEvents = AsyncStream.makeStream() + private let sendEvents = AsyncStream.makeStream() private let readEvents = AsyncStream.makeStream() private let closeEvents = AsyncStream.makeStream() - var sent: AsyncStream { sendEvents.stream } + var sent: AsyncStream { sendEvents.stream } var reads: AsyncStream { readEvents.stream } var closes: AsyncStream { closeEvents.stream } var closed: Bool { lock.withLock { ended } } func send(_ data: Data) async throws { - sendEvents.continuation.yield(()) + sendEvents.continuation.yield(data) if let request = try? JSONDecoder().decode(JSONValue.self, from: data), request["method"] == .string("answer"), let id = request["id"] { push(try JSONEncoder().encode(JSONValue.object(["jsonrpc": .string("2.0"), "id": id, "result": .string("ok")]))) diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift index 547d802e7114..0c6a73e613e9 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift @@ -6,12 +6,12 @@ import Testing @Test func preCancelledRunDoesNotStartOperation() async throws { let starts = Counter() let task = Task { - try await LinkDeadline.run(timeout: .seconds(1), operation: { + withUnsafeCurrentTask { $0?.cancel() } + return try await LinkDeadline.run(timeout: .seconds(1), operation: { starts.increment() return 1 }) } - task.cancel() do { _ = try await task.value Issue.record("A pre-cancelled run succeeded") From 3f924eec76a3d57e73df27c2578c8e8a7ab98219 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 22:54:07 +0000 Subject: [PATCH 3/5] test(ios): give deadline counter a unique helper name Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-557bb07834475125d7ba --- .../Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift index 0c6a73e613e9..e4f55e0206cf 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileTransportTests/LinkDeadlineTests.swift @@ -4,7 +4,7 @@ import Testing @Suite(.timeLimit(.minutes(1))) struct LinkDeadlineTests { @Test func preCancelledRunDoesNotStartOperation() async throws { - let starts = Counter() + let starts = DeadlineInvocationCounter() let task = Task { withUnsafeCurrentTask { $0?.cancel() } return try await LinkDeadline.run(timeout: .seconds(1), operation: { @@ -67,7 +67,7 @@ import Testing } } -private final class Counter: @unchecked Sendable { +private final class DeadlineInvocationCounter: @unchecked Sendable { private let lock = NSLock() private var count = 0 func increment() { lock.withLock { count += 1 } } From 47cd5bf2d4bfe05de5f0053e43ac02bb87306867 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 31 Oct 2026 04:04:48 +0000 Subject: [PATCH 4/5] test(ios): stabilize replay resync and pairing deadlines Keep replay fixtures alive through asynchronous notifications and give the handshake test scheduling room while preserving a post-deadline approval assertion. Agent-Session: https://cmux.com/deeplink/agent?id=aid-557bb07834475125d7ba --- .../Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift | 7 ++++++- .../CmuxMobileLiveTests/LinkRequestDeadlineTests.swift | 6 ++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift index efc382e94070..4d717dcf08d0 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/AcpmuxReplayTests.swift @@ -51,6 +51,10 @@ import Testing let hub = AcpmuxHub() await hub.bind(try await link.openChannel(service: "acpmux", args: [:])) let source = hub.source(sessionID: "s1", summary: ChatOpenSnapshotLiveTests.summary) + // The hub routes through a weak-self notification pump, and its source registry + // is weak too. Keep both owners alive until the separately delivered live event + // reaches each reader, as the app's coordinator and chat store do. + defer { withExtendedLifetime((hub, source)) {} } let (first, firstTask) = ChatOpenSnapshotLiveTests.collect(source.events()) defer { firstTask.cancel() } for _ in 0..<2000 where first.events.isEmpty { await Task.yield() } @@ -76,7 +80,8 @@ import Testing "resync never repeats history into the existing reader") _ = try await hub.request("_test/live", .object([:])) - for _ in 0..<2000 where !ChatOpenSnapshotLiveTests.userMessages(second.events).contains("live") { + for _ in 0..<2000 where !ChatOpenSnapshotLiveTests.userMessages(first.events).contains("live") + || !ChatOpenSnapshotLiveTests.userMessages(second.events).contains("live") { await Task.yield() } #expect(ChatOpenSnapshotLiveTests.userMessages(first.events).contains("live")) diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift index 433337be9445..783677809095 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift @@ -55,7 +55,9 @@ import Testing @Test func approvalCanFinishAfterTheHandshakeDeadline() async throws { var pairing = PairInviteTests.pairing(PairInvites()) - pairing.handshakeTimeout = .milliseconds(10) + // Leave enough room for the first frame to be scheduled while still + // proving that approval has its own budget after the handshake expires. + pairing.handshakeTimeout = .milliseconds(100) pairing.approvalTimeout = .seconds(5) let channel = DeadlineChannel() channel.push(Data(#"{"pending":true}"#.utf8)) @@ -65,7 +67,7 @@ import Testing } var notices = waiting.stream.makeAsyncIterator() _ = await notices.next() - try await Task.sleep(for: .milliseconds(30)) + try await Task.sleep(for: .milliseconds(150)) channel.push(Data(#"{"ok":true}"#.utf8)) #expect(await task.value == .paired(macDevice: "test-mac")) waiting.continuation.finish() From 1c8696ca9805fee0244e4ab49ae4375cece2ad24 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Sat, 31 Oct 2026 04:22:37 +0000 Subject: [PATCH 5/5] fix(ios): preserve chat closure and timeout behavior after base merge Notify all active replay generations when the channel closes, preserve the existing acpmux timeout error contract, and allow a one-second handshake scheduling budget in the approval regression fixture. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-557bb07834475125d7ba --- .../Sources/CmuxMobileLive/AcpmuxConnection.swift | 1 + .../CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift | 4 ++-- .../CmuxMobileLiveTests/LinkRequestDeadlineTests.swift | 8 ++++---- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift index 12a4219ffef3..f3ac95bf5ecb 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxConnection.swift @@ -64,6 +64,7 @@ public actor AcpmuxConnection { rpcLog.debug("-> \(id) \(method, privacy: .public) \(data.count) bytes") return try await LinkDeadline.run( timeout: timeout, + timeoutError: LinkError(code: "timeout", message: "The Mac didn't answer (\(method))"), operation: { try await self.sendAndWait(id: id, data: data) }, onCancel: { await self.fail(id, CancellationError()) } ) diff --git a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift index cb66fc2ba344..db9e4dcc3948 100644 --- a/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift +++ b/cmux-next-ios/CmuxMobileKit/Sources/CmuxMobileLive/AcpmuxHub.swift @@ -117,7 +117,7 @@ public final class AcpmuxHub { self.isReady = false self.fullAccess = false self.lastError = self.lastError ?? "acpmux channel closed" - for entry in self.sources.values { entry.source?.linkDown() } + for source in self.liveSources { source.linkDown() } } do { let initialized = try await connection.request("initialize", .object([ @@ -150,7 +150,7 @@ public final class AcpmuxHub { isReady = false fullAccess = false lastError = LinkController.reason(error) - for entry in sources.values { entry.source?.linkDown() } + for source in liveSources { source.linkDown() } await connection.close() log.error("acpmux bind failed: \(String(describing: error), privacy: .public)") } diff --git a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift index 783677809095..cc95e1348a58 100644 --- a/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift +++ b/cmux-next-ios/CmuxMobileKit/Tests/CmuxMobileLiveTests/LinkRequestDeadlineTests.swift @@ -12,7 +12,7 @@ import Testing _ = try await connection.request("missing", .null, timeout: .milliseconds(20)) Issue.record("A dropped reply succeeded") } catch let error as LinkError { - #expect(error.code == "reply_timeout") + #expect(error.code == "timeout") } // The old response cannot answer the next request. channel.push(Data(#"{"jsonrpc":"2.0","id":1,"result":"late"}"#.utf8)) @@ -57,8 +57,8 @@ import Testing var pairing = PairInviteTests.pairing(PairInvites()) // Leave enough room for the first frame to be scheduled while still // proving that approval has its own budget after the handshake expires. - pairing.handshakeTimeout = .milliseconds(100) - pairing.approvalTimeout = .seconds(5) + pairing.handshakeTimeout = .seconds(1) + pairing.approvalTimeout = .seconds(10) let channel = DeadlineChannel() channel.push(Data(#"{"pending":true}"#.utf8)) let waiting = AsyncStream.makeStream() @@ -67,7 +67,7 @@ import Testing } var notices = waiting.stream.makeAsyncIterator() _ = await notices.next() - try await Task.sleep(for: .milliseconds(150)) + try await Task.sleep(for: .milliseconds(1100)) channel.push(Data(#"{"ok":true}"#.utf8)) #expect(await task.value == .paired(macDevice: "test-mac")) waiting.continuation.finish()