diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/ChatsFeed.swift similarity index 66% rename from Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed.swift rename to Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/ChatsFeed.swift index 315e1566ab24..049944085776 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed.swift +++ b/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? 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, 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 diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/SidebarChatHistoryFeed.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/SidebarChatHistoryFeed.swift index d82eaff70dda..807da40c1fca 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/SidebarChatHistoryFeed.swift +++ b/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? + /// 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? + + /// 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 + 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 { var live: Set = [] @@ -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, 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 } diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatProjects.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatProjects.swift new file mode 100644 index 000000000000..ffe335e4b233 --- /dev/null +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatProjects.swift @@ -0,0 +1,53 @@ +public import Foundation + +/// acpmux's chat index grouped into each harness's project folders (`_acpmux/chat_projects`): +/// every absolute chat folder with the time of its newest chat. acpmux groups its own index, so the +/// app reads a few rows, never every chat (P-RUST: no whole-index mirror in Swift). +public nonisolated struct AcpmuxChatProjects: Sendable, Equatable { + public nonisolated struct Project: Sendable, Equatable { + public let harness: String + public let path: String + public let lastUsedMs: Int64 + + public init(harness: String, path: String, lastUsedMs: Int64) { + self.harness = harness + self.path = path + self.lastUsedMs = lastUsedMs + } + } + + /// The first scan finished. + public var ready: Bool + /// Chats are on (`agents.chats.enabled`). + public var enabled: Bool + /// By harness, then path. + public var projects: [Project] + + public init(ready: Bool, enabled: Bool, projects: [Project]) { + self.ready = ready + self.enabled = enabled + self.projects = projects + } + + init(result: [String: Any]) { + let rows = result["projects"] as? [[String: Any]] ?? [] + self.init(ready: result["ready"] as? Bool ?? false, enabled: result["enabled"] as? Bool ?? true, + projects: rows.compactMap { row in + guard let harness = row["harness"] as? String, let path = row["path"] as? String, + let used = (row["lastUsedMs"] as? NSNumber)?.int64Value else { return nil } + return Project(harness: harness, path: path, lastUsedMs: used) + }) + } +} + +extension AcpmuxChatProjects { + /// `_acpmux/chat_projects` from the daemon at `socketPath`; the socket exchange and the decode + /// run off the caller's actor. No reply cap of its own: the list is one row per chat folder and + /// a person's folder count has no bound here (the daemon's frame limit still holds), and the + /// deadline is the status client's usual one for a local request. + @concurrent public static func read(socketPath: String, deadline: Duration = .seconds(5)) async throws -> AcpmuxChatProjects { + let result = try await AcpmuxStatusClient.call(socketPath: socketPath, method: "_acpmux/chat_projects", deadline: deadline, + limit: .max) + return AcpmuxChatProjects(result: result) + } +} diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatWindows.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatWindows.swift new file mode 100644 index 000000000000..a212873abb78 --- /dev/null +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatWindows.swift @@ -0,0 +1,203 @@ +public import Foundation +public import Observation + +/// The app's windows over acpmux's chat index (`_acpmux/chats_window`, P-RUST 2026-10-30). +/// +/// acpmux owns the chat list: its order, filter, exclusions and paging. A native list here names +/// its window (``Spec``) and holds only that window's rows, as value snapshots; acpmux sends the +/// rows, then one row diff per change: the window's keys in order and only the rows whose value +/// changed, with the next revision. Applying a diff on the main actor is O(window keys + changed +/// rows), whatever the index holds (the old whole-index mirror copied and compared every chat per +/// change on the main actor). Diffs are decoded off the main actor. +/// +/// Each spec sent names the window's next generation (`gen`): acpmux refuses an older one and +/// echoes it in every message, so requests that run out of order never leave a window on an older +/// spec, and a message for an older spec is never applied. +/// +/// Each window is its own observable object: a view that reads one window updates only when that +/// window's rows change, and a window whose rows did not change sets nothing. +@MainActor +public final class AcpmuxChatWindows { + /// What a window shows (the `_acpmux/chats_window` params). + public nonisolated struct Spec: Sendable, Equatable { + public var window: String + public var limit: Int + /// Leaves out a chat another chat's agent started (it folds under that chat). + public var topLevel: Bool + public var exclude: [String] + /// Exactly these chats, in this order (a pinned list); nil: the newest chats. + public var keys: [String]? + + public init(window: String, limit: Int, topLevel: Bool = true, exclude: [String] = [], keys: [String]? = nil) { + self.window = window + self.limit = limit + self.topLevel = topLevel + self.exclude = exclude + self.keys = keys + } + + /// The request's params for generation `generation` of the window's spec. + func params(generation: UInt64) -> [String: any Sendable] { + var params: [String: any Sendable] = ["window": window, "gen": generation, "limit": max(1, limit), "topLevel": topLevel] + if !exclude.isEmpty { params["exclude"] = exclude } + if let keys { params["keys"] = keys } + return params + } + } + + /// One window's rows, newest first (or in its keys' order). + @MainActor @Observable + public final class Window { + public private(set) var rows: [AcpmuxChat] = [] + /// The revision of the last message applied; 0 before the first. + @ObservationIgnored public fileprivate(set) var revision: UInt64 = 0 + + fileprivate func set(rows: [AcpmuxChat]) { + if self.rows != rows { self.rows = rows } + } + } + + /// One decoded `_acpmux/chats_window_changed` (decoded off the main actor). + public nonisolated struct Update: Sendable, Equatable { + public var window: String + /// The generation of the spec it answers; nil from a daemon that names none. + public var generation: UInt64? + public var revision: UInt64 + /// Every row of the window; false: only the rows whose value changed. + public var full: Bool + public var keys: [String] + public var rows: [String: AcpmuxChat] + + public init(window: String, generation: UInt64? = nil, revision: UInt64, full: Bool, keys: [String], rows: [AcpmuxChat]) { + self.window = window + self.generation = generation + self.revision = revision + self.full = full + self.keys = keys + self.rows = Dictionary(rows.map { ($0.id, $0) }, uniquingKeysWith: { $1 }) + } + + public init?(params: [String: Any]) { + guard let window = params["window"] as? String, let revision = (params["revision"] as? NSNumber)?.uint64Value, + let keys = params["keys"] as? [String] else { return nil } + let rows = (params["rows"] as? [[String: Any]] ?? []).compactMap(AcpmuxChat.init(json:)) + self.init(window: window, generation: (params["gen"] as? NSNumber)?.uint64Value, revision: revision, + full: params["full"] as? Bool ?? false, keys: keys, rows: rows) + } + } + + /// Sends one request to acpmux (`method`, `params`); nil while there is no connection. + private var send: (@MainActor (String, [String: any Sendable]) -> Void)? + private var specs: [String: Spec] = [:] + /// Each window's generation: one more per spec the client opens or closes. + private var generations: [String: UInt64] = [:] + private var models: [String: Window] = [:] + /// Windows asked for again: their diffs are ignored until the full message comes. + private var awaitingFull: Set = [] + /// Windows whose last request acpmux refused (busy): asked again at the next ``retryFailed()``. + private var failed: Set = [] + /// The rows the last applied message installed (decoded values taken), for tests and the + /// debug counters: one changed chat is one row. + public private(set) var lastTouchedRows = 0 + + public init() {} + + /// The window named `id` (empty until acpmux answers its spec). + public func window(_ id: String) -> Window { + if let model = models[id] { return model } + let model = Window() + models[id] = model + return model + } + + /// Opens `spec`'s window, or changes its spec; an unchanged spec sends nothing. + public func open(_ spec: Spec) { + guard specs[spec.window] != spec else { return } + specs[spec.window] = spec + generations[spec.window, default: 0] += 1 + _ = window(spec.window) + awaitingFull.remove(spec.window) + failed.remove(spec.window) + ask(spec) + } + + /// Closes window `id`: its rows empty. + public func close(_ id: String) { + guard specs.removeValue(forKey: id) != nil else { return } + generations[id, default: 0] += 1 + awaitingFull.remove(id) + failed.remove(id) + models[id]?.set(rows: []) + send?("_acpmux/chats_window_close", ["window": id, "gen": generations[id] ?? 0]) + } + + /// A connection to acpmux is up: every open window is asked again (a new daemon numbers its + /// revisions from the start, and its first message for each window is full). + public func connected(send: @escaping @MainActor (String, [String: any Sendable]) -> Void) { + self.send = send + for model in models.values { model.revision = 0 } + resync() + } + + /// The connection is gone: windows keep their rows until the next one answers. + public func disconnected() { + send = nil + } + + /// The connection dropped a notification (`_acpmux/lagged`): every window is asked again. + public func resync() { + awaitingFull = [] + failed = [] + for spec in specs.values.sorted(by: { $0.window < $1.window }) { ask(spec) } + } + + /// acpmux answered window `id`'s request with an error: a busy daemon is asked again at the next + /// ``retryFailed()`` (the next index change or reconnect), never on a timer. + public func requestFailed(_ id: String) { + guard specs[id] != nil else { return } + awaitingFull.remove(id) + failed.insert(id) + } + + /// Asks again for every window whose request failed (an index change: the daemon is serving). + public func retryFailed() { + guard !failed.isEmpty else { return } + let retry = failed + failed = [] + for id in retry.sorted() { if let spec = specs[id] { ask(spec) } } + } + + /// Applies one window message: O(keys + changed rows). A message for an older spec is dropped; + /// a diff that does not follow the last revision, or names a row this window never got, asks + /// for the window again. + public func apply(_ update: Update) { // perf-hot: 1ms chats-window-diff (every chat index change a window shows) + guard let spec = specs[update.window], let model = models[update.window] else { return } + if let generation = update.generation, generation != generations[update.window] { return } + if update.full { + awaitingFull.remove(update.window) + } else { + guard !awaitingFull.contains(update.window) else { return } + guard update.revision == model.revision + 1 else { return ask(spec) } + } + var known: [String: AcpmuxChat] = [:] + if !update.full { + known.reserveCapacity(model.rows.count) + for row in model.rows { known[row.id] = row } + } + var rows: [AcpmuxChat] = [] + rows.reserveCapacity(update.keys.count) + for key in update.keys { + guard let row = update.rows[key] ?? known[key] else { return ask(spec) } + rows.append(row) + } + model.revision = update.revision + lastTouchedRows = update.rows.count + model.set(rows: rows) + } + + /// Asks for `spec`'s whole window, once until it comes. + private func ask(_ spec: Spec) { + guard let send, awaitingFull.insert(spec.window).inserted else { return } + send("_acpmux/chats_window", spec.params(generation: generations[spec.window] ?? 0)) + } +} diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatsStore.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatsStore.swift index 628adfe4170b..fca77ffe788c 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatsStore.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AcpmuxChatsStore.swift @@ -26,13 +26,6 @@ public nonisolated struct AcpmuxChatsStore: Sendable, Equatable { return shown } - /// Every chat, newest first, read lazily (a child chat left out): a reader that stops after a - /// page visits only what it reads. - public var newestFirst: some Sequence { - let values = values - return orderedKeys.lazy.compactMap { values[$0] }.filter { $0.parent == nil } - } - /// Replaces the mirror from a page response. acpmux sends the page sorted, newest first: /// its order is kept after one O(n) check (an older daemon's page is sorted here). public mutating func reset(_ result: [String: Any]) { diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentPageChats.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentPageChats.swift index 14430d87e6d1..0f312e53bd22 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentPageChats.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentPageChats.swift @@ -1,3 +1,4 @@ +import CmuxNextAgentLanes import Foundation import CmuxNextActions import CmuxNextAgentPane diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentRecentsFeed.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentRecentsFeed.swift index 442844bfa643..493a2853c7bc 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentRecentsFeed.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/AgentRecentsFeed.swift @@ -1,2 +1,2 @@ // The Recents feed was replaced by ChatsFeed for ALL-CHATS-ON-DEVICE C7. -// AgentRecentsFeed remains a typealias in ChatsFeed.swift for source compatibility. +// AgentRecentsFeed remains a typealias in CmuxNextAgentLanes/ChatsFeed.swift for source compatibility. diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed+Start.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed+Start.swift index 3ff6c1623d9e..5c5b693efa52 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed+Start.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed+Start.swift @@ -24,12 +24,11 @@ extension ChatsFeed { turns.start(localHost: localHost) startChiefTurnStates(tag: services.environment.tag) SidebarSessionTreeFeed.shared.start(workspaces: { services.machines.allWorkspaces.map { $0.0 } }, localHost: localHost) - let feed = ChatsFeed(socketPath: environment.socketPath) - // The session tree's chat history: rows open through the shared Open Chat path. - let history = SidebarChatHistoryFeed.shared - feed.observe(history) { [weak feed] in if let feed { history.chats = feed.store } } - history.start(workspaces: { services.machines.allWorkspaces.map { $0.0 } }, localHost: localHost, - open: { [weak services] key in services?.chatsOpener.open(key) }) + let feed = ChatsFeed(socketPath: environment.socketPath, drained: { DebugTimings.chatsFeedApplied(lines: $0, $1) }) + // The session tree's chat history: two windows acpmux keeps (never the whole index); rows + // open through the shared Open Chat path. + SidebarChatHistoryFeed.shared.start(workspaces: { services.machines.allWorkspaces.map { $0.0 } }, localHost: localHost, + open: { [weak services] key in services?.chatsOpener.open(key) }, windows: feed.windows) return feed } diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsOpenCoordinator.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsOpenCoordinator.swift index 6ac1ceca389f..2a9ca52f1d99 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsOpenCoordinator.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsOpenCoordinator.swift @@ -1,5 +1,6 @@ import AppKit import CmuxNextActions +import CmuxNextAgentLanes import CmuxNextAgentPane import CmuxNextBridge import CmuxNextDaemon diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppDelegate.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppDelegate.swift index 50ef60fc2726..d52e0c4a4cf7 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppDelegate.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppDelegate.swift @@ -1,5 +1,6 @@ import AppKit import CmuxNextActions +import CmuxNextAgentLanes import CmuxNextAgentPane import CmuxNextBrowser import CmuxNextCompat diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppServices.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppServices.swift index bd031574c603..94b04935d0de 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppServices.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/AppServices.swift @@ -1,5 +1,6 @@ import AppKit import CmuxNextActions +import CmuxNextAgentLanes import CmuxNextOnboarding import CmuxNextApps import CmuxNextAgentPane diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/AgentChatsPalettePage.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/AgentChatsPalettePage.swift index 9429f5149bed..307a93efdc04 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/AgentChatsPalettePage.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/AgentChatsPalettePage.swift @@ -1,5 +1,6 @@ import CmuxAgentBrands import CmuxNextActions +import CmuxNextAgentLanes import CmuxNextAgentPane import CmuxNextPalette import Foundation diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/OnboardingHandlers.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/OnboardingHandlers.swift index a9433449043e..f25e26c82b0c 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/OnboardingHandlers.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Handlers/OnboardingHandlers.swift @@ -1,5 +1,6 @@ import AppKit import CmuxNextActions +import CmuxNextAgentLanes import CmuxNextBrowser import CmuxNextCookieImport import CmuxNextDesign diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Projects/ProjectsImport.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Projects/ProjectsImport.swift index 4c892a06e1f5..b7f7229d4fde 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Projects/ProjectsImport.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextApp/Projects/ProjectsImport.swift @@ -1,3 +1,4 @@ +import CmuxNextAgentLanes import CmuxNextAgentPane import CmuxNextDaemon import Foundation @@ -36,33 +37,24 @@ final class ProjectsImport { } } - /// The batches for `chats`, leaving out each harness whose fingerprint is - /// already `sent`. A harness in `sent` with no chat left gets an empty - /// complete list. Sorted by harness, entries by path. Nothing while the - /// index is not `complete` (still scanning, or turned off): a partial - /// list sent as complete would drop the harness from its projects. - nonisolated static func batches(_ chats: [AcpmuxChat], sent: [String: Int], complete: Bool = true) -> [Batch] { - guard complete else { return [] } - var newest: [String: [String: Int64]] = [:] - for chat in chats { - guard let cwd = chat.cwd, cwd.hasPrefix("/") else { continue } - let used = Int64((chat.updatedAt.timeIntervalSince1970 * 1000).rounded()) - newest[chat.harness, default: [:]][cwd] = max(newest[chat.harness]?[cwd] ?? .min, used) + /// The batches for acpmux's grouped `projects`, leaving out each harness whose fingerprint is + /// already `sent`. A harness in `sent` with no project left gets an empty complete list. Sorted + /// by harness, entries by path. Nothing while the index is not ready or chats are off: a + /// partial list sent as complete would drop the harness from its projects. + nonisolated static func batches(_ projects: AcpmuxChatProjects, sent: [String: Int]) -> [Batch] { + guard projects.ready, projects.enabled else { return [] } + var bySource: [String: [Batch.Entry]] = [:] + for project in projects.projects { + bySource[project.harness, default: []].append(Batch.Entry(path: project.path, lastUsedMs: project.lastUsedMs)) } - for source in sent.keys where newest[source] == nil { newest[source] = [:] } - return newest.keys.sorted().compactMap { source in - let entries = (newest[source] ?? [:]).sorted { $0.key < $1.key }.map { Batch.Entry(path: $0.key, lastUsedMs: $0.value) } + for source in sent.keys where bySource[source] == nil { bySource[source] = [] } + return bySource.keys.sorted().compactMap { source in + let entries = (bySource[source] ?? []).sorted { $0.path < $1.path } let batch = Batch(source: source, entries: entries) return sent[source] == batch.fingerprint ? nil : batch } } - /// `batches` over a mirror snapshot, off the main actor: it groups every chat (5000+ on a - /// real Mac) on each index change (cx-9c8m.6). - @concurrent nonisolated static func batches(of store: AcpmuxChatsStore, sent: [String: Int], complete: Bool) async -> [Batch] { - batches(store.chats, sent: sent, complete: complete) - } - private weak var services: AppServices? private var connectionID: ObjectIdentifier? private var sent: [String: Int] = [:] @@ -109,18 +101,22 @@ final class ProjectsImport { let id = ObjectIdentifier(connection) let fresh = connectionID != id if fresh { sent = [:] } - let feed = services.chatsFeed - // A copy-on-write snapshot (O(1) here); the walk over every chat runs off the main actor. - let store = feed?.store ?? AcpmuxChatsStore() - let complete = (feed?.isReady ?? false) && (feed?.isEnabled ?? false) + // acpmux groups its own index (`_acpmux/chat_projects`): a few rows, read off the main + // actor, never a walk over every chat here. + let socket = services.chatsFeed?.socketPath let alreadySent = sent let client = ProjectStateClient(connection: connection) - // task-owner: awaited at once by `sending` below. Only this task holds the snapshot, so it - // is released when the batches are ready, not after the sends: a held snapshot would make - // the next main-actor store change copy all 5000 chats. - let grouping = Task { await Self.batches(of: store, sent: alreadySent, complete: complete) } sending = Task { [weak self] in - let batches = await grouping.value + var batches: [Batch] = [] + if let socket { + do { + batches = Self.batches(try await AcpmuxChatProjects.read(socketPath: socket), sent: alreadySent) + } catch { + // Nothing is sent (a partial list would drop projects). The next index change or + // daemon connection relays again: an event that can make the read succeed. + self?.logger.error("project list read failed: \(String(describing: error), privacy: .public)") + } + } var delivered: [String: Int] = [:] var synced = !fresh guard fresh || !batches.isEmpty else { diff --git a/Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/ChatsWindowMainThreadTests.swift b/Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/ChatsWindowMainThreadTests.swift new file mode 100644 index 000000000000..2fbd3105badc --- /dev/null +++ b/Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/ChatsWindowMainThreadTests.swift @@ -0,0 +1,143 @@ +@testable import CmuxNextAgentLanes +@testable import CmuxNextAgentPane +import CmuxNextSidebar +import Foundation +import Testing + +/// P-RUST item 2 (Lawrence's v2.2.9 hangs of 570 to 1,148 ms, 2026-10-30): with thousands of chats +/// in acpmux's index, one chat change must touch O(1) rows on the main thread. The hangs were the +/// whole-index mirror: `AcpmuxChatsStore.apply` copied every chat (the sidebar history held the same +/// storage) and `SidebarChatHistoryFeed.chats`'s setter compared and freed every chat. Now acpmux +/// keeps the sidebar history's windows (`_acpmux/chats_window`): the main thread holds only their +/// visible rows and applies one row diff per change. +@MainActor @Suite struct ChatsWindowMainThreadTests { + static let indexSize = 3000 + + /// The requests the windows sent (method, params). + @MainActor final class Sent { + var methods: [String] = [] + var params: [[String: any Sendable]] = [] + + func record(_ method: String, _ params: [String: any Sendable]) { + methods.append(method) + self.params.append(params) + } + } + + static func key(_ i: Int) -> String { "claude-code:s\(i)" } + + static func chat(_ i: Int, updatedMs: Int? = nil) -> AcpmuxChat { + let ms = updatedMs ?? 1_800_000_000_000 - i * 1000 + return AcpmuxChat(id: key(i), sessionID: "s\(i)", harness: "claude-code", title: "chat \(i)", cwd: "/work/p\(i % 50)", + updatedAt: Date(timeIntervalSince1970: Double(ms) / 1000)) + } + + /// The history's Earlier window as acpmux answers it: the newest page and one more of `indexSize`. + static func opened(send: @escaping @MainActor (String, [String: any Sendable]) -> Void = { _, _ in }) + -> (AcpmuxChatWindows, SidebarChatHistoryFeed.Inputs) { + let windows = AcpmuxChatWindows() + let inputs = SidebarChatHistoryFeed.Inputs(live: [], pinned: [], limit: SidebarChatHistory.pageSize, folded: false) + windows.open(inputs.earlierSpec) + windows.connected(send: send) + let size = SidebarChatHistory.pageSize + 1 + windows.apply(.init(window: SidebarChatHistoryFeed.earlierWindow, revision: 1, full: true, keys: (0.. gens[0], "each spec names a newer generation: \(gens)") + let older: [String: Any] = ["window": "w", "gen": NSNumber(value: gens.first ?? 0), "revision": 1, "full": true, + "keys": [Self.key(0)], "rows": [["key": Self.key(0), "sessionId": "s0", "harness": "claude-code", + "updatedMs": 1_800_000_000_000]]] + if let update = AcpmuxChatWindows.Update(params: older) { windows.apply(update) } + #expect(windows.window("w").rows.isEmpty, "the older spec's rows were applied") + } + + /// Review F3 (#28055): acpmux refused a window request (busy): the window is not left waiting + /// for a full message that never comes; the next sign that the daemon serves asks again. + @Test func aRefusedRequestIsAskedAgainWhenTheDaemonServes() { + let windows = AcpmuxChatWindows() + let sent = Sent() + windows.connected(send: sent.record) + windows.open(.init(window: "w", limit: 3)) + #expect(sent.methods == ["_acpmux/chats_window"]) + windows.requestFailed("w") + windows.retryFailed() + #expect(sent.methods == ["_acpmux/chats_window", "_acpmux/chats_window"], "a refused window is asked again") + windows.retryFailed() + #expect(sent.methods.count == 2, "once per refusal") + } +} diff --git a/Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/SidebarChatHistoryTests.swift b/Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/SidebarChatHistoryTests.swift index 1855cabc4516..4d29e160ab11 100644 --- a/Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/SidebarChatHistoryTests.swift +++ b/Packages/macOS/CmuxNext/Tests/CmuxNextAppTests/SidebarChatHistoryTests.swift @@ -220,17 +220,21 @@ import Testing #expect(lane("claude", nil)?.chat == nil) } - @Test func theFeedListsTheIndexWithoutChildChatsAndNamesUntitledOnes() { - var store = AcpmuxChatsStore() - store.reset(["chats": [ + /// acpmux's window holds no child chat (`topLevel`); the history names an untitled chat by its + /// folder, else its harness, and leaves out a chat an open workspace shows until acpmux's next + /// window excludes it too. + @Test func theFeedListsTheWindowAndNamesUntitledOnes() { + let json: [[String: Any]] = [ ["key": "claude-code:1", "sessionId": "1", "harness": "claude-code", "title": "Fix the build", "updatedMs": 3_000], ["key": "codex:2", "sessionId": "2", "harness": "codex", "cwd": "/Users/me/cmux", "updatedMs": 2_000], - ["key": "claude-code:3", "sessionId": "3", "harness": "claude-code", "updatedMs": 1_000, "parent": "claude-code:1"], ["key": "claude-code:4", "sessionId": "4", "harness": "claude-code", "updatedMs": 500] - ]]) - let state = SidebarChatHistoryFeed.state(store, live: ["claude-code:1"], pinned: [], limit: 10, folded: false) + ] + let rows = json.compactMap(AcpmuxChat.init(json:)) + let inputs = SidebarChatHistoryFeed.Inputs(live: ["claude-code:1"], pinned: [], limit: 10, folded: false) + #expect(inputs.earlierSpec.topLevel) + #expect(inputs.earlierSpec.exclude == ["claude-code:1"]) + let state = SidebarChatHistoryFeed.state(earlier: rows, pinned: [], inputs: inputs) #expect(state.earlier.map(\.id) == ["codex:2", "claude-code:4"]) #expect(state.earlier.map(\.title) == ["cmux", "claude-code"]) - #expect(Array(store.newestFirst.map(\.id)) == ["claude-code:1", "codex:2", "claude-code:4"]) } } diff --git a/cmux-tui/crates/acpmux/spec/acpmux-protocol.json b/cmux-tui/crates/acpmux/spec/acpmux-protocol.json index b273df97d0b7..cc8c2c5e196d 100644 --- a/cmux-tui/crates/acpmux/spec/acpmux-protocol.json +++ b/cmux-tui/crates/acpmux/spec/acpmux-protocol.json @@ -1114,6 +1114,12 @@ "params": true, "result": true }, + { + "kind": "request", + "name": "_acpmux/chat_projects", + "params": true, + "result": true + }, { "kind": "notification", "name": "_acpmux/chat_changed", @@ -1124,6 +1130,23 @@ "name": "_acpmux/chats_lagged", "params": true }, + { + "kind": "request", + "name": "_acpmux/chats_window", + "params": true, + "result": true + }, + { + "kind": "request", + "name": "_acpmux/chats_window_close", + "params": true, + "result": true + }, + { + "kind": "notification", + "name": "_acpmux/chats_window_changed", + "params": true + }, { "kind": "request", "name": "_acpmux/transfer_abort", diff --git a/cmux-tui/crates/acpmux/src/chats/activity.rs b/cmux-tui/crates/acpmux/src/chats/activity.rs index 0c40f2232e18..f96189f440fe 100644 --- a/cmux-tui/crates/acpmux/src/chats/activity.rs +++ b/cmux-tui/crates/acpmux/src/chats/activity.rs @@ -96,6 +96,11 @@ pub fn activity_change(hub: &Hub, session: &Session) -> Option { pub struct ChatActivity(HashMap); impl ChatActivity { + /// The chat was started by another chat's agent (`parent`): lists fold it under that chat. + pub fn has_parent(&self, key: &str) -> bool { + self.0.get(key).is_some_and(|(_, fields)| !fields.get("parent").is_none_or(Value::is_null)) + } + /// Adds `attention`, `preview` and `parent` to a chat value (`chat_value`); a chat /// no session continues gets nulls, so a client clears what it had. pub fn decorate(&self, chat: &mut Value) { diff --git a/cmux-tui/crates/acpmux/src/chats/mod.rs b/cmux-tui/crates/acpmux/src/chats/mod.rs index 801e75063ee5..22fc0a5ebbe5 100644 --- a/cmux-tui/crates/acpmux/src/chats/mod.rs +++ b/cmux-tui/crates/acpmux/src/chats/mod.rs @@ -18,6 +18,7 @@ mod query; mod settings; pub(crate) mod sources; mod watch; +mod window; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; @@ -37,6 +38,7 @@ pub use open::{StoreProfile, plan_open, store_profiles}; pub use query::{ChatCursor, ChatQuery}; pub use settings::{ChatSettings, SettingsRefusal, probe}; pub use sources::{ChatSources, EnvLookup, launch_roots, login_var, lookup, refusal}; +pub use window::{WindowRows, WindowSpec, WindowState}; /// Work run once the chat index has started. pub type ChatsWaiter = Box) + Send>; @@ -140,6 +142,46 @@ impl ChatService { (page, next, seq) } + /// The rows of a client's window (`_acpmux/chats_window`), read under the index lock: none + /// while chats are off. + pub fn window_rows(&self, spec: &WindowSpec, activity: &ChatActivity) -> WindowRows { + let state = lock(&self.state); + if !state.settings.enabled { + return WindowRows::default(); + } + window::rows(spec, &state.index, activity) + } + + /// `_acpmux/chat_projects`: each harness's project folders, every absolute chat folder with + /// the time of its newest chat (a chat another chat started left out), by harness id then + /// path. acpmux groups its own index, so the app relays a few rows, never every chat. None + /// while chats are off. + pub fn projects(&self, activity: &ChatActivity) -> Option> { + let state = lock(&self.state); + if !state.settings.enabled { + return None; + } + let mut newest: std::collections::BTreeMap<(&str, &str), i64> = + std::collections::BTreeMap::new(); + for chat in state.index.chats() { + let entry = &chat.entry; + let Some(cwd) = entry.cwd.as_deref().filter(|cwd| cwd.starts_with('/')) else { + continue; + }; + if activity.has_parent(&key_text(&entry.key())) { + continue; + } + let used = newest.entry((entry.harness.id(), cwd)).or_insert(i64::MIN); + *used = (*used).max(entry.updated_ms); + } + Some( + newest + .into_iter() + .map(|((harness, path), used)| json!({"harness": harness, "path": path, "lastUsedMs": used})) + .collect(), + ) + } + pub fn get(&self, key: &ChatKey) -> Option { lock(&self.state).index.get(key).cloned() } diff --git a/cmux-tui/crates/acpmux/src/chats/window.rs b/cmux-tui/crates/acpmux/src/chats/window.rs new file mode 100644 index 000000000000..1f40de5968dd --- /dev/null +++ b/cmux-tui/crates/acpmux/src/chats/window.rs @@ -0,0 +1,249 @@ +//! `_acpmux/chats_window`: a client's visible window over the chat index (P-RUST, 2026-10-30). +//! +//! acpmux owns the chat list a client shows: the order, the filter, the exclusions and the paging. +//! A client names a window and its spec and gets the window's rows; after each index or activity +//! change it gets a row diff: the window's keys in order and only the rows whose value changed, +//! with the next revision. A client never holds the whole index (the Mac app's whole-index mirror +//! copied and compared every chat on the main thread per change: hangs of 570 to 1,148 ms). +//! +//! Pure: the server keeps one [`WindowState`] per window and connection (`server/chat_windows.rs`). + +use std::collections::{HashMap, HashSet}; + +use cmux_chat_index::{ChatIndex, IndexedChat}; +use serde_json::{Value, json}; + +use super::{ChatActivity, ChatQuery, chat_value, key_text, parse_key}; + +/// What a window shows: `{window, gen?, query?, harness?, folder?, account?, limit?, topLevel?, +/// exclude?, keys?}`. `gen` is the client's generation of the window's spec: acpmux keeps the +/// highest it saw and echoes it in every message, so requests that run out of order never leave a +/// window on an older spec. With `keys`, exactly those chats in that order (a pinned list); otherwise +/// the newest chats the filter passes, none of `exclude`, at most `limit`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct WindowSpec { + pub window: String, + /// The client's generation of this window's spec (0 from a client that sends none). + pub generation: u64, + filter: ChatQuery, + /// Leaves out a chat another chat's agent started (`parent`): it folds under that chat. + top_level: bool, + exclude: HashSet, + keys: Option>, +} + +impl WindowSpec { + pub fn from_params(params: &Value) -> Result { + let window = params + .get("window") + .and_then(Value::as_str) + .filter(|w| !w.is_empty()) + .ok_or("window must be a non-empty string")? + .to_owned(); + let filter = ChatQuery::from_params(params)?; + let strings = |name: &str| -> Result>, String> { + match params.get(name) { + None | Some(Value::Null) => Ok(None), + Some(Value::Array(items)) => items + .iter() + .map(|item| item.as_str().map(str::to_owned)) + .collect::>>() + .map(Some) + .ok_or_else(|| format!("{name} must be an array of chat keys")), + Some(_) => Err(format!("{name} must be an array of chat keys")), + } + }; + let top_level = match params.get("topLevel") { + None | Some(Value::Null) => false, + Some(Value::Bool(on)) => *on, + Some(_) => return Err("topLevel must be true or false".to_owned()), + }; + let generation = match params.get("gen") { + None | Some(Value::Null) => 0, + Some(value) => value.as_u64().ok_or("gen must be a non-negative number")?, + }; + Ok(Self { + window, + generation, + filter, + top_level, + exclude: strings("exclude")?.unwrap_or_default().into_iter().collect(), + keys: strings("keys")?, + }) + } +} + +/// One row: the chat's key and its value as `_acpmux/chats` sends it (with its activity). +pub type Row = (String, Value); + +/// A window's rows now, and whether more chats pass its filter past them. +#[derive(Clone, Debug, Default, PartialEq)] +pub struct WindowRows { + pub rows: Vec, + pub has_more: bool, +} + +/// The rows of `spec` in `index`: one walk of the index in its order (newest first) that stops +/// once the window is full and one more chat passes; a value is built only for a chat the filter +/// and the exclusions pass. +pub fn rows(spec: &WindowSpec, index: &ChatIndex, activity: &ChatActivity) -> WindowRows { + let decorated = |chat: &IndexedChat| { + let mut value = chat_value(chat); + activity.decorate(&mut value); + value + }; + if let Some(keys) = &spec.keys { + let rows = keys + .iter() + .filter_map(|text| { + let chat = index.get(&parse_key(text)?)?; + Some((text.clone(), decorated(chat))) + }) + .collect(); + return WindowRows { rows, has_more: false }; + } + let mut window = WindowRows::default(); + for chat in index.chats() { + if !spec.filter.matches(chat) { + continue; + } + let key = key_text(&chat.entry.key()); + if spec.exclude.contains(&key) { + continue; + } + let value = decorated(chat); + if spec.top_level && !value.get("parent").is_none_or(Value::is_null) { + continue; + } + if window.rows.len() == spec.filter.limit { + window.has_more = true; + break; + } + window.rows.push((key, value)); + } + window +} + +/// What a window last sent its client, so the next message is a diff against it. +#[derive(Debug)] +pub struct WindowState { + pub spec: WindowSpec, + revision: u64, + /// None until the window's first message: that one sends every row. + sent: Option, +} + +impl WindowState { + /// A new window, or a window whose spec changed: its next message sends every row. The + /// revision goes on from `revision`, so a client never takes an older message for a newer. + pub fn new(spec: WindowSpec, revision: u64) -> Self { + Self { spec, revision, sent: None } + } + + pub fn revision(&self) -> u64 { + self.revision + } + + /// The `_acpmux/chats_window_changed` params that bring the client from what it was sent to + /// `next`, or None when it shows nothing new. The first message is `full` (every row); every + /// later one is a diff: the window's keys in order and the rows whose value changed. + pub fn next(&mut self, next: WindowRows) -> Option { + let keys: Vec<&str> = next.rows.iter().map(|(key, _)| key.as_str()).collect(); + let message = match &self.sent { + None => { + let rows: Vec<&Value> = next.rows.iter().map(|(_, value)| value).collect(); + json!({"full": true, "keys": keys, "rows": rows}) + } + Some(sent) => { + if *sent == next { + return None; + } + let before: HashMap<&str, &Value> = + sent.rows.iter().map(|(key, value)| (key.as_str(), value)).collect(); + let changed: Vec<&Value> = next + .rows + .iter() + .filter(|(key, value)| before.get(key.as_str()) != Some(&value)) + .map(|(_, value)| value) + .collect(); + json!({"full": false, "keys": keys, "rows": changed}) + } + }; + self.revision += 1; + let mut message = message; + if let Some(object) = message.as_object_mut() { + object.insert("window".into(), json!(self.spec.window)); + object.insert("gen".into(), json!(self.spec.generation)); + object.insert("revision".into(), json!(self.revision)); + object.insert("hasMore".into(), json!(next.has_more)); + } + self.sent = Some(next); + Some(message) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn spec(params: Value) -> WindowSpec { + WindowSpec::from_params(¶ms).unwrap() + } + + fn row(key: &str, title: &str) -> Row { + (key.to_owned(), json!({"key": key, "title": title})) + } + + #[test] + fn a_spec_needs_its_window_name_and_reads_its_lists() { + assert!(WindowSpec::from_params(&json!({})).is_err()); + assert!(WindowSpec::from_params(&json!({"window": "w", "exclude": "x"})).is_err()); + assert!(WindowSpec::from_params(&json!({"window": "w", "topLevel": 1})).is_err()); + assert!(WindowSpec::from_params(&json!({"window": "w", "gen": -1})).is_err()); + assert_eq!(spec(json!({"window": "w", "gen": 7})).generation, 7); + let s = spec(json!({"window": "w", "limit": 3, "exclude": ["a:1"], "keys": ["b:2"]})); + assert_eq!(s.filter.limit, 3); + assert!(s.exclude.contains("a:1")); + assert_eq!(s.keys, Some(vec!["b:2".to_owned()])); + } + + #[test] + fn the_first_message_is_full_and_later_ones_carry_only_changed_rows() { + let mut state = WindowState::new(spec(json!({"window": "w"})), 0); + let first = WindowRows { rows: vec![row("a:1", "one"), row("a:2", "two")], has_more: true }; + let full = state.next(first.clone()).unwrap(); + assert_eq!(full["full"], true); + assert_eq!(full["revision"], 1); + assert_eq!(full["gen"], 0); + assert_eq!(full["rows"].as_array().unwrap().len(), 2); + assert_eq!(state.next(first), None, "nothing new, nothing sent"); + let moved = WindowRows { + rows: vec![row("a:2", "two"), row("a:1", "one, renamed")], + has_more: true, + }; + let diff = state.next(moved).unwrap(); + assert_eq!(diff["full"], false); + assert_eq!(diff["revision"], 2); + assert_eq!(diff["keys"], json!(["a:2", "a:1"])); + assert_eq!(diff["rows"], json!([{"key": "a:1", "title": "one, renamed"}])); + let more = WindowRows { + rows: vec![row("a:2", "two"), row("a:1", "one, renamed")], + has_more: false, + }; + let diff = state.next(more).unwrap(); + assert_eq!(diff["rows"], json!([]), "only hasMore changed"); + assert_eq!(diff["hasMore"], false); + } + + #[test] + fn a_replaced_spec_sends_every_row_again_with_a_later_revision() { + let mut state = WindowState::new(spec(json!({"window": "w"})), 0); + state.next(WindowRows { rows: vec![row("a:1", "one")], has_more: false }); + let mut replaced = + WindowState::new(spec(json!({"window": "w", "limit": 5})), state.revision()); + let full = + replaced.next(WindowRows { rows: vec![row("a:1", "one")], has_more: false }).unwrap(); + assert_eq!(full["full"], true); + assert_eq!(full["revision"], 2); + } +} diff --git a/cmux-tui/crates/acpmux/src/protocol/mod.rs b/cmux-tui/crates/acpmux/src/protocol/mod.rs index ce332712edf5..00f7540b3828 100644 --- a/cmux-tui/crates/acpmux/src/protocol/mod.rs +++ b/cmux-tui/crates/acpmux/src/protocol/mod.rs @@ -332,8 +332,13 @@ pub fn registry() -> Vec { untyped("_acpmux/chat_settled", Request), untyped("_acpmux/chat_transcript", Request), untyped("_acpmux/chat_fork", Request), + untyped("_acpmux/chat_projects", Request), untyped("_acpmux/chat_changed", Notification), untyped("_acpmux/chats_lagged", Notification), + // A client's window over the chat index (`server/chat_windows.rs`). + untyped("_acpmux/chats_window", Request), + untyped("_acpmux/chats_window_close", Request), + untyped("_acpmux/chats_window_changed", Notification), // Moving a chat to another daemon (`transfer_out.rs`). untyped("_acpmux/transfer_abort", Request), untyped("_acpmux/transfer_begin", Request), diff --git a/cmux-tui/crates/acpmux/src/protocol/tests.rs b/cmux-tui/crates/acpmux/src/protocol/tests.rs index e5ec38058f3f..e8580bfb1032 100644 --- a/cmux-tui/crates/acpmux/src/protocol/tests.rs +++ b/cmux-tui/crates/acpmux/src/protocol/tests.rs @@ -59,6 +59,7 @@ fn protocol_registry_names_every_served_method() { include_str!("../server/requests.rs"), include_str!("../server/mod.rs"), include_str!("../server/chats.rs"), + include_str!("../server/chat_windows.rs"), include_str!("../server/secrets.rs"), include_str!("../server/peer_forward.rs"), include_str!("../server/remote_guard.rs"), diff --git a/cmux-tui/crates/acpmux/src/server/chat_windows.rs b/cmux-tui/crates/acpmux/src/server/chat_windows.rs new file mode 100644 index 000000000000..508919eed556 --- /dev/null +++ b/cmux-tui/crates/acpmux/src/server/chat_windows.rs @@ -0,0 +1,243 @@ +//! `_acpmux/chats_window` and `_acpmux/chats_window_close`: a connection's windows over the chat +//! index (`chats/window.rs`), kept current by one task per connection. +//! +//! - `_acpmux/chats_window {window, query?, harness?, folder?, account?, limit?, topLevel?, +//! exclude?, keys?}` opens the window or replaces its spec, and answers `{window, ready}`. The +//! rows travel as `_acpmux/chats_window_changed {window, revision, full: true, keys, rows, +//! hasMore}`, queued before the answer, on the same ordered path as every later diff. +//! - Each index change or chat activity change then recomputes the connection's windows, and each +//! window that shows something new gets `_acpmux/chats_window_changed {window, revision, +//! full: false, keys, rows, hasMore}`: its keys in order and only the rows whose value changed. +//! The revision goes up by one per message of that window. A client that sees a revision gap, +//! or an `_acpmux/lagged` (the connection dropped a notification), opens the window again. +//! - `_acpmux/chats_window_close {window, gen?}` stops it. +//! - Each request names the client's generation of the window (`gen`). acpmux keeps the highest +//! it saw per window and refuses an older open or close as `stale`: the daemon runs requests as +//! tasks of their own, so two specs sent back to back can run in either order. Every message +//! echoes `gen`, so the client applies only its current spec's rows. +//! +//! Owned (R6): the connection's one task ends when its connection closes, when its last window +//! closes (a later open starts a new one), and with the runtime at daemon exit; a drop guard +//! counts it out on every path (`live_tasks`). +//! +//! The state is level-triggered: a window is recomputed from the index as it is now, so a lagged +//! or coalesced change can never leave a window behind. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex as StdMutex}; + +use serde_json::{Value, json}; +use tokio::sync::broadcast::error::{RecvError, TryRecvError}; + +use super::Conn; +use crate::chats::{ChatService, WindowSpec, WindowState}; +use crate::hub::Hub; +use crate::rpc::{Message, RpcError}; + +/// A connection's chat windows. +#[derive(Default)] +pub(super) struct ConnWindows { + /// Held only for map changes and a window's `next` and send (never across an index read), so + /// an open, a close or an activity change never waits for a scan. + windows: StdMutex, + /// One recompute at a time (an async lock: a waiter holds no thread), so a recompute that read + /// the index earlier never sends after one that read it later. + refreshing: tokio::sync::Mutex<()>, + /// Wakes the connection's task after a chat activity change or a close. + wake: tokio::sync::Notify, +} + +#[derive(Default)] +struct Windows { + by_name: BTreeMap, + /// The highest generation seen for each window name, opens and closes alike. + generations: HashMap, + /// The connection's task runs. Set and cleared under this lock, so an open never misses a + /// task that is ending, and a task never ends with a window open. + task: bool, +} + +/// Window tasks alive in this process (tests check that none outlives its connection). +static LIVE_TASKS: AtomicUsize = AtomicUsize::new(0); + +/// How many chat window tasks are alive in this process. +pub fn live_tasks() -> usize { + LIVE_TASKS.load(Ordering::Acquire) +} + +/// Counts one task out however it ends: return, connection close, or cancel at runtime drop. +struct Live; + +impl Live { + fn new() -> Self { + LIVE_TASKS.fetch_add(1, Ordering::AcqRel); + Self + } +} + +impl Drop for Live { + fn drop(&mut self) { + LIVE_TASKS.fetch_sub(1, Ordering::AcqRel); + } +} + +fn lock(mutex: &StdMutex) -> std::sync::MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// `_acpmux/chats_window`. +pub(super) async fn open( + hub: &Arc, + conn: &Arc, + service: Option>, + params: &Value, +) -> Result { + let spec = WindowSpec::from_params(params).map_err(RpcError::invalid_params)?; + let window = spec.window.clone(); + { + let mut windows = lock(&conn.chat_windows.windows); + let newest = windows.generations.get(&window).copied().unwrap_or(0); + if spec.generation < newest { + return Ok(json!({"window": window, "gen": spec.generation, "stale": true})); + } + windows.generations.insert(window.clone(), spec.generation); + let revision = windows.by_name.get(&window).map_or(0, WindowState::revision); + windows.by_name.insert(window.clone(), WindowState::new(spec, revision)); + } + let Some(service) = service else { + // The index has not started (the app connects at launch): the window fills when it does. + let (owner, conn) = (hub.clone(), conn.clone()); + hub.when_chats_ready(Box::new(move |service| { + if !conn.out.is_closed() { + start(&owner, service, &conn); + conn.chat_windows.wake.notify_one(); + } + })); + return Ok(json!({"window": window, "ready": false})); + }; + start(hub, &service, conn); + refresh(hub, &service, conn).await?; + Ok(json!({"window": window, "ready": true})) +} + +/// `_acpmux/chats_window_close`. +pub(super) fn close(conn: &Conn, params: &Value) -> Result { + let window = params + .get("window") + .and_then(Value::as_str) + .ok_or_else(|| RpcError::invalid_params("window must be a string"))?; + let generation = match params.get("gen") { + None | Some(Value::Null) => None, + Some(value) => Some( + value + .as_u64() + .ok_or_else(|| RpcError::invalid_params("gen must be a non-negative number"))?, + ), + }; + let closed = { + let mut windows = lock(&conn.chat_windows.windows); + let newest = windows.generations.get(window).copied().unwrap_or(0); + if generation.is_some_and(|generation| generation < newest) { + return Ok(json!({"window": window, "closed": false, "stale": true})); + } + if let Some(generation) = generation { + windows.generations.insert(window.to_owned(), generation); + } + windows.by_name.remove(window).is_some() + }; + // The task sees the last window gone and ends. + conn.chat_windows.wake.notify_one(); + Ok(json!({"window": window, "closed": closed})) +} + +/// A chat's activity changed (`chats::push_activity`): the connection's windows recompute. +pub(super) fn activity_changed(conn: &Conn) { + // No lock: with no task the stored permit only wakes the next task once more. + conn.chat_windows.wake.notify_one(); +} + +/// Starts the connection's task unless it runs: it recomputes the windows after each index change +/// and each wake, until the connection closes or its last window closes. +fn start(hub: &Arc, service: &Arc, conn: &Arc) { + { + let mut windows = lock(&conn.chat_windows.windows); + if windows.task || windows.by_name.is_empty() { + return; + } + windows.task = true; + } + let mut changes = service.subscribe(); + let (hub, service, conn) = (hub.clone(), service.clone(), conn.clone()); + let live = Live::new(); + // task-owner: the connection (R6): it ends with the connection, its last window or the runtime. + tokio::spawn(async move { + let _live = live; + loop { + { + let mut windows = lock(&conn.chat_windows.windows); + if windows.by_name.is_empty() { + windows.task = false; + return; + } + } + tokio::select! { + () = conn.closed() => break, + next = changes.recv() => match next { + // A lagged receiver missed changes, not the state: the recompute reads the index now. + Ok(_) | Err(RecvError::Lagged(_)) => {} + Err(RecvError::Closed) => break, + }, + () = conn.chat_windows.wake.notified() => {} + } + // The batches that came meanwhile: one recompute reads the index after all of them. + while matches!(changes.try_recv(), Ok(_) | Err(TryRecvError::Lagged(_))) {} + if refresh(&hub, &service, &conn).await.is_err() { + // The recompute failed (its blocking task panicked): the client asks again. + conn.send(&Message::notification("_acpmux/chats_lagged", json!({"dropped": 0}))); + break; + } + } + let mut windows = lock(&conn.chat_windows.windows); + windows.by_name.clear(); + windows.task = false; + }); +} + +/// Recomputes every window of `conn` and sends each one's message, if any: the rows are read +/// off the executor with no windows lock held (the index lock may be held by a scan), then each +/// window whose spec is unchanged takes its diff and sends it under the lock. +async fn refresh( + hub: &Arc, + service: &Arc, + conn: &Arc, +) -> Result<(), RpcError> { + let _one = conn.chat_windows.refreshing.lock().await; + let specs: Vec = + lock(&conn.chat_windows.windows).by_name.values().map(|state| state.spec.clone()).collect(); + if specs.is_empty() { + return Ok(()); + } + let activity = hub.chat_activity(); + let service = service.clone(); + let rows = tokio::task::spawn_blocking(move || { + specs + .into_iter() + .map(|spec| (service.window_rows(&spec, &activity), spec)) + .collect::>() + }) + .await + .map_err(|e| RpcError::internal(e.to_string()))?; + let mut windows = lock(&conn.chat_windows.windows); + for (rows, spec) in rows { + // A spec replaced meanwhile is recomputed by its own open, which waits for this one. + let Some(state) = windows.by_name.get_mut(&spec.window).filter(|state| state.spec == spec) + else { + continue; + }; + if let Some(message) = state.next(rows) { + conn.send(&Message::notification("_acpmux/chats_window_changed", message)); + } + } + Ok(()) +} diff --git a/cmux-tui/crates/acpmux/src/server/chats.rs b/cmux-tui/crates/acpmux/src/server/chats.rs index a83bb642b9d5..9e3e01980951 100644 --- a/cmux-tui/crates/acpmux/src/server/chats.rs +++ b/cmux-tui/crates/acpmux/src/server/chats.rs @@ -16,6 +16,12 @@ //! person (`chats/activity.rs`). Every chat value carries `attention` //! (`needsInput`, `failed`, `unread` or null) and `preview` (its live //! session's latest reply, or null). +//! - `_acpmux/chats_window {window, ...filter, limit?, topLevel?, exclude?, keys?}` and +//! `_acpmux/chats_window_close {window}`: a client's visible window, kept current with row +//! diffs (`_acpmux/chats_window_changed`, `chat_windows.rs`). A client that shows a list asks +//! for its window instead of mirroring the whole index. +//! - `_acpmux/chat_projects`: `{ready, enabled, projects: [{harness, path, lastUsedMs}]}`, each +//! harness's project folders grouped by acpmux (the app relays them to the cmux daemon). //! - `_acpmux/chat_open {key, cwd?}`: how the chat opens again (`chats/open.rs`): //! `adopt` with ready `session/new` params, `terminal` with argv/env/cwd, //! or `readOnly`; `needsFolder` when the person must pick the folder. @@ -333,6 +339,21 @@ pub(super) async fn route( }) .await } + "_acpmux/chat_projects" => { + let Some(service) = service else { + return Ok(json!({"ready": false, "enabled": true, "projects": []})); + }; + let activity = hub.chat_activity(); + blocking(move || { + Ok(match service.projects(&activity) { + Some(projects) => json!({"ready": true, "enabled": true, "projects": projects}), + None => json!({"ready": true, "enabled": false, "projects": []}), + }) + }) + .await + } + "_acpmux/chats_window" => super::chat_windows::open(hub, conn, service, ¶ms).await, + "_acpmux/chats_window_close" => super::chat_windows::close(conn, ¶ms), other => Err(RpcError::method_not_found(other)), } } @@ -454,6 +475,8 @@ pub(super) fn push_activity(hub: &Hub, conn: &Conn, ev: &HubEvent) { if rec.dir != "mux" || !changes_activity(&rec.kind) { return; } + // A window shows the chat's activity too: its rows are recomputed. + super::chat_windows::activity_changed(conn); if !hub.chat_index().is_some_and(|service| service.watches(&conn.id)) { return; } diff --git a/cmux-tui/crates/acpmux/src/server/mod.rs b/cmux-tui/crates/acpmux/src/server/mod.rs index ea98d108d554..ac056d572192 100644 --- a/cmux-tui/crates/acpmux/src/server/mod.rs +++ b/cmux-tui/crates/acpmux/src/server/mod.rs @@ -211,6 +211,8 @@ pub struct Conn { pub(super) remote_media: media_forward::RemoteMedia, /// Requests this daemon sent the client, waiting on its answer (`client_request.rs`). client_requests: client_request::ClientRequests, + /// Its windows over the chat index (`chat_windows.rs`). + chat_windows: chat_windows::ConnWindows, /// Its refused Stops that wait for room (`requests/cancel_refused.rs`). cancel_refusals: Arc, } @@ -1468,6 +1470,7 @@ async fn serve_connection_from( }, remote_media: media_forward::RemoteMedia::default(), client_requests: Default::default(), + chat_windows: Default::default(), cancel_refusals: Default::default(), }); tracing::debug!(conn = %conn.id, "client connected"); @@ -1844,6 +1847,8 @@ mod client_request; mod clock_text; pub(crate) use clock_text::iso; mod change_kind; +mod chat_windows; +pub use chat_windows::live_tasks as live_chat_window_tasks; mod chats; pub mod conn_auth; mod ephemeral; diff --git a/cmux-tui/crates/acpmux/tests/chats_window.rs b/cmux-tui/crates/acpmux/tests/chats_window.rs new file mode 100644 index 000000000000..fc270aed752a --- /dev/null +++ b/cmux-tui/crates/acpmux/tests/chats_window.rs @@ -0,0 +1,349 @@ +//! P-RUST item 2 (stall 2026-10-30): acpmux owns the chat list a client shows. A client asks for +//! its visible window (`_acpmux/chats_window`: filter, exclusions, limit) and gets the rows sorted +//! and filtered by acpmux; each index change then reaches it as a row diff with a revision: the +//! window's keys in order and only the rows whose value changed, never the index again. Every store +//! is a synthetic fixture in a temp home. + +#[path = "support/daemon_env.rs"] +mod daemon_env; +// This process starts no agent. +const HERMETIC: daemon_env::Process = daemon_env::Process::deny(); + +use acpmux::chats::ChatSources; +use acpmux::config::{Config, StoreMode}; +use acpmux::hub::Hub; +use acpmux::rpc::Message; +use acpmux::server::{Origin, serve_connection_with}; +use serde_json::{Value, json}; +use std::collections::VecDeque; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; + +const CHATS: usize = 3000; +/// The chats of folder `/work/p`. +const FOLDERS: usize = 50; +const WINDOW: usize = 20; + +struct Home(PathBuf); + +impl Drop for Home { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn session_id(i: usize) -> String { + format!("00000000-0000-4000-8000-{i:012}") +} + +fn key(i: usize) -> String { + format!("claude-code:{}", session_id(i)) +} + +fn record(i: usize, seconds: usize, text: &str) -> Value { + let (h, m, s) = (seconds / 3600, seconds / 60 % 60, seconds % 60); + json!({"type":"user","isSidechain":false,"cwd":format!("/work/p{}", i % FOLDERS),"sessionId":session_id(i), + "timestamp":format!("2026-10-01T{h:02}:{m:02}:{s:02}.000Z"), + "message":{"role":"user","content":text}}) +} + +fn session_file(projects: &Path, i: usize) -> PathBuf { + projects.join(format!("-work-p{}", i % FOLDERS)).join(format!("{}.jsonl", session_id(i))) +} + +fn append(file: &Path, record: &Value) { + use std::io::Write; + let mut f = std::fs::OpenOptions::new().append(true).open(file).unwrap(); + writeln!(f, "{record}").unwrap(); +} + +fn hub() -> Arc { + let mut cfg: Config = serde_json::from_value(json!({ + "harnesses": {"fake": {"argv": ["python3", "-c", "pass"]}}, + "defaultHarness": "fake", + })) + .unwrap(); + cfg.store.mode = StoreMode::Memory; + let store = acpmux::store::open(&cfg.store, Path::new("/nonexistent")).unwrap(); + Hub::new(cfg, store) +} + +/// The connection, the last request id, and the notifications read while waiting for an answer +/// (a window's rows are queued before the answer to its open). +struct Client(mpsc::Sender, mpsc::Receiver, i64, VecDeque); + +impl Client { + async fn next(&mut self) -> Value { + if let Some(v) = self.3.pop_front() { + return v; + } + let line = tokio::time::timeout(Duration::from_secs(30), self.1.recv()) + .await + .expect("a message within 30 s") + .expect("connection open"); + serde_json::from_str(&line).unwrap() + } + + async fn ok(&mut self, m: &str, params: Value) -> Value { + self.2 += 1; + let id = self.2; + self.0.send(Message::request(id, m, params).to_line()).await.unwrap(); + let mut held = VecDeque::new(); + loop { + let v = self.next().await; + if v.get("id") == Some(&json!(id)) { + assert!(v.get("error").is_none(), "{m}: {v}"); + held.extend(self.3.drain(..)); + self.3 = held; + return v["result"].clone(); + } + held.push_back(v); + } + } + + /// The next `_acpmux/chats_window_changed`. + async fn next_window(&mut self) -> Value { + loop { + let v = self.next().await; + if v["method"] == "_acpmux/chats_window_changed" { + return v["params"].clone(); + } + } + } +} + +fn keys(message: &Value) -> Vec { + message["keys"].as_array().unwrap().iter().map(|k| k.as_str().unwrap().to_owned()).collect() +} + +fn row_keys(message: &Value) -> Vec { + message["rows"] + .as_array() + .unwrap() + .iter() + .map(|c| c["key"].as_str().unwrap().to_owned()) + .collect() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_window_over_three_thousand_chats_is_sorted_filtered_and_one_change_is_one_row() { + // Under /tmp, not the temp dir: a worker's temp dir may be on another volume (`/Volumes`), + // where no chat root is ever listed. + let base = PathBuf::from("/tmp").join(format!("acpmux-window-{}", uuid::Uuid::now_v7())); + std::fs::create_dir_all(&base).unwrap(); + let home = Home(std::fs::canonicalize(&base).unwrap()); + let projects = home.0.join(".claude/projects"); + // A Claude chat's time is its file's: one second apart, a day ago, so the order is known and a + // later write puts its chat on top. + let day_ago = std::time::SystemTime::now() - Duration::from_secs(86_400); + for i in 0..CHATS { + let file = session_file(&projects, i); + std::fs::create_dir_all(file.parent().unwrap()).unwrap(); + std::fs::write(&file, format!("{}\n", record(i, i, &format!("chat {i}")))).unwrap(); + let at = day_ago + Duration::from_secs(i as u64); + std::fs::File::options().write(true).open(&file).unwrap().set_modified(at).unwrap(); + } + let hub = hub(); + hub.start_chats(ChatSources { + home: home.0.clone(), + acpmux_home: home.0.join(".acpmux"), + env: Arc::new(|_: &str| None), + launch_roots: Vec::new(), + user_roots: Vec::new(), + }) + .await + .unwrap(); + let (in_tx, in_rx) = mpsc::channel(64); + let (out_tx, out_rx) = mpsc::channel(4096); + tokio::spawn(serve_connection_with(hub.clone(), in_rx, out_tx, Origin::Local)); + let mut c = Client(in_tx, out_rx, 0, VecDeque::new()); + + // Folder p7 holds 60 of the 3000 chats (7, 57, ..., 2957). The window: its newest 20 but the + // one the client excludes (an open workspace shows it), newest first, sorted by acpmux. + let folder: Vec = (0..CHATS).rev().filter(|i| i % FOLDERS == 7).collect(); + let excluded = folder[1]; + let ack = c + .ok( + "_acpmux/chats_window", + json!({"window": "earlier", "folder": "/work/p7", "limit": WINDOW, "topLevel": true, + "exclude": [key(excluded)]}), + ) + .await; + assert_eq!(ack["window"], "earlier", "{ack}"); + let full = c.next_window().await; + assert_eq!(full["window"], "earlier", "{full}"); + assert_eq!(full["full"], true, "{full}"); + let first_revision = full["revision"].as_u64().expect("a revision"); + let expected: Vec = + folder.iter().filter(|&&i| i != excluded).take(WINDOW).map(|&i| key(i)).collect(); + assert_eq!(keys(&full), expected, "sorted newest first, filtered, the excluded chat left out"); + assert_eq!(row_keys(&full), expected, "the open sends every row of the window, no more"); + assert_eq!(full["hasMore"], true, "{full}"); + assert!(full["rows"].as_array().unwrap().iter().all(|row| row["brand"] == "claude")); + + // One change in the window's folder: the oldest p7 chat moves to the top. The diff is the next + // revision, the window's keys in order, and that one row alone. + let moved = folder[folder.len() - 1]; + append(&session_file(&projects, moved), &record(moved, CHATS + 1, "a new prompt")); + let diff = c.next_window().await; + assert_eq!(diff["full"], false, "{diff}"); + assert_eq!(diff["revision"], first_revision + 1, "{diff}"); + let mut after = vec![key(moved)]; + after.extend(expected.iter().take(WINDOW - 1).cloned()); + assert_eq!(keys(&diff), after, "{diff}"); + assert_eq!(row_keys(&diff), vec![key(moved)], "only the changed row travels"); + + // A change outside the window's filter sends nothing: the next message is the next p7 change, + // one revision later. + let other = 8; + append(&session_file(&projects, other), &record(other, CHATS + 2, "another folder")); + let next = folder[folder.len() - 2]; + append(&session_file(&projects, next), &record(next, CHATS + 3, "and another")); + let diff = c.next_window().await; + assert_eq!( + diff["revision"], + first_revision + 2, + "a change outside the window sent a diff: {diff}" + ); + assert_eq!(row_keys(&diff), vec![key(next)], "{diff}"); + assert_eq!(keys(&diff)[..2], [key(next), key(moved)], "{diff}"); + + // Closing the window stops its diffs. + c.ok("_acpmux/chats_window_close", json!({"window": "earlier"})).await; +} + +/// acpmux groups its own index into each harness's project folders (`_acpmux/chat_projects`), so +/// the app relays a few rows to the cmux daemon instead of walking every chat. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn chat_projects_are_each_folder_with_its_newest_chat() { + const N: usize = 120; + let base = PathBuf::from("/tmp").join(format!("acpmux-projects-{}", uuid::Uuid::now_v7())); + std::fs::create_dir_all(&base).unwrap(); + let home = Home(std::fs::canonicalize(&base).unwrap()); + let projects = home.0.join(".claude/projects"); + let day_ago = std::time::SystemTime::now() - Duration::from_secs(86_400); + for i in 0..N { + let file = session_file(&projects, i); + std::fs::create_dir_all(file.parent().unwrap()).unwrap(); + std::fs::write(&file, format!("{}\n", record(i, i, &format!("chat {i}")))).unwrap(); + let at = day_ago + Duration::from_secs(i as u64); + std::fs::File::options().write(true).open(&file).unwrap().set_modified(at).unwrap(); + } + let hub = hub(); + hub.start_chats(ChatSources { + home: home.0.clone(), + acpmux_home: home.0.join(".acpmux"), + env: Arc::new(|_: &str| None), + launch_roots: Vec::new(), + user_roots: Vec::new(), + }) + .await + .unwrap(); + let (in_tx, in_rx) = mpsc::channel(64); + let (out_tx, out_rx) = mpsc::channel(4096); + tokio::spawn(serve_connection_with(hub.clone(), in_rx, out_tx, Origin::Local)); + let mut c = Client(in_tx, out_rx, 0, VecDeque::new()); + + let page = c.ok("_acpmux/chats", json!({"limit": N})).await; + let newest_ms = |folder: usize| { + page["chats"] + .as_array() + .unwrap() + .iter() + .filter(|chat| chat["cwd"] == format!("/work/p{folder}")) + .map(|chat| chat["updatedMs"].as_i64().unwrap()) + .max() + .unwrap() + }; + let result = c.ok("_acpmux/chat_projects", json!({})).await; + assert_eq!(result["ready"], true, "{result}"); + assert_eq!(result["enabled"], true, "{result}"); + let rows = result["projects"].as_array().unwrap(); + assert_eq!(rows.len(), FOLDERS, "one row per folder: {result}"); + let mut paths: Vec = (0..FOLDERS).map(|f| format!("/work/p{f}")).collect(); + paths.sort_unstable(); + let listed: Vec<&str> = rows.iter().map(|row| row["path"].as_str().unwrap()).collect(); + assert_eq!(listed, paths, "by harness, then path"); + for row in rows { + assert_eq!(row["harness"], "claude-code", "{row}"); + let folder: usize = row["path"].as_str().unwrap()["/work/p".len()..].parse().unwrap(); + assert_eq!(row["lastUsedMs"].as_i64().unwrap(), newest_ms(folder), "{row}"); + } +} + +/// Review F1 (#28055): the daemon runs each request as its own task, so two specs of one window sent +/// back to back may be applied in either order. The client names each spec with a generation +/// (`gen`); acpmux keeps the highest it saw, ignores an older spec or close, and echoes `gen` in +/// every message, so the window always ends on the last spec the client sent. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_last_spec_sent_wins_whatever_order_the_requests_run_in() { + const N: usize = 40; + let base = PathBuf::from("/tmp").join(format!("acpmux-window-gen-{}", uuid::Uuid::now_v7())); + std::fs::create_dir_all(&base).unwrap(); + let home = Home(std::fs::canonicalize(&base).unwrap()); + let projects = home.0.join(".claude/projects"); + for i in 0..N { + let file = session_file(&projects, i); + std::fs::create_dir_all(file.parent().unwrap()).unwrap(); + std::fs::write(&file, format!("{}\n", record(i, i, &format!("chat {i}")))).unwrap(); + } + let hub = hub(); + hub.start_chats(ChatSources { + home: home.0.clone(), + acpmux_home: home.0.join(".acpmux"), + env: Arc::new(|_: &str| None), + launch_roots: Vec::new(), + user_roots: Vec::new(), + }) + .await + .unwrap(); + let (in_tx, in_rx) = mpsc::channel(64); + let (out_tx, out_rx) = mpsc::channel(4096); + tokio::spawn(serve_connection_with(hub.clone(), in_rx, out_tx, Origin::Local)); + let mut c = Client(in_tx, out_rx, 0, VecDeque::new()); + + // Many rounds: each sends specs gen 1..=5 (limit = gen) in one burst, then reads the answers. + for round in 0..20u64 { + let window = format!("w{round}"); + for g in 1..=5u64 { + c.2 += 1; + let params = json!({"window": window, "gen": g, "limit": g}); + c.0.send(Message::request(c.2, "_acpmux/chats_window", params).to_line()) + .await + .unwrap(); + } + let last_id = c.2; + let mut last = None; + loop { + let v = c.next().await; + if v["method"] == "_acpmux/chats_window_changed" && v["params"]["window"] == window { + last = Some(v["params"].clone()); + } + if v.get("id") == Some(&json!(last_id)) { + break; + } + } + // Answers may still be in flight after the last id: read until the gen-5 rows come. + let mut last = last.unwrap_or(Value::Null); + while last["gen"] != 5 { + let v = c.next().await; + if v["method"] == "_acpmux/chats_window_changed" && v["params"]["window"] == window { + last = v["params"].clone(); + } + } + assert_eq!(last["full"], true, "round {round}: {last}"); + assert_eq!( + last["keys"].as_array().unwrap().len(), + 5, + "round {round}: the last spec (limit 5) wins: {last}" + ); + // A close of an older generation is ignored; the window stays. + c.ok("_acpmux/chats_window_close", json!({"window": window, "gen": 4})).await; + let ack = + c.ok("_acpmux/chats_window", json!({"window": window, "gen": 3, "limit": 1})).await; + assert_eq!(ack["stale"], true, "round {round}: an older spec is refused as stale: {ack}"); + c.ok("_acpmux/chats_window_close", json!({"window": window, "gen": 6})).await; + } +} diff --git a/cmux-tui/crates/acpmux/tests/chats_window_owned.rs b/cmux-tui/crates/acpmux/tests/chats_window_owned.rs new file mode 100644 index 000000000000..e7c64398d834 --- /dev/null +++ b/cmux-tui/crates/acpmux/tests/chats_window_owned.rs @@ -0,0 +1,160 @@ +//! R6 for `_acpmux/chats_window` (lead 2026-10-30): a connection's window task is owned. It ends +//! when the client closes its last window and when the connection closes, so no task outlives +//! what it serves. Its own test binary: the task count is process-wide. + +#[path = "support/daemon_env.rs"] +mod daemon_env; +// This process starts no agent. +const HERMETIC: daemon_env::Process = daemon_env::Process::deny(); + +use acpmux::chats::ChatSources; +use acpmux::config::{Config, StoreMode}; +use acpmux::hub::Hub; +use acpmux::rpc::Message; +use acpmux::server::{Origin, serve_connection_with}; +use serde_json::{Value, json}; +use std::collections::VecDeque; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; + +/// The chats of folder `/work/p`. +const FOLDERS: usize = 50; + +struct Home(PathBuf); + +impl Drop for Home { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn session_id(i: usize) -> String { + format!("00000000-0000-4000-8000-{i:012}") +} + +fn record(i: usize, seconds: usize, text: &str) -> Value { + let (h, m, s) = (seconds / 3600, seconds / 60 % 60, seconds % 60); + json!({"type":"user","isSidechain":false,"cwd":format!("/work/p{}", i % FOLDERS),"sessionId":session_id(i), + "timestamp":format!("2026-10-01T{h:02}:{m:02}:{s:02}.000Z"), + "message":{"role":"user","content":text}}) +} + +fn session_file(projects: &Path, i: usize) -> PathBuf { + projects.join(format!("-work-p{}", i % FOLDERS)).join(format!("{}.jsonl", session_id(i))) +} + +fn hub() -> Arc { + let mut cfg: Config = serde_json::from_value(json!({ + "harnesses": {"fake": {"argv": ["python3", "-c", "pass"]}}, + "defaultHarness": "fake", + })) + .unwrap(); + cfg.store.mode = StoreMode::Memory; + let store = acpmux::store::open(&cfg.store, Path::new("/nonexistent")).unwrap(); + Hub::new(cfg, store) +} + +/// The connection, the last request id, and the notifications read while waiting for an answer +/// (a window's rows are queued before the answer to its open). +struct Client(mpsc::Sender, mpsc::Receiver, i64, VecDeque); + +impl Client { + async fn next(&mut self) -> Value { + if let Some(v) = self.3.pop_front() { + return v; + } + let line = tokio::time::timeout(Duration::from_secs(30), self.1.recv()) + .await + .expect("a message within 30 s") + .expect("connection open"); + serde_json::from_str(&line).unwrap() + } + + async fn ok(&mut self, m: &str, params: Value) -> Value { + self.2 += 1; + let id = self.2; + self.0.send(Message::request(id, m, params).to_line()).await.unwrap(); + let mut held = VecDeque::new(); + loop { + let v = self.next().await; + if v.get("id") == Some(&json!(id)) { + assert!(v.get("error").is_none(), "{m}: {v}"); + held.extend(self.3.drain(..)); + self.3 = held; + return v["result"].clone(); + } + held.push_back(v); + } + } + + /// The next `_acpmux/chats_window_changed`. + async fn next_window(&mut self) -> Value { + loop { + let v = self.next().await; + if v["method"] == "_acpmux/chats_window_changed" { + return v["params"].clone(); + } + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_window_task_ends_with_its_last_window_and_with_its_connection() { + const N: usize = 10; + let base = PathBuf::from("/tmp").join(format!("acpmux-window-owned-{}", uuid::Uuid::now_v7())); + std::fs::create_dir_all(&base).unwrap(); + let home = Home(std::fs::canonicalize(&base).unwrap()); + let projects = home.0.join(".claude/projects"); + for i in 0..N { + let file = session_file(&projects, i); + std::fs::create_dir_all(file.parent().unwrap()).unwrap(); + std::fs::write(&file, format!("{}\n", record(i, i, &format!("chat {i}")))).unwrap(); + } + let hub = hub(); + hub.start_chats(ChatSources { + home: home.0.clone(), + acpmux_home: home.0.join(".acpmux"), + env: Arc::new(|_: &str| None), + launch_roots: Vec::new(), + user_roots: Vec::new(), + }) + .await + .unwrap(); + assert_eq!(acpmux::server::live_chat_window_tasks(), 0); + let (in_tx, in_rx) = mpsc::channel(64); + let (out_tx, out_rx) = mpsc::channel(4096); + tokio::spawn(serve_connection_with(hub.clone(), in_rx, out_tx, Origin::Local)); + let mut c = Client(in_tx, out_rx, 0, VecDeque::new()); + + // Two windows, one task; the task lives until the last one closes. + for window in ["a", "b"] { + c.ok("_acpmux/chats_window", json!({"window": window, "limit": 3})).await; + assert_eq!(c.next_window().await["window"], window); + } + assert_eq!(acpmux::server::live_chat_window_tasks(), 1, "one task per connection"); + c.ok("_acpmux/chats_window_close", json!({"window": "a"})).await; + assert_eq!(acpmux::server::live_chat_window_tasks(), 1, "window b is still open"); + c.ok("_acpmux/chats_window_close", json!({"window": "b"})).await; + tasks_reach_zero("the last window closed").await; + + // A later open starts a new task; closing the connection ends it. + c.ok("_acpmux/chats_window", json!({"window": "a", "limit": 3})).await; + assert_eq!(c.next_window().await["full"], true, "a reopened window sends every row"); + assert_eq!(acpmux::server::live_chat_window_tasks(), 1); + drop(c); + tasks_reach_zero("the connection closed").await; +} + +/// Waits (bounded) for the task count to reach 0: the task ends on its own executor turn. +async fn tasks_reach_zero(after: &str) { + let ended = tokio::time::timeout(Duration::from_secs(10), async { + while acpmux::server::live_chat_window_tasks() != 0 { + tokio::task::yield_now().await; + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await; + assert!(ended.is_ok(), "a chat window task outlived {after}"); +} diff --git a/plans/cmux-next/layer-ownership.md b/plans/cmux-next/layer-ownership.md index 341b03a29929..77ce989ee356 100644 --- a/plans/cmux-next/layer-ownership.md +++ b/plans/cmux-next/layer-ownership.md @@ -6,9 +6,9 @@ This file answers one question: which LANGUAGE LAYER owns a piece of state or a ## 1. The model -1. The Rust daemon owns all durable state, all shared state and all product rules. "Daemon" means cmux-tui-core and its peer actors in the same process (store, session host, home, config, git/files), plus acpmux for agent sessions until it moves in-process (cx-ncc.27). Each piece of state has exactly one writer. The daemon exposes it as v2 resource operations (cmux-tui/spec/resource-operations-v2.json) and event streams. -2. The Swift app owns presentation and platform: windows, views, focus, input, gestures, animation, and macOS services (Keychain custody, notifications, TCC prompts, NSWorkspace, pasteboard, Chromium runtime). It keeps a read-only mirror of daemon state and sends intents. The only optimistic layer is a pending-edit log that clears when the daemon confirms or rejects (the SidebarPendingEdits pattern, cx-odqn, `Packages/macOS/CmuxNext/Sources/CmuxNextSidebar/Logic/SidebarPendingEdits.swift`). -3. TypeScript and React webviews are views. Data comes through a generated typed API (webviews/src/protocol/generated, the generated page clients, acpmux generated types). A webview holds only local UI state (hover, open or closed, the text in an input, scroll). It holds no business rule and no durable product state (no localStorage or IndexedDB copy of product data). +1. Rust owns every state, list, sort, filter, search and rule (Lawrence 2026-10-30: "move as much as possible into Rust"). The Rust daemon owns all durable state, all shared state and all product rules, and it also owns every list a client shows: the order, the filter, the search, the grouping and the paging run in Rust, and a client asks for its visible window and gets row diffs with a revision (example: `_acpmux/chats_window`, the chat index). "Daemon" means cmux-tui-core and its peer actors in the same process (store, session host, home, config, git/files), plus acpmux for agent sessions until it moves in-process (cx-ncc.27). Each piece of state has exactly one writer. The daemon exposes it as v2 resource operations (cmux-tui/spec/resource-operations-v2.json) and event streams. +2. Swift is only the platform shell: windows, the web view host, focus and first responder, key equivalents, IME and dictation, drag and drop, native menus, Keychain, notifications, TCC prompts, NSWorkspace, pasteboard, and the CEF/WebKit runtime. It sends intents. It never mirrors a whole daemon list: a native Swift list holds only its visible rows, as value snapshots from the Rust window, and applies the row diffs on the main thread in O(changed rows). Swift is never a relay or a mirror for data that a webview shows; the webview reads it from its owner. The only optimistic layer is a pending-edit log that clears when the daemon confirms or rejects (the SidebarPendingEdits pattern, cx-odqn, `Packages/macOS/CmuxNext/Sources/CmuxNextSidebar/Logic/SidebarPendingEdits.swift`). +3. TypeScript and React webviews are views. They ask the Rust owner for their window (sort, filter, search and paging stay in Rust) and never get it through a Swift relay. Data comes through a generated typed API (webviews/src/protocol/generated, the generated page clients, acpmux generated types). A webview holds only local UI state (hover, open or closed, the text in an input, scroll). It holds no business rule and no durable product state (no localStorage or IndexedDB copy of product data). 4. A rule needed in more than one layer (validation limits, policies, catalogs, enums, op names, wire types) lives once, in Rust or in one data file, and reaches Swift and TypeScript by codegen with a CI drift check. Never by a hand copy, and never by a "parity test" that keeps two hand copies equal. 5. Mobile follows the same model through the shared Rust core over uniffi (mobile-rust-core.md, MOBILE-RUST-1 to 5). Swift on iOS and Kotlin on Android hold UI and platform only. @@ -56,7 +56,7 @@ Exceptions that stay (recorded decisions, not violations): - L4. The only optimistic state is a pending-edit log keyed by an idempotency key, cleared on echo or reject. To compute the optimistic view, call the Rust reducer (FFI, as RustSidebarClient does with `cmux-layout-reducer-ffi`) instead of a Swift copy of the reducer. A Swift copy is allowed only behind a property test against the Rust reducer, and only until the FFI path exists. - L5. Shared constants, catalogs and enums live once in Rust or one data file and are generated into Swift and TS. Every generated copy has a `--check` mode run by safe-push or CI. A parity test between two hand copies is a migration aid, not a fix. - L6. Op names and wire types reach Swift and TS by codegen from resource-operations-v2.json (TS and the 7 SDKs already have emitters; Swift has none yet, see V-1). -- L7. Webviews keep no durable state and open no side channel to an owner that bypasses the generated API, unless the channel is the designed transport and its client is generated. +- L7. Webviews keep no durable state and open no side channel to an owner that bypasses the generated API, unless the channel is the designed transport and its client is generated. Data that a webview shows comes from its Rust owner, never through Swift as a relay or a mirror. No layer outside Rust keeps a whole list to sort, filter, search or page it; it asks Rust for the visible window and applies row diffs. - L8. A new mobile feature with a wire contract goes into the Rust mobile core, never into a Swift package that Android would have to copy. - L9. Review subagents cite these rule ids together with code-quality.md G1 and SW11. diff --git a/scripts/cmux-next/main-actor-work-baseline.json b/scripts/cmux-next/main-actor-work-baseline.json index 8d54b7ff93e7..58c3a6680706 100644 --- a/scripts/cmux-next/main-actor-work-baseline.json +++ b/scripts/cmux-next/main-actor-work-baseline.json @@ -8,6 +8,9 @@ "Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/AgentTurnStateFeed.swift": { "a main-actor Task per line or event (batch with MainActorLineBatch)": 1 }, + "Packages/macOS/CmuxNext/Sources/CmuxNextAgentLanes/ChatsFeed.swift": { + "a main-actor Task per line or event (batch with MainActorLineBatch)": 1 + }, "Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AgentPaneContextMenu.swift": { "line split of a whole input": 1 }, @@ -26,9 +29,6 @@ "Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AgentPaneShell.swift": { "process launch": 1 }, - "Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentRecents/ChatsFeed.swift": { - "a main-actor Task per line or event (batch with MainActorLineBatch)": 1 - }, "Packages/macOS/CmuxNext/Sources/CmuxNextApp/AgentTabSeed.swift": { "line split of a whole input": 1 }, diff --git a/scripts/cmux-next/perf-hot-benches.json b/scripts/cmux-next/perf-hot-benches.json index 5f11b8a4c04a..eed92a96461e 100644 --- a/scripts/cmux-next/perf-hot-benches.json +++ b/scripts/cmux-next/perf-hot-benches.json @@ -80,6 +80,11 @@ "what": "every action run's always-on budget record: two clock reads and one ring-slot write (ActionBudgetMonitor.record)", "bench": "CMUX_ACTION_BUDGET_BENCH=1 swift test -c release --filter ActionBudgetMonitorBench (ns per run and per batch), on macOS 26 and 27", "owner": "Performance Lead (PL-instant)" + }, + "chats-window-diff": { + "what": "one acpmux chats window message on the main actor: the window's keys and the changed rows (sidebar chat history; acpmux sorts, filters and pages)", + "bench": "ChatsWindowMainThreadTests (rows touched per change, 3,000-chat index) on the fleet, and the debug hang monitor (debug.hangs) on a many-chat load", + "owner": "Performance Lead" } } } diff --git a/webviews/src/agent-session/acpmux/protocol/generated/methods.ts b/webviews/src/agent-session/acpmux/protocol/generated/methods.ts index e951642bc919..83404b756ed6 100644 --- a/webviews/src/agent-session/acpmux/protocol/generated/methods.ts +++ b/webviews/src/agent-session/acpmux/protocol/generated/methods.ts @@ -169,6 +169,9 @@ export interface AcpmuxRequests { "_acpmux/chat_settled": { params: unknown; result: unknown }; "_acpmux/chat_transcript": { params: unknown; result: unknown }; "_acpmux/chat_fork": { params: unknown; result: unknown }; + "_acpmux/chat_projects": { params: unknown; result: unknown }; + "_acpmux/chats_window": { params: unknown; result: unknown }; + "_acpmux/chats_window_close": { params: unknown; result: unknown }; "_acpmux/transfer_abort": { params: unknown; result: unknown }; "_acpmux/transfer_begin": { params: unknown; result: unknown }; "_acpmux/transfer_commit": { params: unknown; result: unknown }; @@ -252,6 +255,7 @@ export interface AcpmuxNotifications { "_acpmux/process_update": unknown; "_acpmux/chat_changed": unknown; "_acpmux/chats_lagged": unknown; + "_acpmux/chats_window_changed": unknown; "_acpmux/event": SessionEvent; "_acpmux/message": MessageEvent; "_acpmux/session_changed": SessionChanged; @@ -451,8 +455,12 @@ export const ACPMUX_METHODS: ReadonlyArray<{ name: string; kind: AcpmuxMethodKin { name: "_acpmux/chat_settled", kind: "request", acp: false, typed: false }, { name: "_acpmux/chat_transcript", kind: "request", acp: false, typed: false }, { name: "_acpmux/chat_fork", kind: "request", acp: false, typed: false }, + { name: "_acpmux/chat_projects", kind: "request", acp: false, typed: false }, { name: "_acpmux/chat_changed", kind: "notification", acp: false, typed: false }, { name: "_acpmux/chats_lagged", kind: "notification", acp: false, typed: false }, + { name: "_acpmux/chats_window", kind: "request", acp: false, typed: false }, + { name: "_acpmux/chats_window_close", kind: "request", acp: false, typed: false }, + { name: "_acpmux/chats_window_changed", kind: "notification", acp: false, typed: false }, { name: "_acpmux/transfer_abort", kind: "request", acp: false, typed: false }, { name: "_acpmux/transfer_begin", kind: "request", acp: false, typed: false }, { name: "_acpmux/transfer_commit", kind: "request", acp: false, typed: false },