From dc26dd4f3f5473932c7943afe40621501e58966b Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Fri, 22 May 2026 15:59:57 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Stream app operator state over WebSocket","authority":"manual"} --- .../Sources/DecodexApp/AccountPanelView.swift | 9 +- .../Sources/DecodexApp/AccountStore.swift | 124 +++++++--- .../Sources/DecodexApp/DecodexApp.swift | 2 +- .../Sources/DecodexApp/DecodexAppBridge.swift | 2 - .../DecodexApp/DecodexServerBridge.swift | 19 +- .../DecodexApp/OperatorSnapshotModels.swift | 220 ++++++++++++++++++ apps/decodex/src/orchestrator.rs | 1 + apps/decodex/src/orchestrator/entrypoints.rs | 38 +-- 8 files changed, 355 insertions(+), 60 deletions(-) diff --git a/apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift b/apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift index 93cb732d7..5d9a7a28f 100644 --- a/apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift +++ b/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" diff --git a/apps/decodex-app/Sources/DecodexApp/AccountStore.swift b/apps/decodex-app/Sources/DecodexApp/AccountStore.swift index 5f91995b7..5c0a1c932 100644 --- a/apps/decodex-app/Sources/DecodexApp/AccountStore.swift +++ b/apps/decodex-app/Sources/DecodexApp/AccountStore.swift @@ -15,12 +15,12 @@ final class AccountStore: ObservableObject { private let bridge = DecodexAppBridge() private var automaticRefreshTask: Task? - private var automaticOperatorRefreshTask: Task? - static let operatorSnapshotRefreshIntervalSeconds = 10 + private var operatorSnapshotStreamTask: Task? + 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 } } diff --git a/apps/decodex-app/Sources/DecodexApp/DecodexApp.swift b/apps/decodex-app/Sources/DecodexApp/DecodexApp.swift index 5eb69748a..11d04f95b 100644 --- a/apps/decodex-app/Sources/DecodexApp/DecodexApp.swift +++ b/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, *) { diff --git a/apps/decodex-app/Sources/DecodexApp/DecodexAppBridge.swift b/apps/decodex-app/Sources/DecodexApp/DecodexAppBridge.swift index 21a3f72d0..08de47e0d 100644 --- a/apps/decodex-app/Sources/DecodexApp/DecodexAppBridge.swift +++ b/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, diff --git a/apps/decodex-app/Sources/DecodexApp/DecodexServerBridge.swift b/apps/decodex-app/Sources/DecodexApp/DecodexServerBridge.swift index ed026af7b..f2d73975a 100644 --- a/apps/decodex-app/Sources/DecodexApp/DecodexServerBridge.swift +++ b/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(_ 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 } diff --git a/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift b/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift index 88763e4fb..e81300fc2 100644 --- a/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift +++ b/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift @@ -77,6 +77,46 @@ struct OperatorSnapshotResponse: Decodable, Sendable { activeRuns(for: account).count } + func mergingRunActivity(_ activityRuns: [OperatorRunStatus]) -> OperatorSnapshotResponse { + let snapshotRunsByID = activeRuns.reduce(into: [String: OperatorRunStatus]()) { runsByID, run in + runsByID[run.runID] = run + } + let mergedRuns = activityRuns.map { activityRun in + snapshotRunsByID[activityRun.runID]?.mergingActivity(activityRun) ?? activityRun + } + let activeCountsByProject = Dictionary(grouping: mergedRuns.compactMap(\.projectID)) { $0 } + .mapValues(\.count) + let mergedProjects = projects.map { project in + guard let projectID = project.projectID else { + return project + } + + return project.withActiveRunCount(activeCountsByProject[projectID] ?? 0) + } + + return OperatorSnapshotResponse( + warnings: warnings, + projects: mergedProjects, + activeRuns: mergedRuns, + queuedCandidates: queuedCandidates, + postReviewLanes: postReviewLanes + ) + } + + private init( + warnings: [String], + projects: [OperatorProjectStatus], + activeRuns: [OperatorRunStatus], + queuedCandidates: [OperatorQueuedIssueStatus], + postReviewLanes: [OperatorPostReviewLaneStatus] + ) { + self.warnings = warnings + self.projects = projects + self.activeRuns = activeRuns + self.queuedCandidates = queuedCandidates + self.postReviewLanes = postReviewLanes + } + private var isAPIOnlySnapshot: Bool { warnings.contains("automation_disabled") && projects.allSatisfy { $0.connectorState == "api_only" } @@ -129,6 +169,7 @@ struct OperatorSnapshotResponse: Decodable, Sendable { } struct OperatorProjectStatus: Decodable, Sendable { + let projectID: String? let connectorState: String? let activeRunCount: Int let queuedCandidateCount: Int @@ -138,7 +179,22 @@ struct OperatorProjectStatus: Decodable, Sendable { let cleanupBlockedCount: Int let cleanupPendingCount: Int + func withActiveRunCount(_ count: Int) -> OperatorProjectStatus { + OperatorProjectStatus( + projectID: projectID, + connectorState: connectorState, + activeRunCount: count, + queuedCandidateCount: queuedCandidateCount, + postReviewLaneCount: postReviewLaneCount, + waitingLaneCount: waitingLaneCount, + attentionCount: attentionCount, + cleanupBlockedCount: cleanupBlockedCount, + cleanupPendingCount: cleanupPendingCount + ) + } + enum CodingKeys: String, CodingKey { + case projectID = "project_id" case connectorState = "connector_state" case activeRunCount = "active_run_count" case queuedCandidateCount = "queued_candidate_count" @@ -152,6 +208,7 @@ struct OperatorProjectStatus: Decodable, Sendable { init(from decoder: Decoder) throws { let container = try decoder.container(keyedBy: CodingKeys.self) + projectID = try container.decodeIfPresent(String.self, forKey: .projectID) connectorState = try container.decodeIfPresent(String.self, forKey: .connectorState) activeRunCount = try container.decodeIfPresent(Int.self, forKey: .activeRunCount) ?? 0 queuedCandidateCount = try container.decodeIfPresent(Int.self, forKey: .queuedCandidateCount) ?? 0 @@ -161,6 +218,28 @@ struct OperatorProjectStatus: Decodable, Sendable { cleanupBlockedCount = try container.decodeIfPresent(Int.self, forKey: .cleanupBlockedCount) ?? 0 cleanupPendingCount = try container.decodeIfPresent(Int.self, forKey: .cleanupPendingCount) ?? 0 } + + private init( + projectID: String?, + connectorState: String?, + activeRunCount: Int, + queuedCandidateCount: Int, + postReviewLaneCount: Int, + waitingLaneCount: Int, + attentionCount: Int, + cleanupBlockedCount: Int, + cleanupPendingCount: Int + ) { + self.projectID = projectID + self.connectorState = connectorState + self.activeRunCount = activeRunCount + self.queuedCandidateCount = queuedCandidateCount + self.postReviewLaneCount = postReviewLaneCount + self.waitingLaneCount = waitingLaneCount + self.attentionCount = attentionCount + self.cleanupBlockedCount = cleanupBlockedCount + self.cleanupPendingCount = cleanupPendingCount + } } struct OperatorQueuedIssueStatus: Decodable, Sendable { @@ -274,6 +353,54 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable { self.account?.matches(account) == true } + func mergingActivity(_ activity: OperatorRunStatus) -> OperatorRunStatus { + OperatorRunStatus( + projectID: activity.projectID ?? projectID, + runID: activity.runID, + issueID: activity.issueID ?? issueID, + issueIdentifier: activity.issueIdentifier ?? issueIdentifier, + title: mergedTitle(from: activity), + status: activity.status ?? status, + attemptStatus: activity.attemptStatus ?? attemptStatus, + attemptNumber: activity.attemptNumber ?? attemptNumber, + phase: activity.phase ?? phase, + waitReason: activity.waitReason ?? waitReason, + currentOperation: activity.currentOperation ?? currentOperation, + threadStatus: activity.threadStatus ?? threadStatus, + idleForSeconds: activity.idleForSeconds ?? idleForSeconds, + protocolIdleForSeconds: activity.protocolIdleForSeconds ?? protocolIdleForSeconds, + updatedAt: activity.updatedAt ?? updatedAt, + lastProgressAt: activity.lastProgressAt ?? lastProgressAt, + nextRetryAt: activity.nextRetryAt ?? nextRetryAt, + lastEventType: activity.lastEventType ?? lastEventType, + eventCount: activity.eventCount ?? eventCount, + processAlive: activity.processAlive ?? processAlive, + activeLease: activity.activeLease ?? activeLease, + branchName: activity.branchName ?? branchName, + worktreePath: activity.worktreePath ?? worktreePath, + suspectedStall: activity.suspectedStall || suspectedStall, + childAgentActivity: activity.childAgentActivity ?? childAgentActivity, + account: activity.account ?? account, + accounts: activity.accounts.isEmpty ? accounts : activity.accounts + ) + } + + private func mergedTitle(from activity: OperatorRunStatus) -> String? { + if let title = activity.title, !title.isEmpty, !activity.titleIsOperationFallback { + return title + } + + return title ?? activity.title + } + + private var titleIsOperationFallback: Bool { + guard let title, !title.isEmpty else { + return false + } + + return title == readable(currentOperation ?? phase ?? "") + } + enum CodingKeys: String, CodingKey { case projectID = "project_id" case runID = "run_id" @@ -338,6 +465,91 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable { account = try container.decodeIfPresent(OperatorRunAccountSummary.self, forKey: .account) accounts = try container.decodeIfPresent([OperatorRunAccountSummary].self, forKey: .accounts) ?? [] } + + private init( + projectID: String?, + runID: String, + issueID: String?, + issueIdentifier: String?, + title: String?, + status: String?, + attemptStatus: String?, + attemptNumber: Int?, + phase: String?, + waitReason: String?, + currentOperation: String?, + threadStatus: String?, + idleForSeconds: Int?, + protocolIdleForSeconds: Int?, + updatedAt: String?, + lastProgressAt: String?, + nextRetryAt: String?, + lastEventType: String?, + eventCount: Int?, + processAlive: Bool?, + activeLease: Bool?, + branchName: String?, + worktreePath: String?, + suspectedStall: Bool, + childAgentActivity: OperatorChildAgentActivity?, + account: OperatorRunAccountSummary?, + accounts: [OperatorRunAccountSummary] + ) { + self.projectID = projectID + self.runID = runID + self.issueID = issueID + self.issueIdentifier = issueIdentifier + self.title = title + self.status = status + self.attemptStatus = attemptStatus + self.attemptNumber = attemptNumber + self.phase = phase + self.waitReason = waitReason + self.currentOperation = currentOperation + self.threadStatus = threadStatus + self.idleForSeconds = idleForSeconds + self.protocolIdleForSeconds = protocolIdleForSeconds + self.updatedAt = updatedAt + self.lastProgressAt = lastProgressAt + self.nextRetryAt = nextRetryAt + self.lastEventType = lastEventType + self.eventCount = eventCount + self.processAlive = processAlive + self.activeLease = activeLease + self.branchName = branchName + self.worktreePath = worktreePath + self.suspectedStall = suspectedStall + self.childAgentActivity = childAgentActivity + self.account = account + self.accounts = accounts + } +} + +struct OperatorDashboardSocketEvent: Decodable, Sendable { + let type: String + let payload: OperatorDashboardSocketPayload? +} + +struct OperatorDashboardSocketPayload: Decodable, Sendable { + let emittedAtUnixEpoch: Int64? + let snapshotPublishedAtUnixEpoch: Int64? + let snapshot: OperatorSnapshotResponse? + let activeRuns: [OperatorRunStatus]? + + var emittedAt: Date? { + date(fromUnixEpoch: emittedAtUnixEpoch) + } + + var snapshotPublishedAt: Date? { + date(fromUnixEpoch: snapshotPublishedAtUnixEpoch) + } + + enum CodingKeys: String, CodingKey { + case emittedAtUnixEpoch + case snapshotPublishedAtUnixEpoch + case snapshot + case activeRuns + } } struct OperatorChildAgentActivity: Decodable, Sendable { @@ -474,3 +686,11 @@ private func readable(_ value: String) -> String { return words.joined(separator: " ") } + +private func date(fromUnixEpoch value: Int64?) -> Date? { + guard let value else { + return nil + } + + return Date(timeIntervalSince1970: TimeInterval(value)) +} diff --git a/apps/decodex/src/orchestrator.rs b/apps/decodex/src/orchestrator.rs index 772062241..6623c0b52 100644 --- a/apps/decodex/src/orchestrator.rs +++ b/apps/decodex/src/orchestrator.rs @@ -86,6 +86,7 @@ const OPERATOR_DASHBOARD_WS_CLIENT_MESSAGE_MAX_BYTES: usize = 64 * 1_024; const OPERATOR_STATE_HEADER_TERMINATOR: &[u8] = b"\r\n\r\n"; const OPERATOR_DASHBOARD_WS_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(20); const OPERATOR_RUN_ACTIVITY_STREAM_INTERVAL: Duration = Duration::from_secs(1); +const OPERATOR_API_ONLY_SNAPSHOT_STREAM_INTERVAL: Duration = Duration::from_secs(1); const PULL_REQUEST_REVIEW_STATE_QUERY: &str = r#" query($owner: String!, $name: String!, $number: Int!, $reviewThreadsAfter: String) { repository(owner: $owner, name: $name) { diff --git a/apps/decodex/src/orchestrator/entrypoints.rs b/apps/decodex/src/orchestrator/entrypoints.rs index 49fc80cc3..4e3104918 100644 --- a/apps/decodex/src/orchestrator/entrypoints.rs +++ b/apps/decodex/src/orchestrator/entrypoints.rs @@ -125,21 +125,13 @@ pub(crate) fn run_control_plane(request: ServeRequest<'_>) -> Result<()> { let runtime_db_path = runtime::runtime_db_path()?; let global_config_path = runtime::global_config_path()?; let project_config_dir = runtime::project_config_dir()?; - let snapshot = run_control_plane_api_only_tick(&state_store)?; - - if let Err(error) = operator_state_endpoint.publish_snapshot(&snapshot) { - let _ = error; - - tracing::warn!( - "Operator snapshot publish failed; sensitive runtime details were withheld from control-plane logs." - ); - } tracing::info!( listen_address = %operator_state_endpoint.listen_address(), path = OPERATOR_DASHBOARD_ALIAS_ENDPOINT_PATH, ws_path = OPERATOR_DASHBOARD_WS_ENDPOINT_PATH, api_only = true, + stream_interval_s = OPERATOR_API_ONLY_SNAPSHOT_STREAM_INTERVAL.as_secs(), runtime_db_path = %runtime_db_path.display(), global_config_path = %global_config_path.display(), project_config_dir = %project_config_dir.display(), @@ -147,7 +139,11 @@ pub(crate) fn run_control_plane(request: ServeRequest<'_>) -> Result<()> { ); loop { - thread::park(); + let tick_started_at = Instant::now(); + let snapshot = run_control_plane_api_only_tick(&state_store)?; + + publish_operator_snapshot(&operator_state_endpoint, &snapshot); + sleep_until_next_tick(OPERATOR_API_ONLY_SNAPSHOT_STREAM_INTERVAL, tick_started_at); } } @@ -200,14 +196,7 @@ pub(crate) fn run_control_plane(request: ServeRequest<'_>) -> Result<()> { let snapshot = run_control_plane_tick(&state_store, &mut project_runtimes)?; - if let Err(error) = operator_state_endpoint.publish_snapshot(&snapshot) { - let _ = error; - - tracing::warn!( - "Operator snapshot publish failed; sensitive runtime details were withheld from control-plane logs." - ); - } - + publish_operator_snapshot(&operator_state_endpoint, &snapshot); sleep_until_next_tick(poll_interval, tick_started_at); } } @@ -343,6 +332,19 @@ pub(crate) fn run_diagnose(request: DiagnoseRequest<'_>) -> Result<()> { Ok(()) } +fn publish_operator_snapshot( + operator_state_endpoint: &OperatorStateEndpoint, + snapshot: &OperatorStatusSnapshot, +) { + if let Err(error) = operator_state_endpoint.publish_snapshot(snapshot) { + let _ = error; + + tracing::warn!( + "Operator snapshot publish failed; sensitive runtime details were withheld from control-plane logs." + ); + } +} + fn run_control_plane_maintenance(trigger: &'static str) { match maintenance::run_auto_safe_prune() { Ok(report) => {