From 21ec7f9afd944e8c3d697328a5ae09c93280c33c Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Thu, 11 Jun 2026 16:23:16 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Fix Decodex app running task stream","authority":"manual"} --- .../Sources/DecodexApp/AccountPanelView.swift | 89 ++++- .../Sources/DecodexApp/AccountStore.swift | 74 ++-- .../DashboardWebSocketConnection.swift | 330 ++++++++++++++++++ .../Sources/DecodexApp/DecodexApp.swift | 2 +- .../DecodexApp/OperatorSnapshotModels.swift | 61 ++++ .../DecodexAppTests/AccountModelTests.swift | 97 +++++ .../DecodexServerBridgeTests.swift | 14 + 7 files changed, 616 insertions(+), 51 deletions(-) create mode 100644 apps/decodex-app/Sources/DecodexApp/DashboardWebSocketConnection.swift diff --git a/apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift b/apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift index 3c09d500c..55dc03b76 100644 --- a/apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift +++ b/apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift @@ -279,6 +279,10 @@ struct AccountPanelView: View { header accountSummary + if globalActiveRuns.isEmpty == false { + globalRunSummary + } + if telemetryMatrixIsVisible { AccountTelemetryMatrixView( aggregate: accountProfileAggregate, @@ -605,6 +609,9 @@ struct AccountPanelView: View { if store.accountList?.usageProbeError != nil { height += AccountPanelLayout.sectionSpacing + AccountPanelLayout.noticeHeight } + if globalActiveRuns.isEmpty == false { + height += AccountPanelLayout.sectionSpacing + AccountRunStripLayout.globalSummaryHeight + } return height } @@ -659,6 +666,30 @@ struct AccountPanelView: View { + CGFloat(rows.count - 1) * AccountPanelLayout.telemetryRowSpacing } + private var globalActiveRuns: [OperatorRunStatus] { + store.operatorSnapshot?.activeRuns ?? [] + } + + private var globalRunSummary: some View { + VStack(alignment: .leading, spacing: 6) { + HStack(alignment: .firstTextBaseline, spacing: 6) { + Image(systemName: "play.circle.fill") + .font(PanelFont.accountDetail) + .foregroundStyle(PanelPalette.routeAccent(colorScheme)) + .frame(width: 12) + Text("\(globalActiveRuns.count) active") + .font(PanelFont.accountDetail) + .foregroundStyle(PanelPalette.primaryText(colorScheme)) + Spacer(minLength: 0) + } + + AccountRunSummaryView(runs: globalActiveRuns, currentTime: currentTime) + } + .padding(.horizontal, 8) + .padding(.vertical, 7) + .modernGlassSurface(cornerRadius: 9, depth: .row) + } + private func displayName(for account: CodexAccount) -> String { if emailsHidden { return AccountDisplay.aliases(for: store.accounts)[account.id] @@ -837,7 +868,7 @@ struct AccountRowView: View { } if runs.isEmpty == false { - AccountRunSummaryView(runs: runs) + AccountRunSummaryView(runs: runs, currentTime: currentTime) } if account.hasUsageSummary { @@ -887,6 +918,7 @@ struct AccountRowView: View { struct AccountRunSummaryView: View { let runs: [OperatorRunStatus] + let currentTime: Date @State private var placementStore = AccountRunStripPlacementStore() @State private var scrollProxy = AccountRunStripScrollProxy() @State private var scrollMetrics = AccountRunStripMetrics() @@ -916,7 +948,7 @@ struct AccountRunSummaryView: View { ) { HStack(spacing: 5) { ForEach(runs) { run in - AccountRunChipView(run: run) + AccountRunChipView(run: run, currentTime: currentTime) .modifier( AccountRunChipPlacementReporter( runID: run.id, @@ -986,6 +1018,7 @@ struct AccountRunSummaryView: View { } private enum AccountRunStripLayout { + static let globalSummaryHeight: CGFloat = 42 static let contentCoordinateSpace = "account-run-strip-content" static let dragActivationDistance: CGFloat = 1 static let edgeControlSpacing: CGFloat = 4 @@ -1850,6 +1883,7 @@ private enum AccountRunChipLayout { struct AccountRunChipView: View { let run: OperatorRunStatus + let currentTime: Date @Environment(\.colorScheme) private var colorScheme @State private var isHovered = false @State private var showsPopover = false @@ -1889,7 +1923,7 @@ struct AccountRunChipView: View { cancelPopover() } .popover(isPresented: $showsPopover, arrowEdge: .trailing) { - OperatorLanePopoverView(run: run) + OperatorLanePopoverView(run: run, currentTime: currentTime) .fixedSize(horizontal: true, vertical: false) } } @@ -2833,6 +2867,7 @@ struct NoticeView: View { struct OperatorLanePopoverView: View { let run: OperatorRunStatus + let currentTime: Date var body: some View { VStack(alignment: .leading, spacing: 6) { @@ -2846,7 +2881,7 @@ struct OperatorLanePopoverView: View { OperatorLaneProgressReadoutRow( title: "Model", percent: bucketPercent(modelBucket), - elapsed: formatActivityDuration(modelBucket.wallSeconds) ?? "0s", + elapsed: formatActivityDuration(bucketWallSeconds(modelBucket)) ?? "0s", total: formatActivityDuration(totalWallSeconds) ?? "0s", barShare: bucketShare(modelBucket) ) @@ -2883,6 +2918,14 @@ struct OperatorLanePopoverView: View { } private var currentSummary: String { + if run.processAlive == false { + if let idle = formatActivityDuration(run.inactiveDurationSeconds) { + return "Stopped · idle \(idle)" + } + + return "Stopped" + } + guard let activity else { return "Waiting for child activity" } @@ -2890,7 +2933,7 @@ struct OperatorLanePopoverView: View { let label = panelTrimmed(activity.currentDetail) ?? panelTrimmed(activity.currentBucket).map(rawPanelToken) ?? "Active" - if let elapsed = formatActivityDuration(activity.currentElapsedSeconds) { + if let elapsed = formatActivityDuration(activity.currentElapsedSeconds(at: currentTime)) { return "\(rawPanelToken(label)) · \(elapsed)" } @@ -2923,7 +2966,7 @@ struct OperatorLanePopoverView: View { var items = [ OperatorLaneReadoutItem( label: "wall", - value: formatActivityDuration(activity.wallSeconds) ?? "0s" + value: formatActivityDuration(activity.wallSeconds(at: currentTime)) ?? "0s" ), OperatorLaneReadoutItem( label: "events", @@ -2956,7 +2999,11 @@ struct OperatorLanePopoverView: View { } private var modelBucket: OperatorChildAgentBucket? { - orderedBuckets.first { bucket in + guard run.processAlive != false else { + return nil + } + + return orderedBuckets.first { bucket in bucket.name.caseInsensitiveCompare("Model") == .orderedSame } } @@ -2975,8 +3022,10 @@ struct OperatorLanePopoverView: View { if leftPriority != rightPriority { return leftPriority < rightPriority } - if left.wallSeconds != right.wallSeconds { - return left.wallSeconds > right.wallSeconds + let leftWallSeconds = bucketWallSeconds(left) + let rightWallSeconds = bucketWallSeconds(right) + if leftWallSeconds != rightWallSeconds { + return leftWallSeconds > rightWallSeconds } if left.eventCount != right.eventCount { return left.eventCount > right.eventCount @@ -3040,8 +3089,8 @@ struct OperatorLanePopoverView: View { private var totalWallSeconds: Int { max( 1, - activity?.wallSeconds ?? 0, - bucketRows.reduce(0) { $0 + max(0, $1.wallSeconds) } + activity?.wallSeconds(at: currentTime) ?? 0, + bucketRows.reduce(0) { $0 + max(0, bucketWallSeconds($1)) } ) } @@ -3049,8 +3098,10 @@ struct OperatorLanePopoverView: View { let normalizedName = bucket.name.lowercased() var items = [OperatorLaneReadoutItem]() - if normalizedName.contains("tracker"), bucket.wallSeconds > 0 { - items.append(OperatorLaneReadoutItem(label: "wall", value: formatActivityDuration(bucket.wallSeconds) ?? "0s")) + let wallSeconds = bucketWallSeconds(bucket) + + if normalizedName.contains("tracker"), wallSeconds > 0 { + items.append(OperatorLaneReadoutItem(label: "wall", value: formatActivityDuration(wallSeconds) ?? "0s")) } if bucket.eventCount > 0 { items.append(OperatorLaneReadoutItem(label: "events", value: formatCompactCount(bucket.eventCount))) @@ -3098,15 +3149,21 @@ struct OperatorLanePopoverView: View { } private func bucketShare(_ bucket: OperatorChildAgentBucket) -> CGFloat { - guard bucket.wallSeconds > 0 else { + let wallSeconds = bucketWallSeconds(bucket) + + guard wallSeconds > 0 else { return 0 } - return min(1, max(0.02, CGFloat(bucket.wallSeconds) / CGFloat(max(1, totalWallSeconds)))) + return min(1, max(0.02, CGFloat(wallSeconds) / CGFloat(max(1, totalWallSeconds)))) } private func bucketPercent(_ bucket: OperatorChildAgentBucket) -> Int { - Int((Double(bucket.wallSeconds) / Double(max(1, totalWallSeconds)) * 100).rounded()) + Int((Double(bucketWallSeconds(bucket)) / Double(max(1, totalWallSeconds)) * 100).rounded()) + } + + private func bucketWallSeconds(_ bucket: OperatorChildAgentBucket) -> Int { + activity?.wallSeconds(for: bucket, at: currentTime) ?? bucket.wallSeconds } } diff --git a/apps/decodex-app/Sources/DecodexApp/AccountStore.swift b/apps/decodex-app/Sources/DecodexApp/AccountStore.swift index d475dfdab..d4d2005e9 100644 --- a/apps/decodex-app/Sources/DecodexApp/AccountStore.swift +++ b/apps/decodex-app/Sources/DecodexApp/AccountStore.swift @@ -1,5 +1,10 @@ import AppKit import Foundation +import OSLog + +private let accountStoreLog = Logger(subsystem: "ink.hack.DecodexApp", category: "AccountStore") +private let operatorSnapshotReconnectInitialDelay: UInt64 = 1_000_000_000 +private let operatorSnapshotReconnectMaxDelay: UInt64 = 30_000_000_000 @MainActor final class AccountStore: ObservableObject { @@ -145,63 +150,62 @@ final class AccountStore: ObservableObject { return } - operatorSnapshotStreamTask = Task { [weak self] in + operatorSnapshotStreamTask = makeOperatorSnapshotStreamTask() + } + + private func makeOperatorSnapshotStreamTask() -> Task { + Task { [weak self] in await self?.runOperatorSnapshotStream() } } private func runOperatorSnapshotStream() async { + var reconnectDelay = operatorSnapshotReconnectInitialDelay + while Task.isCancelled == false { do { try await connectOperatorSnapshotStream() + reconnectDelay = operatorSnapshotReconnectInitialDelay } catch { - liveRunActivity = nil + accountStoreLog.warning("Operator snapshot stream dropped: \(error.localizedDescription, privacy: .public)") } do { - try await Task.sleep(nanoseconds: 1_000_000_000) + try await Task.sleep(nanoseconds: reconnectDelay) } catch { return } + reconnectDelay = min(operatorSnapshotReconnectMaxDelay, reconnectDelay * 2) } } 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 == false { - 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 + let socket = DashboardWebSocketConnection(url: url) - switch message { - case .string(let text): - guard let textData = text.data(using: .utf8) else { - throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket event is not UTF-8") + try await withTaskCancellationHandler { + do { + try await socket.connect() + while Task.isCancelled == false { + let data = try await socket.readMessageData() + do { + let event = try JSONDecoder().decode(OperatorDashboardSocketEvent.self, from: data) + applyOperatorDashboardEvent(event) + } catch { + accountStoreLog.debug("Skipped dashboard WebSocket message bytes=\(data.count, privacy: .public) error=\(error.localizedDescription, privacy: .public)") + continue + } + } + await socket.close() + } catch { + await socket.close() + throw error + } + } onCancel: { + Task { + await socket.close() } - - 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) } func applyOperatorDashboardEvent(_ event: OperatorDashboardSocketEvent) { @@ -230,6 +234,8 @@ final class AccountStore: ObservableObject { liveRunActivity = activity if let operatorSnapshot { self.operatorSnapshot = activity.merging(into: operatorSnapshot) + } else { + operatorSnapshot = OperatorSnapshotResponse.activeRunsOnly(activeRuns) } operatorSnapshotUpdatedAt = activity.emittedAt default: diff --git a/apps/decodex-app/Sources/DecodexApp/DashboardWebSocketConnection.swift b/apps/decodex-app/Sources/DecodexApp/DashboardWebSocketConnection.swift new file mode 100644 index 000000000..52d5c754b --- /dev/null +++ b/apps/decodex-app/Sources/DecodexApp/DashboardWebSocketConnection.swift @@ -0,0 +1,330 @@ +import Foundation +import Network +import Security + +actor DashboardWebSocketConnection { + private let url: URL + private var connection: NWConnection? + private var buffer = Data() + + init(url: URL) { + self.url = url + } + + func connect() async throws { + guard let host = url.host, let portValue = url.port else { + throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket URL is missing host or port") + } + guard let port = NWEndpoint.Port(rawValue: UInt16(portValue)) else { + throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket URL port is invalid") + } + + let connection = NWConnection(host: NWEndpoint.Host(host), port: port, using: .tcp) + self.connection = connection + + try await start(connection) + try await sendHandshake(host: host, port: portValue) + try await readHandshakeResponse() + } + + func close() { + connection?.cancel() + connection = nil + buffer.removeAll(keepingCapacity: false) + } + + func readMessageData() async throws -> Data { + while true { + let frame = try await readFrame() + + switch frame.opcode { + case 0x1, 0x2: + return frame.payload + case 0x8: + throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket closed") + case 0x9: + try await sendPong(frame.payload) + case 0xA: + continue + default: + continue + } + } + } + + private func start(_ connection: NWConnection) async throws { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + let resumeBox = DashboardConnectionResumeBox(continuation: continuation) + + connection.stateUpdateHandler = { state in + switch state { + case .ready: + resumeBox.resume(.success(())) + case .failed(let error): + resumeBox.resume(.failure(error)) + case .cancelled: + resumeBox.resume(.failure(DecodexAppBridgeError.invalidResponse("dashboard WebSocket connection cancelled"))) + default: + break + } + } + connection.start(queue: .global(qos: .userInitiated)) + } + } + + private func sendHandshake(host: String, port: Int) async throws { + let key = websocketKey() + let path = websocketRequestPath() + + try await send(Self.handshakeRequest(host: host, port: port, path: path, key: key)) + } + + static func handshakeRequest(host: String, port: Int, path: String, key: String) -> Data { + let lines = [ + "GET \(path) HTTP/1.1", + "Host: \(host):\(port)", + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: \(key)", + "Sec-WebSocket-Version: 13", + "", + "", + ] + + return Data(lines.joined(separator: "\r\n").utf8) + } + + private func readHandshakeResponse() async throws { + let delimiter = Data("\r\n\r\n".utf8) + + while buffer.range(of: delimiter) == nil { + try await receiveMore() + } + + guard let range = buffer.range(of: delimiter) else { + throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket handshake response is incomplete") + } + + let headerData = buffer[.. WebSocketFrame { + while buffer.count < 2 { + try await receiveMore() + } + + let firstByte = buffer[buffer.startIndex] + let secondByte = buffer[buffer.index(after: buffer.startIndex)] + let opcode = firstByte & 0x0F + let masked = (secondByte & 0x80) != 0 + var length = UInt64(secondByte & 0x7F) + var headerLength = 2 + + if length == 126 { + while buffer.count < 4 { + try await receiveMore() + } + length = UInt64(readUInt16(offset: 2)) + headerLength = 4 + } else if length == 127 { + while buffer.count < 10 { + try await receiveMore() + } + length = readUInt64(offset: 2) + headerLength = 10 + } + + let maskLength = masked ? 4 : 0 + let payloadLength = try checkedPayloadLength(length) + let frameLength = headerLength + maskLength + payloadLength + + while buffer.count < frameLength { + try await receiveMore() + } + + let maskStart = headerLength + let payloadStart = headerLength + maskLength + var payload = buffer.subdata(in: payloadStart..) in + connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { + data, + _, + isComplete, + error in + if let error { + continuation.resume(throwing: error) + return + } + if let data, data.isEmpty == false { + continuation.resume(returning: data) + return + } + if isComplete { + continuation.resume(throwing: DecodexAppBridgeError.invalidResponse("dashboard WebSocket ended")) + return + } + + continuation.resume(returning: Data()) + } + } + + if data.isEmpty { + return + } + + buffer.append(data) + } + + private func sendPong(_ payload: Data) async throws { + try await send(clientFrame(opcode: 0xA, payload: payload)) + } + + private func send(_ data: Data) async throws { + guard let connection else { + throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket is not connected") + } + + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + connection.send(content: data, completion: .contentProcessed { error in + if let error { + continuation.resume(throwing: error) + } else { + continuation.resume(returning: ()) + } + }) + } + } + + private func websocketRequestPath() -> String { + var path = url.path.isEmpty ? "/" : url.path + if let query = url.query, query.isEmpty == false { + path += "?\(query)" + } + + return path + } + + private func websocketKey() -> String { + randomData(byteCount: 16).base64EncodedString() + } + + private func clientFrame(opcode: UInt8, payload: Data) -> Data { + var output = Data() + let length = payload.count + + output.append(0x80 | opcode) + if length <= 125 { + output.append(0x80 | UInt8(length)) + } else if length <= UInt16.max { + output.append(0x80 | 126) + output.append(UInt8((length >> 8) & 0xFF)) + output.append(UInt8(length & 0xFF)) + } else { + output.append(0x80 | 127) + output.append(contentsOf: UInt64(length).bigEndianBytes) + } + + let mask = randomData(byteCount: 4) + output.append(mask) + for index in payload.indices { + output.append(payload[index] ^ mask[index % 4]) + } + + return output + } + + private func checkedPayloadLength(_ length: UInt64) throws -> Int { + guard length <= UInt64(Int.max) else { + throw DecodexAppBridgeError.invalidResponse("dashboard WebSocket frame is too large") + } + + return Int(length) + } + + private func readUInt16(offset: Int) -> UInt16 { + let first = UInt16(buffer[offset]) + let second = UInt16(buffer[offset + 1]) + + return (first << 8) | second + } + + private func readUInt64(offset: Int) -> UInt64 { + var value: UInt64 = 0 + for byte in buffer[offset.. Data { + var bytes = [UInt8](repeating: 0, count: byteCount) + let status = SecRandomCopyBytes(kSecRandomDefault, bytes.count, &bytes) + if status != errSecSuccess { + for index in bytes.indices { + bytes[index] = UInt8.random(in: UInt8.min...UInt8.max) + } + } + + return Data(bytes) + } +} + +private struct WebSocketFrame { + let opcode: UInt8 + let payload: Data +} + +private final class DashboardConnectionResumeBox: @unchecked Sendable { + private let lock = NSLock() + private var resumed = false + private let continuation: CheckedContinuation + + init(continuation: CheckedContinuation) { + self.continuation = continuation + } + + func resume(_ result: Result) { + lock.lock() + defer { + lock.unlock() + } + guard resumed == false else { + return + } + resumed = true + continuation.resume(with: result) + } +} + +private extension UInt64 { + var bigEndianBytes: [UInt8] { + withUnsafeBytes(of: bigEndian) { bytes in + Array(bytes) + } + } +} diff --git a/apps/decodex-app/Sources/DecodexApp/DecodexApp.swift b/apps/decodex-app/Sources/DecodexApp/DecodexApp.swift index 328c66e41..01a5440bb 100644 --- a/apps/decodex-app/Sources/DecodexApp/DecodexApp.swift +++ b/apps/decodex-app/Sources/DecodexApp/DecodexApp.swift @@ -88,8 +88,8 @@ struct DecodexApp: App { _store = StateObject(wrappedValue: accountStore) Task { - await accountStore.refreshIfNeeded() accountStore.startAutomaticRefresh() + await accountStore.refreshIfNeeded() } } diff --git a/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift b/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift index 7f5b49158..d36d3e752 100644 --- a/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift +++ b/apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift @@ -137,6 +137,16 @@ struct OperatorSnapshotResponse: Decodable, Sendable { ) } + static func activeRunsOnly(_ activeRuns: [OperatorRunStatus]) -> OperatorSnapshotResponse { + OperatorSnapshotResponse( + warnings: [], + projects: [], + activeRuns: activeRuns, + queuedCandidates: [], + postReviewLanes: [] + ) + } + private init( warnings: [String], projects: [OperatorProjectStatus], @@ -374,6 +384,7 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable { || attemptStatus == "waiting_for_review" || status == "manual_attention" || status == "blocked" + || processAlive == false } var isWaiting: Bool { @@ -382,6 +393,16 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable { || phase?.contains("waiting") == true } + var inactiveDurationSeconds: Int? { + let candidates = [ + idleForSeconds, + protocolIdleForSeconds, + childAgentActivity?.currentElapsedSeconds, + ].compactMap { $0 } + + return candidates.max() + } + func isAssigned(to account: CodexAccount) -> Bool { self.account?.matches(account) == true || accounts.contains { $0.isSelected && $0.matches(account) } @@ -649,6 +670,7 @@ struct OperatorChildAgentActivity: Decodable, Sendable { let currentBucket: String? let currentDetail: String? let currentElapsedSeconds: Int? + let currentStartedUnixEpoch: Int64? let eventCount: Int let inputTokensCumulative: Int let inputTokensCurrent: Int? @@ -664,6 +686,7 @@ struct OperatorChildAgentActivity: Decodable, Sendable { case currentBucket = "current_bucket" case currentDetail = "current_detail" case currentElapsedSeconds = "current_elapsed_seconds" + case currentStartedUnixEpoch = "current_started_unix_epoch" case eventCount = "event_count" case inputTokensCumulative = "input_tokens_cumulative" case inputTokensCurrent = "input_tokens_current" @@ -682,6 +705,7 @@ struct OperatorChildAgentActivity: Decodable, Sendable { currentBucket = try container.decodeIfPresent(String.self, forKey: .currentBucket) currentDetail = try container.decodeIfPresent(String.self, forKey: .currentDetail) currentElapsedSeconds = try container.decodeIfPresent(Int.self, forKey: .currentElapsedSeconds) + currentStartedUnixEpoch = try container.decodeIfPresent(Int64.self, forKey: .currentStartedUnixEpoch) eventCount = try container.decodeIfPresent(Int.self, forKey: .eventCount) ?? 0 inputTokensCumulative = try container.decodeIfPresent(Int.self, forKey: .inputTokensCumulative) ?? 0 inputTokensCurrent = try container.decodeIfPresent(Int.self, forKey: .inputTokensCurrent) @@ -693,6 +717,43 @@ struct OperatorChildAgentActivity: Decodable, Sendable { wallSeconds = try container.decodeIfPresent(Int.self, forKey: .wallSeconds) ?? 0 buckets = try container.decodeIfPresent([OperatorChildAgentBucket].self, forKey: .buckets) ?? [] } + + func currentElapsedSeconds(at now: Date) -> Int? { + var candidates = [Int]() + if let currentElapsedSeconds { + candidates.append(currentElapsedSeconds) + } + if let currentStartedUnixEpoch { + let liveElapsed = Int(now.timeIntervalSince1970.rounded(.down)) - Int(currentStartedUnixEpoch) + + candidates.append(max(0, liveElapsed)) + } + + return candidates.max() + } + + func wallSeconds(at now: Date) -> Int { + wallSeconds + currentElapsedDelta(at: now) + } + + func wallSeconds( + for bucket: OperatorChildAgentBucket, + at now: Date + ) -> Int { + guard let currentBucket, bucket.name.caseInsensitiveCompare(currentBucket) == .orderedSame else { + return bucket.wallSeconds + } + + return bucket.wallSeconds + currentElapsedDelta(at: now) + } + + private func currentElapsedDelta(at now: Date) -> Int { + guard let baselineElapsed = currentElapsedSeconds, let liveElapsed = currentElapsedSeconds(at: now) else { + return 0 + } + + return max(0, liveElapsed - baselineElapsed) + } } struct OperatorChildAgentBucket: Decodable, Identifiable, Sendable { diff --git a/apps/decodex-app/Tests/DecodexAppTests/AccountModelTests.swift b/apps/decodex-app/Tests/DecodexAppTests/AccountModelTests.swift index 19fce0ef1..923d505ef 100644 --- a/apps/decodex-app/Tests/DecodexAppTests/AccountModelTests.swift +++ b/apps/decodex-app/Tests/DecodexAppTests/AccountModelTests.swift @@ -92,6 +92,70 @@ final class AccountModelTests: XCTestCase { XCTAssertTrue(due.accessibility.contains("reset due now")) } + func testOperatorChildActivityAdvancesCurrentElapsedFromStartedAt() throws { + let payload = """ + { + "current_bucket": "Model", + "current_detail": "model output", + "current_elapsed_seconds": 5, + "current_started_unix_epoch": 100, + "wall_seconds": 20, + "buckets": [ + { + "name": "Model", + "wall_seconds": 15 + }, + { + "name": "Tool", + "wall_seconds": 5 + } + ] + } + """.data(using: .utf8)! + + let activity = try JSONDecoder().decode(OperatorChildAgentActivity.self, from: payload) + let modelBucket = try XCTUnwrap(activity.buckets.first { $0.name == "Model" }) + let toolBucket = try XCTUnwrap(activity.buckets.first { $0.name == "Tool" }) + let now = Date(timeIntervalSince1970: 110) + + XCTAssertEqual(activity.currentElapsedSeconds(at: now), 10) + XCTAssertEqual(activity.wallSeconds(at: now), 25) + XCTAssertEqual(activity.wallSeconds(for: modelBucket, at: now), 20) + XCTAssertEqual(activity.wallSeconds(for: toolBucket, at: now), 5) + } + + func testStoppedActiveRunUsesInactiveDurationAndAttentionTone() throws { + let payload = """ + { + "run_id": "pub-1524-attempt-2", + "issue_identifier": "PUB-1524", + "status": "running", + "phase": "executing", + "process_alive": false, + "idle_for_seconds": 20815, + "protocol_idle_for_seconds": 20816, + "child_agent_activity": { + "current_bucket": "Model", + "current_detail": "waiting after completed item", + "current_elapsed_seconds": 20840, + "wall_seconds": 788, + "buckets": [ + { + "name": "Model", + "wall_seconds": 21389 + } + ] + } + } + """.data(using: .utf8)! + + let run = try JSONDecoder().decode(OperatorRunStatus.self, from: payload) + + XCTAssertFalse(run.countsAsRunning) + XCTAssertTrue(run.hasAttentionTone) + XCTAssertEqual(run.inactiveDurationSeconds, 20840) + } + func testOperatorSnapshotAssignsCodexAccountRunsToAccountRows() throws { let assignedAccount = makeAccount( status: "available", @@ -324,6 +388,39 @@ final class AccountModelTests: XCTestCase { XCTAssertEqual(store.operatorSnapshot?.activeRuns(for: account).map(\.runID), ["run-old"]) } + @MainActor + func testRunActivityBeforeSnapshotCreatesVisibleActiveRuns() throws { + let account = makeAccount( + status: "available", + email: "copy@example.com", + accountFingerprint: "...123456" + ) + let store = AccountStore() + + try store.applyOperatorDashboardEvent(dashboardEvent( + type: "runActivity", + payload: """ + { + "emittedAtUnixEpoch": 30, + "activeRunsComplete": true, + "activeRuns": [ + { + "run_id": "run-live", + "issue_identifier": "XY-672", + "account": { + "email": "copy@example.com", + "account_fingerprint": "...123456" + } + } + ] + } + """ + )) + + XCTAssertEqual(store.operatorSnapshot?.activeRuns.map(\.runID), ["run-live"]) + XCTAssertEqual(store.operatorSnapshot?.activeRuns(for: account).map(\.runID), ["run-live"]) + } + func testPartialRunActivityPreservesSnapshotActiveRuns() throws { let account = makeAccount( status: "available", diff --git a/apps/decodex-app/Tests/DecodexAppTests/DecodexServerBridgeTests.swift b/apps/decodex-app/Tests/DecodexAppTests/DecodexServerBridgeTests.swift index 010a910e9..0a34c2a4b 100644 --- a/apps/decodex-app/Tests/DecodexAppTests/DecodexServerBridgeTests.swift +++ b/apps/decodex-app/Tests/DecodexAppTests/DecodexServerBridgeTests.swift @@ -12,4 +12,18 @@ final class DecodexServerBridgeTests: XCTestCase { ] ) } + + func testDashboardWebSocketHandshakeRequestEndsWithHeaderTerminator() { + let request = DashboardWebSocketConnection.handshakeRequest( + host: "127.0.0.1", + port: 8192, + path: "/dashboard/control", + key: "test-key" + ) + let text = String(decoding: request, as: UTF8.self) + + XCTAssertTrue(text.hasSuffix("\r\n\r\n")) + XCTAssertTrue(text.contains("GET /dashboard/control HTTP/1.1\r\n")) + XCTAssertTrue(text.contains("Sec-WebSocket-Key: test-key\r\n")) + } }