Stream app operator state over WebSocket

apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift

@@ -261,8 +261,7 @@ struct AccountPanelView: View {
 			if let snapshot = store.operatorSnapshot, snapshot.shouldDisplayInPanel {
 				OperatorStatusStripView(
 					snapshot: snapshot,
-					updatedAt: store.operatorSnapshotUpdatedAt,
-					refreshIntervalSeconds: AccountStore.operatorSnapshotRefreshIntervalSeconds
+					updatedAt: store.operatorSnapshotUpdatedAt
 				)
 			}
 
@@ -1531,7 +1530,6 @@ struct NoticeView: View {
 struct OperatorStatusStripView: View {
 	let snapshot: OperatorSnapshotResponse
 	let updatedAt: Date?
-	let refreshIntervalSeconds: Int
 	@Environment(\.colorScheme) private var colorScheme
 	@State private var showsAllLanes = false
 
@@ -1576,7 +1574,6 @@ struct OperatorStatusStripView: View {
 						.monospacedDigit()
 				}
 				.frame(height: 16)
-				.help("Operator snapshot refreshes every \(refreshIntervalSeconds) seconds.")
 			}
 		}
 		.padding(.horizontal, 6)
@@ -1628,12 +1625,12 @@ struct OperatorStatusStripView: View {
 
 	private var refreshMeta: String {
 		guard let updatedAt else {
-			return "\(refreshIntervalSeconds)s refresh"
+			return "WS live"
 		}
 
 		let age = max(0, Int(Date().timeIntervalSince(updatedAt).rounded()))
 		if age < 2 {
-			return "\(refreshIntervalSeconds)s refresh"
+			return "live"
 		}
 
 		return "\(age)s ago"

apps/decodex-app/Sources/DecodexApp/AccountStore.swift

@@ -15,12 +15,12 @@ final class AccountStore: ObservableObject {
 
 	private let bridge = DecodexAppBridge()
 	private var automaticRefreshTask: Task<Void, Never>?
-	private var automaticOperatorRefreshTask: Task<Void, Never>?
-	static let operatorSnapshotRefreshIntervalSeconds = 10
+	private var operatorSnapshotStreamTask: Task<Void, Never>?
+	private var pendingRunActivity: [OperatorRunStatus]?
 
 	deinit {
 		automaticRefreshTask?.cancel()
-		automaticOperatorRefreshTask?.cancel()
+		operatorSnapshotStreamTask?.cancel()
 	}
 
 	var isInitialLoading: Bool {
@@ -87,9 +87,7 @@ final class AccountStore: ObservableObject {
 			)
 			notice = nil
 			await refreshFastMode()
-			await refreshOperatorSnapshot()
 		} catch {
-			operatorSnapshot = nil
 			notice = error.localizedDescription
 		}
 	}
@@ -102,19 +100,6 @@ final class AccountStore: ObservableObject {
 		await refresh()
 	}
 
-	func refreshOperatorSnapshot() async {
-		do {
-			operatorSnapshot = try await bridge.runJSON(
-				.operatorSnapshot,
-				as: OperatorSnapshotResponse.self
-			)
-			operatorSnapshotUpdatedAt = Date()
-		} catch {
-			operatorSnapshot = nil
-			operatorSnapshotUpdatedAt = nil
-		}
-	}
-
 	func openWebUI() async {
 		do {
 			let url = try await DecodexServerBridge.shared.dashboardURL()
@@ -152,26 +137,103 @@ final class AccountStore: ObservableObject {
 			}
 		}
 
-		startAutomaticOperatorRefresh()
+		startOperatorSnapshotStream()
 	}
 
-	private func startAutomaticOperatorRefresh() {
-		guard automaticOperatorRefreshTask == nil else {
+	func startOperatorSnapshotStream() {
+		guard operatorSnapshotStreamTask == nil else {
 			return
 		}
 
-		automaticOperatorRefreshTask = Task { [weak self] in
-			while !Task.isCancelled {
-				do {
-					try await Task.sleep(
-						nanoseconds: UInt64(Self.operatorSnapshotRefreshIntervalSeconds) * 1_000_000_000
-					)
-				} catch {
-					return
-				}
+		operatorSnapshotStreamTask = Task { [weak self] in
+			await self?.runOperatorSnapshotStream()
+		}
+	}
+
+	private func runOperatorSnapshotStream() async {
+		while !Task.isCancelled {
+			do {
+				try await connectOperatorSnapshotStream()
+			} catch {
+				operatorSnapshot = nil
+				operatorSnapshotUpdatedAt = nil
+			}
+
+			do {
+				try await Task.sleep(nanoseconds: 1_000_000_000)
+			} catch {
+				return
+			}
+		}
+	}
+
+	private func connectOperatorSnapshotStream() async throws {
+		let url = try await DecodexServerBridge.shared.dashboardWebSocketURL()
+		let socket = URLSession.shared.webSocketTask(with: url)
+
+		socket.resume()
+		defer {
+			socket.cancel(with: .normalClosure, reason: nil)
+		}
+
+		while !Task.isCancelled {
+			let event = try await receiveOperatorDashboardEvent(from: socket)
+
+			applyOperatorDashboardEvent(event)
+		}
+	}
+
+	private func receiveOperatorDashboardEvent(
+		from socket: URLSessionWebSocketTask
+	) async throws -> OperatorDashboardSocketEvent {
+		let message = try await socket.receive()
+		let data: Data
+
+		switch message {
+		case .string(let text):
+			guard let textData = text.data(using: .utf8) else {
+				throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket event is not UTF-8")
+			}
+
+			data = textData
+		case .data(let binary):
+			data = binary
+		@unknown default:
+			throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket event type is unsupported")
+		}
+
+		return try JSONDecoder().decode(OperatorDashboardSocketEvent.self, from: data)
+	}
+
+	private func applyOperatorDashboardEvent(_ event: OperatorDashboardSocketEvent) {
+		guard let payload = event.payload else {
+			return
+		}
+
+		switch event.type {
+		case "snapshot":
+			guard let snapshot = payload.snapshot else {
+				return
+			}
+
+			if let pendingRunActivity {
+				operatorSnapshot = snapshot.mergingRunActivity(pendingRunActivity)
+			} else {
+				operatorSnapshot = snapshot
+			}
+			operatorSnapshotUpdatedAt = payload.snapshotPublishedAt ?? Date()
+		case "runActivity":
+			guard let activeRuns = payload.activeRuns else {
+				return
+			}
 
-				await self?.refreshOperatorSnapshot()
+			pendingRunActivity = activeRuns
+			if let operatorSnapshot {
+				self.operatorSnapshot = operatorSnapshot.mergingRunActivity(activeRuns)
 			}
+			operatorSnapshotUpdatedAt = payload.emittedAt ?? Date()
+		default:
+			break
 		}
 	}
 

apps/decodex-app/Sources/DecodexApp/DecodexApp.swift

@@ -61,7 +61,7 @@ struct DecodexApp: App {
 		let content = AccountPanelView(store: store, loginWindowState: loginWindowState)
 			.task {
 				await store.refreshIfNeeded()
-				await store.refreshOperatorSnapshot()
+				store.startOperatorSnapshotStream()
 			}
 
 		if #available(macOS 15.0, *) {

apps/decodex-app/Sources/DecodexApp/DecodexAppBridge.swift

@@ -73,8 +73,6 @@ struct AppBridgeRequest: Encodable, Sendable {
 		AppBridgeRequest(operation: "codex_fast_mode_set", enabled: enabled)
 	}
 
-	static let operatorSnapshot = AppBridgeRequest(operation: "operator_snapshot")
-
 	private init(
 		operation: String,
 		selector: String? = nil,

apps/decodex-app/Sources/DecodexApp/DecodexServerBridge.swift

@@ -29,6 +29,23 @@ actor DecodexServerBridge {
 		return baseURL.appendingPathComponent("dashboard")
 	}
 
+	func dashboardWebSocketURL() async throws -> URL {
+		let baseURL = try await ensureServer()
+		guard var components = URLComponents(url: baseURL, resolvingAgainstBaseURL: false) else {
+			throw DecodexAppBridgeError.invalidResponse("Decodex server URL is invalid")
+		}
+
+		components.scheme = baseURL.scheme == "https" ? "wss" : "ws"
+		components.path = "/dashboard/control"
+		components.query = nil
+
+		guard let url = components.url else {
+			throw DecodexAppBridgeError.invalidResponse("Decodex dashboard WebSocket URL is invalid")
+		}
+
+		return url
+	}
+
 	func run<T: Decodable & Sendable>(_ request: AppBridgeRequest, as type: T.Type) async throws -> T {
 		guard let route = try request.serverRoute() else {
 			throw DecodexAppBridgeError.invalidResponse("request is not supported by Decodex server")
@@ -272,8 +289,6 @@ extension AppBridgeRequest {
 			return try jsonPost("api/accounts/import")
 		case "account_use":
 			return try jsonPost("api/accounts/use")
-		case "operator_snapshot":
-			return ServerRoute(method: "GET", path: "api/operator-snapshot", body: nil)
 		default:
 			return nil
 		}