import Foundation import HostRegistry import SessionCore /// T-iOS-15 · Third fan-out branch: session-level side effects that belong to /// neither terminal nor gate UI. /// /// - `.adopted` → remember the SERVER-issued id (always adopt it — attaching /// an unknown UUID yields a brand-new session) and persist it per host via /// `LastSessionStore` (TerminalViewModel's doc assigns this persistence to /// the T-iOS-15 wiring). This id is also what a background→foreground /// rebuild re-attaches to. /// - `.gate` → forward pending (gate != nil) into /// `SessionListViewModel.setPendingApproval` — the ⚠ overlay hook the list /// VM documents ("The T-iOS-15 wiring forwards gate/status events into /// setPendingApproval"). Gates arriving before adoption have no session to /// attribute and are dropped (cannot happen on the real wire: `attached` is /// always first — defensive only). /// - `.exited` → clear the ⚠ overlay AND the persisted lastSessionId: a dead /// session must not stay flagged nor be offered as "继续上次" (documented /// decision — otherwise cold start would silently spawn a NEW session via /// the unknown-UUID attach path). /// /// A detach (stream finish without exit) deliberately KEEPS the ⚠ overlay: /// the server still holds the gate; the list poll prunes it if the session /// disappears. @MainActor final class SessionActivityBridge { private(set) var adoptedSessionId: UUID? private let events: AsyncStream private let hostId: UUID private let lastSessionStore: any LastSessionStore private let onPendingChanged: @MainActor (UUID, Bool) -> Void private var consumeTask: Task? // Deterministic test barrier (same pattern as the W3 ViewModels). private(set) var processedEventCount = 0 private var eventWaiters: [CountWaiter] = [] private struct CountWaiter { let target: Int let continuation: CheckedContinuation } init( events: AsyncStream, hostId: UUID, lastSessionStore: any LastSessionStore, onPendingChanged: @escaping @MainActor (UUID, Bool) -> Void ) { self.events = events self.hostId = hostId self.lastSessionStore = lastSessionStore self.onPendingChanged = onPendingChanged } /// Begin consuming. Idempotent — the branch has exactly one consumer. func start() { guard consumeTask == nil else { return } consumeTask = Task { [weak self] in guard let events = self?.events else { return } for await event in events { guard let self else { return } self.apply(event) } self?.releaseAllWaiters() } } func stop() { consumeTask?.cancel() consumeTask = nil releaseAllWaiters() } // MARK: - Event application private func apply(_ event: SessionEvent) { switch event { case .adopted(let sessionId): adoptedSessionId = sessionId lastSessionStore.setLastSessionId(sessionId, host: hostId) case .gate(let gate): forwardPending(gate != nil) case .exited: forwardPending(false) lastSessionStore.setLastSessionId(nil, host: hostId) case .connection, .output, .telemetry, .digest: break // terminal / gate UI domain } processedEventCount += 1 resumeEventWaiters() } private func forwardPending(_ pending: Bool) { guard let adoptedSessionId else { return } onPendingChanged(adoptedSessionId, pending) } // MARK: - Test barrier /// Suspends until at least `eventCount` events have been applied. func waitUntilProcessed(eventCount target: Int) async { await withCheckedContinuation { continuation in guard processedEventCount < target else { continuation.resume() return } eventWaiters = eventWaiters + [CountWaiter(target: target, continuation: continuation)] } } private func resumeEventWaiters() { let satisfied = eventWaiters.filter { $0.target <= processedEventCount } eventWaiters = eventWaiters.filter { $0.target > processedEventCount } for waiter in satisfied { waiter.continuation.resume() } } private func releaseAllWaiters() { let all = eventWaiters eventWaiters = [] for waiter in all { waiter.continuation.resume() } } }