diff --git a/Sources/CoalesceLatestPublisher.swift b/Sources/CoalesceLatestPublisher.swift new file mode 100644 index 000000000000..5299ef4fb719 --- /dev/null +++ b/Sources/CoalesceLatestPublisher.swift @@ -0,0 +1,158 @@ +import Combine +import Foundation + +// CoalesceLatestPublisher and CoalesceLatestInner are one operator: the +// Inner is the publisher's subscription and shares its file-private access +// boundary. Splitting the Inner into its own file would force widening +// `private` to `internal` for an implementation detail, so the pair +// intentionally lives in this single file. + +// MARK: - Leading-edge coalescing + +extension Publisher where Failure == Never { + /// Coalesces bursts while keeping the leading edge synchronous. + /// + /// Combine's `throttle` schedules every emission, including the first, + /// onto the scheduler, so even an isolated value is deferred to the next + /// run-loop turn and a subscriber never observes a synchronous emission. + /// The sidebar's immediate observation contract requires the opposite: + /// the current-state replay a subscriber receives from `@Published` + /// upstreams, and the first change after an idle period, must both arrive + /// in the same run-loop turn; only the tail of a burst may be deferred. + /// + /// Semantics per subscription: + /// - The first value (the `@Published` replay of current state) is + /// forwarded synchronously and does not open a coalesce window, so a + /// change made right after subscribing is still synchronous. + /// - A value arriving when no window is open is forwarded synchronously + /// and opens a window of `interval`. + /// - Values arriving inside an open window are coalesced: the latest one + /// is emitted when the window closes (on `scheduler`), which opens the + /// next window. + /// + /// Not thread-safe: intended for main-thread streams with `RunLoop.main`. + /// Downstream demand is ignored (sink-style subscribers only). + func coalesceLatest( + for interval: Context.SchedulerTimeType.Stride, + scheduler: Context + ) -> AnyPublisher { + CoalesceLatestPublisher(upstream: self, interval: interval, scheduler: scheduler) + .eraseToAnyPublisher() + } +} + +private struct CoalesceLatestPublisher: Publisher + where Upstream.Failure == Never { + typealias Output = Upstream.Output + typealias Failure = Never + + let upstream: Upstream + let interval: Context.SchedulerTimeType.Stride + let scheduler: Context + + func receive(subscriber: S) where S.Input == Output, S.Failure == Never { + upstream.subscribe(CoalesceLatestInner( + downstream: subscriber, + interval: interval, + scheduler: scheduler + )) + } +} + +private final class CoalesceLatestInner: Subscriber, Subscription + where Downstream.Failure == Never { + typealias Input = Downstream.Input + typealias Failure = Never + + private let downstream: Downstream + private let interval: Context.SchedulerTimeType.Stride + private let scheduler: Context + private var upstreamSubscription: Subscription? + private var hasReceivedReplay = false + private var windowStart: Context.SchedulerTimeType? + private var pendingValue: Input? + private var trailingScheduled = false + private var isCancelled = false + + init(downstream: Downstream, interval: Context.SchedulerTimeType.Stride, scheduler: Context) { + self.downstream = downstream + self.interval = interval + self.scheduler = scheduler + } + + func receive(subscription: Subscription) { + upstreamSubscription = subscription + downstream.receive(subscription: self) + subscription.request(.unlimited) + } + + func receive(_ input: Input) -> Subscribers.Demand { + guard !isCancelled else { return .none } + if !hasReceivedReplay { + hasReceivedReplay = true + _ = downstream.receive(input) + return .none + } + let now = scheduler.now + if let start = windowStart, now < start.advanced(by: interval) { + pendingValue = input + scheduleTrailingEmission(at: start.advanced(by: interval)) + } else { + // If a trailing emission was scheduled but its callback is + // overdue (main run loop stalled past the deadline), this newer + // value supersedes the stale pending one; drop it so the late + // callback cannot emit it out of order after this value. + pendingValue = nil + windowStart = now + _ = downstream.receive(input) + } + return .none + } + + func receive(completion: Subscribers.Completion) { + guard !isCancelled else { return } + if let value = pendingValue { + pendingValue = nil + _ = downstream.receive(value) + } + downstream.receive(completion: completion) + } + + private func scheduleTrailingEmission(at deadline: Context.SchedulerTimeType) { + guard !trailingScheduled else { return } + trailingScheduled = true + scheduler.schedule(after: deadline) { [weak self] in + self?.emitTrailing() + } + } + + private func emitTrailing() { + trailingScheduled = false + guard !isCancelled, let value = pendingValue else { return } + // An overdue callback may fire inside a window that a newer leading + // value opened; hold the pending value until that window's own + // deadline instead of emitting early. + if let start = windowStart { + let deadline = start.advanced(by: interval) + if scheduler.now < deadline { + scheduleTrailingEmission(at: deadline) + return + } + } + pendingValue = nil + windowStart = scheduler.now + _ = downstream.receive(value) + } + + func request(_ demand: Subscribers.Demand) { + // Downstream demand is intentionally ignored; this operator backs + // sink-style Void observation streams with unlimited demand. + } + + func cancel() { + isCancelled = true + pendingValue = nil + upstreamSubscription?.cancel() + upstreamSubscription = nil + } +} diff --git a/Sources/ContentView.swift b/Sources/ContentView.swift index 7411eeec29b1..789b5226db5b 100644 --- a/Sources/ContentView.swift +++ b/Sources/ContentView.swift @@ -10969,11 +10969,8 @@ struct VerticalTabsSidebar: View { return } - extensionSidebarImmediateObservationPublisher = Publishers.MergeMany( - tabs.map { $0.sidebarImmediateObservationPublisher } - ) - .receive(on: RunLoop.main) - .eraseToAnyPublisher() + extensionSidebarImmediateObservationPublisher = + Workspace.mergedImmediateObservationPublisher(for: tabs) extensionSidebarDebouncedObservationPublisher = Publishers.MergeMany( tabs.map { $0.sidebarObservationPublisher } ) diff --git a/Sources/WorkspaceSidebarObservation.swift b/Sources/WorkspaceSidebarObservation.swift index 4c304562d14a..58fe421d6ac6 100644 --- a/Sources/WorkspaceSidebarObservation.swift +++ b/Sources/WorkspaceSidebarObservation.swift @@ -60,6 +60,19 @@ private struct SidebarObservationState: Equatable { } extension Workspace { + // Leading-edge coalescing for the immediate sidebar observation stream. + // Every subscription (a sidebar row, the MergeMany extension-sidebar + // aggregate) fires a full makeWorkspaceSnapshot() rebuild per emission. + // Agents (e.g. Codex) rewrite a workspace title every turn, and + // removeDuplicates() cannot collapse distinct titles, so without coalescing + // each rewrite drives a snapshot rebuild per consumer per workspace. + // coalesceLatest (below) keeps the first change in a burst synchronous + // (a user pin/color/title edit stays immediate, which Combine's throttle + // cannot guarantee because it schedules every emission onto the scheduler) + // and collapses the tail of the burst into one trailing emission per window. + // See https://github.com/manaflow-ai/cmux/issues/4127. + static let sidebarImmediateObservationCoalesceInterval: RunLoop.SchedulerTimeType.Stride = .milliseconds(50) + func makeSidebarImmediateObservationPublisher() -> AnyPublisher { let workspaceFields = Publishers.CombineLatest4( $title, @@ -87,10 +100,30 @@ extension Workspace { ) } .removeDuplicates() + .coalesceLatest( + for: Self.sidebarImmediateObservationCoalesceInterval, + scheduler: RunLoop.main + ) .map { _ in () } .eraseToAnyPublisher() } + /// Merged immediate observation across workspaces for the extension + /// sidebar. Coalesced again across the merge: per-workspace coalescing + /// caps each stream, but N workspaces bursting concurrently would still + /// re-render the whole extension sidebar once per workspace per window. + /// The leading edge stays synchronous, so a lone change is as immediate + /// as before. + static func mergedImmediateObservationPublisher(for workspaces: [Workspace]) -> AnyPublisher { + Publishers.MergeMany(workspaces.map { $0.sidebarImmediateObservationPublisher }) + .receive(on: RunLoop.main) + .coalesceLatest( + for: sidebarImmediateObservationCoalesceInterval, + scheduler: RunLoop.main + ) + .eraseToAnyPublisher() + } + func makeSidebarObservationPublisher() -> AnyPublisher { let workspaceFields = Publishers.CombineLatest4( $currentDirectory, diff --git a/cmux.xcodeproj/project.pbxproj b/cmux.xcodeproj/project.pbxproj index dd9f6338c557..63a05d2d17fd 100644 --- a/cmux.xcodeproj/project.pbxproj +++ b/cmux.xcodeproj/project.pbxproj @@ -407,6 +407,7 @@ 5E55200000000000000000B3 /* CmuxWindowing in Frameworks */ = {isa = PBXBuildFile; productRef = 5E55200000000000000000B2 /* CmuxWindowing */; }; E30750000000000000000004 /* CmuxWorkspaceDefinition.swift in Sources */ = {isa = PBXBuildFile; fileRef = E30750000000000000000003 /* CmuxWorkspaceDefinition.swift */; }; E3B7A3000000000000000103 /* CmuxWorkspaces in Frameworks */ = {isa = PBXBuildFile; productRef = E3B7A3000000000000000102 /* CmuxWorkspaces */; }; + C0A1E5CE01C0A1E5CE01A001 /* CoalesceLatestPublisher.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0A1E5CE01C0A1E5CE01A002 /* CoalesceLatestPublisher.swift */; }; A9F200000000000000000016 /* CodexAppServerQueuedInput.swift in Sources */ = {isa = PBXBuildFile; fileRef = A9F100000000000000000016 /* CodexAppServerQueuedInput.swift */; }; A9E02000000000000000000E /* CodexAppServerSession.swift in Sources */ = {isa = PBXBuildFile; fileRef = A9E01000000000000000000E /* CodexAppServerSession.swift */; }; A9E040000000000000000002 /* CodexAppServerSessionTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = A9E040000000000000000001 /* CodexAppServerSessionTests.swift */; }; @@ -1651,6 +1652,7 @@ C0DE58990000000000000002 /* CmuxWebViewKeyDownReentryTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CmuxWebViewKeyDownReentryTests.swift; sourceTree = ""; }; 43F90FAF3FD44F11BF547BE9 /* CmuxWebViewMouseNavigationButtonTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CmuxWebViewMouseNavigationButtonTests.swift; sourceTree = ""; }; E30750000000000000000003 /* CmuxWorkspaceDefinition.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CmuxWorkspaceDefinition.swift; sourceTree = ""; }; + C0A1E5CE01C0A1E5CE01A002 /* CoalesceLatestPublisher.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CoalesceLatestPublisher.swift; sourceTree = ""; }; A9F100000000000000000016 /* CodexAppServerQueuedInput.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Panels/CodexAppServerQueuedInput.swift; sourceTree = ""; }; A9E01000000000000000000E /* CodexAppServerSession.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Panels/CodexAppServerSession.swift; sourceTree = ""; }; A9E040000000000000000001 /* CodexAppServerSessionTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CodexAppServerSessionTests.swift; sourceTree = ""; }; @@ -2919,6 +2921,7 @@ A5001520 /* PostHogAnalytics.swift */, A5001416 /* Workspace.swift */, 6799A0026799A0026799A002 /* WorkspaceSidebarAgentRuntimeObservationModel.swift */, + C0A1E5CE01C0A1E5CE01A002 /* CoalesceLatestPublisher.swift */, 5659A0025659A0025659A002 /* WorkspaceSidebarObservation.swift */, 5659B0025659B0025659B002 /* WorkspaceSidebarLogEntryLimitProvider.swift */, D7AB3605C10DEF0000000004 /* WorkspaceCloseTabsBatching.swift */, @@ -4438,6 +4441,7 @@ C67540030000000000000001 /* CmuxWebView+SubframeDownloadIntentScript.swift in Sources */, A5001500 /* CmuxWebView.swift in Sources */, E30750000000000000000004 /* CmuxWorkspaceDefinition.swift in Sources */, + C0A1E5CE01C0A1E5CE01A001 /* CoalesceLatestPublisher.swift in Sources */, A9F200000000000000000016 /* CodexAppServerQueuedInput.swift in Sources */, A9E02000000000000000000E /* CodexAppServerSession.swift in Sources */, C4041001000000000000001B /* CommandClickFileOpenRouter.swift in Sources */, diff --git a/cmuxTests/WorkspaceSidebarObservationTests.swift b/cmuxTests/WorkspaceSidebarObservationTests.swift index 791bda6b9417..fa140cae821d 100644 --- a/cmuxTests/WorkspaceSidebarObservationTests.swift +++ b/cmuxTests/WorkspaceSidebarObservationTests.swift @@ -118,6 +118,116 @@ struct WorkspaceSidebarObservationTests { ) } + @Test func sidebarImmediateObservationPublisherDeliversFirstChangeSynchronously() { + let workspace = Workspace() + + var publishCount = 0 + let cancellable = workspace.sidebarImmediateObservationPublisher.sink { + publishCount += 1 + } + defer { cancellable.cancel() } + publishCount = 0 + + workspace.title = "User Edit" + + #expect( + publishCount == 1, + "The first immediate-field change after subscribing must reach the sidebar in the same run-loop turn; coalescing may only defer the tail of a burst." + ) + } + + @Test func sidebarImmediateObservationPublisherCoalescesTitleBursts() { + let workspace = Workspace() + + var publishCount = 0 + let cancellable = workspace.sidebarImmediateObservationPublisher.sink { + publishCount += 1 + } + defer { cancellable.cancel() } + publishCount = 0 + + for turn in 0..<20 { + workspace.title = "Agent Turn \(turn)" + } + + #expect( + publishCount == 1, + "A synchronous burst of distinct titles must deliver only its leading edge immediately." + ) + + // Generous pump so the 50ms trailing emission fires deterministically. + RunLoop.main.run(until: Date().addingTimeInterval(0.3)) + + #expect( + publishCount == 2, + "A coalesced burst must settle with exactly one trailing emission carrying the latest state." + ) + } + + @Test func coalesceLatestKeepsLeadingEdgeSynchronousAndEmitsLatestTrailing() { + let subject = PassthroughSubject() + var received: [Int] = [] + let cancellable = subject + .coalesceLatest(for: .milliseconds(50), scheduler: RunLoop.main) + .sink { received.append($0) } + defer { cancellable.cancel() } + + // First value models the @Published current-state replay: forwarded + // synchronously without opening a coalesce window. + subject.send(1) + #expect(received == [1]) + + // First change is the synchronous leading edge and opens the window. + subject.send(2) + #expect(received == [1, 2]) + + // Burst inside the window coalesces to the latest value. + subject.send(3) + subject.send(4) + subject.send(5) + #expect(received == [1, 2]) + + RunLoop.main.run(until: Date().addingTimeInterval(0.3)) + #expect(received == [1, 2, 5]) + + // After the window closes and the trailing window expires, the next + // value is synchronous again. + subject.send(6) + #expect(received == [1, 2, 5, 6]) + } + + @Test func coalesceLatestDropsStalePendingValueWhenLeadingSupersedesOverdueTrailing() { + let scheduler = VirtualCoalesceScheduler() + let subject = PassthroughSubject() + var received: [Int] = [] + let cancellable = subject + .coalesceLatest(for: .milliseconds(50), scheduler: scheduler) + .sink { received.append($0) } + defer { cancellable.cancel() } + + subject.send(1) // replay: forwarded, no window + subject.send(2) // leading edge: opens window + subject.send(3) // pending trailing value for the open window + #expect(received == [1, 2]) + #expect(scheduler.scheduledActionCount == 1) + + // The deadline passes WITHOUT the scheduled callback running, + // modeling a stalled main run loop with an overdue timer. + scheduler.advance(by: 0.12) + subject.send(4) // deadline passed: new leading edge must supersede 3 + + #expect( + received == [1, 2, 4], + "A newer leading value after an overdue deadline must drop the stale pending value." + ) + + scheduler.runScheduledActions() + #expect( + received == [1, 2, 4], + "The overdue trailing callback must not emit the superseded stale value out of order." + ) + } + @Test func sidebarObservationPublisherIgnoresRemoteHeartbeatOnlyChanges() { let workspace = Workspace() @@ -146,3 +256,51 @@ private final class ObservationChangeFlag: @unchecked Sendable { fired = true } } + +// Deterministic Combine scheduler for coalesceLatest tests: `now` only moves +// via advance(by:), and scheduled actions run only when runScheduledActions() +// is called, so overdue-timer interleavings are exact instead of wall-clock. +private final class VirtualCoalesceScheduler: Scheduler { + typealias SchedulerTimeType = RunLoop.SchedulerTimeType + typealias SchedulerOptions = Never + + private(set) var now = SchedulerTimeType(Date(timeIntervalSinceReferenceDate: 0)) + var minimumTolerance: SchedulerTimeType.Stride { .seconds(0) } + private var scheduledActions: [() -> Void] = [] + + var scheduledActionCount: Int { scheduledActions.count } + + func advance(by seconds: TimeInterval) { + now = SchedulerTimeType(now.date.addingTimeInterval(seconds)) + } + + func runScheduledActions() { + let actions = scheduledActions + scheduledActions = [] + actions.forEach { $0() } + } + + func schedule(options: Never?, _ action: @escaping () -> Void) { + action() + } + + func schedule( + after date: SchedulerTimeType, + tolerance: SchedulerTimeType.Stride, + options: Never?, + _ action: @escaping () -> Void + ) { + scheduledActions.append(action) + } + + func schedule( + after date: SchedulerTimeType, + interval: SchedulerTimeType.Stride, + tolerance: SchedulerTimeType.Stride, + options: Never?, + _ action: @escaping () -> Void + ) -> Cancellable { + scheduledActions.append(action) + return AnyCancellable {} + } +}