diff --git a/apps/ade-cli/src/tuiClient/__tests__/remoteLauncher.test.ts b/apps/ade-cli/src/tuiClient/__tests__/remoteLauncher.test.ts index 015832c44..9b5e1ff31 100644 --- a/apps/ade-cli/src/tuiClient/__tests__/remoteLauncher.test.ts +++ b/apps/ade-cli/src/tuiClient/__tests__/remoteLauncher.test.ts @@ -533,21 +533,29 @@ describe("ade code remote launcher", () => { children.push(child); return child; }); - const startedAt = Date.now(); - - await expect(openRemoteRpcSession({ + const target = { ...legacyAccountTarget(), id: "offline-explicit-ssh", name: "Offline workstation", sshUser: "arul", - }, { + }; + const routeCount = new Set([ + `${target.hostname}:${target.port ?? ""}`, + ...(target.routes ?? []).map((route) => `${route.hostname}:${route.port ?? ""}`), + ]).size; + const maxRouteRuntimeAttempts = + routeCount * remoteRuntimeLayoutCandidates(process.env).length * 2; + const startedAt = Date.now(); + + await expect(openRemoteRpcSession(target, { totalTimeoutMs: 80, attemptTimeoutMs: 50, spawnProcess, })).rejects.toThrow(/bounded route\/runtime combinations.*deadline/i); expect(Date.now() - startedAt).toBeLessThan(1_000); - expect(spawnProcess.mock.calls.length).toBeLessThanOrEqual(2); + expect(spawnProcess.mock.calls.length).toBeGreaterThan(0); + expect(spawnProcess.mock.calls.length).toBeLessThanOrEqual(maxRouteRuntimeAttempts); expect(children.every((child) => child.kill.mock.calls.length > 0)).toBe(true); }); diff --git a/apps/ios/ADE/Models/RemoteModels.swift b/apps/ios/ADE/Models/RemoteModels.swift index 8b726af21..ceb637bb5 100644 --- a/apps/ios/ADE/Models/RemoteModels.swift +++ b/apps/ios/ADE/Models/RemoteModels.swift @@ -2266,8 +2266,14 @@ extension AgentChatEvent { case options case id case origin + case sampleId case newConversationId case stillQueuedUuids + case cancelledUuids + case recoveryId + case messageCount + case expiresAt + case stopMode case commandUuid case preview case goal @@ -2487,12 +2493,17 @@ extension AgentChatEvent { turnId: try container.decodeIfPresent(String.self, forKey: .turnId) ) case "context_usage": + let usage = try container.decode(AgentChatContextUsage.self, forKey: .usage) + let turnId = try container.decodeIfPresent(String.self, forKey: .turnId) + let origin = try container.decodeIfPresent(String.self, forKey: .origin) + let state = try container.decodeIfPresent(String.self, forKey: .state) + let sampleId = try container.decodeIfPresent(Int.self, forKey: .sampleId) self = .contextUsage( - usage: try container.decode(AgentChatContextUsage.self, forKey: .usage), - turnId: try container.decodeIfPresent(String.self, forKey: .turnId), - origin: try container.decodeIfPresent(String.self, forKey: .origin), - state: try container.decodeIfPresent(String.self, forKey: .state), - sampleId: try container.decodeIfPresent(Int.self, forKey: .sampleId) + usage: usage, + turnId: turnId, + origin: origin, + state: state, + sampleId: sampleId ) case "conversation_reset": self = .conversationReset( @@ -2504,16 +2515,22 @@ extension AgentChatEvent { cancelledUuids: try container.decodeIfPresent([String].self, forKey: .cancelledUuids) ) case "queue_recovery": + let state = try container.decode(String.self, forKey: .state) + let messageCount = try container.decode(Int.self, forKey: .messageCount) + let expiresAt = try container.decode(String.self, forKey: .expiresAt) + let stopMode = try container.decode(String.self, forKey: .stopMode) + let turnId = try container.decodeIfPresent(String.self, forKey: .turnId) + let recoveryId = try container.decode(String.self, forKey: .recoveryId) self = .systemNotice( noticeKind: .queueRecovery, - message: try container.decode(String.self, forKey: .state), + message: state, detail: .object([ - "messageCount": .number(Double(try container.decode(Int.self, forKey: .messageCount))), - "expiresAt": .string(try container.decode(String.self, forKey: .expiresAt)), - "stopMode": .string(try container.decode(String.self, forKey: .stopMode)), + "messageCount": .number(Double(messageCount)), + "expiresAt": .string(expiresAt), + "stopMode": .string(stopMode), ]), - turnId: try container.decodeIfPresent(String.self, forKey: .turnId), - steerId: try container.decode(String.self, forKey: .recoveryId) + turnId: turnId, + steerId: recoveryId ) case "command_lifecycle": self = .commandLifecycle( diff --git a/apps/ios/ADE/Services/SyncService.swift b/apps/ios/ADE/Services/SyncService.swift index eb0e5dd33..b424efbed 100644 --- a/apps/ios/ADE/Services/SyncService.swift +++ b/apps/ios/ADE/Services/SyncService.swift @@ -2811,6 +2811,11 @@ final class SyncService: ObservableObject { let startedAt: TimeInterval } + private struct PendingChatUnsubscribe { + let task: Task + let scheduledAt: Date + } + private struct PendingOutboundChangeset { var payload: SyncChangesetBatchPayload var sentAt: TimeInterval @@ -2918,6 +2923,13 @@ final class SyncService: ObservableObject { private var supportsProjectCatalog = false private var supportsProjectActions = false private var supportsChatStreaming = false + private let chatSnapshotRequestCoalescingInterval: TimeInterval = 5 + private let chatEventUnsubscribeRetentionLimit = 4 + private var recentFullChatSnapshotRequestBySession: [ + String: (uptime: TimeInterval, connectionGeneration: UInt64) + ] = [:] + private var pendingChatUnsubscribesBySession: [String: PendingChatUnsubscribe] = [:] + private var chatSubscriptionsNeedingRemoteActivation: Set = [] private var supportsChangesetAck = false private var relayAuthorizationLease: SyncRelayAuthorizationLease? private var relayReauthorizationTask: Task? @@ -8329,41 +8341,144 @@ final class SyncService: ObservableObject { isRemoteActionQueueable(chatActionName(projectAction, sessionId: sessionId)) } - func subscribeToChatEvents(sessionId: String, requestSnapshot: Bool = false, maxBytes: Int? = nil) async throws { + /// Returns true when the requested subscription state is backed by an + /// envelope dispatched now or, for a full snapshot, a recently dispatched + /// envelope that is still inside the coalescing window. + @discardableResult + func subscribeToChatEvents( + sessionId: String, + requestSnapshot: Bool = false, + maxBytes: Int? = nil + ) async throws -> Bool { let trimmedSessionId = sessionId.trimmingCharacters(in: .whitespacesAndNewlines) - guard !trimmedSessionId.isEmpty else { return } + guard !trimmedSessionId.isEmpty else { return false } let wasSubscribed = subscribedChatSessionIds.contains(trimmedSessionId) - if !wasSubscribed { - subscribedChatSessionIds.insert(trimmedSessionId) - localStateRevision += 1 + retainChatEventSubscription(sessionId: trimmedSessionId) + let needsRemoteActivation = chatSubscriptionsNeedingRemoteActivation.contains(trimmedSessionId) + let requestUptime = ProcessInfo.processInfo.systemUptime + let canSendChatSubscription = canSendLiveRequests() && supportsChatStreaming + let snapshotRequestWasRecentlySent: Bool + if requestSnapshot, + canSendChatSubscription, + let recentRequest = recentFullChatSnapshotRequestBySession[trimmedSessionId], + recentRequest.connectionGeneration == connectionGeneration, + requestUptime - recentRequest.uptime < chatSnapshotRequestCoalescingInterval { + snapshotRequestWasRecentlySent = true + } else { + snapshotRequestWasRecentlySent = false + } + if snapshotRequestWasRecentlySent { + return true } - if canSendLiveRequests() && supportsChatStreaming && (!wasSubscribed || requestSnapshot) { + if canSendChatSubscription + && (!wasSubscribed || needsRemoteActivation || requestSnapshot) { // Explicit snapshot requests must not advertise a resume point — the // caller wants the full history, not a delta replay. let payload = chatSubscriptionPayload(sessionId: trimmedSessionId, maxBytes: maxBytes, includeSinceSeq: !requestSnapshot) - syncChatLog.notice( - "chat_subscribe_send session=\(trimmedSessionId, privacy: .public) requestSnapshot=\(requestSnapshot, privacy: .public) wasSubscribed=\(wasSubscribed, privacy: .public) maxBytes=\((payload["maxBytes"] as? Int) ?? -1, privacy: .public) sinceSeq=\((payload["sinceSeq"] as? Int) ?? -1, privacy: .public) reducedLoad=\(self.prefersReducedSyncLoad, privacy: .public) state=\(self.connectionState.rawValue, privacy: .public)" - ) - sendEnvelope( + guard sendEnvelope( type: "chat_subscribe", requestId: nil, payload: payload + ) else { return false } + if requestSnapshot { + recentFullChatSnapshotRequestBySession[trimmedSessionId] = ( + uptime: requestUptime, + connectionGeneration: connectionGeneration + ) + } + syncChatLog.notice( + "chat_subscribe_send session=\(trimmedSessionId, privacy: .public) requestSnapshot=\(requestSnapshot, privacy: .public) wasSubscribed=\(wasSubscribed, privacy: .public) maxBytes=\((payload["maxBytes"] as? Int) ?? -1, privacy: .public) sinceSeq=\((payload["sinceSeq"] as? Int) ?? -1, privacy: .public) reducedLoad=\(self.prefersReducedSyncLoad, privacy: .public) state=\(self.connectionState.rawValue, privacy: .public)" ) + chatSubscriptionsNeedingRemoteActivation.remove(trimmedSessionId) + return true } + return false } - func requestFullChatEventSnapshot(sessionId: String) async throws { - try await subscribeToChatEvents(sessionId: sessionId, requestSnapshot: true, maxBytes: syncChatSubscriptionMaxBytes) + /// Mark a visible chat as locally desired even when the host is unreachable. + /// This synchronously cancels pending warm-cache eviction so reconnect + /// restoration cannot omit a chat that is still on screen. + func retainChatEventSubscription(sessionId: String) { + let trimmedSessionId = sessionId.trimmingCharacters(in: .whitespacesAndNewlines) + guard !trimmedSessionId.isEmpty else { return } + cancelDelayedChatUnsubscribe(sessionId: trimmedSessionId) + if subscribedChatSessionIds.insert(trimmedSessionId).inserted { + chatSubscriptionsNeedingRemoteActivation.insert(trimmedSessionId) + localStateRevision += 1 + } + } + + @discardableResult + func requestFullChatEventSnapshot(sessionId: String) async throws -> Bool { + try await subscribeToChatEvents( + sessionId: sessionId, + requestSnapshot: true, + maxBytes: syncChatSubscriptionMaxBytes + ) + } + + func isFullChatEventSnapshotPending(sessionId: String) -> Bool { + let trimmedSessionId = sessionId.trimmingCharacters(in: .whitespacesAndNewlines) + guard canSendLiveRequests(), + let recentRequest = recentFullChatSnapshotRequestBySession[trimmedSessionId] + else { + return false + } + return recentRequest.connectionGeneration == connectionGeneration } func unsubscribeFromChatEvents(sessionId: String) async throws { let trimmedSessionId = sessionId.trimmingCharacters(in: .whitespacesAndNewlines) guard !trimmedSessionId.isEmpty else { return } - guard subscribedChatSessionIds.contains(trimmedSessionId) else { return } - subscribedChatSessionIds.remove(trimmedSessionId) + cancelDelayedChatUnsubscribe(sessionId: trimmedSessionId) + performChatEventUnsubscribe(sessionId: trimmedSessionId) + } + + /// Keep a recently-viewed project chat warm briefly so switching back can + /// reuse its render-ready transcript and sequence watermark. Retention is + /// bounded to four chats; older pending subscriptions are released first. + func scheduleChatEventUnsubscribe( + sessionId: String, + delayNanoseconds: UInt64 = 120_000_000_000 + ) { + let trimmedSessionId = sessionId.trimmingCharacters(in: .whitespacesAndNewlines) + guard !trimmedSessionId.isEmpty, + subscribedChatSessionIds.contains(trimmedSessionId) + else { return } + + cancelDelayedChatUnsubscribe(sessionId: trimmedSessionId) + while pendingChatUnsubscribesBySession.count >= chatEventUnsubscribeRetentionLimit, + let oldestSessionId = pendingChatUnsubscribesBySession.min(by: { + $0.value.scheduledAt < $1.value.scheduledAt + })?.key { + cancelDelayedChatUnsubscribe(sessionId: oldestSessionId) + performChatEventUnsubscribe(sessionId: oldestSessionId) + } + + let task = Task { @MainActor [weak self] in + try? await Task.sleep(nanoseconds: delayNanoseconds) + guard let self, !Task.isCancelled else { return } + self.pendingChatUnsubscribesBySession.removeValue(forKey: trimmedSessionId) + self.performChatEventUnsubscribe(sessionId: trimmedSessionId) + } + pendingChatUnsubscribesBySession[trimmedSessionId] = PendingChatUnsubscribe( + task: task, + scheduledAt: Date() + ) + } + + private func cancelDelayedChatUnsubscribe(sessionId: String) { + pendingChatUnsubscribesBySession.removeValue(forKey: sessionId)?.task.cancel() + } + + private func performChatEventUnsubscribe(sessionId: String) { + guard subscribedChatSessionIds.contains(sessionId) else { return } + subscribedChatSessionIds.remove(sessionId) + chatSubscriptionsNeedingRemoteActivation.remove(sessionId) + recentFullChatSnapshotRequestBySession.removeValue(forKey: sessionId) localStateRevision += 1 if canSendLiveRequests() && supportsChatStreaming { - sendEnvelope(type: "chat_unsubscribe", requestId: nil, payload: chatSubscriptionPayload(sessionId: trimmedSessionId)) + sendEnvelope(type: "chat_unsubscribe", requestId: nil, payload: chatSubscriptionPayload(sessionId: sessionId)) } } @@ -14346,6 +14461,7 @@ final class SyncService: ObservableObject { let dict = payload as? [String: Any], let snapshot = try? decode(dict, as: SyncChatSubscribeSnapshotPayload.self), subscribedChatSessionIds.contains(snapshot.sessionId) { + recentFullChatSnapshotRequestBySession.removeValue(forKey: snapshot.sessionId) let resumed = (dict["resumed"] as? Bool) == true let previousLastSeq = chatEventLastSeqBySession[snapshot.sessionId] if !resumed { @@ -14731,12 +14847,13 @@ final class SyncService: ObservableObject { } } + @discardableResult private func sendEnvelope( type: String, requestId: String?, payload: Any, projectIdOverride: String? = nil - ) { + ) -> Bool { #if DEBUG if capturesOutboundEnvelopesForTesting { let projectId = syncNormalizedCommandScopeValue(projectIdOverride) @@ -14747,12 +14864,12 @@ final class SyncService: ObservableObject { let response = capturedRefreshResponseForTesting(type: type, payload: payload) { resolve(requestId: requestId, result: .success(response)) } - return + return true } #endif - guard let socket else { return } + guard let socket else { return false } let sendSocket = socket - guard let payloadData = try? adeJSONData(withJSONObject: payload) else { return } + guard let payloadData = try? adeJSONData(withJSONObject: payload) else { return false } var envelope: [String: Any] if payloadData.count >= compressionThresholdBytes { @@ -14783,7 +14900,7 @@ final class SyncService: ObservableObject { guard let data = try? adeJSONData(withJSONObject: envelope), let text = String(data: data, encoding: .utf8) - else { return } + else { return false } sendSocket.send(.string(text)) { error in if let error { @@ -14792,6 +14909,7 @@ final class SyncService: ObservableObject { } } } + return true } private func awaitSocketOpen(_ task: URLSessionWebSocketTask) async throws { @@ -15771,6 +15889,7 @@ final class SyncService: ObservableObject { requestId: nil, payload: payload ) + chatSubscriptionsNeedingRemoteActivation.remove(sessionId) } } @@ -16139,7 +16258,13 @@ final class SyncService: ObservableObject { // tagging lanes with another project's PRs. laneGithubPrItems = [] laneGithubPrItemsFetchedAt = nil + for pendingUnsubscribe in pendingChatUnsubscribesBySession.values { + pendingUnsubscribe.task.cancel() + } + pendingChatUnsubscribesBySession.removeAll() subscribedChatSessionIds.removeAll() + chatSubscriptionsNeedingRemoteActivation.removeAll() + recentFullChatSnapshotRequestBySession.removeAll() // Turn-active hints are scoped to the live connection's event stream — // a stale "running" hint must not survive a project switch or reconnect. chatTurnActiveHintBySession.removeAll() diff --git a/apps/ios/ADE/Views/Work/WorkChatSessionView.swift b/apps/ios/ADE/Views/Work/WorkChatSessionView.swift index 19db587e2..f0653dcf5 100644 --- a/apps/ios/ADE/Views/Work/WorkChatSessionView.swift +++ b/apps/ios/ADE/Views/Work/WorkChatSessionView.swift @@ -155,6 +155,7 @@ struct WorkChatSessionView: View { @State var latestPinTask: Task? @State var latestPinGeneration = 0 @State var assistantPreviewCache = WorkAssistantPreviewCache() + @State private var contextUsageViewModelCache = WorkContextUsageViewModelCache() @State var assistantLineBudgets: [String: Int] = [:] @State var composerSettingMutationInFlight = false @State var composerSettingMutationGeneration = 0 @@ -807,8 +808,10 @@ struct WorkChatSessionView: View { WorkChatComposerCard( chatSummary: chatSummaryContext, - usageViewModel: workContextUsageViewModel( + usageViewModel: contextUsageViewModelCache.value( + sessionId: session.id, transcript: transcript, + transcriptRenderSignature: transcriptRenderSignature, provider: chatSummaryContext.provider, fallbackContextWindow: chatSummaryContext.contextWindowFallback ), diff --git a/apps/ios/ADE/Views/Work/WorkErrorAndMessageHelpers.swift b/apps/ios/ADE/Views/Work/WorkErrorAndMessageHelpers.swift index d5764a950..87423c30e 100644 --- a/apps/ios/ADE/Views/Work/WorkErrorAndMessageHelpers.swift +++ b/apps/ios/ADE/Views/Work/WorkErrorAndMessageHelpers.swift @@ -1795,7 +1795,18 @@ func pendingWorkInputItemIds(from transcript: [WorkChatEnvelope]) -> Set } func sortedWorkChatEnvelopes(_ transcript: [WorkChatEnvelope]) -> [WorkChatEnvelope] { - transcript.sorted { lhs, rhs in + guard transcript.count > 1 else { return transcript } + let isAlreadySorted = transcript.indices.dropFirst().allSatisfy { index in + let lhs = transcript[transcript.index(before: index)] + let rhs = transcript[index] + if lhs.timestamp == rhs.timestamp { + return (lhs.sequence ?? 0) <= (rhs.sequence ?? 0) + } + return lhs.timestamp <= rhs.timestamp + } + guard !isAlreadySorted else { return transcript } + + return transcript.sorted { lhs, rhs in if lhs.timestamp == rhs.timestamp { return (lhs.sequence ?? 0) < (rhs.sequence ?? 0) } diff --git a/apps/ios/ADE/Views/Work/WorkRootComponents.swift b/apps/ios/ADE/Views/Work/WorkRootComponents.swift index 4d8386f0a..b9f88220e 100644 --- a/apps/ios/ADE/Views/Work/WorkRootComponents.swift +++ b/apps/ios/ADE/Views/Work/WorkRootComponents.swift @@ -1018,7 +1018,7 @@ private struct WorkSessionRowRenderSignature: Equatable { let laneAhead: Int let laneBehind: Int let activityTimestamp: String - let previewSource: String? + let previewText: String? let pinned: Bool let pullRequestNumber: Int? let pullRequestState: String? @@ -1069,7 +1069,7 @@ private struct WorkSessionRowRenderSignature: Equatable { self.laneBehind = lane?.status.behind ?? 0 self.activityTimestamp = workSessionActivityTimestamp(session: session, summary: chatSummary) let canonical = workCanonicalSessionState(session: session, summary: chatSummary) - self.previewSource = workSessionRowPreviewSource( + self.previewText = workSessionRowPreviewSource( session: session, chatSummary: chatSummary, isSettled: canonical.phase == .settled @@ -1248,13 +1248,13 @@ struct WorkSessionRow: View, Equatable { WorkSessionStatusCapsule(badge: badge) } Spacer(minLength: 6) - Text(relativeTimestampCompact(workSessionActivityTimestamp(session: session, summary: chatSummary))) + Text(relativeTimestampCompact(renderSignature.activityTimestamp)) .font(.caption2.monospacedDigit()) .foregroundStyle(ADEColor.textMuted) .lineLimit(1) } - if let preview = workSessionPreviewText(rowPreviewSource) { + if let preview = renderSignature.previewText { Text(preview) .font(.caption2) .foregroundStyle(ADEColor.textMuted) @@ -1387,14 +1387,6 @@ struct WorkSessionRow: View, Equatable { return workSnoozeWakeLabel(session.snoozedUntil) } - var rowPreviewSource: String? { - workSessionRowPreviewSource( - session: session, - chatSummary: chatSummary, - isSettled: isSettled - ) - } - var isPendingSyncCreation: Bool { workIsPendingChatCreationSession(session) } diff --git a/apps/ios/ADE/Views/Work/WorkSessionDestinationView.swift b/apps/ios/ADE/Views/Work/WorkSessionDestinationView.swift index 3ef7d874d..746d4cc05 100644 --- a/apps/ios/ADE/Views/Work/WorkSessionDestinationView.swift +++ b/apps/ios/ADE/Views/Work/WorkSessionDestinationView.swift @@ -258,8 +258,7 @@ struct WorkLiveTranscriptCache { return [] } - if canAppend(sessionId: sessionId, events: events) { - let appendedEvents = Array(events.dropFirst(eventCount)) + if let appendedEvents = appendedEvents(sessionId: sessionId, events: events) { if !appendedEvents.isEmpty { let appendedTranscript = makeWorkChatTranscript(from: appendedEvents) recentDeltaTranscript = appendedTranscript @@ -285,20 +284,57 @@ struct WorkLiveTranscriptCache { return transcript } - private func canAppend( + /// Returns only events that arrived after the previously-rendered tail. + /// + /// The sync service keeps a capped ring. Once it reaches that cap, every new + /// event removes one item from the head while the count stays constant. The + /// old count/head check treated that ordinary slide as a full rebuild, which + /// replayed the entire raw event window through the streaming text merger on + /// every tick. Anchor on the previous tail instead: it remains immediately + /// before the true delta during a normal ring slide. + private func appendedEvents( sessionId: String, events: [AgentChatEventEnvelope] - ) -> Bool { - guard self.sessionId == sessionId, - eventCount <= events.count - else { return false } + ) -> [AgentChatEventEnvelope]? { + guard self.sessionId == sessionId else { return nil } if eventCount == 0 { - return true + return events + } + + guard let tailEvent else { return nil } + + if tailEvent == events.last { + return eventCount == events.count && headEvent == events.first ? [] : nil + } + + // Streaming normally appends one event at a time. Walk backward across + // only the new sequence suffix, then validate the exact previous tail at + // the boundary. This keeps the hot path O(delta), not O(ring size). + if let tailSequence = tailEvent.sequence, + let latestSequence = events.last?.sequence, + latestSequence > tailSequence { + var appendStart = events.endIndex + while appendStart > events.startIndex { + let candidateIndex = events.index(before: appendStart) + guard let candidateSequence = events[candidateIndex].sequence, + candidateSequence > tailSequence + else { break } + appendStart = candidateIndex + } + if appendStart > events.startIndex { + let anchorIndex = events.index(before: appendStart) + if events[anchorIndex] == tailEvent { + return Array(events[appendStart...]) + } + } } - return headEvent == events.first - && tailEvent == events[eventCount - 1] + // Older hosts may omit sequence numbers. Exact tail identity still makes a + // capped ring slide safely incremental. + guard let tailIndex = events.lastIndex(of: tailEvent) else { return nil } + let appendStart = events.index(after: tailIndex) + return Array(events[appendStart...]) } } @@ -308,6 +344,7 @@ private struct WorkChatTranscriptPresentationCacheEntry { var olderTranscriptCursor: Int? var transcriptCursorKind: String? var olderChatEventHistoryCursor: Int? + var initialTranscriptTailHydrated: Bool var storedAt: Date var hasVisibleTranscript: Bool { @@ -319,6 +356,7 @@ private struct WorkChatTranscriptPresentationCacheEntry { private var workChatTranscriptPresentationCacheBySession: [String: WorkChatTranscriptPresentationCacheEntry] = [:] private let workChatTranscriptPresentationCacheLimit = 8 +private let workChatOpeningSnapshotRetryInterval: TimeInterval = 30 func workChatTranscriptEntriesByIndexForRestoredPresentation( fallbackEntries: [AgentChatTranscriptEntry], @@ -330,6 +368,25 @@ func workChatTranscriptEntriesByIndexForRestoredPresentation( ) } +func workChatShouldRequestOpeningSnapshot( + alreadySubscribed: Bool, + openingSnapshotRequestedAtUptime: TimeInterval?, + forceFreshTranscriptOnOpen: Bool, + initialTranscriptTailHydrated: Bool, + hasVisiblePresentation: Bool, + hasCachedEventHistory: Bool, + nowUptime: TimeInterval = ProcessInfo.processInfo.systemUptime, + retryInterval: TimeInterval = workChatOpeningSnapshotRetryInterval +) -> Bool { + if !alreadySubscribed { return true } + if let openingSnapshotRequestedAtUptime, + nowUptime - openingSnapshotRequestedAtUptime < retryInterval { + return false + } + if forceFreshTranscriptOnOpen && !initialTranscriptTailHydrated { return true } + return !hasVisiblePresentation && !hasCachedEventHistory +} + /// Resolve the scroll-back cursor a chat event history snapshot implies. /// /// `hasOlderHistory` is authoritative when the host sends it: it is derived @@ -402,6 +459,78 @@ struct WorkSessionDestinationView: View { /// composer. Model, reasoning, fast mode, and identity live in CTO settings. var compactComposer = false + @MainActor + init( + sessionId: String, + initialOpeningPrompt: String?, + initialOpeningPromptDispatchHandled: Bool = false, + initialOpeningDeliveryState: String? = nil, + initialOpeningAttachments: [AgentChatFileRef] = [], + initialSession: TerminalSessionSummary?, + initialChatSummary: AgentChatSessionSummary?, + initialTranscript: [WorkChatEnvelope]?, + transitionNamespace: Namespace.ID?, + isLive: Bool, + navigationChrome: WorkSessionNavigationChrome, + forceFreshTranscriptOnOpen: Bool = false, + showsLaneActions: Bool = true, + navigationTitleOverride: String? = nil, + lanes: [LaneSummary] = [], + crossProjectContext: WorkChatCrossProjectContext? = nil, + personalChat: Bool = false, + compactComposer: Bool = false + ) { + self.sessionId = sessionId + self.initialOpeningPrompt = initialOpeningPrompt + self.initialOpeningPromptDispatchHandled = initialOpeningPromptDispatchHandled + self.initialOpeningDeliveryState = initialOpeningDeliveryState + self.initialOpeningAttachments = initialOpeningAttachments + self.initialSession = initialSession + self.initialChatSummary = initialChatSummary + self.initialTranscript = initialTranscript + self.transitionNamespace = transitionNamespace + self.isLive = isLive + self.navigationChrome = navigationChrome + self.forceFreshTranscriptOnOpen = forceFreshTranscriptOnOpen + self.showsLaneActions = showsLaneActions + self.navigationTitleOverride = navigationTitleOverride + self.lanes = lanes + self.crossProjectContext = crossProjectContext + self.personalChat = personalChat + self.compactComposer = compactComposer + + let providedTranscript = initialTranscript ?? [] + let cachedPresentation = forceFreshTranscriptOnOpen || !providedTranscript.isEmpty + ? nil + : workChatTranscriptPresentationCacheBySession[sessionId] + let seededTranscript = providedTranscript.isEmpty + ? (cachedPresentation?.transcript ?? []) + : providedTranscript + let seededFallbackEntries = seededTranscript.isEmpty + ? (cachedPresentation?.fallbackEntries ?? []) + : [] + + _session = State(initialValue: initialSession) + _chatSummary = State(initialValue: initialChatSummary) + _lastKnownChatSummary = State(initialValue: initialChatSummary) + _transcript = State(initialValue: seededTranscript) + _transcriptRenderSignature = State(initialValue: workChatEnvelopeListRenderSignature(seededTranscript)) + _fallbackEntries = State(initialValue: seededFallbackEntries) + _fallbackEntriesRenderSignature = State(initialValue: workFallbackEntriesRenderSignature(seededFallbackEntries)) + _transcriptEntriesByIndex = State(initialValue: workChatTranscriptEntriesByIndexForRestoredPresentation( + fallbackEntries: seededFallbackEntries, + cursorKind: cachedPresentation?.transcriptCursorKind + )) + _olderTranscriptCursor = State(initialValue: cachedPresentation?.olderTranscriptCursor) + _transcriptCursorKind = State(initialValue: cachedPresentation?.transcriptCursorKind) + _olderChatEventHistoryCursor = State(initialValue: cachedPresentation?.olderChatEventHistoryCursor) + _initialTranscriptTailHydrated = State( + initialValue: !providedTranscript.isEmpty + || cachedPresentation?.initialTranscriptTailHydrated == true + ) + _openingTranscriptSnapshotRequestedAtUptime = State(initialValue: nil) + } + /// Whether this view is a cross-project "quick look" (see `crossProjectContext`). var isCrossProject: Bool { crossProjectContext != nil } var isRemoteOnlyChat: Bool { isCrossProject || personalChat } @@ -488,6 +617,8 @@ struct WorkSessionDestinationView: View { @State var lastCanonicalTranscriptRefreshAt = Date.distantPast @State var lastArtifactRefreshAt = Date.distantPast @State var initialTranscriptTailHydrated = false + @State var openingTranscriptSnapshotRequestedAtUptime: TimeInterval? + @State var initialLoadCompleted = false @State var openingLoadInFlight = false @State var emptyTranscriptHydrationInFlight = false @State var canonicalTranscriptRefreshInFlight = false @@ -528,17 +659,24 @@ struct WorkSessionDestinationView: View { transcriptCursorKind = nil olderChatEventHistoryCursor = nil initialTranscriptTailHydrated = false + openingTranscriptSnapshotRequestedAtUptime = nil } @MainActor func cacheCurrentTranscriptPresentationIfNeeded() { guard !transcript.isEmpty || !fallbackEntries.isEmpty else { return } + // The mapped transcript and canonical fallback represent the same visible + // conversation in two shapes. Retaining both doubled the largest chat + // arrays after navigating back to Work. Keep the render-ready transcript + // when available; retain fallback rows only for fallback-only sessions. + let cachedFallbackEntries = transcript.isEmpty ? fallbackEntries : [] workChatTranscriptPresentationCacheBySession[sessionId] = WorkChatTranscriptPresentationCacheEntry( transcript: transcript, - fallbackEntries: fallbackEntries, + fallbackEntries: cachedFallbackEntries, olderTranscriptCursor: olderTranscriptCursor, transcriptCursorKind: transcriptCursorKind, olderChatEventHistoryCursor: olderChatEventHistoryCursor, + initialTranscriptTailHydrated: initialTranscriptTailHydrated, storedAt: Date() ) guard workChatTranscriptPresentationCacheBySession.count > workChatTranscriptPresentationCacheLimit else { return } @@ -568,13 +706,13 @@ struct WorkSessionDestinationView: View { fallbackEntries: cached.fallbackEntries, cursorKind: cached.transcriptCursorKind ) + initialTranscriptTailHydrated = cached.initialTranscriptTailHydrated if !cached.transcript.isEmpty { setTranscript(cached.transcript) } if !cached.fallbackEntries.isEmpty { setFallbackEntries(cached.fallbackEntries) } - initialTranscriptTailHydrated = true } @MainActor @@ -1005,25 +1143,47 @@ struct WorkSessionDestinationView: View { // user interaction cannot race the async load task and accidentally // fall back to the active project. registerChatCommandScope() + if !isRemoteOnlyChat, + let currentSession = session ?? initialSession, + isChatSession(currentSession) { + syncService.retainChatEventSubscription(sessionId: sessionId) + } } .task { // Cross-project "quick look": register the foreign scope BEFORE load() // so every transcript/summary/send routes to that project without // switching the phone's active project. registerChatCommandScope() - mainChatRenderEpoch = 0 liveTranscriptCache.reset(sessionId: sessionId) - resetTranscriptHistoryState() - session = initialSession - chatSummary = initialChatSummary - setTranscript(initialTranscript ?? []) - seedTranscriptFromPresentationCacheIfNeeded() + if forceFreshTranscriptOnOpen { + resetTranscriptHistoryState() + session = initialSession + chatSummary = initialChatSummary + lastKnownChatSummary = initialChatSummary + setTranscript(initialTranscript ?? []) + setFallbackEntries([]) + } else { + if session == nil { + session = initialSession + } + if chatSummary == nil { + chatSummary = initialChatSummary + } + if lastKnownChatSummary == nil { + lastKnownChatSummary = initialChatSummary + } + if transcript.isEmpty && fallbackEntries.isEmpty { + setTranscript(initialTranscript ?? []) + seedTranscriptFromPresentationCacheIfNeeded() + } + } if initialOpeningPromptNeedsManualRetry, let initialOpeningPrompt { composerDraftRestore = WorkChatComposerDraftRestore(text: initialOpeningPrompt) openingDeliveryWarning = SyncRequestTimeout.chatSendMessage } stageInitialOpeningPromptEchoIfNeeded() await load() + initialLoadCompleted = true await sendInitialOpeningPromptIfNeeded() refreshChatInfoSnapshots() // Remote subagent probing hits the host; skip the eager pass for a @@ -1052,6 +1212,9 @@ struct WorkSessionDestinationView: View { .task(id: emptyTranscriptHydrationKey) { await hydrateEmptyTranscriptFromHostIfNeeded() } + .task(id: openingSnapshotRetryKey) { + await retryUnacknowledgedOpeningSnapshotIfNeeded() + } .task(id: sessionRowObservationKey) { // A cross-project quick look has no local DB row for this session (only // the active project is mirrored) — status comes from the streamed @@ -1122,16 +1285,21 @@ struct WorkSessionDestinationView: View { cleanupLoadedArtifactContent() let wasCrossProject = isCrossProject let wasPersonalChat = personalChat - Task { @MainActor in - try? await syncService.unsubscribeFromChatEvents(sessionId: sessionId) - // Preserve routing through the unsubscribe payload, then drop it so - // a later ordinary project chat with the same id cannot inherit the - // foreign/runtime scope. - if wasCrossProject { - syncService.clearCrossProjectChatScope(sessionId: sessionId) - } else if wasPersonalChat { - syncService.clearPersonalChatScope(sessionId: sessionId) + if wasCrossProject || wasPersonalChat { + Task { @MainActor in + try? await syncService.unsubscribeFromChatEvents(sessionId: sessionId) + // Preserve routing through the unsubscribe payload, then drop it + // so a later ordinary project chat with the same id cannot inherit + // the foreign/runtime scope. + if wasCrossProject { + syncService.clearCrossProjectChatScope(sessionId: sessionId) + } else { + syncService.clearPersonalChatScope(sessionId: sessionId) + } } + } else if let currentSession = session ?? initialSession, + isChatSession(currentSession) { + syncService.scheduleChatEventUnsubscribe(sessionId: sessionId) } } } @@ -1158,12 +1326,20 @@ struct WorkSessionDestinationView: View { .environmentObject(syncService) } } else { - ADEEmptyStateView( - symbol: "bubble.left.and.bubble.right", - title: "Session unavailable", - message: "This session is no longer cached on the phone. Reconnect and refresh Work to restore it." - ) - .adeScreenBackground() + if initialLoadCompleted { + ADEEmptyStateView( + symbol: "bubble.left.and.bubble.right", + title: "Session unavailable", + message: "This session is no longer cached on the phone. Reconnect and refresh Work to restore it." + ) + .adeScreenBackground() + } else { + ProgressView() + .controlSize(.large) + .frame(maxWidth: .infinity, maxHeight: .infinity) + .adeScreenBackground() + .accessibilityLabel("Opening session") + } } } @@ -1241,12 +1417,17 @@ struct WorkSessionDestinationView: View { "chat.dispatchSteer", sessionId: session.id ) - let restoreCancelledQueueAction: (@MainActor (String) async -> Void)? = syncService.supportsChatRemoteAction( + let restoreCancelledQueueAction: (@MainActor (String) async -> Void)? + if syncService.supportsChatRemoteAction( "chat.restoreCancelledQueue", sessionId: session.id - ) ? { recoveryId in - await restoreCancelledQueue(recoveryId: recoveryId) - } : nil + ) { + restoreCancelledQueueAction = { recoveryId in + await restoreCancelledQueue(recoveryId: recoveryId) + } + } else { + restoreCancelledQueueAction = nil + } return WorkChatSessionView( session: WorkChatSessionRenderContext(session), chatSummaryContext: WorkChatSummaryRenderContext(composerChatSummary), @@ -1380,6 +1561,10 @@ struct WorkSessionDestinationView: View { "\(session?.id ?? sessionId)-empty:\(transcript.isEmpty)-fallback:\(fallbackEntries.isEmpty)-host:\(hostReachable)-opening:\(openingLoadInFlight)-local:\(syncService.localStateRevision)" } + var openingSnapshotRetryKey: String { + "\(sessionId)-request:\(openingTranscriptSnapshotRequestedAtUptime ?? -1)-cross:\(isCrossProject)-reachable:\(isLiveAndReachable)" + } + var selectedSubagentPollingKey: String { guard let selectedSubagentSnapshot, selectedSubagentSnapshot.status == .running @@ -1555,40 +1740,97 @@ struct WorkSessionDestinationView: View { await loadTranscript(forceRemote: true, preferLightweight: false) } + @MainActor + func retryUnacknowledgedOpeningSnapshotIfNeeded() async { + guard isCrossProject, + isLiveAndReachable, + transcript.isEmpty, + fallbackEntries.isEmpty, + let requestedAtUptime = openingTranscriptSnapshotRequestedAtUptime, + syncService.isFullChatEventSnapshotPending(sessionId: sessionId) + else { return } + + let elapsed = ProcessInfo.processInfo.systemUptime - requestedAtUptime + let remaining = max(0, workChatOpeningSnapshotRetryInterval - elapsed) + if remaining > 0 { + try? await Task.sleep(nanoseconds: UInt64(remaining * 1_000_000_000)) + } + guard !Task.isCancelled, + isCrossProject, + isLiveAndReachable, + transcript.isEmpty, + fallbackEntries.isEmpty, + openingTranscriptSnapshotRequestedAtUptime == requestedAtUptime, + syncService.isFullChatEventSnapshotPending(sessionId: sessionId) + else { return } + + openingTranscriptSnapshotRequestedAtUptime = nil + await loadTranscript(forceRemote: true, preferLightweight: false) + } + @MainActor func loadTranscript(forceRemote: Bool, preferLightweight: Bool = false) async { seedTranscriptFromPresentationCacheIfNeeded() - let forceOpeningTranscriptRefresh = forceFreshTranscriptOnOpen && !initialTranscriptTailHydrated - let status = normalizedWorkChatSessionStatus(session: session ?? initialSession, summary: chatSummary ?? initialChatSummary) let transcriptStatus = workChatTranscriptPreferenceStatus( sessionStatus: status, liveTurnActiveHint: syncService.chatTurnActiveHint(sessionId: sessionId) ) let reducedActiveLiveStream = preferLightweight && transcriptStatus == "active" + var requestedOpeningSnapshotThisLoad = false if forceRemote, let currentSession = session ?? initialSession, isChatSession(currentSession) { let alreadySubscribed = syncService.subscribedChatSessionIds.contains(sessionId) let hasReusablePresentation = workChatTranscriptPresentationCacheBySession[sessionId]?.hasVisibleTranscript == true let hasCachedEventHistory = !syncService.chatEventHistory(sessionId: sessionId).isEmpty - let needsOpeningSnapshot = forceOpeningTranscriptRefresh - || (transcript.isEmpty && fallbackEntries.isEmpty && !hasReusablePresentation && !hasCachedEventHistory) + let hasVisiblePresentation = !transcript.isEmpty || !fallbackEntries.isEmpty || hasReusablePresentation + let needsOpeningSnapshot = workChatShouldRequestOpeningSnapshot( + alreadySubscribed: alreadySubscribed, + openingSnapshotRequestedAtUptime: openingTranscriptSnapshotRequestedAtUptime, + forceFreshTranscriptOnOpen: forceFreshTranscriptOnOpen, + initialTranscriptTailHydrated: initialTranscriptTailHydrated, + hasVisiblePresentation: hasVisiblePresentation, + hasCachedEventHistory: hasCachedEventHistory + ) + requestedOpeningSnapshotThisLoad = needsOpeningSnapshot if status == "active" { // First visit subscribes (the host answers with a snapshot or a // sinceSeq replay). Once subscribed, live chat_event push plus the // host's transcript pump cover continuity — re-requesting a full // byte-capped snapshot on every 8s poll was redundant wire traffic // and a full dedupe/sort merge on the phone mid-stream. - try? await syncService.subscribeToChatEvents( - sessionId: sessionId, - requestSnapshot: !alreadySubscribed || needsOpeningSnapshot - ) + do { + let snapshotRequestDispatched = try await syncService.subscribeToChatEvents( + sessionId: sessionId, + requestSnapshot: needsOpeningSnapshot + ) + if needsOpeningSnapshot && snapshotRequestDispatched { + openingTranscriptSnapshotRequestedAtUptime = ProcessInfo.processInfo.systemUptime + } + } catch { + // Leave the retry latch open. A later poll can recover if the + // transport changed while this request was being dispatched. + } } else if needsOpeningSnapshot { // Active streaming stays on reduced snapshots for performance, but an // idle detail view must reconcile against a full event snapshot. A // reduced JSONL tail can start mid-message and render as a broken // final transcript until the canonical transcript fetch lands. - try? await syncService.requestFullChatEventSnapshot(sessionId: sessionId) + do { + let snapshotRequestDispatched = try await syncService.requestFullChatEventSnapshot( + sessionId: sessionId + ) + if snapshotRequestDispatched { + openingTranscriptSnapshotRequestedAtUptime = ProcessInfo.processInfo.systemUptime + } + } catch { + // Leave the retry latch open; the next poll can try again. + } + } else { + // Reopening an already-warm idle chat must still cancel its pending + // delayed unsubscribe. No envelope is sent when the subscription is + // already active; this only preserves live delivery while visible. + _ = try? await syncService.subscribeToChatEvents(sessionId: sessionId) } // Quick looks stay on the chat_subscribe snapshot/tail — the canonical @@ -1601,7 +1843,7 @@ struct WorkSessionDestinationView: View { || transcript.isEmpty || transcriptStatus != "active" || !initialTranscriptTailHydrated - || forceOpeningTranscriptRefresh + || needsOpeningSnapshot ) if shouldHydrateCanonicalEventTail { do { @@ -1639,6 +1881,9 @@ struct WorkSessionDestinationView: View { ) let liveDeltaTranscript = liveTranscriptCache.recentDeltaTranscript let liveTranscriptWasRebuilt = liveTranscriptCache.recentTranscriptWasRebuilt + if !liveTranscript.isEmpty { + initialTranscriptTailHydrated = true + } var fallbackTranscript: [WorkChatEnvelope] = [] var eventTranscript: [WorkChatEnvelope] = [] var fetchedFallbackEntries: [AgentChatTranscriptEntry] = [] @@ -1660,7 +1905,7 @@ struct WorkSessionDestinationView: View { let shouldFetchFallback = !isCrossProject && !reducedActiveLiveStream && ( - forceOpeningTranscriptRefresh + requestedOpeningSnapshotThisLoad || needsInitialTailHydration || !preferLightweight || (liveTranscript.isEmpty && transcript.isEmpty) diff --git a/apps/ios/ADE/Views/Work/WorkTimelineHelpers.swift b/apps/ios/ADE/Views/Work/WorkTimelineHelpers.swift index addb6b1a2..f36e158af 100644 --- a/apps/ios/ADE/Views/Work/WorkTimelineHelpers.swift +++ b/apps/ios/ADE/Views/Work/WorkTimelineHelpers.swift @@ -3563,6 +3563,48 @@ func workContextUsageViewModel( ) } +final class WorkContextUsageViewModelCache { + private struct Key: Equatable { + let sessionId: String + let transcriptRenderSignature: Int + let provider: String + let fallbackContextWindow: Int? + } + + private struct Entry { + let key: Key + let value: WorkContextUsageViewModel? + } + + private var entry: Entry? + + func value( + sessionId: String, + transcript: [WorkChatEnvelope], + transcriptRenderSignature: Int, + provider: String, + fallbackContextWindow: Int? + ) -> WorkContextUsageViewModel? { + let key = Key( + sessionId: sessionId, + transcriptRenderSignature: transcriptRenderSignature, + provider: provider, + fallbackContextWindow: fallbackContextWindow + ) + if entry?.key == key { + return entry?.value + } + + let value = workContextUsageViewModel( + transcript: transcript, + provider: provider, + fallbackContextWindow: fallbackContextWindow + ) + entry = Entry(key: key, value: value) + return value + } +} + private func makeWorkContextUsageViewModel( usage: WorkUsageSummary, provider: String, diff --git a/apps/ios/ADETests/ADETests.swift b/apps/ios/ADETests/ADETests.swift index 7ef033a3f..017c5a4f4 100644 --- a/apps/ios/ADETests/ADETests.swift +++ b/apps/ios/ADETests/ADETests.swift @@ -4571,6 +4571,289 @@ final class ADETests: XCTestCase { XCTAssertEqual(service.localStateRevision, unsubscribedRevision) } + @MainActor + func testRapidFullChatSnapshotRequestsAreCoalesced() async throws { + let service = SyncService(database: makeDatabase(baseURL: makeTemporaryDirectory())) + try service.applyHelloPayloadForTesting([ + "brain": [ + "deviceId": "host-1", + "deviceName": "Mac Studio", + ], + "features": [ + "chatStreaming": true, + ], + ]) + service.configureConnectedTransportForTesting() + service.beginOutboundEnvelopeCaptureForTesting() + defer { service.endOutboundEnvelopeCaptureForTesting() } + + let firstRequestDispatched = try await service.requestFullChatEventSnapshot(sessionId: "session-1") + let secondRequestCoalesced = try await service.requestFullChatEventSnapshot(sessionId: "session-1") + let thirdRequestCoalesced = try await service.requestFullChatEventSnapshot(sessionId: "session-1") + XCTAssertTrue(firstRequestDispatched) + XCTAssertTrue(secondRequestCoalesced) + XCTAssertTrue(thirdRequestCoalesced) + + XCTAssertEqual(service.subscribedChatSessionIds, Set(["session-1"])) + XCTAssertEqual(service.capturedOutboundEnvelopeCountForTesting(type: "chat_subscribe"), 1) + XCTAssertEqual(service.localStateRevision, 1) + XCTAssertTrue(service.isFullChatEventSnapshotPending(sessionId: "session-1")) + + service.disconnect(clearCredentials: false) + let offlineRequestDispatched = try await service.requestFullChatEventSnapshot(sessionId: "session-1") + XCTAssertFalse(offlineRequestDispatched) + XCTAssertFalse(service.isFullChatEventSnapshotPending(sessionId: "session-1")) + XCTAssertEqual(service.capturedOutboundEnvelopeCountForTesting(type: "chat_subscribe"), 1) + } + + func testOpeningSnapshotRequestDoesNotRepeatAfterDispatch() { + XCTAssertTrue(workChatShouldRequestOpeningSnapshot( + alreadySubscribed: false, + openingSnapshotRequestedAtUptime: nil, + forceFreshTranscriptOnOpen: false, + initialTranscriptTailHydrated: false, + hasVisiblePresentation: false, + hasCachedEventHistory: false + )) + XCTAssertTrue(workChatShouldRequestOpeningSnapshot( + alreadySubscribed: true, + openingSnapshotRequestedAtUptime: nil, + forceFreshTranscriptOnOpen: true, + initialTranscriptTailHydrated: false, + hasVisiblePresentation: true, + hasCachedEventHistory: true + )) + XCTAssertFalse(workChatShouldRequestOpeningSnapshot( + alreadySubscribed: true, + openingSnapshotRequestedAtUptime: 100, + forceFreshTranscriptOnOpen: true, + initialTranscriptTailHydrated: false, + hasVisiblePresentation: true, + hasCachedEventHistory: true, + nowUptime: 120 + )) + XCTAssertFalse(workChatShouldRequestOpeningSnapshot( + alreadySubscribed: true, + openingSnapshotRequestedAtUptime: nil, + forceFreshTranscriptOnOpen: false, + initialTranscriptTailHydrated: false, + hasVisiblePresentation: true, + hasCachedEventHistory: false + )) + XCTAssertTrue(workChatShouldRequestOpeningSnapshot( + alreadySubscribed: true, + openingSnapshotRequestedAtUptime: 100, + forceFreshTranscriptOnOpen: true, + initialTranscriptTailHydrated: false, + hasVisiblePresentation: true, + hasCachedEventHistory: true, + nowUptime: 131 + )) + } + + func testContextUsageViewModelCacheInvalidatesOnlyForRelevantInputs() throws { + let firstUsage = WorkUsageSummary( + turnCount: 1, + inputTokens: 100, + outputTokens: 20, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalTokens: 120, + contextWindow: 1_000, + costUsd: 0 + ) + let secondUsage = WorkUsageSummary( + turnCount: 1, + inputTokens: 500, + outputTokens: 20, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalTokens: 520, + contextWindow: 1_000, + costUsd: 0 + ) + let firstTranscript = [ + WorkChatEnvelope( + sessionId: "session-1", + timestamp: "2026-07-27T12:00:00.000Z", + sequence: 1, + event: .tokens(usage: firstUsage, turnId: "turn-1", itemId: nil) + ), + ] + let secondTranscript = [ + WorkChatEnvelope( + sessionId: "session-1", + timestamp: "2026-07-27T12:00:01.000Z", + sequence: 2, + event: .tokens(usage: secondUsage, turnId: "turn-1", itemId: nil) + ), + ] + let crossSessionTranscript = [ + WorkChatEnvelope( + sessionId: "session-2", + timestamp: "2026-07-27T12:00:01.000Z", + sequence: 2, + event: .tokens(usage: secondUsage, turnId: "turn-1", itemId: nil) + ), + ] + let cache = WorkContextUsageViewModelCache() + + let first = try XCTUnwrap(cache.value( + sessionId: "session-1", + transcript: firstTranscript, + transcriptRenderSignature: 1, + provider: "codex", + fallbackContextWindow: nil + )) + let cached = try XCTUnwrap(cache.value( + sessionId: "session-1", + transcript: secondTranscript, + transcriptRenderSignature: 1, + provider: "codex", + fallbackContextWindow: nil + )) + let crossSession = try XCTUnwrap(cache.value( + sessionId: "session-2", + transcript: crossSessionTranscript, + transcriptRenderSignature: 1, + provider: "codex", + fallbackContextWindow: nil + )) + let refreshed = try XCTUnwrap(cache.value( + sessionId: "session-1", + transcript: secondTranscript, + transcriptRenderSignature: 2, + provider: "codex", + fallbackContextWindow: nil + )) + + XCTAssertEqual(cached.usedTokens, first.usedTokens) + XCTAssertNotEqual(crossSession.usedTokens, first.usedTokens) + XCTAssertNotEqual(refreshed.usedTokens, first.usedTokens) + } + + @MainActor + func testReopeningChatOfflineCancelsDelayedUnsubscribe() async throws { + let service = SyncService(database: makeDatabase(baseURL: makeTemporaryDirectory())) + try service.applyHelloPayloadForTesting([ + "brain": [ + "deviceId": "host-1", + "deviceName": "Mac Studio", + ], + "features": [ + "chatStreaming": true, + ], + ]) + service.configureConnectedTransportForTesting() + service.beginOutboundEnvelopeCaptureForTesting() + defer { service.endOutboundEnvelopeCaptureForTesting() } + + try await service.subscribeToChatEvents(sessionId: "session-1") + service.resetOutboundEnvelopeCaptureForTesting() + service.scheduleChatEventUnsubscribe( + sessionId: "session-1", + delayNanoseconds: 2_000_000 + ) + service.disconnect(clearCredentials: false) + service.retainChatEventSubscription(sessionId: "session-1") + try await Task.sleep(nanoseconds: 10_000_000) + + XCTAssertEqual(service.subscribedChatSessionIds, Set(["session-1"])) + XCTAssertEqual(service.capturedOutboundEnvelopeCountForTesting(type: "chat_unsubscribe"), 0) + + try await service.unsubscribeFromChatEvents(sessionId: "session-1") + } + + @MainActor + func testReopeningChatAfterWarmEvictionResubscribesRemotely() async throws { + let service = SyncService(database: makeDatabase(baseURL: makeTemporaryDirectory())) + try service.applyHelloPayloadForTesting([ + "brain": [ + "deviceId": "host-1", + "deviceName": "Mac Studio", + ], + "features": [ + "chatStreaming": true, + ], + ]) + service.configureConnectedTransportForTesting() + service.beginOutboundEnvelopeCaptureForTesting() + defer { service.endOutboundEnvelopeCaptureForTesting() } + + try await service.subscribeToChatEvents(sessionId: "session-1") + service.resetOutboundEnvelopeCaptureForTesting() + service.scheduleChatEventUnsubscribe( + sessionId: "session-1", + delayNanoseconds: 2_000_000 + ) + let evictionDeadline = Date().addingTimeInterval(1) + while !service.subscribedChatSessionIds.isEmpty, Date() < evictionDeadline { + try await Task.sleep(nanoseconds: 1_000_000) + } + + XCTAssertTrue(service.subscribedChatSessionIds.isEmpty) + XCTAssertEqual(service.capturedOutboundEnvelopeCountForTesting(type: "chat_unsubscribe"), 1) + + service.resetOutboundEnvelopeCaptureForTesting() + service.retainChatEventSubscription(sessionId: "session-1") + try await service.subscribeToChatEvents(sessionId: "session-1") + + XCTAssertEqual(service.subscribedChatSessionIds, Set(["session-1"])) + XCTAssertEqual(service.capturedOutboundEnvelopeCountForTesting(type: "chat_subscribe"), 1) + + try await service.unsubscribeFromChatEvents(sessionId: "session-1") + } + + func testLiveTranscriptCacheAppendsAcrossCappedRingSlides() { + func envelope(_ sequence: Int) -> AgentChatEventEnvelope { + AgentChatEventEnvelope( + sessionId: "session-1", + timestamp: String(format: "2026-07-27T00:00:%02d.%03dZ", (sequence / 1_000) % 60, sequence % 1_000), + event: .text( + text: "chunk-\(sequence)", + messageId: "message-1", + turnId: "turn-1", + itemId: "item-1" + ), + sequence: sequence, + provenance: nil + ) + } + + var cache = WorkLiveTranscriptCache() + let initialEvents = (1...1_000).map(envelope) + _ = cache.transcript(for: "session-1", events: initialEvents) + + let slidEvents = (2...1_001).map(envelope) + _ = cache.transcript(for: "session-1", events: slidEvents) + + XCTAssertFalse(cache.recentTranscriptWasRebuilt) + XCTAssertEqual(cache.recentDeltaTranscript.count, 1) + guard case .assistantText(let text, _, _) = cache.recentDeltaTranscript.first?.event else { + return XCTFail("Expected the new ring-tail event to map to one assistant delta.") + } + XCTAssertEqual(text, "chunk-1001") + } + + func testLiveTranscriptCacheRebuildsWhenPreviousTailFallsOutOfWindow() { + func envelope(_ sequence: Int) -> AgentChatEventEnvelope { + AgentChatEventEnvelope( + sessionId: "session-1", + timestamp: String(format: "2026-07-27T00:01:%02d.%03dZ", (sequence / 1_000) % 60, sequence % 1_000), + event: .activity(activity: .thinking, detail: "event-\(sequence)", turnId: "turn-1"), + sequence: sequence, + provenance: nil + ) + } + + var cache = WorkLiveTranscriptCache() + _ = cache.transcript(for: "session-1", events: (1...1_000).map(envelope)) + _ = cache.transcript(for: "session-1", events: (2_000...2_999).map(envelope)) + + XCTAssertTrue(cache.recentTranscriptWasRebuilt) + XCTAssertTrue(cache.recentDeltaTranscript.isEmpty) + } + @MainActor func testCredentialClearingRemovesHostBoundTerminalHistoryAndDeliveryState() throws { let service = SyncService(database: makeDatabase(baseURL: makeTemporaryDirectory())) @@ -17796,9 +18079,10 @@ final class ADETests: XCTestCase { command: "npm test", cwd: "/repo", output: "", + status: .completed, itemId: "command-1", - status: "completed", exitCode: 0, + durationMs: nil, turnId: nil ) ), diff --git a/docs/features/sync-and-multi-device/ios-companion.md b/docs/features/sync-and-multi-device/ios-companion.md index 81bd5b631..711c34765 100644 --- a/docs/features/sync-and-multi-device/ios-companion.md +++ b/docs/features/sync-and-multi-device/ios-companion.md @@ -817,7 +817,7 @@ Implemented envelope types on iOS: | `terminal_subscribe` / `terminal_unsubscribe` / `terminal_data` | Phone ↔ runtime | Terminal streaming; `unsubscribe` is sent when a Work terminal screen disappears so the phone stops accumulating buffer for off-screen sessions. `terminal_data.offset` is the lifetime logical UTF-8 end offset (null only for untracked/no-transcript or failed-write streams), so physical transcript rollover does not rewind it. The phone drops duplicates, trims UTF-8 overlap, and on a gap launches one guarded resubscribe from its watermark instead of rendering out of order. `terminal_subscribe.sinceOffset` returns an append-only `delta: true` snapshot when the retained logical window still covers the request; otherwise a full snapshot replaces local state even when its end offset equals the watermark. The host's bounded snapshot barrier queues live data/exits during capture and recaptures rather than flushing a gap. Snapshots also report `startOffset`/`endOffset`, plus `live: false` when no PTY backs the session so the phone shows a resume bar instead of accepting keystrokes | | `terminal_history` | Phone → runtime | On-demand scrollback paging: `{ sessionId, beforeOffset, maxBytes? }` returns retained transcript bytes `[startOffset, endOffset)` ending at/before `beforeOffset` (page start scanned forward to a newline/ESC and UTF-8 boundary; `atStart: true` means the oldest **retained logical offset**, which may be greater than zero after rollover). Requires an active `terminal_subscribe` | | `terminal_input` / `terminal_input_ack` / `terminal_resize` | Phone ↔ runtime | Input is queued in order only after the terminal snapshot is ready. ACK-capable hosts receive a stable `inputId`; the phone sends one item at a time, waits 8 seconds, and retries with 0.5/1/2-second backoff within the host-advertised lease (four total attempts). The host dedupes `(device, session, inputId)` before writing, so a lost ack cannot type twice. `not_subscribed` is the only retryable rejection: iOS re-subscribes through the snapshot barrier and resends the same id. Other errors fail that item and continue the queue. Legacy hosts receive one-shot input without an id or ambiguous retry. Mobile resizes are non-authoritative: the runtime restores the last desktop size when the final phone detaches | -| `chat_subscribe` / `chat_event` | Phone → runtime / runtime → phone | Agent chat transcript streaming; `chat_subscribe` carries `sinceSeq` so the runtime can replay exactly the missed events from its per-session buffer instead of re-sending a snapshot. The subscribe ack carries `turnActive` from the live agent chat service so a phone subscribing mid-turn renders the stop button and working indicator immediately — the byte-capped snapshot tail may have dropped the turn's `status: started` event, and the synced session row arrives via the slower changeset pump. The phone keeps the hint current from live `status` / `done` events, drops it when a full ack omits the flag (older host / no live summary), and clears it on project switch / reconnect resets. Incoming chat events bump a UI revision through a leading-edge coalescer (~150 ms window: the first event after a quiet period renders immediately, bursts batch); turn-state flips bypass the coalescer entirely so the stop button reacts instantly. On strained relay connections, the Work detail view stays subscribed to `chat_event` but skips heavyweight `chat.getChatEventHistory` and fallback transcript fetches while the turn is active; idle refresh reconciles the canonical transcript. When the host advertises `crossProjectChat`, `chat_subscribe` / `chat_unsubscribe` also carry an optional `projectId` / `projectRootPath` override so the Hub can open a chat in a **foreign** project read-only (transcript streamed straight off that project's `.ade` JSONL) without switching the phone's active project — see the Hub and Lane-data-projection sections. A `session_meta_updated` `chat_event` carrying a client's permission/interaction/mode change is folded into the cached summary via `applyChatSessionMetaModeUpdateIfNeeded` (decoded through `AgentChatSessionMetaModeUpdate`, a lenient all-optional-string type that no-ops for the bare title/manuallyNamed events older hosts send), so the open composer's mode pill updates live without a refetch | +| `chat_subscribe` / `chat_event` | Phone → runtime / runtime → phone | Agent chat transcript streaming; `chat_subscribe` carries `sinceSeq` so the runtime can replay exactly the missed events from its per-session buffer instead of re-sending a snapshot. Explicit full-snapshot subscribes omit `sinceSeq`; rapid duplicates are coalesced for five seconds or until the snapshot ack arrives. Ordinary project-chat subscriptions stay warm for 120 seconds after leaving a detail screen (at most four pending inactive chats), and reopening synchronously cancels eviction even while offline. Personal and cross-project scopes still unsubscribe immediately so their routing scope can be cleared safely. The subscribe ack carries `turnActive` from the live agent chat service so a phone subscribing mid-turn renders the stop button and working indicator immediately — the byte-capped snapshot tail may have dropped the turn's `status: started` event, and the synced session row arrives via the slower changeset pump. The phone keeps the hint current from live `status` / `done` events, drops it when a full ack omits the flag (older host / no live summary), and clears it on project switch / reconnect resets. Incoming chat events bump a UI revision through a leading-edge coalescer (~150 ms window: the first event after a quiet period renders immediately, bursts batch); turn-state flips bypass the coalescer entirely so the stop button reacts instantly. On strained relay connections, the Work detail view stays subscribed to `chat_event` but skips heavyweight `chat.getChatEventHistory` and fallback transcript fetches while the turn is active; idle refresh reconciles the canonical transcript. When the host advertises `crossProjectChat`, `chat_subscribe` / `chat_unsubscribe` also carry an optional `projectId` / `projectRootPath` override so the Hub can open a chat in a **foreign** project read-only (transcript streamed straight off that project's `.ade` JSONL) without switching the phone's active project — see the Hub and Lane-data-projection sections. A `session_meta_updated` `chat_event` carrying a client's permission/interaction/mode change is folded into the cached summary via `applyChatSessionMetaModeUpdateIfNeeded` (decoded through `AgentChatSessionMetaModeUpdate`, a lenient all-optional-string type that no-ops for the bare title/manuallyNamed events older hosts send), so the open composer's mode pill updates live without a refetch | | `chat_subscribe` with `chatScope: "personal"` | Phone → runtime / runtime → phone | Explicit projectless transcript/event subscription. `SyncService` marks the session personal, omits project id/root, routes send/steer/approval/update/lifecycle and scheduled-work Cancel/Pause calls to `personalChats.*`, and loads image bytes through `personalChats.getImageDataUrl`. Missing project scope alone never selects this path. | | `roster_subscribe` / `roster_unsubscribe` / `roster_snapshot` / `roster_delta` | Phone → runtime / runtime → phone | All-projects session roster feed backing the Hub: agent chats, their attached shell rows, and standalone CLI (tracked terminal) sessions — live **and** ended. Subscribe (optionally with `sinceSeq`) yields a full `roster_snapshot` then incremental `roster_delta` upserts (`changed` = whole project entries) / `removed` project ids. Un-booted projects carry disk-derived status only (a booted scope also overlays PTY liveness for CLI rows); transcripts load on demand when a chat opens. Additive lifecycle fields carry `settledAt`, `statusNote`, `attentionRequestedAt`, `attentionMessage`, `lastTurnFailedAt`, and `exitCode`; disk readers return nulls against legacy databases that do not have those columns. `toolType` passes through so the phone routes chat rows to the chat surface and CLI rows to the terminal — a CLI row must never take the cross-project chat quick-look (it has no chat JSONL and would render blank) | | `envelope_chunk` | Runtime → phone | Slice of an oversized encoded envelope (>720 KB); the phone reassembles by `chunkId`/`index` before normal decode. `SyncEnvelopeChunkAssembler` enforces a 32 MiB reassembly byte cap (`maxChunkedSyncEnvelopeBytes`) and drops chunk sets with inconsistent `total`s so a malformed or oversized stream cannot grow phone memory unbounded | @@ -1997,6 +1997,13 @@ different machine's cached limits. which caps retained events at `chatEventHistoryMaxEvents = 1_000` (up from the previous 500-event cap) so very long chats don't evict their own recent turns on reconnect. +- **The capped live-event ring advances from the previous tail.** + `WorkLiveTranscriptCache` treats the previously rendered tail envelope as + the continuity anchor and maps only the newer suffix when the 1,000-event + ring slides. Count/head equality is not a continuity check once the ring is + full. A missing tail or out-of-order replacement forces a rebuild; ordinary + one-event slides must remain O(delta) so old streaming text is not replayed + and duplicated on every tick. - **Chat-event snapshot decode is element-lossy, not all-or-nothing.** The `events` array on every chat snapshot payload (`AgentChatEventHistorySnapshot`, `SyncChatSubscribeSnapshotPayload`, @@ -2094,6 +2101,13 @@ different machine's cached limits. a calm "N agents stopped when you interrupted" line that expands to a per-agent list, and tapping a row reopens that subagent's detail. - **Long Work chats must keep row work and root polling cheap.** The + destination seeds its render-ready presentation cache synchronously before + the first SwiftUI body; the async opening load fills only missing state unless + a force-fresh open was requested, and “Session unavailable” appears only + after that authoritative load completes. Cached mapped transcript rows and + canonical fallback rows are mutually exclusive, avoiding duplicate retained + arrays. Work-list rows render their preview and activity timestamp from the + equatable render signature instead of recomputing them in `body`. The Work chat detail keeps the full timeline snapshot preview-free, then attaches cached initial assistant-message previews only to the visible presentation rows. That avoids splitting or line-counting huge hidden