fix(swift-ios): retain thread subscriptions across navigation

apps/swift-ios/Core/T3Client.swift

@@ -638,6 +638,29 @@ public actor T3Client {
         )
     }
 
+    /// Passive shells start with a full snapshot on each socket subscription.
+    /// Registering its identity atomically prevents late socket results from
+    /// acquiring authority over a replacement connection.
+    public func shellEventsOnCurrentConnection() async throws -> (
+        events: AsyncThrowingStream<ShellStreamItem, Error>, connectionID: UUID
+    ) {
+        try await rpc.subscribeOnCurrentConnection(
+            RPCMethod.subscribeShell.rawValue,
+            payload: .object(["requestCompletionMarker": .bool(true)]),
+            as: ShellStreamItem.self
+        )
+    }
+
+    public func shellEventBatchesOnCurrentConnection() async throws -> (
+        events: AsyncThrowingStream<[ShellStreamItem], Error>, connectionID: UUID
+    ) {
+        try await rpc.subscribeBatchesOnCurrentConnection(
+            RPCMethod.subscribeShell.rawValue,
+            payload: .object(["requestCompletionMarker": .bool(true)]),
+            as: ShellStreamItem.self
+        )
+    }
+
     public func threadEventBatches(
         threadID: String,
         after sequence: Int? = nil,

apps/swift-ios/Features/Chat/ThreadDetailView.swift

@@ -14,6 +14,7 @@ public struct ThreadDetailView: View {
     let submitMessage: (FeatureMessageSubmission) async -> Bool
     let onNavigateBack: () -> Void
     private let draftStore: FeatureComposerDraftStore
+    private let managesThreadPresentation: Bool
 
     @State private var draft = ""
     @State private var composerContext: OrchestrationMessageContext?
@@ -52,20 +53,22 @@ public struct ThreadDetailView: View {
         thread: FeatureThread,
         submitMessage: @escaping (FeatureMessageSubmission) async -> Bool,
         onNavigateBack: @escaping () -> Void = {},
-        draftStore: FeatureComposerDraftStore = .shared
+        draftStore: FeatureComposerDraftStore = .shared,
+        managesThreadPresentation: Bool = true
     ) {
         self.model = model
         self.thread = thread
         self.submitMessage = submitMessage
         self.onNavigateBack = onNavigateBack
         self.draftStore = draftStore
+        self.managesThreadPresentation = managesThreadPresentation
     }
 
     private var threadContent: some View {
         Group {
             if let detail {
                 timeline(detail)
-            } else if isLoading {
+            } else if isOpening {
                 FeatureThreadOpeningView()
             } else {
                 ContentUnavailableView {
@@ -102,16 +105,27 @@ public struct ThreadDetailView: View {
             }
         }
         .task(id: thread.id) {
+            guard managesThreadPresentation else { return }
             isLoading = true
-            _ = await model.detail(for: thread.id, force: true)
-            isLoading = false
+            await model.runThreadPresentation(id: thread.id) {
+                isLoading = false
+            }
         }
         .task(id: thread.id) {
             // A cached thread can already show its composer while the server
             // is catching up. Local drafts must not wait for that request.
-            await model.checkRewindRecovery(for: currentThread)
-            guard !didRestoreDraft else { return }
-            await restoreDraft(from: composerDraft, key: draftKey)
+            // When a thread is reopened in the compact split view, SwiftUI can
+            // report this view as disappeared right after it appears while it
+            // stays on screen. That cancels this task and it never re-runs, so
+            // the composer would stay busy. Finish the restore outside the
+            // view's task lifetime; each thread has its own view identity, so
+            // a late restore cannot reach another thread's composer.
+            let restore = Task { @MainActor in
+                await model.checkRewindRecovery(for: currentThread)
+                guard !didRestoreDraft else { return }
+                await restoreDraft(from: composerDraft, key: draftKey)
+            }
+            await restore.value
         }
         .task(id: pullRequestObservationID) {
             await observeThreadPullRequest()
@@ -133,7 +147,7 @@ public struct ThreadDetailView: View {
         .onChange(of: threadConnectionState) { _, state in
             if state == .connected,
                case .failed = model.detailLoadStates[thread.id],
-               !isLoading {
+               !isOpening {
                 reloadThread()
             }
         }
@@ -143,7 +157,6 @@ public struct ThreadDetailView: View {
             }
         }
         .onDisappear {
-            model.releaseThread(thread.id)
             persistDraftBeforeLeaving()
         }
         .sheet(item: $toolSurface) { surface in
@@ -277,6 +290,12 @@ public struct ThreadDetailView: View {
         }
     }
 
+    private var isOpening: Bool {
+        if managesThreadPresentation { return isLoading }
+        if case .failed = model.detailLoadStates[thread.id] { return false }
+        return detail == nil || model.detailLoadStates[thread.id] == .loading
+    }
+
     private var detail: FeatureThreadDetail? {
         model.details[thread.id]
     }
@@ -683,10 +702,8 @@ public struct ThreadDetailView: View {
     }
 
     private func reloadThread() {
-        isLoading = true
-        Task {
-            _ = await model.detail(for: thread.id, force: true, fresh: true)
-            isLoading = false
+        if model.refreshThreadPresentation(id: thread.id, onLoaded: { isLoading = false }) {
+            isLoading = true
         }
     }
 
@@ -699,7 +716,7 @@ public struct ThreadDetailView: View {
         ThreadRefreshPresentation.resolve(
             loadState: model.detailLoadStates[thread.id],
             connectionState: threadConnectionState,
-            isOpening: isLoading,
+            isOpening: isOpening,
             syncState: model.threadSyncStates[thread.id]
         )
     }

apps/swift-ios/Features/Root/FeatureRootModel.swift

@@ -90,11 +90,19 @@ public final class FeatureRootModel {
         draftStore: draftStore
     )
     private var pendingSubmissionsByID: [String: FeatureQueuedSubmission] = [:]
+    /// Accepted messages not yet present in a server transcript, by message ID.
+    private var deliveredAwaitingDetail: [String: (threadID: String, message: FeatureMessage)] = [:]
+    /// Message IDs in each thread's latest server transcript.
+    private var serverMessageIDs: [String: Set<String>] = [:]
     private var activeSubmissionCounts: [String: Int] = [:]
     private var pendingThreadsByID: [String: FeatureThread] = [:]
     private var pendingSettlementMutations: [String: PendingSettlementMutation] = [:]
     private var pendingCompletionSubmissionIDs: Set<String> = []
     private var pendingDiscardSubmissionIDs: Set<String> = []
+    @ObservationIgnored private var activeDetailPresentation: (
+        threadID: String, owner: UUID, lifetime: AsyncStream<Void>.Continuation,
+        refreshTask: Task<Void, Never>?
+    )?
     private var detailRecency: [String] = []
     private var detailLoadGeneration: UInt64 = 0
     private var detailLoadRevisions: [String: UInt64] = [:]
@@ -149,6 +157,7 @@ public final class FeatureRootModel {
 
     func applicationDidEnterBackground(at date: Date = .now) {
         backgroundedAt = date
+        client.suspendForBackground()
     }
 
     func applicationDidBecomeActive(at date: Date = .now) async {
@@ -721,6 +730,55 @@ public final class FeatureRootModel {
         }
     }
 
+    /// Holds selected-thread transport for the owning presentation task, including after loading.
+    func runThreadPresentation(id: String, onLoaded: @MainActor () -> Void = {}) async {
+        guard !Task.isCancelled else { return }
+        if let previous = activeDetailPresentation {
+            releaseThread(previous.threadID)
+        }
+        let owner = UUID()
+        let lifetime = AsyncStream<Void>.makeStream()
+        activeDetailPresentation = (id, owner, lifetime.continuation, nil)
+        await withTaskCancellationHandler {
+            defer { releaseThreadPresentation(id: id, owner: owner) }
+            _ = await detail(for: id, force: true)
+            guard !Task.isCancelled,
+                  activeDetailPresentation?.owner == owner else { return }
+            onLoaded()
+            for await _ in lifetime.stream {}
+        } onCancel: {
+            // Cancellation must release even when the detail read is still suspended.
+            Task { @MainActor [weak self] in
+                self?.releaseThreadPresentation(id: id, owner: owner)
+            }
+        }
+    }
+
+    /// Retries belong to the current presentation and cannot outlive a close or replacement.
+    @discardableResult
+    func refreshThreadPresentation(
+        id: String, fresh: Bool = true, onLoaded: @escaping @MainActor () -> Void = {}
+    ) -> Bool {
+        guard let presentation = activeDetailPresentation, presentation.threadID == id else { return false }
+        presentation.refreshTask?.cancel()
+        let owner = presentation.owner
+        activeDetailPresentation?.refreshTask = Task { [weak self] in
+            guard let self, !Task.isCancelled,
+                  self.activeDetailPresentation?.owner == owner else { return }
+            _ = await self.detail(for: id, force: true, fresh: fresh)
+            guard !Task.isCancelled,
+                  self.activeDetailPresentation?.owner == owner else { return }
+            onLoaded()
+        }
+        return true
+    }
+
+    private func releaseThreadPresentation(id: String, owner: UUID) {
+        guard activeDetailPresentation?.threadID == id,
+              activeDetailPresentation?.owner == owner else { return }
+        releaseThread(id)
+    }
+
     public func detail(for id: String, force: Bool = false, fresh: Bool = false) async -> FeatureThreadDetail? {
         if !force, let cached = details[id] {
             return cached
@@ -799,6 +857,11 @@ public final class FeatureRootModel {
 
     /// Ends any selected-thread transport work when its detail view closes.
     public func releaseThread(_ id: String) {
+        if let presentation = activeDetailPresentation, presentation.threadID == id {
+            activeDetailPresentation = nil
+            presentation.refreshTask?.cancel()
+            presentation.lifetime.finish()
+        }
         client.releaseThread(id: id)
         markDetailRecentlyUsed(id)
         evictOldThreadDetailsIfNeeded()
@@ -1317,6 +1380,7 @@ public final class FeatureRootModel {
     }
 
     private func removeThread(id: String) {
+        forgetDeliveredMessages(threadID: id)
         guard let index = snapshot.threads.firstIndex(where: { $0.id == id }) else { return }
         let projectID = snapshot.threads[index].projectID
         snapshot.threads.remove(at: index)
@@ -1449,15 +1513,18 @@ public final class FeatureRootModel {
         }
     }
 
+    /// `isServerTranscript` is false for details built locally, such as a
+    /// restored outbox or a pending creation; they cannot confirm delivery.
     private func store(
         _ incoming: FeatureThreadDetail,
-        invalidatesInFlightLoad: Bool = true
+        invalidatesInFlightLoad: Bool = true,
+        isServerTranscript: Bool = true
     ) {
         var incoming = retainingLocalAttachmentPreviews(in: incoming)
         incoming.thread = retainingPendingSettlement(in: incoming.thread)
         let id = incoming.thread.id
         acknowledgeDeliveredMessages(incoming.messages)
-        let prepared = addingPendingMessages(to: incoming)
+        let prepared = addingPendingMessages(to: incoming, isServerTranscript: isServerTranscript)
         let next = details[id].map { current in
             FeatureThreadDetail(
                 thread: prepared.thread,
@@ -1484,7 +1551,7 @@ public final class FeatureRootModel {
         incoming.thread = retainingPendingSettlement(in: incoming.thread)
         let id = incoming.thread.id
         acknowledgeDeliveredMessages(incoming.messages)
-        let next = addingPendingMessages(to: incoming)
+        let next = addingPendingMessages(to: incoming, isServerTranscript: true)
         details[id] = next
         markDetailRecentlyUsed(id)
         bumpDetailLoadRevision(id: id)
@@ -1523,7 +1590,18 @@ public final class FeatureRootModel {
         return true
     }
 
-    private func removeDetail(id: String) {
+    private func forgetDeliveredMessages(threadID: String) {
+        deliveredAwaitingDetail = deliveredAwaitingDetail.filter { $0.value.threadID != threadID }
+        serverMessageIDs[threadID] = nil
+    }
+
+    private func removeDetail(id: String, forgettingDeliveredMessages: Bool = true) {
+        if forgettingDeliveredMessages {
+            forgetDeliveredMessages(threadID: id)
+        } else {
+            // The next server transcript records these again; keep the cache bounded.
+            serverMessageIDs[id] = nil
+        }
         if details.removeValue(forKey: id) != nil {
             detailRecency.removeAll { $0 == id }
         }
@@ -1536,6 +1614,8 @@ public final class FeatureRootModel {
     }
 
     private func clearDetails() {
+        deliveredAwaitingDetail.removeAll()
+        serverMessageIDs.removeAll()
         detailLoadGeneration &+= 1
         detailLoadRevisions.removeAll()
         storedDetailLoadRequestRevisions.removeAll()
@@ -1571,7 +1651,8 @@ public final class FeatureRootModel {
         while details.count > Self.maximumRetainedThreadDetails,
               let candidate = detailRecency.first(where: { !protected.contains($0) }) {
             detailRecency.removeAll { $0 == candidate }
-            removeDetail(id: candidate)
+            // Eviction only drops the cache; accepted messages must survive a reopen.
+            removeDetail(id: candidate, forgettingDeliveredMessages: false)
         }
     }
 
@@ -1614,7 +1695,7 @@ public final class FeatureRootModel {
                 if snapshot.threads.contains(where: { $0.id == submission.threadID }) {
                     pendingSubmissionsByID[submission.id] = submission
                     if let detail = details[submission.threadID] {
-                        store(addingPendingMessages(to: detail))
+                        store(detail, isServerTranscript: false)
                     }
                     continue
                 }
@@ -1645,7 +1726,7 @@ public final class FeatureRootModel {
             }
             pendingSubmissionsByID[submission.id] = submission
             if let detail = details[submission.threadID] {
-                store(addingPendingMessages(to: detail))
+                store(detail, isServerTranscript: false)
             }
         }
     }
@@ -1708,7 +1789,7 @@ public final class FeatureRootModel {
         store(FeatureThreadDetail(
             thread: thread,
             messages: [queuedMessage(for: submission)]
-        ))
+        ), isServerTranscript: false)
     }
 
     private func provider(id: String?, environmentID: String) -> FeatureProvider? {
@@ -1737,16 +1818,30 @@ public final class FeatureRootModel {
         )
     }
 
-    private func addingPendingMessages(to incoming: FeatureThreadDetail) -> FeatureThreadDetail {
+    private func addingPendingMessages(
+        to incoming: FeatureThreadDetail,
+        isServerTranscript: Bool
+    ) -> FeatureThreadDetail {
+        let existing = Set(incoming.messages.map(\.id))
+        // A delivered message stays visible until a server transcript includes it.
+        // Its refresh no longer blocks the send, so an older read can arrive first.
+        if isServerTranscript {
+            serverMessageIDs[incoming.thread.id] = existing
+            for (id, delivered) in deliveredAwaitingDetail
+            where delivered.threadID == incoming.thread.id && existing.contains(id) {
+                deliveredAwaitingDetail.removeValue(forKey: id)
+            }
+        }
+        let delivered = deliveredAwaitingDetail.values
+            .filter { $0.threadID == incoming.thread.id && !existing.contains($0.message.id) }
+            .map(\.message)
         let queued = pendingSubmissionsByID.values
-            .filter { $0.threadID == incoming.thread.id }
-            .sorted { $0.identity.createdAt < $1.identity.createdAt }
-        guard !queued.isEmpty else { return incoming }
+            .filter { $0.threadID == incoming.thread.id && !existing.contains($0.identity.messageID) }
+            .map(queuedMessage(for:))
+        guard !queued.isEmpty || !delivered.isEmpty else { return incoming }
         var result = incoming
-        let existing = Set(result.messages.map(\.id))
-        result.messages.append(contentsOf: queued.lazy
-            .filter { !existing.contains($0.identity.messageID) }
-            .map(queuedMessage(for:)))
+        // A newer send can be delivered while an older one waits to retry; keep send order.
+        result.messages.append(contentsOf: (delivered + queued).sorted { $0.createdAt < $1.createdAt })
         return result
     }
 
@@ -1823,6 +1918,12 @@ public final class FeatureRootModel {
         }
         pendingCompletionSubmissionIDs.remove(submission.id)
         pendingSubmissionsByID.removeValue(forKey: submission.id)
+        // A transcript that already includes it confirmed delivery; do not retain it.
+        if serverMessageIDs[submission.threadID]?.contains(submission.identity.messageID) != true {
+            var message = queuedMessage(for: submission)
+            message.state = .complete
+            deliveredAwaitingDetail[submission.identity.messageID] = (submission.threadID, message)
+        }
         setAttachmentOutboxOwnership(false, for: submission)
         pendingThreadsByID.removeValue(forKey: submission.threadID)
         markQueuedMessageDelivered(submission)