Chats: acpmux owns the chat list; the sidebar history holds only its window (P0 stall)

Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed.swift → Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/ChatsFeed.swift

@@ -1,26 +1,38 @@
 import CmuxNextAgentActivity
-import CmuxNextAgentPane
-import CmuxNextWakeups
-import Foundation
+public import CmuxNextAgentPane
+public import CmuxNextWakeups
+public import Foundation
 
-/// The device-wide acpmux chat index for sidebar and palette clients.
+/// The app's connection to acpmux's device-wide chat index: the windows native lists show
+/// (``AcpmuxChatWindows``: acpmux sorts, filters and pages; the main thread holds only visible rows
+/// and applies row diffs), and, until the palette, Open Chat and the New Tab cards ask acpmux
+/// themselves (P-RUST step 2), the index mirror they read.
 @MainActor
-final class ChatsFeed {
-    /// The mirror (a copy-on-write value: taking it is O(1), so readers that
-    /// need every chat take it off the main actor).
-    private(set) var store = AcpmuxChatsStore()
+public final class ChatsFeed {
+    /// The windows over the index that native lists show (the sidebar's chat history).
+    public let windows = AcpmuxChatWindows()
+    /// The mirror. Only this feed holds it: a change applies in place (O(log n)), never a copy of
+    /// every chat (v2.2.9: 1,148 ms in `apply` while the sidebar history held the same storage).
+    /// No reader may keep it; whole-index work runs in acpmux (`_acpmux/chat_projects`).
+    private var store = AcpmuxChatsStore()
     /// Every chat, newest first. O(n) per read: per-change readers use
     /// `chat(id:)` and `newest(_:)`; a publish no longer builds this list.
-    var chats: [AcpmuxChat] { store.chats }
-    private(set) var isEnabled = true
-    private(set) var isReady = false
+    public var chats: [AcpmuxChat] { store.chats }
+    public private(set) var isEnabled = true
+    public private(set) var isReady = false
     private let socket: String
+    /// Each drain's line count and main-actor time (the App's debug timings).
+    private let drained: @MainActor (Int, Duration) -> Void
+    /// The acpmux socket this feed reads (ProjectsImport asks it for the grouped projects).
+    public var socketPath: String { socket }
     private var observers: [(owner: () -> AnyObject?, changed: @MainActor () -> Void)] = []
     private var subscription: AgentActivityLineConnection?
     private var directoryWatch: (any DispatchSourceFileSystemObject)?
     private var watchedDirectory: String?
     private var reconnect: Task<Void, Never>?
     private var requestID = 2
+    /// The open window requests by JSON-RPC id, so an error answer names its window.
+    private var windowRequests: [Int: String] = [:]
     /// The seq of the last index change in the mirror (acpmux numbers every change; the page
     /// names the last one it holds). Nil from a daemon that does not number them.
     private var seq: UInt64?
@@ -34,13 +46,19 @@ final class ChatsFeed {
     private var early: [(change: AcpmuxChatsStore.Change, seq: UInt64?)] = []
     private var backoff = Backoff(initial: .milliseconds(250), maximum: .seconds(30))
 
-    init(socketPath: String) { socket = socketPath }
+    /// - Parameters:
+    ///   - socketPath: acpmux's socket.
+    ///   - drained: Each drain's line count and main-actor time (the App's debug timings).
+    public init(socketPath: String, drained: @escaping @MainActor (Int, Duration) -> Void = { _, _ in }) {
+        socket = socketPath
+        self.drained = drained
+    }
 
     /// One chat by key, O(1).
-    func chat(id: String) -> AcpmuxChat? { store.chat(id: id) }
+    public func chat(id: String) -> AcpmuxChat? { store.chat(id: id) }
 
     /// The newest `limit` chats, O(limit).
-    func newest(_ limit: Int) -> [AcpmuxChat] { store.newest(limit) }
+    public func newest(_ limit: Int) -> [AcpmuxChat] { store.newest(limit) }
 
     // cx-t7hu: not an isolated deinit; its runtime hop traps when released inside an action's task-local scope.
     // Each handle here is Sendable and stops from any thread: a release off main cleans up too.
@@ -51,14 +69,14 @@ final class ChatsFeed {
     }
 
     /// Starts one long-lived watch for the first live observer.
-    func observe(_ owner: AnyObject, _ changed: @escaping @MainActor () -> Void) {
+    public func observe(_ owner: AnyObject, _ changed: @escaping @MainActor () -> Void) {
         observers.append(({ [weak owner] in owner }, changed))
         if subscription == nil, directoryWatch == nil, reconnect == nil { connect() }
     }
 
     /// Keeps the mirror current for the palette chats page and Open Chat while
     /// no sidebar shows Chats: one idle push connection, no timers.
-    func keepCurrent() {
+    public func keepCurrent() {
         observe(self) {}
     }
 
@@ -79,20 +97,39 @@ final class ChatsFeed {
         connection.start(send: Self.watchRequest(limit: 5000), decode: Message.init(line:)) { [weak self, weak connection] drain in
             if let connection { self?.apply(drain, from: connection) }
         }
+        // The windows' specs go after the watch request, on the same ordered connection.
+        windows.connected { [weak self, weak connection] method, params in
+            guard let self, let connection, subscription === connection else { return }
+            send(method, params, on: connection)
+        }
     }
 
     /// One decoded line of the watch.
-    nonisolated enum Message: Sendable {
+    public nonisolated enum Message: Sendable {
         /// A page response: the whole mirror, decoded off the main actor in acpmux's order.
         case page(store: AcpmuxChatsStore, ready: Bool, enabled: Bool, seq: UInt64?)
         /// One index change and its seq (nil for activity, which is not an index change).
         case change(AcpmuxChatsStore.Change, seq: UInt64?)
+        /// A window's rows or row diff (`_acpmux/chats_window_changed`).
+        case window(AcpmuxChatWindows.Update)
+        /// The answer to request `id` (a window's open or close), and whether it is an error.
+        case answered(id: Int, failed: Bool)
+        /// The daemon dropped a notification or the watch lagged: list and ask every window again.
         case lagged
 
-        init?(line: Data) {
+        public init?(line: Data) {
             guard let message = try? JSONSerialization.jsonObject(with: line) as? [String: Any] else { return nil }
+            if let id = (message["id"] as? NSNumber)?.intValue, id >= 3, message["error"] != nil {
+                self = .answered(id: id, failed: true)
+                return
+            }
             if let id = (message["id"] as? NSNumber)?.intValue, id == 2 || id >= 3,
                let result = message["result"] as? [String: Any] {
+                // A window request's answer: its rows come as `_acpmux/chats_window_changed`.
+                guard result["window"] == nil else {
+                    self = .answered(id: id, failed: false)
+                    return
+                }
                 var store = AcpmuxChatsStore()
                 store.reset(result)
                 self = .page(store: store, ready: result["ready"] as? Bool ?? false, enabled: result["enabled"] as? Bool ?? true,
@@ -104,7 +141,14 @@ final class ChatsFeed {
                 let params = message["params"] as? [String: Any]
                 guard let change = params.flatMap(AcpmuxChatsStore.change(from:)) else { return nil }
                 self = .change(change, seq: (params?["seq"] as? NSNumber)?.uint64Value)
-            case "_acpmux/chats_lagged":
+            case "_acpmux/chats_window_changed":
+                guard let update = (message["params"] as? [String: Any]).flatMap(AcpmuxChatWindows.Update.init(params:)) else { return nil }
+                self = .window(update)
+            case "_acpmux/lagged" where (message["params"] as? [String: Any])?["reason"] as? String == "message_too_large":
+                // A message past one frame never fits, so asking again would send it again forever:
+                // the next change of the index or of its window sends that window again.
+                return nil
+            case "_acpmux/chats_lagged", "_acpmux/lagged":
                 self = .lagged
             default:
                 return nil
@@ -131,7 +175,7 @@ final class ChatsFeed {
     private func apply(_ drain: MainActorLineDrain<Message>, from connection: AgentActivityLineConnection) {
         guard subscription === connection else { return }
         let started = ContinuousClock.now
-        defer { DebugTimings.chatsFeedApplied(lines: drain.values.count, ContinuousClock.now - started) }
+        defer { drained(drain.values.count, ContinuousClock.now - started) }
         var changed = false
         var lagged = false
         for message in drain.values {
@@ -156,12 +200,20 @@ final class ChatsFeed {
                 }
                 if !applyChange(change, seq: changeSeq) { lagged = true }
                 changed = true
+            case .window(let update):
+                windows.apply(update)
+            case .answered(let id, let failed):
+                if let window = windowRequests.removeValue(forKey: id), failed { windows.requestFailed(window) }
             case .lagged:
                 lagged = true
+                windows.resync()
             }
         }
         // Dropped lines (a full batch) leave the mirror incomplete, like a lag: fetch a fresh page.
         if lagged || drain.overflowed, !drain.closed { requestList() }
+        if drain.overflowed, !drain.closed { windows.resync() }
+        // The daemon answers again: a window it refused (busy) is asked again.
+        if changed { windows.retryFailed() }
         if changed { publish() }
         if drain.closed { lost(connection) }
     }
@@ -191,13 +243,18 @@ final class ChatsFeed {
             return
         }
         pageInFlight = true
+        send("_acpmux/chats", ["limit": 5000], on: subscription)
+    }
+
+    /// One request on the watch's connection.
+    private func send(_ method: String, _ params: [String: any Sendable], on connection: AgentActivityLineConnection) {
         requestID += 1
-        let request: [String: Any] = ["jsonrpc": "2.0", "id": requestID, "method": "_acpmux/chats",
-                                       "params": ["limit": 5000]]
+        if method == "_acpmux/chats_window", let window = params["window"] as? String { windowRequests[requestID] = window }
+        let request: [String: Any] = ["jsonrpc": "2.0", "id": requestID, "method": method, "params": params as [String: Any]]
         guard let data = try? JSONSerialization.data(withJSONObject: request) else { return }
         var line = data
         line.append(0x0A)
-        subscription.send(line)
+        connection.send(line)
     }
 
     private func publish() {
@@ -208,6 +265,8 @@ final class ChatsFeed {
     private func lost(_ connection: AgentActivityLineConnection) {
         guard subscription === connection else { return }
         subscription = nil
+        windowRequests = [:]
+        windows.disconnected()
         reconnect?.cancel()
         reconnect = Task { [weak self] in
             guard var backoff = self?.backoff else { return }
@@ -239,4 +298,4 @@ final class ChatsFeed {
 }
 
 /// Compatibility for code and extensions that still name the old feed.
-typealias AgentRecentsFeed = ChatsFeed
+public typealias AgentRecentsFeed = ChatsFeed

Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/SidebarChatHistoryFeed.swift

@@ -7,40 +7,92 @@ public import CmuxNextDaemon
 import CmuxNextDesign
 public import CmuxNextSidebar
 import Foundation
-public import Observation
 
 /// Feeds the session tree's chat history (`sidebar.sessionTree`, hq#2569 #6) from acpmux's chat
-/// index (the app's Chats feed mirror, metadata only): the pinned chats, and Earlier, the newest
-/// chats that no open workspace of this Mac shows, one page at a time. A workspace shows a chat
-/// when one of its agent tabs runs an acpmux session that continues it (the lane's `chat`).
-/// Off, it leaves `SidebarChatHistory.state` nil.
-@MainActor @Observable
+/// index: the pinned chats, and Earlier, the newest chats that no open workspace of this Mac shows,
+/// one page at a time. A workspace shows a chat when one of its agent tabs runs an acpmux session
+/// that continues it (the lane's `chat`). Off, it leaves `SidebarChatHistory.state` nil.
+///
+/// acpmux owns the list (P-RUST, layer-ownership.md rules 1-3): Earlier and Pinned are two windows
+/// over its index (``AcpmuxChatWindows``), sorted, filtered and paged in Rust. The main thread holds
+/// only their visible rows and applies one row diff per change; it never holds, copies or compares
+/// the whole index (v2.2.9: 730 ms in the old whole-index setter's `==`).
+@MainActor
 public final class SidebarChatHistoryFeed {
     public static let shared = SidebarChatHistoryFeed()
 
-    /// The chat index mirror. The app sets it on each index change (a copy-on-write value).
-    public var chats = AcpmuxChatsStore()
-    @ObservationIgnored private var task: Task<Void, Never>?
+    /// Earlier's window: the newest top-level chats that no open workspace shows and no pin holds.
+    public nonisolated static let earlierWindow = "sidebar.history.earlier"
+    /// Pinned's window: the pinned chats, in pin order.
+    public nonisolated static let pinnedWindow = "sidebar.history.pinned"
 
-    /// Follows the setting, the index, the lanes, every machine's `workspaces` and the user's pins
-    /// from now on; a row opens its chat with `open` (the app's Open Chat path).
+    private var task: Task<Void, Never>?
+
+    /// Follows the setting, acpmux's windows, the lanes, every machine's `workspaces` and the
+    /// user's pins from now on; a row opens its chat with `open` (the app's Open Chat path).
     public func start(workspaces: @escaping @MainActor @Sendable () -> [WorkspaceModel], localHost: String?,
-                      open: @escaping @MainActor (String) -> Void, lanes: AgentLaneStore = .shared,
-                      history: SidebarChatHistory = .shared) {
+                      open: @escaping @MainActor (String) -> Void, windows: AcpmuxChatWindows,
+                      lanes: AgentLaneStore = .shared, history: SidebarChatHistory = .shared) {
         guard task == nil else { return }
         history.onOpen = open
-        // task-owner: the app (one feed for the app's life); event-driven (Observation)
-        task = Task { [self] in
-            for await state in ObservationStream({ [self] () -> SidebarChatHistoryState? in
+        let earlier = windows.window(Self.earlierWindow)
+        let pinned = windows.window(Self.pinnedWindow)
+        // task-owner: the app (one feed for the app's life); event-driven (Observation). One value
+        // per change of what the history shows besides the index (the windows' specs for acpmux)
+        // or of a window's rows (the state, O(visible rows)).
+        task = Task {
+            for await shown in ObservationStream({ () -> Shown? in
                 guard DesignSettings.shared.sidebarSections.sessionTree else { return nil }
-                let live = Self.liveChats(workspaces(), lanes: lanes.lanes, localHost: localHost)
-                return Self.state(chats, live: live, pinned: history.pinned, limit: history.limit, folded: history.earlierFolded)
-            }) where history.state != state {
-                history.state = state
+                let inputs = Inputs(live: Self.liveChats(workspaces(), lanes: lanes.lanes, localHost: localHost),
+                                    pinned: history.pinned, limit: history.limit, folded: history.earlierFolded)
+                return Shown(inputs: inputs, state: Self.state(earlier: earlier.rows, pinned: pinned.rows, inputs: inputs))
+            }) {
+                if let shown {
+                    windows.open(shown.inputs.earlierSpec)
+                    windows.open(shown.inputs.pinnedSpec)
+                } else {
+                    windows.close(Self.earlierWindow)
+                    windows.close(Self.pinnedWindow)
+                }
+                if history.state != shown?.state { history.state = shown?.state }
             }
         }
     }
 
+    /// The history's inputs and its state from them and the windows' rows.
+    nonisolated struct Shown: Equatable, Sendable {
+        var inputs: Inputs
+        var state: SidebarChatHistoryState
+    }
+
+    /// What the history shows, besides the index: the chats open workspaces show, the pins, the page.
+    nonisolated struct Inputs: Equatable, Sendable {
+        var live: Set<String>
+        var pinned: [String]
+        var limit: Int
+        var folded: Bool
+
+        /// Earlier's window: one more row than the page, so the history knows Show More; one row
+        /// while folded, so its header knows whether Earlier has a chat.
+        var earlierSpec: AcpmuxChatWindows.Spec {
+            AcpmuxChatWindows.Spec(window: SidebarChatHistoryFeed.earlierWindow, limit: folded ? 1 : max(0, limit) + 1,
+                                   topLevel: true, exclude: live.union(pinned).sorted())
+        }
+
+        var pinnedSpec: AcpmuxChatWindows.Spec {
+            AcpmuxChatWindows.Spec(window: SidebarChatHistoryFeed.pinnedWindow, limit: max(1, pinned.count), topLevel: false,
+                                   keys: pinned)
+        }
+    }
+
+    /// The history of the two windows' rows. The local skip (`live`, `pinned`) holds only until
+    /// acpmux answers a new spec, so a chat just pinned never lists twice.
+    static func state(earlier: [AcpmuxChat], pinned: [AcpmuxChat], inputs: Inputs) -> SidebarChatHistoryState {
+        let pinnedByID = Dictionary(pinned.map { ($0.id, historyChat($0)) }, uniquingKeysWith: { first, _ in first })
+        return SidebarChatHistoryState.make(newest: earlier.lazy.map(historyChat), chat: { pinnedByID[$0] }, live: inputs.live,
+                                            pinned: inputs.pinned, limit: inputs.limit, folded: inputs.folded)
+    }
+
     /// The chats an open workspace shows: those its agent tabs' acpmux sessions on this Mac continue.
     static func liveChats(_ workspaces: [WorkspaceModel], lanes: [String: AgentActivityLane], localHost: String?) -> Set<String> {
         var live: Set<String> = []
@@ -71,12 +123,6 @@ public final class SidebarChatHistoryFeed {
         }
     }
 
-    /// The history of `index`: reads the index lazily, newest first, only as far as the page needs.
-    static func state(_ index: AcpmuxChatsStore, live: Set<String>, pinned: [String], limit: Int, folded: Bool) -> SidebarChatHistoryState {
-        SidebarChatHistoryState.make(newest: index.newestFirst.lazy.map(historyChat), chat: { index.chat(id: $0).map(historyChat) },
-                                     live: live, pinned: pinned, limit: limit, folded: folded)
-    }
-
     /// A chat as the history lists it: its title, else its folder's name, else its harness.
     static func historyChat(_ chat: AcpmuxChat) -> SidebarHistoryChat {
         let folder = chat.cwd.map { ($0 as NSString).lastPathComponent }.flatMap { $0.isEmpty ? nil : $0 }