// FlowSessionManager.swift
// OSGKeyboard · Main App
//
// Session Owner for TypeWhisper-style Flow dictation: continuous
// `.playAndRecord` capture for the whole session, utterance gating for
// ASR and cloud LLM polish, with App Group result delivery.

import AVFoundation
import Foundation
import OSGKeyboardHostSupport
import OSGKeyboardShared
import Speech
import SwiftUI
import UIKit

struct FlowUtteranceStartToken: Equatable, Sendable {
    let generation: UInt64
    let utteranceId: UUID
}

enum FlowUtteranceLifecyclePolicy {
    static func canContinueStart(
        token: FlowUtteranceStartToken,
        currentGeneration: UInt64,
        currentUtteranceId: UUID?,
        terminalUtteranceIds: Set<UUID>
    ) -> Bool {
        currentGeneration == token.generation
            && currentUtteranceId == token.utteranceId
            && !terminalUtteranceIds.contains(token.utteranceId)
    }
}

enum FlowPiPActivationTrigger: String, Sendable {
    case coldStart
    case foreground
    case existingSession
    case startupRecovery
    case unexpectedStop
    case healthCheck
    case manualRetry
}

enum FlowPiPReconciliationDecision: Equatable, Sendable {
    case noSessionIntent
    case alreadyActive
    case waitForForeground
    case startRecovery
}

/// Guarantees that every user-visible session warning (the home-screen
/// "cannot start voice session" toast) leaves a report behind.
///
/// Most warning sites already write their own, better-labelled report on the
/// following lines. This gate lets the catch-all skip those and fire only for
/// the paths nobody instrumented — which is exactly where a silent toast comes
/// from.
struct FlowSessionWarningReportGate: Equatable, Sendable {
    /// An explicit report this recent is assumed to describe the same incident.
    static let duplicateGrace: TimeInterval = 1.5

    private var lastReportedWarning: String?

    mutating func shouldReport(
        warning: String,
        lastExplicitReportAt: Date?,
        now: Date
    ) -> Bool {
        guard lastReportedWarning != warning else { return false }
        lastReportedWarning = warning
        guard let lastExplicitReportAt else { return true }
        return now.timeIntervalSince(lastExplicitReportAt) >= Self.duplicateGrace
    }

    /// A cleared warning ends the episode: the same text later is a new toast.
    mutating func warningCleared() {
        lastReportedWarning = nil
    }
}

/// One startup-failure report per PiP-loss episode.
///
/// While the app is backgrounded the heartbeat re-evaluates the lost PiP every
/// second; without this gate a single incident would write dozens of identical
/// reports and evict everything else from the 40-report budget.
struct FlowPiPLossReportGate: Equatable, Sendable {
    private var reportedForCurrentEpisode = false

    mutating func shouldReport() -> Bool {
        guard !reportedForCurrentEpisode else { return false }
        reportedForCurrentEpisode = true
        return true
    }

    /// A healthy PiP closes the episode, so the next loss reports again.
    mutating func pipBecameActive() {
        reportedForCurrentEpisode = false
    }
}

enum FlowPiPRecoveryPolicy {
    static let maxAttempts = 4
    static let totalBudget: TimeInterval = 5

    static func reconciliationDecision(
        wantsActiveSession: Bool,
        isAppForeground: Bool,
        isPictureInPictureActive: Bool
    ) -> FlowPiPReconciliationDecision {
        guard wantsActiveSession else { return .noSessionIntent }
        if isPictureInPictureActive { return .alreadyActive }
        return isAppForeground ? .startRecovery : .waitForForeground
    }

    static func canContinue(
        operation: UInt64,
        currentOperation: UInt64,
        wantsActiveSession: Bool,
        isAppForeground: Bool,
        taskIsCancelled: Bool
    ) -> Bool {
        operation == currentOperation
            && wantsActiveSession
            && isAppForeground
            && !taskIsCancelled
    }

    /// Earliest start of each attempt, measured from the start of recovery.
    /// Audio activation failures return almost instantly, so gaps measured
    /// after each failure would spend every attempt within ~1 s, while a reset
    /// media server needs ~3 s before activation succeeds again.
    static func attemptOffset(beforeAttempt attempt: Int) -> TimeInterval {
        switch attempt {
        case 1: return 0
        case 2: return 0.5
        case 3: return 1.5
        default: return 3
        }
    }

    static func hostTimeout(attempt: Int, remainingBudget: TimeInterval) -> TimeInterval {
        min(attempt == 1 ? 1.5 : 0.25, max(0, remainingBudget))
    }

    static func activeTimeout(remainingBudget: TimeInterval) -> TimeInterval {
        min(1.4, max(0, remainingBudget))
    }
}

@MainActor
final class FlowSessionManager: ObservableObject {
    private struct PrefilledAIRequest {
        let question: String
        let conversationID: UUID?
        let webPageURL: URL?
        let taskKind: ManagedGatewayTaskKind
        let requestSource: ManagedGatewayRequestSource?
        let requestPurpose: ManagedGatewayRequestPurpose?
        let oobeFeature: ManagedGatewayOOBEFeature?
        let thinkingEnabled: Bool
    }

    @Published private(set) var isActive = false
    @Published private(set) var isStarting = false
    @Published private(set) var sessionExpiresAt: Date?
    /// Non-nil when continuous capture failed or permissions are missing.
    @Published private(set) var sessionWarning: String? {
        didSet {
            guard let sessionWarning else {
                sessionWarningReportGate.warningCleared()
                return
            }
            guard sessionWarning != oldValue else { return }
            // Deferred by one runloop turn on purpose: the paths that raise a
            // warning keep mutating lifecycle state — and usually write their
            // own, better-labelled report — on the following lines. Reporting
            // synchronously here would snapshot half-updated state and
            // double-count the same incident.
            Task { @MainActor [weak self] in
                self?.reportSessionWarningIfUnreported(sessionWarning)
            }
        }
    }

    private var sessionWarningReportGate = FlowSessionWarningReportGate()
    @Published private(set) var pipLifecycleState: FlowPiPLifecycleState = .inactive {
        didSet {
            // A healthy PiP closes the current loss episode, so the next loss
            // reports again instead of being deduped against a stale one.
            if pipLifecycleState == .active { pipLossReportGate.pipBecameActive() }
        }
    }

    private var pipLossReportGate = FlowPiPLossReportGate()

    private let capture = FlowContinuousCapture()
    private let pipController: any FlowPictureInPictureControlling
    private let store = AppGroupStore()
    private let analyticsClient: any AnalyticsClient
    /// Cloud / local polish via the user-configured LLM provider.
    private let polisher: PolishingService
    /// AI-mode turns are intentionally process-local and never persisted.
    private let aiConversations = AIConversationStore()
    private let webPageContentService: WebPageContentService
    /// Cached ASR instance. The only on-device backend is iOS
    /// `SpeechAnalyzer`, which has no warm-up step — we can hand the
    /// factory-built service straight back without going through the
    /// old `OnDeviceModelWarmup` registry.
    private var sessionASR: ASRService?
    /// Tracks which engine mode `sessionASR` was created for.
    private var sessionASREngineMode: String?
    /// Locale id last passed to `warmup(locale:)`.
    private var sessionASRWarmedLocaleID: String?
    private var asr: ASRService {
        if let sessionASR { return sessionASR }
        let service = ASRServiceFactory.make(store: store)
        sessionASR = service
        return service
    }

    private var pollingTask: Task<Void, Never>?
    private var heartbeatTask: Task<Void, Never>?
    private var levelTask: Task<Void, Never>?
    private var startTask: Task<Void, Never>?
    private var audioPrimeTask: Task<Bool, Never>?
    private var audioPrimeID: UUID?
    private var audioPrimeCancellationRequested = false
    private var startupAudioHealthTask: Task<Void, Never>?
    private var didRunStartupAudioHealthCheck = false
    private var commandObserver: FlowSessionDarwinObserver?
    /// Alive for the whole process, not just an active session: the keyboard
    /// asks for a companion dump exactly when no session exists.
    private var diagnosticsDumpObserver: FlowSessionDarwinObserver?
    /// Last recording state the poll loop observed — logs only on transition.
    private var lastObservedRecordingState: FlowSessionKeys.RecordingState = .idle
    private var activeSessionId: UUID?
    private var currentUtteranceId: UUID?
    /// Claimed synchronously before any async capture work begins.
    private var startingUtteranceId: UUID?
    private var startTransactionDeadlineAt: TimeInterval?
    /// Monotonic token captured by every async worker for one utterance.
    private var utteranceGeneration: UInt64 = 0
    /// Terminal delivery is idempotent: only the first path may write a result.
    private var terminalUtteranceIds: Set<UUID> = []
    /// Cursor context captured by the keyboard at the final insertion point.
    private var pendingFieldContext: FlowFieldContext?
    /// Dictation vs explicit last-input editing for the live utterance.
    private var currentUtteranceMode: FlowUtteranceMode = .dictation
    private var pendingEditSourceText: String?
    private var pendingSourceHistoryEntryID: UUID?
    private var pendingSourceHistoryEntryRevision: Int64?
    private var pendingAIConversationID: UUID?
    private var pendingManagedRequestPurpose: ManagedGatewayRequestPurpose?
    private var pendingManagedOOBEFeature: ManagedGatewayOOBEFeature?
    private var pendingProcessingDeadlineAt: TimeInterval?
    private var pendingStopUtteranceId: UUID?
    private var currentCommandSeq: Int64 = 0
    private var lastHandledCommandSeq: Int64 = 0
    /// Throttles AI-mode LLM draft writes into App Group.
    private var aiAnswerStreamThrottle = AIAnswerStreamThrottle()
    /// Published so Home / debug UI can show "recording" instead of a false "ready".
    @Published private(set) var isUtteranceRecording = false
    /// True from `stopped` until the result/error is written back to App Group.
    @Published private(set) var isUtteranceProcessing = false
    private var finalizeTask: Task<Void, Never>?
    private var asrTask: Task<Void, Never>?
    private let analyticsOperations = FlowAnalyticsOperationRegistry()
    private var utteranceSafetyTask: Task<Void, Never>?
    private var chunkedPipeline: ChunkedUtterancePipeline?
    private var currentPartial = ""
    private var lastFinal = ""
    private var lastFinalWithPauseMarks = ""
    /// Partial stitched text captured when the user stops recording.
    private var bestPartialSnapshot = ""
    /// Full utterance PCM for batch ASR fallback after pipelined chunking.
    private var utterancePCMSamples: [Float] = []
    private var chunkWarnings: [String] = []
    private var asrCompletedGeneration: UInt64?
    private var asrFailureMessage: String?
    private var lastReadyTraceSignature = ""
    private var lastCommandFingerprint = ""
    private var lastIgnoredCommandSignature = ""
    /// Wall-clock span of the current mic-open utterance (excludes LLM polish).
    private var utteranceRecordingStartedAt: Date?
    /// True while the host app scene is `.active` — drives foreground renewal.
    private var isAppForeground = false
    /// True while handling a keyboard-initiated `startflow` cold start.
    private var isColdStartHandoff = false
    /// User intent outlives a transient PiP loss; only explicit teardown clears it.
    private var wantsActiveSession = false
    /// Invalidates late async completions after retry, stop, or process teardown.
    private var pipRecoveryOperation: UInt64 = 0
    var shouldDeferHostHeavyWork: Bool {
        isStarting
            || startingUtteranceId != nil
            || isUtteranceRecording
            || isUtteranceProcessing
            || hasUnacknowledgedTerminalResult()
    }

    func attachPiPHostView(_ view: UIView) {
        pipController.attachHostView(view)
    }

    /// Guards the once-per-process launch reconciliation (scene reconnects
    /// recreate the `@StateObject`-owned manager within the same process).
    private static var didRunLaunchReconciliation = false

    init(
        analyticsClient: any AnalyticsClient = NoopAnalyticsClient(),
        pipController: any FlowPictureInPictureControlling = FlowPictureInPictureController(),
        webPageContentService: WebPageContentService = .shared
    ) {
        self.analyticsClient = analyticsClient
        self.polisher = PolishingService(analyticsClient: analyticsClient)
        self.pipController = pipController
        self.webPageContentService = webPageContentService
        // Sessions are (re)started explicitly on app foreground via
        // `activateOnForeground()`. We deliberately do NOT silently reattach a
        // stored session here — after a force-quit that would resurrect capture
        // without the user re-opening.
        //
        // Launch reconciliation: a brand-new process can never own an
        // in-flight session, so whatever the previous generation persisted
        // (force-quit skips `applicationWillTerminate` entirely when the app
        // was suspended) is void. Rotating the generation token also lets the
        // keyboard invalidate stale ready snapshots instantly instead of
        // waiting out the 60 s heartbeat-zombie window.
        //
        // Once per PROCESS, not per manager: iOS can disconnect and later
        // reconnect the sole scene without killing the process, which
        // recreates the `@StateObject` (and thus this init). Re-rotating then
        // would wipe live state that belongs to this very process.
        if AppGroup.isAvailable, !Self.didRunLaunchReconciliation {
            Self.didRunLaunchReconciliation = true
            let previous = FlowSessionBridge.rotateHostGeneration()
            if previous != nil || FlowSessionBridge.isSessionActive() {
                FlowSessionBridge.clearFlowStateOnHostLaunch()
                FlowSessionDarwin.postSessionChanged()
                debug("launch reconciliation: voided previous-generation Flow state")
            }
        }

        capture.onEngineLiveChanged = { [weak self] live in
            guard let self else { return }
            self.refreshHostReady()
            if !live, self.isUtteranceRecording {
                self.failUtterance(
                    message: AppL10n.string("flow.error.audioUnavailable"),
                    kind: .audioUnavailable
                )
            }
        }
        // A system interruption (call / Siri) stops audio frames mid-utterance;
        // fail fast so the user is not silently recording into a gap.
        capture.onInterruptionBegan = { [weak self] in
            guard let self, self.isUtteranceRecording else { return }
            self.failUtterance(
                message: AppL10n.string("flow.error.recognitionInterrupted"),
                kind: .recognitionInterrupted
            )
        }
        pipController.onUnexpectedStop = { [weak self] in
            guard let self, self.wantsActiveSession else { return }
            self.handleUnexpectedPiPStop()
        }
        FlowTerminationCoordinator.register(self)
        startDiagnosticsDumpObserver()
    }

    /// The keyboard extension is the process that witnesses "the session never
    /// came up", but only the host knows why it never published ready. When the
    /// keyboard writes a terminal start-failure report it pings us here so the
    /// host's own breadcrumb window lands next to it in the same export.
    private func startDiagnosticsDumpObserver() {
        diagnosticsDumpObserver = FlowSessionDarwinObserver(
            notificationName: FlowSessionDarwin.diagnosticsDumpNotificationName
        ) { [weak self] in
            guard let self else { return }
            guard let request = FlowFailureDiagnostics.consumeHostSnapshotRequest() else {
                return
            }
            FlowDiagnostics.persistStartupFailure(
                reason: "keyboardReported:\(request.reason)",
                context: self.startupFailureContext(
                    trigger: "keyboardDumpRequest",
                    attemptCount: 0,
                    elapsed: Date().timeIntervalSince(request.requestedAt)
                )
            )
        }
    }

    // MARK: - Public

    /// Starts a Flow session: permissions → continuous capture → App Group active.
    func startSession(
        duration: TimeInterval? = nil,
        coldStart: Bool = false,
        reason: String = "unspecified"
    ) {
        OSGDiag.log(
            "startSession.request reason=\(reason) coldStart=\(coldStart) "
                + "active=\(isActive) starting=\(isStarting) \(OSGDiag.memoryTag())",
            category: "flow"
        )
        traceState(
            "startSession.request",
            extra: "coldStart=\(coldStart) reason=\(reason)"
        )
        guard AppGroup.isAvailable else {
            debug("cannot start flow session: App Group unavailable")
            FlowDiagnostics.persistStartupFailure(
                reason: "appGroupUnavailable",
                context: startupFailureContext(
                    trigger: reason,
                    attemptCount: 0,
                    elapsed: 0
                )
            )
            return
        }

        reconcilePersistedFlowStateBeforeStart()

        // Healthy / busy sessions must not flash the cold-start overlay when a
        // spurious `startflow` arrives (e.g. keyboard finalize race).
        if coldStart, isActive {
            extendSession(duration: duration)
            refreshHostReady()
            let decision = FlowHandoffPolicy.coldStartOverlayDecision(
                sessionIsActive: true,
                hostIsReady: FlowSessionBridge.isHostReady(),
                isUtteranceBusy: isUtteranceRecording || isUtteranceProcessing
            )
            switch decision {
            case .silence:
                traceState(
                    "startSession.coldStart.silenced",
                    extra: FlowSessionBridge.isHostReady() ? "reason=alreadyReady" : "reason=busy"
                )
                return
            case .present:
                isColdStartHandoff = true
                Task { @MainActor [weak self] in
                    await self?.prepareExistingSessionForColdStartReturn()
                }
                return
            }
        }

        if coldStart {
            isColdStartHandoff = true
        }

        if isActive {
            wantsActiveSession = true
            extendSession(duration: duration)
            if pipController.isPictureInPictureActive {
                pipLifecycleState = .active
                refreshHostReady()
            } else {
                ensurePiPActive(
                    trigger: coldStart ? .coldStart : .existingSession,
                    duration: duration
                )
            }
            return
        }

        wantsActiveSession = true
        ensurePiPActive(
            trigger: coldStart ? .coldStart : .foreground,
            duration: duration
        )
    }

    /// Clears App Group Flow state left behind when the host process was killed
    /// or the device rebooted while the session flag was still set.
    private func reconcilePersistedFlowStateBeforeStart() {
        switch FlowOrphanRecordingReconciler.decide(
            isHostStale: FlowSessionBridge.isHostStale(),
            isActive: isActive,
            recordingState: FlowSessionBridge.recordingState()
        ) {
        case .clearZombieSession:
            if isActive {
                endSession()
            } else {
                FlowSessionBridge.clearFlowState()
            }
            debug("reconciled zombie persisted Flow state")
        case .clearOrphanedRecording(let orphaned):
            FlowSessionBridge.setRecordingState(.idle)
            FlowSessionBridge.clearPendingTranscription()
            debug("cleared orphaned keyboard recording state: \(orphaned.rawValue)")
        case .none:
            break
        }
    }

    /// Foreground entry for Flow.
    ///
    /// Default is **light** for audio work: automatically arm the low-profile
    /// PiP, but do not
    /// start capture or ASR. The transparent PiP is intentionally cheap; the
    /// 170–220 MB capture/model path still starts only on explicit speech.
    ///
    /// Pass `startCapture: true` for explicit user intent (Home “Start”, etc.).
    /// Keyboard mic cold-start uses `osgkeyboard://startflow` → `startSession`.
    func activateOnForeground(
        reason: String = "unspecified",
        startCapture: Bool = false
    ) {
        FlowDiagnostics.log(
            "activateOnForeground reason=\(reason) startCapture=\(startCapture) "
                + "onboardingDone=\(ProviderConfig.shared.hasCompletedOnboarding) "
                + "\(OSGDiag.memoryTag())"
        )
        guard AppGroup.isAvailable else {
            FlowDiagnostics.log("activateOnForeground aborted reason=appGroupUnavailable")
            FlowDiagnostics.persistStartupFailure(
                reason: "appGroupUnavailable",
                context: startupFailureContext(
                    trigger: "activateOnForeground:\(reason)",
                    attemptCount: 0,
                    elapsed: 0
                )
            )
            return
        }
        guard AppPermissions.flowRequirementsMet else {
            FlowDiagnostics.log("activateOnForeground aborted reason=permissions")
            sessionWarning = permissionWarningMessage()
            FlowSessionBridge.setHostReady(false)
            FlowDiagnostics.persistStartupFailure(
                reason: "permissionsMissing",
                context: startupFailureContext(
                    trigger: "activateOnForeground:\(reason)",
                    attemptCount: 0,
                    elapsed: 0
                )
            )
            return
        }

        startSession(
            reason: !startCapture
                ? "activateOnForeground.autoPiP:\(reason)"
                : "activateOnForeground:\(reason)"
        )
    }

    /// User-requested recovery always discards the rejected AVKit generation.
    func retryPiPRecovery() {
        guard AppGroup.isAvailable else {
            FlowDiagnostics.log("manual PiP recovery aborted reason=appGroupUnavailable")
            FlowDiagnostics.persistStartupFailure(
                reason: "appGroupUnavailable",
                context: startupFailureContext(
                    trigger: FlowPiPActivationTrigger.manualRetry.rawValue,
                    attemptCount: 0,
                    elapsed: 0
                )
            )
            return
        }
        guard AppPermissions.flowRequirementsMet else {
            sessionWarning = permissionWarningMessage()
            FlowSessionBridge.setHostReady(false)
            FlowDiagnostics.log("manual PiP recovery aborted reason=permissions")
            FlowDiagnostics.persistStartupFailure(
                reason: "permissionsMissing",
                context: startupFailureContext(
                    trigger: FlowPiPActivationTrigger.manualRetry.rawValue,
                    attemptCount: 0,
                    elapsed: 0
                )
            )
            return
        }
        wantsActiveSession = true
        ensurePiPActive(trigger: .manualRetry, duration: nil, forceReset: true)
    }

    private func completeColdStartHandoff() {
        isColdStartHandoff = false
        if isActive {
            refreshHostReady()
        }
    }

    /// 强杀专用同步 teardown：不等待 ASR/LLM。
    func prepareForProcessTermination() {
        debug("prepareForProcessTermination")
        if isUtteranceRecording || isUtteranceProcessing,
           claimTerminal(utteranceId: currentUtteranceId) {
            storeCurrentError(
                AppL10n.string("flow.error.recognitionInterrupted"),
                kind: .recognitionInterrupted,
                status: .aborted
            )
        }
        analyticsOperations.cancelAll()

        wantsActiveSession = false
        pipRecoveryOperation &+= 1
        pipLifecycleState = .inactive
        isColdStartHandoff = false
        startTask?.cancel()
        startTask = nil
        startupAudioHealthTask?.cancel()
        startupAudioHealthTask = nil
        audioPrimeTask?.cancel()
        audioPrimeTask = nil
        audioPrimeID = nil
        audioPrimeCancellationRequested = false
        commandObserver = nil
        pollingTask?.cancel()
        pollingTask = nil
        heartbeatTask?.cancel()
        heartbeatTask = nil
        levelTask?.cancel()
        levelTask = nil
        finalizeTask?.cancel()
        finalizeTask = nil
        utteranceSafetyTask?.cancel()
        utteranceSafetyTask = nil
        asrTask?.cancel()
        asrTask = nil
        chunkedPipeline = nil

        if isUtteranceRecording || isUtteranceProcessing {
            capture.cancelUtterance()
            // `asr` is a computed property that ALLOCATES a fresh ASRService
            // when `sessionASR` is nil — never do that inside the ~5 s
            // termination window; only cancel an instance that exists.
            sessionASR?.cancel()
        }

        capture.cancelUtterance()
        if capture.running {
            capture.stop()
        }
        pipController.stop()

        ScreenWakeLock.release()

        sessionASR?.cancel()
        sessionASR = nil
        sessionASREngineMode = nil
        sessionASRWarmedLocaleID = nil

        if isActive || FlowSessionBridge.isSessionActive() {
            FlowSessionBridge.markSessionInactive()
            FlowSessionDarwin.postSessionChanged()
        }

        activeSessionId = nil
        currentUtteranceId = nil
        pendingFieldContext = nil
        currentCommandSeq = 0
        lastHandledCommandSeq = 0
        isUtteranceRecording = false
        isUtteranceProcessing = false
        isActive = false
        isStarting = false
        sessionExpiresAt = nil
        sessionWarning = nil
        currentPartial = ""
        lastFinal = ""
        lastFinalWithPauseMarks = ""
        bestPartialSnapshot = ""
        utterancePCMSamples = []
        chunkWarnings = []
        FlowSessionBridge.setHostReady(false)
    }

    func endSession() {
        guard isActive || isStarting || wantsActiveSession else { return }
        debug("Flow session ended")
        if isUtteranceRecording || isUtteranceProcessing,
           claimTerminal(utteranceId: currentUtteranceId) {
            storeCurrentError(
                AppL10n.string("flow.error.recognitionInterrupted"),
                kind: .recognitionInterrupted,
                status: .aborted
            )
        }
        analyticsOperations.cancelAll()

        wantsActiveSession = false
        pipRecoveryOperation &+= 1
        pipLifecycleState = .inactive
        isColdStartHandoff = false
        startTask?.cancel()
        startTask = nil
        startupAudioHealthTask?.cancel()
        startupAudioHealthTask = nil
        audioPrimeTask?.cancel()
        audioPrimeTask = nil
        audioPrimeID = nil
        audioPrimeCancellationRequested = false
        commandObserver = nil
        pollingTask?.cancel()
        pollingTask = nil
        heartbeatTask?.cancel()
        heartbeatTask = nil
        levelTask?.cancel()
        levelTask = nil
        finalizeTask?.cancel()
        finalizeTask = nil
        utteranceSafetyTask?.cancel()
        utteranceSafetyTask = nil

        if isUtteranceRecording || isUtteranceProcessing {
            capture.cancelUtterance()
            asrTask?.cancel()
            Task { await chunkedPipeline?.cancel() }
            asr.cancel()
        }
        asrTask = nil
        chunkedPipeline = nil
        activeSessionId = nil
        currentUtteranceId = nil
        pendingFieldContext = nil
        currentCommandSeq = 0
        lastHandledCommandSeq = 0
        isUtteranceRecording = false
        isUtteranceProcessing = false

        capture.stop()
        pipController.stop()
        ScreenWakeLock.release()
        sessionASR = nil
        sessionASREngineMode = nil
        sessionASRWarmedLocaleID = nil
        Task { await aiConversations.removeAll() }
        FlowSessionBridge.markSessionInactive()
        FlowSessionDarwin.postSessionChanged()
        isActive = false
        isStarting = false
        sessionExpiresAt = nil
        sessionWarning = nil
        currentPartial = ""
        lastFinal = ""
        lastFinalWithPauseMarks = ""
    }

    func extendSession(duration: TimeInterval? = nil) {
        _ = duration
        refreshHostReady()
    }

    /// Called from `OSGKeyboardApp` when `scenePhase` changes.
    func setAppForeground(_ foreground: Bool) {
        isAppForeground = foreground
    }

    /// Full scene lifecycle — keeps Flow + ASR alive across app switches.
    func handleScenePhase(_ phase: ScenePhase) {
        switch phase {
        case .active:
            setAppForeground(true)
            resumeAfterForeground()
            if wantsActiveSession, !pipController.isPictureInPictureActive {
                ensurePiPActive(trigger: .foreground, duration: nil)
            }
        case .inactive:
            setAppForeground(false)
            writeHeartbeatIfActive()
            if isActive, pipController.isPictureInPictureActive {
                Task { @MainActor [weak self] in
                    await self?.pipController.prepareForBackgroundAutoStart()
                }
            } else if wantsActiveSession {
                pausePiPRecoveryUntilForeground()
            }
        case .background:
            setAppForeground(false)
            if isActive, pipController.isPictureInPictureActive {
                Task { @MainActor [weak self] in
                    await self?.pipController.prepareForBackgroundAutoStart()
                }
            } else if wantsActiveSession {
                pausePiPRecoveryUntilForeground()
            }
            if isColdStartHandoff {
                completeColdStartHandoff()
            }
            beginBackgroundKeepAlive()
        @unknown default:
            break
        }
    }

    private func writeHeartbeatIfActive() {
        guard isActive else { return }
        FlowSessionBridge.writeHeartbeat()
    }

    private func beginBackgroundKeepAlive() {
        guard isActive else { return }
        FlowSessionBridge.writeHeartbeat()
    }

    private func handleUnexpectedPiPStop() {
        guard wantsActiveSession else { return }
        traceState("pip.unexpectedStop")
        // Root cause of the "session never comes up while I stay in another
        // app" report. Persist HERE, not later: without PiP there is nothing
        // keeping this process alive, so iOS suspends it within seconds and the
        // in-memory breadcrumb window dies with it. This is the one event that
        // explains the whole failure, and it must survive the suspension.
        reportPiPLossIfNeeded(origin: "unexpectedStop")
        pipLifecycleState = .waitingForForeground
        sessionWarning = nil
        if isUtteranceRecording {
            failUtterance(
                message: AppL10n.string("flow.error.recognitionInterrupted"),
                kind: .recognitionInterrupted
            )
        }
        FlowSessionBridge.setHostReady(false)
        refreshHostReady()
        // Do not immediately fight another app for the system-owned PiP slot.
        // The scene-active path or the existing heartbeat performs reconciliation.
    }

    private func pausePiPRecoveryUntilForeground() {
        startTask?.cancel()
        startTask = nil
        pipRecoveryOperation &+= 1
        pipController.resetGeneration()
        isStarting = false
        pipLifecycleState = .waitingForForeground
        refreshHostReady()
        reportPiPLossIfNeeded(origin: "recoveryPausedUntilForeground")
    }

    /// Catch-all: the home screen is showing a failure toast, so a report must
    /// exist for it no matter which code path raised the warning.
    private func reportSessionWarningIfUnreported(_ warning: String) {
        // Cleared or replaced while we waited — whatever the user sees now will
        // have triggered its own pass.
        guard sessionWarning == warning else { return }
        guard sessionWarningReportGate.shouldReport(
            warning: warning,
            lastExplicitReportAt: FlowFailureDiagnostics.lastReportPersistedAt(),
            now: Date()
        ) else { return }

        var context = startupFailureContext(
            trigger: "sessionWarningToast",
            attemptCount: 0,
            elapsed: 0
        )
        context["warningMessage"] = warning
        FlowDiagnostics.persistStartupFailure(
            reason: "sessionWarningShown",
            context: context
        )
    }

    /// Persists the moment this process gives up on the PiP keep-alive.
    ///
    /// Restarting PiP requires the scene to be active, so while the user stays
    /// in another app this is terminal: the session cannot come back until they
    /// open the app by hand. The keyboard cannot see the cause — it only sees
    /// `ready=false` — so the host has to record it while it still can.
    private func reportPiPLossIfNeeded(origin: String) {
        guard wantsActiveSession, pipLossReportGate.shouldReport() else { return }
        var context = startupFailureContext(
            trigger: "pipLoss:\(origin)",
            attemptCount: 0,
            elapsed: 0
        )
        context["recoverableWithoutForeground"] = isAppForeground ? "true" : "false"
        FlowDiagnostics.persistStartupFailure(
            reason: isAppForeground ? "pipLostInForeground" : "pipLostWhileBackgrounded",
            context: context
        )
    }

    private func resumeAfterForeground() {
        guard isActive else {
            return
        }

        FlowSessionBridge.writeHeartbeat()

        Task { @MainActor [weak self] in
            await self?.reactivateCaptureIfNeeded()
            // iOS `SpeechAnalyzer` is bundled with the OS; no
            // on-device weights to reload after a background trip.
            self?.bindSessionASRIfNeeded()
            // ASR warmup stays on first mic press — never on foreground bounce.
        }
    }

    private func reactivateCaptureIfNeeded() async {
        guard isActive else { return }
        // PiP releases the mic between utterances (and after drain while
        // processing). Only reassert when an utterance is actively recording
        // with capture already running — otherwise a foreground bounce was
        // cold-starting the mic mid-finalize (`!pri` / session churn).
        guard isUtteranceRecording, capture.running else {
            refreshHostReady()
            return
        }
        // A system interruption (call / Siri) may be in progress. Probe it:
        // `setActive(true)` inside `reassertIfRunning` fails while the
        // interruption is live and succeeds once it ends — which also covers
        // the documented case where iOS never delivers `.ended` (the latch
        // must not depend on that notification, or the session is dead until
        // its TTL). While the probe fails we deliberately do NOT stop or
        // rebuild: tearing the engine down would remove the observers the
        // `.ended` rebuild relies on and churn the shared session mid-call.
        if capture.isInterrupted {
            guard await capture.reassertIfRunning(), !capture.isInterrupted else { return }
        }

        if capture.running {
            let reasserted = await capture.reassertIfRunning()
            if reasserted, await capture.awaitAudioFlowing(timeout: 2) {
                sessionWarning = nil
                refreshHostReady()
                return
            }
            // The await above is a suspension point: the session may have
            // ended (expiry, user, teardown) while we waited. Never restart
            // the microphone for a session that no longer exists.
            guard isActive, !Task.isCancelled else { return }
            // Never tear capture down underneath a live utterance either — a
            // stalled route transition mid-recording must surface through the
            // utterance pipeline (safety timer / empty-transcript error), not
            // as a silent stop that truncates the take with no error at all.
            guard !isUtteranceRecording, !isUtteranceProcessing else {
                refreshHostReady()
                return
            }
            // Reassert failed, or the engine reports running yet produces no
            // frames (zombie after suspend / mediaserverd reset) — fall
            // through to a full rebuild instead of leaving it half-dead.
            capture.stop()
            debug("capture zombie after foreground — rebuilding")
        }

        do {
            try await capture.start()
            sessionWarning = nil
            debug("capture restarted after foreground")
            refreshHostReady()
        } catch {
            let message = (error as? LocalizedError)?.errorDescription ?? error.localizedDescription
            sessionWarning = message
            debug("capture restart failed: \(message)")
            refreshHostReady()
            FlowDiagnostics.persistStartupFailure(
                reason: "captureRestartFailed",
                context: startupFailureContext(
                    trigger: "resumeAfterForeground",
                    attemptCount: 0,
                    elapsed: 0
                ).merging(
                    ["captureError": Self.safeErrorLogMetadata(error)]
                ) { _, new in new }
            )
        }
    }

    /// Publish whether the keyboard can start a new utterance without jumping to the host app.
    private func refreshHostReady() {
        guard isActive else {
            // Not active yet. Distinguish "no host session at all" from
            // "received a start request and warming up" so the keyboard can
            // tell an alive-but-starting host from a dead one and hold its
            // start budget instead of timing out. When the start is parked
            // because the app is not foreground (PiP cannot arm from the
            // background), advertise `.waitingForForeground` so the keyboard
            // prompts the user to bring the app forward.
            let startupReason: FlowReadySnapshot.Reason
            if wantsActiveSession {
                startupReason = isAppForeground ? .starting : .waitingForForeground
            } else {
                startupReason = .noSession
            }
            FlowSessionBridge.writeReadySnapshot(
                FlowReadySnapshot(
                    sessionId: activeSessionId,
                    ready: false,
                    reason: startupReason,
                    heartbeatAt: Date().timeIntervalSince1970,
                    engineMode: store.engineMode,
                    localeId: store.localeId,
                    hostGeneration: FlowSessionBridge.currentHostGeneration()
                )
            )
            return
        }

        // Invariant: an active host (this process owns a live session) must keep
        // the App Group `flowSessionActive` flag set. It can be violated from the
        // outside — a host suspended after a PiP drop stops writing heartbeats, so
        // the keyboard's `clearIfHostStale()` reads it as dead and clears the flag
        // while the process is merely asleep. On resume `isActive` is still true,
        // but `isHostReachable()`/`isHostReady()` stay false forever because they
        // gate on that flag, so every startflow is rejected as
        // `coldStartHostReadyNotPublished`. Re-assert it here — the single publish
        // chokepoint every path (and the 1 s heartbeat loop) funnels through — so
        // the divergence heals within a second, before the user presses mic.
        if !FlowSessionBridge.isSessionActive() {
            FlowSessionBridge.reassertSessionActive()
            FlowSessionDarwin.postSessionChanged()
            debug("re-asserted flowSessionActive flag cleared by keyboard stale-host cleanup")
        }

        let pollingAlive = pollingTask != nil && pollingTask?.isCancelled != true
        let hasRecentAudio = capture.engineHasRecentAudio(maxAge: 2)
        let hasPendingDelivery = hasUnacknowledgedTerminalResult()
        let canAcceptUtterance = pipController.isPictureInPictureActive
            && pollingAlive
            && startingUtteranceId == nil
            && !isUtteranceRecording
            && !isUtteranceProcessing
            && sessionWarning == nil
            && !capture.isInterrupted
            && !hasPendingDelivery

        let reason: FlowReadySnapshot.Reason
        if canAcceptUtterance {
            reason = .ready
        } else if startingUtteranceId != nil {
            reason = .waitingForAudioProof
        } else if sessionWarning != nil {
            reason = .error
        } else if isUtteranceRecording {
            reason = .recording
        } else if isUtteranceProcessing {
            reason = .processing
        } else if hasPendingDelivery {
            reason = .awaitingDelivery
        } else if !pipController.isPictureInPictureActive {
            reason = .starting
        } else if pipController.isPictureInPictureActive {
            // PiP is already up — transient gates (!polling / interruption)
            // are not a cold start. Prefer awaitingDelivery-adjacent idle over
            // `.starting` so the keyboard never flashes a false “starting” state.
            reason = .awaitingDelivery
        } else {
            reason = .starting
        }

        let now = Date().timeIntervalSince1970
        FlowSessionBridge.writeReadySnapshot(
            FlowReadySnapshot(
                sessionId: activeSessionId,
                ready: canAcceptUtterance,
                reason: reason,
                heartbeatAt: now,
                readyAt: canAcceptUtterance ? now : nil,
                audioProofAt: hasRecentAudio ? now : nil,
                engineMode: store.engineMode,
                localeId: store.localeId,
                busyUtteranceId: startingUtteranceId
                    ?? (isUtteranceRecording || isUtteranceProcessing
                        ? currentUtteranceId
                        : (hasPendingDelivery ? FlowSessionBridge.latestResult()?.utteranceId : nil)),
                hostGeneration: FlowSessionBridge.currentHostGeneration()
            )
        )
        let signature = [
            canAcceptUtterance ? "ready=1" : "ready=0",
            "reason=\(reason.rawValue)",
            capture.engineIsLive ? "engine=live" : "engine=dead",
            hasRecentAudio ? "audio=fresh" : "audio=stale",
            isUtteranceRecording ? "recording=1" : "recording=0",
            isUtteranceProcessing ? "processing=1" : "processing=0",
            startingUtteranceId == nil ? "starting=0" : "starting=1",
            sessionWarning == nil ? "warning=0" : "warning=1"
        ].joined(separator: "|")
        if signature != lastReadyTraceSignature {
            lastReadyTraceSignature = signature
            traceState("hostReady.update", extra: signature)
        }
    }

    /// Home preview field gained focus while this app is the Flow host.
    /// Reactivate capture and refresh the App Group ready contract so the
    /// custom keyboard extension sees green immediately.
    func refreshForInlineKeyboardFocus() async {
        guard isActive else { return }
        refreshHostReady()
        FlowSessionBridge.writeHeartbeat()
    }

    // MARK: - Session start

    private func ensurePiPActive(
        trigger: FlowPiPActivationTrigger,
        duration: TimeInterval?,
        forceReset: Bool = false
    ) {
        switch FlowPiPRecoveryPolicy.reconciliationDecision(
            wantsActiveSession: wantsActiveSession,
            isAppForeground: isAppForeground,
            isPictureInPictureActive: pipController.isPictureInPictureActive
        ) {
        case .noSessionIntent:
            // Silent before: the keyboard just waited out its 8 s budget with
            // nothing on either timeline explaining why.
            traceState(
                "pipRecovery.skipped",
                extra: "trigger=\(trigger.rawValue) reason=noSessionIntent"
            )
            return
        case .alreadyActive:
            pipLifecycleState = .active
            sessionWarning = nil
            if isActive {
                refreshHostReady()
            } else {
                activateFlowSessionAfterPiPProof(duration: duration)
            }
            handleColdStartAfterSessionReady()
            return
        case .waitForForeground:
            pipLifecycleState = .waitingForForeground
            FlowSessionBridge.setHostReady(false)
            refreshHostReady()
            // The most likely cold-start miss: the URL was routed before iOS
            // reported the scene as active, so the start is deferred and the
            // keyboard sees nothing but `hostReady=false` until it times out.
            traceState(
                "pipRecovery.deferred",
                extra: "trigger=\(trigger.rawValue) reason=waitForForeground"
            )
            return
        case .startRecovery:
            break
        }

        guard startTask == nil else {
            traceState(
                "pipRecovery.coalesced",
                extra: "trigger=\(trigger.rawValue)"
            )
            return
        }

        // Claim the flag synchronously: on a cold start the URL router and
        // `activateOnForeground()` both fire in the same runloop turn, and
        // setting it inside the async body let two start bodies interleave.
        isStarting = true
        pipRecoveryOperation &+= 1
        let operation = pipRecoveryOperation
        let recoveringExistingSession = isActive
        let resetFirstGeneration = forceReset
            || recoveringExistingSession
            || pipController.generation > 0
        pipLifecycleState = recoveringExistingSession
            ? .recovering(attempt: 1, total: FlowPiPRecoveryPolicy.maxAttempts)
            : .preparing(attempt: 1, total: FlowPiPRecoveryPolicy.maxAttempts)

        startTask = Task { @MainActor [weak self] in
            await self?.startSessionAsync(
                duration: duration,
                trigger: trigger,
                operation: operation,
                resetFirstGeneration: resetFirstGeneration
            )
        }
    }

    /// Sleeps until the next scheduled PiP attempt, waking early once the
    /// system media server finishes a reset so the retry lands right away.
    private func waitForNextPiPAttempt(seconds: TimeInterval) async {
        let wakeAt = Date().addingTimeInterval(seconds)
        let resetCount = FlowAudioSessionCoordinator.shared.mediaServicesResetCount
        while Date() < wakeAt, !Task.isCancelled {
            if FlowAudioSessionCoordinator.shared.mediaServicesResetCount != resetCount {
                traceState("pipRecovery.wokeOnMediaServicesReset")
                return
            }
            try? await Task.sleep(nanoseconds: 50_000_000)
        }
    }

    private func startSessionAsync(
        duration: TimeInterval?,
        trigger: FlowPiPActivationTrigger,
        operation: UInt64,
        resetFirstGeneration: Bool
    ) async {
        // `isStarting` was claimed synchronously in `ensurePiPActive()`.
        defer {
            if operation == pipRecoveryOperation {
                isStarting = false
                startTask = nil
            }
        }
        guard !Task.isCancelled else { return }
        traceState("startSessionAsync.begin", extra: "trigger=\(trigger.rawValue)")
        sessionWarning = nil
        // Advertise "warming" the moment arming begins. PiP arming can take up
        // to `FlowPiPRecoveryPolicy.totalBudget`; without this the keyboard sees
        // `.noSession` for that whole window and cannot tell it apart from a
        // dead host, so it burns its start budget in the dark.
        refreshHostReady()
        let recoveryStartedAt = Date()

        guard AppPermissions.flowRequirementsMet else {
            sessionWarning = permissionWarningMessage()
            pipLifecycleState = .failed(.notPossible)
            traceState("startSessionAsync.blocked", extra: "reason=permissions")
            FlowSessionBridge.setHostReady(false)
            isColdStartHandoff = false
            FlowDiagnostics.persistStartupFailure(
                reason: "permissionsMissing",
                context: startupFailureContext(
                    trigger: trigger.rawValue,
                    attemptCount: 0,
                    elapsed: Date().timeIntervalSince(recoveryStartedAt)
                )
            )
            return
        }

        let attemptsStartedAt = Date()
        let deadline = attemptsStartedAt.addingTimeInterval(FlowPiPRecoveryPolicy.totalBudget)
        var lastFailure: FlowPiPStartFailure = .timedOut
        var attemptsPerformed = 0

        for attempt in 1...FlowPiPRecoveryPolicy.maxAttempts {
            guard FlowPiPRecoveryPolicy.canContinue(
                operation: operation,
                currentOperation: pipRecoveryOperation,
                wantsActiveSession: wantsActiveSession,
                isAppForeground: isAppForeground,
                taskIsCancelled: Task.isCancelled
            ) else {
                if operation == pipRecoveryOperation,
                   wantsActiveSession,
                   !Task.isCancelled,
                   !isAppForeground {
                    pipLifecycleState = .waitingForForeground
                    FlowSessionBridge.setHostReady(false)
                }
                traceState(
                    "startSessionAsync.aborted",
                    extra: "trigger=\(trigger.rawValue) attempt=\(attempt) "
                        + "at=loopTop reason=\(pipRecoveryAbortReason(operation: operation))"
                )
                return
            }

            let wait = FlowPiPRecoveryPolicy.attemptOffset(beforeAttempt: attempt)
                - Date().timeIntervalSince(attemptsStartedAt)
            if wait > 0 {
                await waitForNextPiPAttempt(seconds: wait)
            }
            guard FlowPiPRecoveryPolicy.canContinue(
                operation: operation,
                currentOperation: pipRecoveryOperation,
                wantsActiveSession: wantsActiveSession,
                isAppForeground: isAppForeground,
                taskIsCancelled: Task.isCancelled
            ) else {
                traceState(
                    "startSessionAsync.aborted",
                    extra: "trigger=\(trigger.rawValue) attempt=\(attempt) "
                        + "at=afterRetryDelay reason=\(pipRecoveryAbortReason(operation: operation))"
                )
                return
            }

            if resetFirstGeneration || attempt > 1 {
                pipController.resetGeneration()
            }

            let remaining = deadline.timeIntervalSinceNow
            guard remaining > 0 else { break }
            attemptsPerformed = attempt
            pipLifecycleState = isActive
                ? .recovering(attempt: attempt, total: FlowPiPRecoveryPolicy.maxAttempts)
                : .preparing(attempt: attempt, total: FlowPiPRecoveryPolicy.maxAttempts)
            traceState(
                "pipRecovery.attempt",
                extra: "trigger=\(trigger.rawValue) attempt=\(attempt) generation=\(pipController.generation)"
            )

            let outcome = await pipController.startAndWait(
                hostTimeout: FlowPiPRecoveryPolicy.hostTimeout(
                    attempt: attempt,
                    remainingBudget: remaining
                ),
                activeTimeout: FlowPiPRecoveryPolicy.activeTimeout(
                    remainingBudget: deadline.timeIntervalSinceNow
                )
            )
            let recoveryElapsed = FlowPiPRecoveryPolicy.totalBudget
                - max(0, deadline.timeIntervalSinceNow)
            guard FlowPiPRecoveryPolicy.canContinue(
                operation: operation,
                currentOperation: pipRecoveryOperation,
                wantsActiveSession: wantsActiveSession,
                isAppForeground: isAppForeground,
                taskIsCancelled: Task.isCancelled
            ) else {
                traceState(
                    "startSessionAsync.aborted",
                    extra: "trigger=\(trigger.rawValue) attempt=\(attempt) "
                        + "at=afterStartAndWait outcome=\(outcome) "
                        + "reason=\(pipRecoveryAbortReason(operation: operation))"
                )
                return
            }

            switch outcome {
            case .started:
                sessionWarning = nil
                pipLifecycleState = .active
                if isActive {
                    refreshHostReady()
                } else {
                    activateFlowSessionAfterPiPProof(duration: duration)
                }
                traceState(
                    "startSessionAsync.ready",
                    extra: "trigger=\(trigger.rawValue) attempt=\(attempt) generation=\(pipController.generation)"
                )
                handleColdStartAfterSessionReady()
                debug("Flow session started (PiP keep-alive), mic released between utterances")
                return
            case .failed(let failure):
                lastFailure = failure
                traceState(
                    "startSessionAsync.failed",
                    extra: "reason=pipUnavailable trigger=\(trigger.rawValue) "
                        + "attempt=\(attempt) failure=\(failure) "
                        + "elapsed=\(String(format: "%.2f", recoveryElapsed))"
                )
            }
        }

        pipController.resetGeneration()
        pipLifecycleState = .failed(lastFailure)
        sessionWarning = AppL10n.string(lastFailure.localizationKey)
        FlowSessionBridge.setHostReady(false)
        refreshHostReady()
        debug("PiP keep-alive failed after bounded recovery: \(lastFailure)")
        FlowDiagnostics.persistStartupFailure(
            reason: String(describing: lastFailure),
            context: startupFailureContext(
                trigger: trigger.rawValue,
                attemptCount: attemptsPerformed,
                elapsed: Date().timeIntervalSince(recoveryStartedAt)
            )
        )
    }

    private func activateFlowSessionAfterPiPProof(duration: TimeInterval?) {
        let sessionId = activeSessionId ?? UUID()
        activeSessionId = sessionId
        lastHandledCommandSeq = 0
        FlowSessionBridge.markSessionActivePersistent(sessionId: sessionId)
        FlowSessionDarwin.postSessionChanged()
        isActive = true
        wantsActiveSession = true
        pipLifecycleState = .active
        // Low-profile PiP is a system-owned keep-alive surface. Keeping the
        // display awake wastes far more power than the static PiP itself.
        ScreenWakeLock.release()
        sessionExpiresAt = nil

        startHeartbeat()
        startCommandObserver()
        startPolling()
        startLevelPublishing()

        bindSessionASRIfNeeded()
        // ASR warmup deferred to beginUtterance (first mic press).

        refreshHostReady()
        scheduleStartupAudioHealthCheck()
        traceState("activateFlowSessionAfterPiPProof.done")
    }

    private func scheduleStartupAudioHealthCheck() {
        guard !didRunStartupAudioHealthCheck else { return }
        didRunStartupAudioHealthCheck = true
        startupAudioHealthTask?.cancel()
        startupAudioHealthTask = Task { @MainActor [weak self] in
            guard let self else { return }
            // Let deterministic Rime deployment claim the launch memory peak.
            // The health probe is opportunistic and must never delay app UI.
            try? await Task.sleep(nanoseconds: 300_000_000)
            let heavyDeadline = Date().addingTimeInterval(8)
            while FlowSessionBridge.isHostHeavy(), Date() < heavyDeadline {
                guard self.isActive, !Task.isCancelled else { return }
                try? await Task.sleep(nanoseconds: 200_000_000)
            }
            guard self.isActive,
                  !Task.isCancelled,
                  AppPermissions.flowRequirementsMet,
                  !FlowSessionBridge.isHostHeavy(),
                  self.startingUtteranceId == nil,
                  !self.isUtteranceRecording,
                  !self.isUtteranceProcessing,
                  !self.hasUnacknowledgedTerminalResult() else {
                FlowDiagnostics.log("startup audio health check skipped")
                return
            }
            await self.runStartupAudioHealthCheck()
        }
    }

    private func runStartupAudioHealthCheck() async {
        let healthID = UUID()
        startAudioPrime(id: healthID, origin: "startupHealth")
        guard let task = audioPrimeTask else { return }
        _ = await task.value
        guard isActive, !Task.isCancelled else { return }

        // If a user utterance or touch prime took ownership, it has priority.
        guard audioPrimeID == healthID,
              startingUtteranceId == nil,
              !isUtteranceRecording,
              !isUtteranceProcessing else {
            FlowDiagnostics.log("startup audio health check adopted by user")
            return
        }

        var flowing = await capture.awaitAudioFlowing(timeout: 1.2)
        if !flowing {
            // One bounded rebuild exercises the same soft-dead recovery as a
            // real first utterance, before the user is waiting on it.
            capture.stop(releaseSession: false)
            audioPrimeTask = nil
            startAudioPrime(id: healthID, origin: "startupHealthRebuild")
            if let rebuild = audioPrimeTask {
                _ = await rebuild.value
                flowing = await capture.awaitAudioFlowing(timeout: 1.0)
            }
        }

        guard audioPrimeID == healthID,
              startingUtteranceId == nil,
              !isUtteranceRecording,
              !isUtteranceProcessing else {
            FlowDiagnostics.log("startup audio health rebuild adopted by user")
            return
        }
        audioPrimeID = nil
        audioPrimeTask = nil
        audioPrimeCancellationRequested = false
        if capture.running {
            capture.stop(releaseSession: false)
        }
        _ = await pipController.reassertKeepAliveAudioSession()
        refreshHostReady()
        FlowDiagnostics.log(
            "startup audio health check done flowing=\(flowing ? 1 : 0)"
        )
    }

    private func prepareExistingSessionForColdStartReturn() async {
        guard isColdStartHandoff, isActive else { return }
        sessionWarning = nil
        if !pipController.isPictureInPictureActive {
            ensurePiPActive(trigger: .existingSession, duration: nil)
            return
        }
        pipLifecycleState = .active
        refreshHostReady()
        handleColdStartAfterSessionReady()
    }

    @MainActor
    private func handleColdStartAfterSessionReady() {
        guard isColdStartHandoff, isActive else { return }

        refreshHostReady()
        guard FlowSessionBridge.isHostReady() else {
            // Busy ≠ broken: a startflow arriving mid-utterance finds a healthy
            // session that is simply recording/processing. Treating it as a
            // PiP failure could restart capture and kill the live utterance.
            if isUtteranceRecording || isUtteranceProcessing {
                completeColdStartHandoff()
                debug("cold-start handoff ignored: session busy with an utterance")
                return
            }
            let message = AppL10n.string("flow.session.error.notPossible")
            sessionWarning = message
            FlowSessionBridge.setHostReady(false)
            if !pipController.isPictureInPictureActive {
                scheduleColdStartRecovery(duration: nil)
            }
            debug("cold-start blocked: host ready contract not published")
            // This is the toast the user reports as "cannot start voice
            // session": the keyboard jumped us here and we still could not
            // publish ready.
            FlowDiagnostics.persistStartupFailure(
                reason: "coldStartHostReadyNotPublished",
                context: startupFailureContext(
                    trigger: "coldStartHandoff",
                    attemptCount: 0,
                    elapsed: 0
                )
            )
            return
        }

        let hostEntry = HostReturnService.pendingHostEntry()
        guard hostEntry != nil else {
            completeColdStartHandoff()
            return
        }
        scheduleAutoReturnToHostIfNeeded(hostEntry: hostEntry)
    }

    private func scheduleAutoReturnToHostIfNeeded(hostEntry: HostAppEntry?) {
        let skipSwitch = true
        guard skipSwitch, hostEntry != nil else { return }
        Task { @MainActor [weak self] in
            try? await Task.sleep(nanoseconds: 450_000_000)
            guard let self, self.isColdStartHandoff, self.isActive,
                  FlowSessionBridge.isHostReady() else { return }
            if HostReturnService.openPendingHostIfPossible() {
                self.completeColdStartHandoff()
            }
        }
    }

    /// Actively rebuilds the audio pipeline after a failed cold start instead
    /// of passively waiting for frames that a dead engine will never produce.
    /// Escalates per attempt: reassert the session → full engine rebuild →
    /// bounce the audio session and rebuild. Force-quit relaunches routinely
    /// inherit stale mediaserverd state that only a rebuild clears.
    private func scheduleColdStartRecovery(duration: TimeInterval?) {
        // The failed controller has been fully torn down. Give the app
        // scene and Pegasus service a short settling interval before the
        // one bounded cold-start retry creates a new controller generation.
        ensurePiPActive(
            trigger: .startupRecovery,
            duration: duration,
            forceReset: true
        )
    }

    private func bindSessionASRIfNeeded(force: Bool = false) {
        let engineMode = store.engineMode
        if !force,
           sessionASR != nil,
           sessionASREngineMode == engineMode {
            return
        }
        sessionASR?.cancel()
        sessionASR = ASRServiceFactory.make(store: store)
        sessionASREngineMode = engineMode
        sessionASRWarmedLocaleID = nil
    }

    private func scheduleASRWarmup() {
        let localeID = SpeechLocaleResolver.resolve(store.localeId).identifier(.bcp47)
        guard sessionASRWarmedLocaleID != localeID else { return }
        guard HostMemoryBudget.gate("asr.warmup", category: "asr") else { return }
        OSGDiag.log("scheduleASRWarmup \(OSGDiag.memoryTag())", category: "asr")
        FlowSessionBridge.setHostHeavy(true)
        Task { @MainActor [weak self] in
            defer { FlowSessionBridge.setHostHeavy(false) }
            await self?.warmupASRIfNeeded()
        }
    }

    private func warmupASRIfNeeded() async {
        OSGDiag.log("warmupASRIfNeeded.begin \(OSGDiag.memoryTag())", category: "asr")
        bindSessionASRIfNeeded()
        let locale = SpeechLocaleResolver.resolve(store.localeId)
        let localeID = locale.identifier(.bcp47)
        guard sessionASRWarmedLocaleID != localeID else {
            OSGDiag.log("warmupASRIfNeeded.skip alreadyWarmed=\(localeID)", category: "asr")
            return
        }
        await asr.warmup(locale: locale)
        sessionASRWarmedLocaleID = localeID
        FlowDiagnostics.log("ASR warmup complete locale=\(localeID)")
        OSGDiag.log("warmupASRIfNeeded.done locale=\(localeID) \(OSGDiag.memoryTag())", category: "asr")
    }

    private func permissionWarningMessage() -> String {
        if AppPermissions.micStatus != .granted {
            return AppL10n.string("flow.error.micRequired")
        }
        return AppL10n.string("flow.error.speechRequired")
    }

    // MARK: - Polling

    private func startCommandObserver() {
        commandObserver = FlowSessionDarwinObserver(
            notificationName: FlowSessionDarwin.commandNotificationName
        ) { [weak self] in
            self?.handleKeyboardSignal()
        }
    }

    private func startPolling() {
        pollingTask?.cancel()
        lastObservedRecordingState = FlowSessionBridge.recordingState()
        FlowDiagnostics.log(
            "polling started: initialRecordingState=\(lastObservedRecordingState.rawValue) " +
            "container=\(AppGroup.containerPathForDiagnostics)"
        )
        pollingTask = Task { @MainActor [weak self] in
            while !Task.isCancelled {
                self?.handleKeyboardSignal()
                try? await Task.sleep(nanoseconds: 500_000_000)
            }
        }
    }

    private func handleKeyboardSignal() {
        FlowSessionBridge.reloadFromDisk()
        consumeHistoryMutationOutbox()
        var commands = FlowSessionBridge.commands(after: lastHandledCommandSeq)
        if commands.isEmpty, let latest = FlowSessionBridge.latestCommand(),
           latest.commandSeq > lastHandledCommandSeq {
            commands = [latest]
        }
        guard !commands.isEmpty else {
            lastCommandFingerprint = ""
            consumeAckIfNeeded()
            return
        }
        for command in commands {
            let fingerprint = "\(command.sessionId.uuidString)|\(command.utteranceId.uuidString)|\(command.action.rawValue)|\(command.commandSeq)"
            guard fingerprint != lastCommandFingerprint else { continue }
            lastCommandFingerprint = fingerprint
            handleFlowCommand(command)
        }
        consumeAckIfNeeded()
    }

    private func consumeHistoryMutationOutbox() {
        for mutation in HistoryMutationOutbox.pending() {
            let entry = SpeechHistoryStore.shared.applyHistoryMutation(mutation)
            if mutation.usageCategory == .ai, let text = mutation.text {
                UsageStatisticsStore.shared.recordAIInsertion(
                    text: text,
                    commitID: mutation.id
                )
            }
            HistoryMutationReceiptStore.save(
                HistoryMutationReceipt(
                    mutationID: mutation.id,
                    entryID: entry?.id,
                    revision: entry?.revision
                )
            )
            HistoryMutationOutbox.acknowledge(mutation.id)
        }
    }

    private func consumeAckIfNeeded() {
        guard let ack = FlowSessionBridge.latestAck(),
              let result = FlowSessionBridge.latestResult(),
              ack.sessionId == result.sessionId,
              ack.utteranceId == result.utteranceId,
              ack.commandSeq == result.commandSeq,
              ack.revision == nil
                || result.revision == nil
                || ack.revision == result.revision else {
            return
        }
        FlowSessionBridge.clearResult()
        // Ack clears the terminal result; republish ready immediately so the
        // keyboard does not linger on awaitingDelivery / starting between the
        // 500 ms poll and the next heartbeat. If processing is still true for
        // this same utterance, the writer forgot to drop the gate — heal here.
        if FlowHostAckGatePolicy.shouldDropProcessingGate(
            ackUtteranceId: ack.utteranceId,
            currentUtteranceId: currentUtteranceId,
            isUtteranceProcessing: isUtteranceProcessing
        ) {
            completeFinalizeCleanup(
                sessionId: ack.sessionId,
                utteranceId: ack.utteranceId
            )
        } else {
            refreshHostReady()
        }
    }

    private func hasUnacknowledgedTerminalResult() -> Bool {
        guard let result = FlowSessionBridge.latestResult(),
              result.status == .final
                || result.status == .error
                || result.status == .aborted
                || result.status == .timeout else {
            return false
        }
        guard let ack = FlowSessionBridge.latestAck() else { return true }
        return ack.sessionId != result.sessionId
            || ack.utteranceId != result.utteranceId
            || ack.commandSeq != result.commandSeq
    }

    private var hostUtteranceState: FlowHostUtteranceState {
        if let startingUtteranceId {
            return .starting(startingUtteranceId)
        }
        if isUtteranceRecording, let currentUtteranceId {
            return .recording(currentUtteranceId)
        }
        if isUtteranceProcessing, let currentUtteranceId {
            return .processing(currentUtteranceId)
        }
        return .idle
    }

    private func storeRejectedStart(
        _ command: FlowCommand,
        message: String,
        status: FlowResult.Status
    ) {
        FlowSessionBridge.writeResult(
            FlowResult(
                sessionId: command.sessionId,
                utteranceId: command.utteranceId,
                commandSeq: command.commandSeq,
                status: status,
                text: message,
                errorKind: .audioUnavailable,
                hostGeneration: FlowSessionBridge.currentHostGeneration(),
                revision: Self.resultRevision(),
                utteranceMode: command.utteranceMode,
                aiConversationID: command.aiConversationID
            )
        )
        traceState(
            "startRecording.rejected",
            extra: "status=\(status.rawValue) utterance=\(command.utteranceId.uuidString.prefix(8))"
        )
    }

    private func handleFlowCommand(_ command: FlowCommand) {
        switch FlowCommandGatekeeper.decide(
            commandSessionId: command.sessionId,
            commandSeq: command.commandSeq,
            activeSessionId: activeSessionId,
            lastHandledCommandSeq: lastHandledCommandSeq
        ) {
        case .rejectWrongSession:
            traceIgnoredCommand(
                reason: "staleSession",
                command: command,
                detail: "commandSession=\(command.sessionId)"
            )
            return
        case .rejectStaleSeq:
            traceIgnoredCommand(
                reason: "seqNotIncreasing",
                command: command,
                detail: "last=\(lastHandledCommandSeq)"
            )
            return
        case .accept:
            break
        }
        lastHandledCommandSeq = command.commandSeq
        lastIgnoredCommandSignature = ""

        FlowDiagnostics.log(
            "command \(command.action.rawValue) seq=\(command.commandSeq) "
                + "utterance=\(command.utteranceId) "
                + "mode=\(command.resolvedUtteranceMode.rawValue) "
                + "editSourceChars=\(command.editSourceText?.count ?? 0)"
        )

        switch command.action {
        case .startRecording:
            guard command.resolvedUtteranceMode != .unsupportedLegacy else {
                storeRejectedStart(
                    command,
                    message: AppL10n.string("flow.error.recognitionInterrupted"),
                    status: .error
                )
                traceState(
                    "startRecording.rejected",
                    extra: "reason=unsupportedLegacyMode"
                )
                return
            }
            let startDecision = FlowStartTransactionPolicy.decide(
                incomingUtteranceID: command.utteranceId,
                deadlineAt: command.startDeadlineAt,
                hostState: hostUtteranceState
            )
            switch startDecision {
            case .idempotent:
                traceState(
                    "startRecording.idempotent",
                    extra: "utterance=\(command.utteranceId.uuidString.prefix(8))"
                )
                refreshHostReady()
                return
            case .rejectBusy:
                storeRejectedStart(
                    command,
                    message: AppL10n.string("flow.error.recognitionInterrupted"),
                    status: .error
                )
                return
            case .rejectExpired:
                storeRejectedStart(
                    command,
                    message: AppL10n.string("flow.coldStart.error.audioTimeout"),
                    status: .timeout
                )
                return
            case .accept:
                break
            }
            guard prepareUtteranceIdentity(
                utteranceId: command.utteranceId,
                commandSeq: command.commandSeq
            ) else { return }
            let deadlineAt = command.startDeadlineAt
                ?? Date().timeIntervalSince1970 + FlowSessionKeys.utteranceStartBudget
            startingUtteranceId = command.utteranceId
            startTransactionDeadlineAt = deadlineAt
            FlowSessionBridge.writeStartTransaction(
                FlowStartTransaction(
                    sessionID: command.sessionId,
                    utteranceID: command.utteranceId,
                    deadlineAt: deadlineAt,
                    phase: .starting
                )
            )
            currentUtteranceMode = command.resolvedUtteranceMode
            pendingManagedRequestPurpose = command.managedRequestPurpose
            pendingManagedOOBEFeature = command.managedOOBEFeature
            pendingAIConversationID = currentUtteranceMode == .aiQuestion
                ? command.aiConversationID
                : nil
            if currentUtteranceMode == .editLastInput {
                let source = command.editSourceText?
                    .trimmingCharacters(in: .whitespacesAndNewlines)
                pendingEditSourceText = source?.isEmpty == false ? source : nil
                pendingSourceHistoryEntryID = command.sourceHistoryEntryID
                pendingSourceHistoryEntryRevision = command.sourceHistoryEntryRevision
            } else {
                pendingEditSourceText = nil
                pendingSourceHistoryEntryID = nil
                pendingSourceHistoryEntryRevision = nil
            }
            guard let startUtteranceId = currentUtteranceId else { return }
            let startToken = FlowUtteranceStartToken(
                generation: utteranceGeneration,
                utteranceId: startUtteranceId
            )
            Task { @MainActor [weak self] in
                await self?.handleStartRecordingCommand(
                    utteranceId: command.utteranceId,
                    commandSeq: command.commandSeq,
                    startToken: startToken,
                    deadlineAt: deadlineAt
                )
            }
        case .stopRecording:
            guard currentUtteranceId == command.utteranceId else { return }
            pendingFieldContext = command.fieldContext
            pendingProcessingDeadlineAt = command.processingDeadlineAt
                ?? (currentUtteranceMode == .editLastInput
                    ? Date().timeIntervalSince1970
                        + FlowSessionKeys.editLastInputHostProcessingBudget
                    : nil)
            FlowDiagnostics.log(
                "field context received before/after=" +
                "\(command.fieldContext?.precedingText?.count ?? 0)/" +
                "\(command.fieldContext?.followingText?.count ?? 0)"
            )
            if isUtteranceRecording {
                endUtterance()
            } else if !isUtteranceProcessing {
                // A very short press can place start+stop in the journal before
                // the async capture-start task gets a MainActor turn.
                pendingStopUtteranceId = command.utteranceId
                debug("stop command queued until capture start")
            }
        case .abort:
            guard currentUtteranceId == command.utteranceId else { return }
            abortUtterance()
        case .prewarm:
            // No utterance identity — warm SpeechAnalyzer / cloud prep only.
            scheduleASRWarmup()
            FlowDiagnostics.log("prewarm ASR requested seq=\(command.commandSeq)")
        case .primeAudio:
            beginAudioPrime(command)
        case .cancelPrimeAudio:
            cancelAudioPrime(command)
        case .endAIConversation:
            if currentUtteranceMode == .aiQuestion,
               isUtteranceProcessing || isUtteranceRecording {
                abortUtterance()
            }
            if let conversationID = command.aiConversationID {
                Task { await aiConversations.removeConversation(conversationID) }
            }
        case .submitAIQuestion:
            guard command.resolvedUtteranceMode == .aiQuestion else {
                storeRejectedStart(
                    command,
                    message: AppL10n.string("flow.error.aiQuestionFailed"),
                    status: .error
                )
                return
            }
            let question = command.aiQuestionText?
                .trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
            guard !question.isEmpty else {
                storeRejectedStart(
                    command,
                    message: AppL10n.string("flow.error.aiQuestionFailed"),
                    status: .error
                )
                return
            }
            let startDecision = FlowStartTransactionPolicy.decide(
                incomingUtteranceID: command.utteranceId,
                deadlineAt: command.startDeadlineAt,
                hostState: hostUtteranceState
            )
            switch startDecision {
            case .idempotent:
                refreshHostReady()
                return
            case .rejectBusy, .rejectExpired:
                let status: FlowResult.Status =
                    startDecision == .rejectExpired ? .timeout : .error
                storeRejectedStart(
                    command,
                    message: AppL10n.string("flow.error.recognitionInterrupted"),
                    status: status
                )
                return
            case .accept:
                break
            }
            guard prepareUtteranceIdentity(
                utteranceId: command.utteranceId,
                commandSeq: command.commandSeq
            ) else { return }
            currentUtteranceMode = .aiQuestion
            pendingManagedRequestPurpose = command.managedRequestPurpose
            pendingManagedOOBEFeature = command.managedOOBEFeature
            pendingAIConversationID = command.aiConversationID
            pendingEditSourceText = nil
            pendingSourceHistoryEntryID = nil
            pendingSourceHistoryEntryRevision = nil
            startingUtteranceId = nil
            startTransactionDeadlineAt = nil
            FlowSessionBridge.clearStartTransaction()
            isUtteranceRecording = false
            isUtteranceProcessing = true
            refreshHostReady()
            let conversationID = command.aiConversationID
            let utteranceId = command.utteranceId
            let commandSeq = command.commandSeq
            let sessionId = command.sessionId
            let taskKind = command.aiTaskKind ?? .aiQuestion
            let thinkingEnabled = command.aiThinkingEnabled ?? true
            Task { @MainActor [weak self] in
                await self?.answerPrefilledAIQuestion(
                    request: PrefilledAIRequest(
                        question: question,
                        conversationID: conversationID,
                        webPageURL: command.aiWebPageURL,
                        taskKind: taskKind,
                        requestSource: command.managedRequestSource,
                        requestPurpose: command.managedRequestPurpose,
                        oobeFeature: command.managedOOBEFeature,
                        thinkingEnabled: thinkingEnabled
                    ),
                    sessionId: sessionId,
                    utteranceId: utteranceId,
                    commandSeq: commandSeq
                )
            }
        }
    }

    /// Clipboard body for an explicitly spoken clipboard request. Opt-in
    /// history is the authorization gate, and while the keyboard is visible its
    /// newest stored item is the live pasteboard. Unlike the idle hint cards,
    /// naming the clipboard out loud is not bound to the 30s hint window.
    private func clipboardMaterialForAIQuestion(store: AppGroupStore) -> String? {
        guard store.clipboardHistoryEnabled else { return nil }
        return ClipboardHistoryStore().newestEntry?.text
    }

    /// Skip ASR and answer a prefilled AI hint / typed question.
    private func answerPrefilledAIQuestion(
        request: PrefilledAIRequest,
        sessionId: UUID,
        utteranceId: UUID,
        commandSeq: Int64
    ) async {
        // Hint-card / prefilled questions never go through `finalizeUtterance`,
        // so this path must drop the processing gate itself. Leaving it set
        // keeps `reason=processing` forever and the keyboard mic unclickable.
        defer {
            completeFinalizeCleanup(
                sessionId: sessionId,
                utteranceId: utteranceId
            )
        }
        guard let conversationID = request.conversationID else {
            claimAndStoreTerminal(utteranceId: utteranceId) {
                storeFinalizedError(
                    AppL10n.string("flow.error.aiQuestionFailed"),
                    kind: .generic,
                    sessionId: sessionId,
                    utteranceId: utteranceId,
                    commandSeq: commandSeq
                )
            }
            return
        }

        // Do not publish the prompt as a "transcript" — clipboard skills
        // send an XML envelope that must never appear above the mic.
        let pipelineStore = AppGroupStore()
        do {
            let resolvedQuestion = try await resolvedQuestion(for: request)
            let taskKind = if pipelineStore.credentialSource == .managed,
                              request.requestPurpose == nil {
                ManagedGatewayQuestionRouter.taskKind(
                    for: resolvedQuestion,
                    requestedTaskKind: request.taskKind
                )
            } else {
                request.taskKind
            }
            let service = try AIQuestionService.configured(
                store: pipelineStore,
                conversations: aiConversations,
                taskKind: taskKind,
                requestSource: request.requestSource,
                requestPurpose: request.requestPurpose,
                oobeFeature: request.oobeFeature,
                thinkingEnabled: request.thinkingEnabled,
                analyticsClient: analyticsClient,
                analyticsFeature: Self.analyticsFeature(for: request.taskKind)
            )
            aiAnswerStreamThrottle = AIAnswerStreamThrottle()
            let answer = try await service.answer(
                question: resolvedQuestion,
                conversationID: conversationID,
                targetLocaleID: pipelineStore.translationTargetLocaleId
            ) { [weak self] partial in
                Task { @MainActor in
                    self?.publishStreamingAIAnswerIfNeeded(
                        partial,
                        sessionId: sessionId,
                        utteranceId: utteranceId,
                        commandSeq: commandSeq,
                        aiConversationID: conversationID,
                        force: partial.isEmpty
                    )
                }
            }
            // Abort during the LLM await must not commit or deliver.
            guard canStoreTerminal(for: utteranceId) else { return }
            publishStreamingAIAnswerIfNeeded(
                answer,
                sessionId: sessionId,
                utteranceId: utteranceId,
                commandSeq: commandSeq,
                aiConversationID: conversationID,
                force: true
            )
            await service.commitSuccessfulTurn(
                question: resolvedQuestion,
                answer: answer,
                conversationID: conversationID
            )
            claimAndStoreTerminal(utteranceId: utteranceId) {
                storeFinalizedResult(
                    answer,
                    warning: nil,
                    sessionId: sessionId,
                    utteranceId: utteranceId,
                    commandSeq: commandSeq,
                    aiConversationID: conversationID
                )
            }
        } catch {
            claimAndStoreTerminal(utteranceId: utteranceId) {
                storeFinalizedError(
                    Self.aiQuestionFailureMessage(for: error),
                    kind: .generic,
                    sessionId: sessionId,
                    utteranceId: utteranceId,
                    commandSeq: commandSeq,
                    aiConversationID: conversationID
                )
            }
        }
    }

    private func resolvedQuestion(
        for request: PrefilledAIRequest
    ) async throws -> String {
        guard let url = request.webPageURL else { return request.question }
        let webpage = try await webPageContentService.fetchSummarySource(from: url)
        return AIClipboardPrompt.compose(
            instruction: request.question,
            material: webpage.summaryMaterial
        )
    }

    private func beginAudioPrime(_ command: FlowCommand) {
        guard startingUtteranceId == nil,
              !isUtteranceRecording,
              !isUtteranceProcessing,
              !hasUnacknowledgedTerminalResult() else {
            return
        }
        bindSessionASRIfNeeded()
        if let cloudASR = sessionASR as? CloudASRService {
            // Same locale source as `beginUtterance`, so the prime is adoptable.
            cloudASR.primeUtteranceStreaming(
                locale: SpeechLocaleResolver.resolve(store.localeId)
            )
        }
        startAudioPrime(id: command.utteranceId, origin: "micTouch")
    }

    private func startAudioPrime(id: UUID, origin: String) {
        if audioPrimeTask != nil || capture.engineHasRecentAudio(maxAge: 2) {
            audioPrimeID = id
            audioPrimeCancellationRequested = false
            return
        }
        audioPrimeID = id
        audioPrimeCancellationRequested = false
        let task = Task { @MainActor [weak self] in
            guard let self else { return false }
            return await self.startCaptureForPiPUtteranceIfNeeded()
        }
        audioPrimeTask = task
        Task { @MainActor [weak self] in
            let started = await task.value
            guard let self else { return }
            guard self.audioPrimeID == id else {
                if self.audioPrimeCancellationRequested,
                   self.startingUtteranceId == nil,
                   !self.isUtteranceRecording,
                   !self.isUtteranceProcessing,
                   self.capture.running {
                    self.capture.stop(releaseSession: false)
                    _ = await self.pipController.reassertKeepAliveAudioSession()
                }
                self.audioPrimeTask = nil
                self.audioPrimeCancellationRequested = false
                self.refreshHostReady()
                return
            }
            self.audioPrimeTask = nil
            if self.audioPrimeCancellationRequested,
               self.startingUtteranceId == nil,
               !self.isUtteranceRecording,
               !self.isUtteranceProcessing {
                self.audioPrimeID = nil
                self.audioPrimeCancellationRequested = false
                if self.capture.running {
                    self.capture.stop(releaseSession: false)
                    _ = await self.pipController.reassertKeepAliveAudioSession()
                }
                self.refreshHostReady()
                return
            }
            self.refreshHostReady()
            FlowDiagnostics.log(
                "audio prime completed started=\(started ? 1 : 0) "
                    + "origin=\(origin) utterance=\(id.uuidString.prefix(8))"
            )
        }
    }

    private func cancelAudioPrime(_ command: FlowCommand) {
        guard audioPrimeID == command.utteranceId,
              startingUtteranceId == nil,
              !isUtteranceRecording,
              !isUtteranceProcessing else {
            return
        }
        (sessionASR as? CloudASRService)?.cancelPrimedUtteranceStreaming()
        audioPrimeCancellationRequested = true
        // `capture.start()` is not cancellation-safe. Never stop it mid-start;
        // the completion waiter performs cleanup unless a real utterance adopts
        // this same single-flight task first.
        guard audioPrimeTask == nil else { return }
        audioPrimeID = nil
        audioPrimeCancellationRequested = false
        capture.stop(releaseSession: false)
        Task { @MainActor [weak self] in
            guard let self else { return }
            _ = await self.pipController.reassertKeepAliveAudioSession()
            self.refreshHostReady()
        }
    }

    private func storeCurrentPartial(
        _ text: String,
        generation: UInt64,
        utteranceId: UUID?
    ) {
        guard utteranceGeneration == generation,
              currentUtteranceId == utteranceId,
              let activeSessionId,
              let currentUtteranceId,
              !terminalUtteranceIds.contains(currentUtteranceId) else { return }
        let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !trimmed.isEmpty else { return }
        FlowSessionBridge.writeResult(
            FlowResult(
                sessionId: activeSessionId,
                utteranceId: currentUtteranceId,
                commandSeq: currentCommandSeq,
                status: .partial,
                text: trimmed,
                rawText: trimmed,
                hostGeneration: FlowSessionBridge.currentHostGeneration(),
                revision: Self.resultRevision(),
                utteranceMode: currentUtteranceMode,
                aiConversationID: pendingAIConversationID
            )
        )
    }

    private func storeCurrentError(
        _ message: String,
        kind: FlowSessionKeys.TranscriptionErrorKind = .generic,
        status: FlowResult.Status = .error
    ) {
        finishUtteranceAnalytics(
            utteranceID: currentUtteranceId,
            success: false,
            failureCategory: Self.transcriptionFailureCategory(
                kind: kind,
                status: status
            )
        )
        guard let activeSessionId, let currentUtteranceId else { return }
        FlowSessionBridge.writeResult(
            FlowResult(
                sessionId: activeSessionId,
                utteranceId: currentUtteranceId,
                commandSeq: currentCommandSeq,
                status: status,
                text: message,
                errorKind: kind,
                hostGeneration: FlowSessionBridge.currentHostGeneration(),
                revision: Self.resultRevision(),
                utteranceMode: currentUtteranceMode,
                aiConversationID: pendingAIConversationID
            )
        )
    }

    private func handleStartRecordingCommand(
        utteranceId: UUID?,
        commandSeq: Int64,
        startToken: FlowUtteranceStartToken,
        deadlineAt: TimeInterval
    ) async {
        guard !Task.isCancelled,
              canContinueStart(startToken),
              Date().timeIntervalSince1970 < deadlineAt else {
            failStartIfCurrent(startToken)
            return
        }
        refreshHostReady()
        // Keep the utterance gate closed until the route is stable and the
        // tap has produced a real frame. The rolling three-second preroll
        // preserves speech spoken during this short readiness window.
        guard await ensureCaptureStartedForUtterance() else {
            guard !Task.isCancelled, canContinueStart(startToken) else { return }
            failUtterance(
                message: AppL10n.string("flow.coldStart.error.audioTimeout"),
                kind: .audioUnavailable
            )
            return
        }
        guard !Task.isCancelled, canContinueStart(startToken) else {
            releaseOrphanedCaptureIfNeeded()
            return
        }
        guard Date().timeIntervalSince1970 < deadlineAt else {
            failStartIfCurrent(startToken)
            return
        }
        let micReady: Bool
        if capture.engineHasRecentAudio(maxAge: 2) {
            micReady = true
        } else {
            let firstBudget = min(
                4,
                max(0, deadlineAt - Date().timeIntervalSince1970)
            )
            var flowing = await capture.awaitAudioFlowing(
                timeout: firstBudget
            )
            let remaining = deadlineAt - Date().timeIntervalSince1970
            if !flowing, remaining > 1 {
                // First cold capture after PiP arm often proves audio late — one rebuild.
                debug("PiP audio proof timeout — one capture rebuild before failing")
                capture.stop(releaseSession: false)
                _ = await startCaptureForPiPUtteranceIfNeeded()
                flowing = await capture.awaitAudioFlowing(
                    timeout: min(2, max(0, deadlineAt - Date().timeIntervalSince1970))
                )
            }
            micReady = flowing && Date().timeIntervalSince1970 < deadlineAt
        }
        guard !Task.isCancelled, canContinueStart(startToken) else {
            releaseOrphanedCaptureIfNeeded()
            return
        }
        if !micReady {
            failUtterance(
                message: AppL10n.string("flow.coldStart.error.audioTimeout"),
                kind: .audioUnavailable
            )
            return
        }
        beginUtterance(
            utteranceId: utteranceId,
            commandSeq: commandSeq,
            requireRecentAudio: false
        )
        if pendingStopUtteranceId == currentUtteranceId {
            pendingStopUtteranceId = nil
            endUtterance()
        }
    }

    private func failStartIfCurrent(_ token: FlowUtteranceStartToken) {
        guard canContinueStart(token) else { return }
        failUtterance(
            message: AppL10n.string("flow.coldStart.error.audioTimeout"),
            kind: .audioUnavailable
        )
    }

    private func ensureCaptureStartedForUtterance() async -> Bool {
        audioPrimeCancellationRequested = false
        audioPrimeID = nil
        if let task = audioPrimeTask {
            let started = await task.value
            audioPrimeTask = nil
            if started || capture.engineHasRecentAudio(maxAge: 2) {
                return true
            }
        }
        return await startCaptureForPiPUtteranceIfNeeded()
    }

    /// Start capture for a PiP utterance without blocking on the first frame.
    /// Cold first start after relaunch often needs one rebuild (VPIO / -66635).
    private func startCaptureForPiPUtteranceIfNeeded() async -> Bool {
        if capture.engineHasRecentAudio(maxAge: 2) {
            return true
        }
        do {
            try await capture.start()
            if capture.engineIsLive || capture.engineHasRecentAudio(maxAge: 2) {
                return true
            }
            debug("PiP utterance capture soft-dead after start — one rebuild retry")
            capture.stop(releaseSession: false)
            try? await Task.sleep(nanoseconds: 120_000_000)
            try await capture.start()
            // Do not accept `running && !engineLive` — that is the silent-mic failure mode.
            return capture.engineIsLive || capture.engineHasRecentAudio(maxAge: 2)
        } catch {
            debug("PiP utterance capture start failed: \(error.localizedDescription) — retry once")
            capture.stop(releaseSession: false)
            try? await Task.sleep(nanoseconds: 150_000_000)
            do {
                try await capture.start()
                return capture.engineIsLive || capture.engineHasRecentAudio(maxAge: 2)
            } catch {
                debug("PiP utterance capture retry failed: \(error.localizedDescription)")
                return false
            }
        }
    }

    private func releaseCaptureAfterPiPUtteranceIfNeeded() {
        guard capture.running else { return }
        guard !isUtteranceRecording, !isUtteranceProcessing else { return }
        capture.stop(releaseSession: false)
        Task { @MainActor [weak self] in
            guard let self else { return }
            _ = await self.pipController.reassertKeepAliveAudioSession()
            self.refreshHostReady()
        }
        pipController.updateWaveformLevels([])
    }

    private func releaseOrphanedCaptureIfNeeded() {
        guard currentUtteranceId == nil,
              !isUtteranceRecording,
              !isUtteranceProcessing else { return }
        traceState("capture.start.discarded", extra: "reason=staleUtterance")
        releaseCaptureAfterPiPUtteranceIfNeeded()
    }

    private func canContinueStart(_ token: FlowUtteranceStartToken) -> Bool {
        FlowUtteranceLifecyclePolicy.canContinueStart(
            token: token,
            currentGeneration: utteranceGeneration,
            currentUtteranceId: currentUtteranceId,
            terminalUtteranceIds: terminalUtteranceIds
        )
    }

    private func beginUtterance(
        utteranceId: UUID? = nil,
        commandSeq: Int64 = 0,
        requireRecentAudio: Bool = true
    ) {
        if let utteranceId, terminalUtteranceIds.contains(utteranceId) {
            traceState(
                "beginUtterance.ignored",
                extra: "reason=terminalId utterance=\(utteranceId.uuidString.prefix(8))"
            )
            return
        }
        if requireRecentAudio {
            guard capture.engineHasRecentAudio(maxAge: 2) else {
                traceState("beginUtterance.blocked", extra: "reason=audioNotRecent")
                failUtterance(
                    message: AppL10n.string("flow.error.audioUnavailable"),
                    kind: .audioUnavailable
                )
                return
            }
        } else {
            // PiP cold path: capture was just started; open the gate so the
            // first tap frames enter the ASR stream instead of preroll only.
            guard capture.running || capture.engineIsLive else {
                traceState("beginUtterance.blocked", extra: "reason=captureNotRunning")
                failUtterance(
                    message: AppL10n.string("flow.error.audioUnavailable"),
                    kind: .audioUnavailable
                )
                return
            }
        }
        guard !isUtteranceProcessing else {
            traceState("beginUtterance.ignored", extra: "reason=processing")
            debug("beginUtterance ignored — previous utterance still processing")
            return
        }
        bindSessionASRIfNeeded()
        let expectedEngine = store.engineMode
        if sessionASREngineMode != expectedEngine {
            traceState(
                "beginUtterance.rebindMismatch",
                extra: "expectedEngine=\(expectedEngine) boundEngine=\(sessionASREngineMode ?? "nil")"
            )
            bindSessionASRIfNeeded(force: true)
        }

        // Usually already warm from session start; refresh without blocking the mic gate.
        scheduleASRWarmup()

        if currentUtteranceId != utteranceId || currentUtteranceId == nil {
            guard prepareUtteranceIdentity(
                utteranceId: utteranceId,
                commandSeq: commandSeq
            ) else { return }
        }
        let generation = utteranceGeneration
        let taskUtteranceId = currentUtteranceId
        currentCommandSeq = commandSeq
        currentPartial = ""
        lastFinal = ""
        lastFinalWithPauseMarks = ""
        bestPartialSnapshot = ""
        utterancePCMSamples = []
        chunkWarnings = []
        asrCompletedGeneration = nil
        asrFailureMessage = nil

        let localeId = store.localeId
        FlowSessionBridge.setTranscriptionLanguage(localeId)
        FlowSessionBridge.clearPendingTranscription()

        let locale = SpeechLocaleResolver.resolve(localeId)
        let stream = capture.beginUtterance()
        let cloudService = asr as? CloudASRService
        let useStreaming =
            store.engineMode == "cloud"
            && (
                cloudService?.supportsUtteranceStreaming
                    ?? CloudASRModelCatalog.supportsTrueStreamingASR(for: store.asrProviderId)
            )
        let pipeline: ChunkedUtterancePipeline?
        if useStreaming {
            chunkedPipeline = nil
            pipeline = nil
        } else {
            let created = ChunkedUtterancePipeline(asr: asr, locale: locale)
            chunkedPipeline = created
            pipeline = created
        }

        isUtteranceRecording = true
        startingUtteranceId = nil
        startTransactionDeadlineAt = nil
        if let activeSessionId, let currentUtteranceId {
            FlowSessionBridge.writeStartTransaction(
                FlowStartTransaction(
                    sessionID: activeSessionId,
                    utteranceID: currentUtteranceId,
                    deadlineAt: Date().timeIntervalSince1970,
                    phase: .recording
                )
            )
        }
        utteranceRecordingStartedAt = Date()
        startUtteranceSafetyTimer()
        refreshHostReady()
        FlowDiagnostics.log(
            "beginUtterance engine=\(store.engineMode) " +
            "asrType=\(type(of: asr)) streaming=\(useStreaming) " +
            "localCustomLM=\(store.localASRCustomLanguageModelEnabled) " +
            "max=\(Int(FlowSessionKeys.maxUtteranceDuration))s"
        )

        let cloudASRForStreaming = useStreaming ? cloudService : nil
        if let taskUtteranceId {
            // A healed delivery gate may allow a replacement utterance before
            // an old finalize task unwinds. Close any old value task first.
            analyticsOperations.cancelAll()
            analyticsOperations.start(
                utteranceID: taskUtteranceId,
                feature: FlowAnalyticsFeatureMapping.recordingFeature(
                    for: currentUtteranceMode
                ),
                executionMode: analyticsExecutionMode,
                client: analyticsClient
            )
        }

        asrTask = Task.detached(priority: .userInitiated) { [weak manager = self] in
            let outcome: ChunkedUtterancePipelineOutcome
            if let cloud = cloudASRForStreaming {
                outcome = await cloud.transcribeUtteranceStreaming(stream: stream, locale: locale) { partial in
                    Task { @MainActor in
                        guard let manager else { return }
                        guard manager.utteranceGeneration == generation,
                              manager.currentUtteranceId == taskUtteranceId else { return }
                        manager.currentPartial = partial
                        manager.storeCurrentPartial(
                            partial,
                            generation: generation,
                            utteranceId: taskUtteranceId
                        )
                    }
                }
            } else if let pipeline {
                outcome = await pipeline.transcribe(stream: stream) { partial in
                    Task { @MainActor in
                        guard let manager else { return }
                        guard manager.utteranceGeneration == generation,
                              manager.currentUtteranceId == taskUtteranceId else { return }
                        manager.currentPartial = partial
                        manager.storeCurrentPartial(
                            partial,
                            generation: generation,
                            utteranceId: taskUtteranceId
                        )
                    }
                }
            } else {
                outcome = .failure(SharedL10n.string("error.asr.noSpeech"))
            }
            // Re-bind `manager` inside the `@MainActor` block so the
            // weak reference is captured under the right isolation. Swift
            // 6 strict concurrency otherwise complains about a
            // task-isolated reference escaping into a main-actor closure.
            await MainActor.run { [weak manager] in
                guard let manager else { return }
                guard manager.utteranceGeneration == generation,
                      manager.currentUtteranceId == taskUtteranceId else {
                    return
                }
                manager.asrCompletedGeneration = generation
                FlowDiagnostics.log(
                    "asr finished streaming=\(useStreaming) partialLen=\(manager.currentPartial.count) " +
                    "finalPending=\(manager.lastFinal.isEmpty)"
                )
                switch outcome {
                case .success(let success):
                    FlowTrace.transcript(
                        "asr.outcome",
                        success.text,
                        "engine=\(manager.store.engineMode) streaming=\(useStreaming ? 1 : 0) "
                            + "warnings=\(success.chunkWarnings.count)"
                    )
                    manager.lastFinal = success.text
                    manager.lastFinalWithPauseMarks = success.textWithPauseMarks
                    manager.chunkWarnings = success.chunkWarnings
                    manager.currentPartial = ""
                case .failure(let message):
                    manager.asrFailureMessage = message
                    manager.debug(
                        "asr error category=asrFailure errorBytes=\(message.utf8.count)"
                    )
                    FlowTrace.warn(
                        "asr.outcome.failed",
                        "engine=\(manager.store.engineMode) streaming=\(useStreaming ? 1 : 0) "
                            + "partialLen=\(manager.currentPartial.count) "
                            + "bestPartialLen=\(manager.bestPartialSnapshot.count) "
                            + "errorCategory=asrFailure errorBytes=\(message.utf8.count)"
                    )
                    // Prefer any non-empty partial over a hard no-speech failure.
                    // Clearing `bestPartialSnapshot` here would race finalize
                    // into an empty transcript even when ASR had usable text.
                    let recovery = [
                        manager.currentPartial,
                        manager.bestPartialSnapshot
                    ]
                        .map { $0.trimmingCharacters(in: .whitespacesAndNewlines) }
                        .first(where: { !$0.isEmpty })
                    // Without a partial, keep recording: capture is independent
                    // of ASR, and finalize retries the full PCM through batch
                    // ASR (`recognitionFailed`) before reporting the error.
                    if let recovery {
                        manager.lastFinal = recovery
                        manager.debug("asr error recovered via partial len=\(recovery.count)")
                    } else if manager.isUtteranceRecording {
                        manager.debug("asr error while recording — deferring to batch fallback")
                    }
                case .cancelled:
                    // A provider-side cancellation while the mic is still open
                    // has no later successful finalize path. Close the
                    // utterance now so its analytics STARTED cannot be orphaned.
                    let hasRecoverableText = !manager.currentPartial
                        .trimmingCharacters(in: .whitespacesAndNewlines).isEmpty
                        || !manager.bestPartialSnapshot
                        .trimmingCharacters(in: .whitespacesAndNewlines).isEmpty
                    if manager.isUtteranceRecording, !hasRecoverableText {
                        manager.failUtterance(
                            message: AppL10n.string(
                                "flow.error.recognitionInterrupted"
                            ),
                            kind: .recognitionInterrupted
                        )
                    }
                }
            }
        }

        debug("utterance recording started")
    }

    @discardableResult
    private func prepareUtteranceIdentity(
        utteranceId: UUID?,
        commandSeq: Int64
    ) -> Bool {
        let resolved = utteranceId ?? UUID()
        guard !terminalUtteranceIds.contains(resolved) else {
            traceState(
                "utteranceIdentity.rejected",
                extra: "reason=terminalId utterance=\(resolved.uuidString.prefix(8))"
            )
            return false
        }
        guard currentUtteranceId != resolved else {
            currentCommandSeq = commandSeq
            return true
        }
        utteranceGeneration &+= 1
        currentUtteranceId = resolved
        currentCommandSeq = commandSeq
        return true
    }

    private func startUtteranceSafetyTimer() {
        utteranceSafetyTask?.cancel()
        let utteranceId = currentUtteranceId
        utteranceSafetyTask = Task { @MainActor [weak self] in
            let timeout = FlowSessionKeys.maxUtteranceDuration + 10
            try? await Task.sleep(nanoseconds: UInt64(timeout * 1_000_000_000))
            guard let self, !Task.isCancelled else { return }
            guard self.isUtteranceRecording, self.currentUtteranceId == utteranceId else { return }
            self.storeCurrentError(
                AppL10n.string("flow.error.recognitionInterrupted"),
                kind: .recognitionInterrupted,
                status: .timeout
            )
            self.abortUtterance()
            self.debug("utterance safety timer aborted stale recording")
        }
    }

    private func endUtterance() {
        guard isUtteranceRecording else { return }

        let duration = utteranceRecordingStartedAt.map { Date().timeIntervalSince($0) } ?? 0
        if currentUtteranceMode != .editLastInput,
           duration < FlowEmptyTapSkipPolicy.maxDurationSeconds {
            let samples = capture.utterancePCMSnapshot()
            let peak = FlowEmptyTapSkipPolicy.peakAbs(samples)
            if FlowEmptyTapSkipPolicy.shouldSkip(
                durationSeconds: duration,
                sampleCount: samples.count,
                peakAmplitude: peak
            ) {
                FlowTrace.pipeline(
                    "utterance.skipEmptyTap",
                    "duration=\(String(format: "%.3f", duration))s "
                        + "samples=\(samples.count) peak=\(String(format: "%.4f", peak))"
                )
                abortUtterance(kind: .discardedEmpty, message: "")
                return
            }
        }

        // Close the mic gate first, then mark processing before dropping the
        // recording flag so the poll loop cannot start a second utterance.
        isUtteranceRecording = false
        isUtteranceProcessing = true
        FlowSessionBridge.clearStartTransaction()
        utteranceSafetyTask?.cancel()
        utteranceSafetyTask = nil
        refreshHostReady()

        // Snapshot pipelined partial before drain — fallback if the final chunk ASR drops tail text.
        bestPartialSnapshot = currentPartial.trimmingCharacters(in: .whitespacesAndNewlines)

        // Do NOT cancel `asrTask` or `asr` — drain trailing PCM, then finalize.

        // Capture ids now: a cancelled finalize must still clear *this*
        // utterance's processing gate even if currentUtteranceId was cleared
        // by a racing fail/abort path.
        let drainingSessionId = activeSessionId
        let drainingUtteranceId = currentUtteranceId
        let drainingCommandSeq = currentCommandSeq
        let drainingGeneration = utteranceGeneration
        finalizeTask?.cancel()
        finalizeTask = Task { @MainActor [weak self] in
            guard let self else { return }
            let drainReport = await self.capture.endUtteranceAndDrain()
            FlowDiagnostics.logDrain(drainReport)
            self.utterancePCMSamples = self.capture.consumeUtteranceSamples()
            FlowTrace.pipeline(
                "utterance.pcmCollected",
                "samples=\(self.utterancePCMSamples.count) "
                    + "seconds=\(FlowTrace.seconds(samples: self.utterancePCMSamples.count)) "
                    + "rms=\(FlowTrace.rms(self.utterancePCMSamples)) "
                    + "capture[\(self.capture.frameReport().summary)]"
            )
            self.capture.stop(releaseSession: false)
            _ = await self.pipController.reassertKeepAliveAudioSession()
            self.pipController.updateWaveformLevels([])
            await self.finalizeUtterance(
                sessionId: drainingSessionId,
                utteranceId: drainingUtteranceId,
                commandSeq: drainingCommandSeq,
                generation: drainingGeneration
            )
        }
        debug("utterance stopped, draining tail")
    }

    private func abortUtterance(
        kind: FlowSessionKeys.TranscriptionErrorKind = .recognitionInterrupted,
        message: String? = nil
    ) {
        let utteranceId = currentUtteranceId
        if claimTerminal(utteranceId: utteranceId) {
            storeCurrentError(
                message ?? AppL10n.string("flow.error.recognitionInterrupted"),
                kind: kind,
                status: .aborted
            )
        }
        analyticsOperations.cancel(utteranceID: utteranceId)
        isUtteranceRecording = false
        isUtteranceProcessing = false
        startingUtteranceId = nil
        startTransactionDeadlineAt = nil
        FlowSessionBridge.clearStartTransaction()
        utteranceRecordingStartedAt = nil
        utteranceSafetyTask?.cancel()
        utteranceSafetyTask = nil
        finalizeTask?.cancel()
        finalizeTask = nil
        asrTask?.cancel()
        Task { await chunkedPipeline?.cancel() }
        chunkedPipeline = nil
        asr.cancel()
        capture.cancelUtterance()
        releaseCaptureAfterPiPUtteranceIfNeeded()
        currentPartial = ""
        lastFinal = ""
        lastFinalWithPauseMarks = ""
        bestPartialSnapshot = ""
        utterancePCMSamples = []
        chunkWarnings = []
        pendingFieldContext = nil
        clearPendingInstructionState()
        utteranceGeneration &+= 1
        currentUtteranceId = nil
        currentCommandSeq = 0
        refreshHostReady()
        debug("utterance aborted")
    }

    private func failUtterance(
        message: String,
        kind: FlowSessionKeys.TranscriptionErrorKind = .asrFailed
    ) {
        guard claimTerminal(utteranceId: currentUtteranceId) else { return }
        isUtteranceRecording = false
        isUtteranceProcessing = false
        startingUtteranceId = nil
        startTransactionDeadlineAt = nil
        FlowSessionBridge.clearStartTransaction()
        utteranceRecordingStartedAt = nil
        utteranceSafetyTask?.cancel()
        utteranceSafetyTask = nil
        finalizeTask?.cancel()
        finalizeTask = nil
        asrTask?.cancel()
        Task { await chunkedPipeline?.cancel() }
        chunkedPipeline = nil
        asr.cancel()
        capture.cancelUtterance()
        releaseCaptureAfterPiPUtteranceIfNeeded()
        currentPartial = ""
        lastFinal = ""
        lastFinalWithPauseMarks = ""
        bestPartialSnapshot = ""
        utterancePCMSamples = []
        chunkWarnings = []
        storeCurrentError(message, kind: kind)
        pendingFieldContext = nil
        clearPendingInstructionState()
        utteranceGeneration &+= 1
        currentUtteranceId = nil
        currentCommandSeq = 0
        refreshHostReady()
        debug("utterance failed: \(message)")
    }

    private func claimTerminal(utteranceId: UUID?) -> Bool {
        guard let utteranceId, !terminalUtteranceIds.contains(utteranceId) else {
            return false
        }
        terminalUtteranceIds.insert(utteranceId)
        return true
    }

    /// After an `await`, refuse to write if this utterance was aborted or replaced.
    private func canStoreTerminal(for utteranceId: UUID?) -> Bool {
        guard let utteranceId else { return false }
        return FlowTerminalStorePolicy.canStore(
            currentUtteranceId: currentUtteranceId,
            finishedUtteranceId: utteranceId,
            alreadyTerminal: terminalUtteranceIds.contains(utteranceId)
        )
    }

    /// Claim + write with no `await` between them, so abort cannot sneak a
    /// second payload in after `claimTerminal`.
    private func claimAndStoreTerminal(utteranceId: UUID?, store: () -> Void) {
        guard canStoreTerminal(for: utteranceId) else { return }
        guard claimTerminal(utteranceId: utteranceId) else { return }
        store()
    }

    private func clearPendingInstructionState() {
        pendingEditSourceText = nil
        pendingSourceHistoryEntryID = nil
        pendingSourceHistoryEntryRevision = nil
        pendingAIConversationID = nil
        pendingManagedRequestPurpose = nil
        pendingManagedOOBEFeature = nil
        pendingProcessingDeadlineAt = nil
        currentUtteranceMode = .dictation
    }

    private func finalizeUtterance(
        sessionId finalizeSessionId: UUID?,
        utteranceId finalizeUtteranceId: UUID?,
        commandSeq finalizeCommandSeq: Int64,
        generation finalizeGeneration: UInt64
    ) async {
        let pipelineStarted = Date()
        let fieldContext = pendingFieldContext
        let utteranceMode = currentUtteranceMode
        let editSourceText = pendingEditSourceText
        let sourceHistoryEntryID = pendingSourceHistoryEntryID
        let sourceHistoryEntryRevision = pendingSourceHistoryEntryRevision
        let aiConversationID = pendingAIConversationID
        let managedRequestPurpose = pendingManagedRequestPurpose
        let managedOOBEFeature = pendingManagedOOBEFeature
        let processingDeadlineAt = pendingProcessingDeadlineAt
        let asrEngineMode = sessionASREngineMode ?? store.engineMode
        // ALWAYS clear the processing gate for this utterance. The previous
        // guard required currentUtteranceId to still match; a racing
        // fail/abort/cancel path could nil the id (or leave processing stuck)
        // and then skip refreshHostReady — keyboard stayed white forever
        // while host logs still said "utterance finalized".
        defer {
            // Any path that returns without an explicit terminal event is a
            // cancellation from the user's perspective. The registry is keyed
            // by utterance ID, so this cannot finish a newer recording.
            analyticsOperations.cancel(utteranceID: finalizeUtteranceId)
            pendingFieldContext = nil
            pendingEditSourceText = nil
            pendingSourceHistoryEntryID = nil
            pendingSourceHistoryEntryRevision = nil
            pendingAIConversationID = nil
            pendingManagedRequestPurpose = nil
            pendingManagedOOBEFeature = nil
            pendingProcessingDeadlineAt = nil
            if currentUtteranceMode == utteranceMode {
                currentUtteranceMode = .dictation
            }
            completeFinalizeCleanup(
                sessionId: finalizeSessionId,
                utteranceId: finalizeUtteranceId
            )
        }

        let asrWait = asrWaitTimeout()
        FlowDiagnostics.log(
            "finalize start asrWait=\(Int(asrWait))s engine=\(store.engineMode)"
        )

        let normalASRDeadline = Date().addingTimeInterval(asrWait)
        let asrDeadline = processingDeadlineAt.map {
            min(normalASRDeadline, Date(timeIntervalSince1970: $0))
        } ?? normalASRDeadline
        while Date() < asrDeadline {
            if !lastFinal.isEmpty { break }
            if asrCompletedGeneration == finalizeGeneration { break }
            if asrTask?.isCancelled == true { break }
            // Honour cooperative cancel so a replaced finalize exits promptly,
            // but still run defer cleanup (unlike an early `return` mid-polish
            // that used to leave processing=true when ids no longer matched).
            if Task.isCancelled { break }
            try? await Task.sleep(nanoseconds: 100_000_000)
        }

        // The recognizer never reported back within the wait: treat it as a
        // failure so batch ASR can still recover the captured audio.
        let asrStalled = lastFinal.isEmpty
            && asrCompletedGeneration != finalizeGeneration
            && !Task.isCancelled
        if lastFinal.isEmpty {
            FlowDiagnostics.log(
                "ASR wait elapsed — cancelling ASR task and using best available transcript "
                    + "stalled=\(asrStalled ? 1 : 0)"
            )
            asrTask?.cancel()
            Task { await chunkedPipeline?.cancel() }
        }

        let asrElapsed = Date().timeIntervalSince(pipelineStarted)
        FlowDiagnostics.log("ASR phase done in \(String(format: "%.1f", asrElapsed))s finalLen=\(lastFinal.count)")
        if utteranceMode == .aiQuestion {
            FlowTrace.pipeline(
                "aiQuestion.asrReady",
                "finalLen=\(lastFinal.count) partialLen=\(bestPartialSnapshot.count)"
            )
        } else {
            FlowTrace.transcript(
                "asr.beforeGuard",
                lastFinal,
                "stage=stitchedFinal engine=\(store.engineMode) "
                    + "elapsed=\(String(format: "%.2f", asrElapsed))s"
            )
            FlowTrace.transcript(
                "asr.bestPartial",
                bestPartialSnapshot,
                "stage=partialSnapshot"
            )
        }

        var text = UtteranceTranscriptGuard.resolve(
            stitchedFinal: lastFinal,
            partialSnapshot: bestPartialSnapshot
        )
        if text.isEmpty {
            text = currentPartial.trimmingCharacters(in: .whitespacesAndNewlines)
        }

        let wantsBatchFallback = UtteranceBatchFallbackPolicy.shouldRunBatchFallback(
            stitchedFinal: lastFinal,
            partialSnapshot: bestPartialSnapshot,
            recognitionFailed: asrFailureMessage != nil,
            recognitionStalled: asrStalled
        )
        FlowTrace.pipeline(
            "batchFallback.decision",
            "wanted=\(wantsBatchFallback ? 1 : 0) stalled=\(asrStalled ? 1 : 0) "
                + "pcmSamples=\(utterancePCMSamples.count) "
                + "pcmRms=\(FlowTrace.rms(utterancePCMSamples)) "
                + "stitchedLen=\(lastFinal.count) partialLen=\(bestPartialSnapshot.count) "
                + "resolvedLen=\(text.count)"
        )
        let hasBatchFallbackBudget = processingDeadlineAt.map {
            Date().timeIntervalSince1970 + FlowSessionKeys.batchASRFallbackTimeout < $0
        } ?? true
        if wantsBatchFallback,
           hasBatchFallbackBudget,
           !utterancePCMSamples.isEmpty {
            text = await runBatchASRFallback(currentText: text)
        }
        let rawTextForPolish = text == lastFinal && !lastFinalWithPauseMarks.isEmpty
            ? lastFinalWithPauseMarks
            : text
        utterancePCMSamples = []
        guard !text.isEmpty else {
            let wasCancelled = (asrTask?.isCancelled == true || Task.isCancelled)
            let key = wasCancelled
                ? "flow.error.recognitionInterrupted"
                : "flow.error.noSpeech"
            let kind: FlowSessionKeys.TranscriptionErrorKind =
                asrFailureMessage == nil
                ? (wasCancelled ? .recognitionInterrupted : .noSpeech)
                : .asrFailed
            FlowDiagnostics.log("finalize failed: empty transcript after \(String(format: "%.1f", asrElapsed))s")
            FlowTrace.warn(
                "finalize.emptyTranscript",
                "engine=\(store.engineMode) elapsed=\(String(format: "%.2f", asrElapsed))s "
                    + "kind=\(kind.rawValue) asrCancelled=\(asrTask?.isCancelled == true ? 1 : 0) "
                    + "capture[\(capture.frameReport().summary)]"
            )
            utteranceRecordingStartedAt = nil
            guard claimTerminal(utteranceId: finalizeUtteranceId) else { return }
            storeFinalizedError(
                asrFailureMessage ?? AppL10n.string(key),
                kind: kind,
                sessionId: finalizeSessionId,
                utteranceId: finalizeUtteranceId,
                commandSeq: finalizeCommandSeq
            )
            return
        }
        if utteranceMode != .aiQuestion {
            finishUtteranceAnalytics(
                utteranceID: finalizeUtteranceId,
                success: true
            )
        }

        // Re-read App Group at finalize so dictionary and translation changes
        // from the keyboard extension are visible before local correction,
        // polish, or translation.
        let pipelineStore = AppGroupStore()
        let postASR = FlowASRPostProcessor.process(
            text: text,
            textForPolish: rawTextForPolish,
            engineMode: asrEngineMode,
            dictionary: pipelineStore.personalDictionary
        )
        text = postASR.text
        let textForPolish = postASR.textForPolish

        storeRawCandidate(
            text,
            sessionId: finalizeSessionId,
            utteranceId: finalizeUtteranceId,
            commandSeq: finalizeCommandSeq
        )
        let recordingDuration = consumeRecordingDuration()
        if utteranceMode == .editLastInput {
            EditUsageMetricsStore.recordInstructionDuration(recordingDuration)
        }

        let engineMode = asrEngineMode
        let chunkNote = Self.chunkWarningMessage(chunkWarnings)
        if utteranceMode == .aiQuestion {
            guard let aiConversationID else {
                guard claimTerminal(utteranceId: finalizeUtteranceId) else { return }
                storeFinalizedError(
                    AppL10n.string("flow.error.aiQuestionFailed"),
                    kind: .generic,
                    sessionId: finalizeSessionId,
                    utteranceId: finalizeUtteranceId,
                    commandSeq: finalizeCommandSeq
                )
                return
            }

            let spoken = AIClipboardPrompt.resolveSpoken(
                question: text,
                material: clipboardMaterialForAIQuestion(store: pipelineStore)
            )
            guard case .ready(let question) = spoken else {
                guard claimTerminal(utteranceId: finalizeUtteranceId) else { return }
                storeFinalizedError(
                    AppL10n.string("flow.error.clipboardUnavailable"),
                    kind: .generic,
                    sessionId: finalizeSessionId,
                    utteranceId: finalizeUtteranceId,
                    commandSeq: finalizeCommandSeq,
                    aiConversationID: aiConversationID
                )
                return
            }

            do {
                let taskKind = if pipelineStore.credentialSource == .managed,
                                  managedRequestPurpose == nil {
                    // Classify the post-ASR spoken text before clipboard
                    // resolution can wrap it in an internal prompt envelope.
                    ManagedGatewayQuestionRouter.taskKind(for: text)
                } else {
                    ManagedGatewayTaskKind.aiQuestion
                }
                let service = try AIQuestionService.configured(
                    store: pipelineStore,
                    conversations: aiConversations,
                    taskKind: taskKind,
                    requestPurpose: managedRequestPurpose,
                    oobeFeature: managedOOBEFeature,
                    analyticsClient: analyticsClient,
                    analyticsFeature: .aiAssistant
                )
                aiAnswerStreamThrottle = AIAnswerStreamThrottle()
                let publishUtteranceId = finalizeUtteranceId
                let publishSessionId = finalizeSessionId
                let publishCommandSeq = finalizeCommandSeq
                let publishConversationID = aiConversationID
                let answer = try await service.answer(
                    question: question,
                    conversationID: aiConversationID,
                    targetLocaleID: pipelineStore.translationTargetLocaleId,
                    analyticsOperation: analyticsOperations.operation(
                        for: finalizeUtteranceId
                    )
                ) { [weak self] partial in
                    Task { @MainActor in
                        self?.publishStreamingAIAnswerIfNeeded(
                            partial,
                            sessionId: publishSessionId,
                            utteranceId: publishUtteranceId,
                            commandSeq: publishCommandSeq,
                            aiConversationID: publishConversationID,
                            force: partial.isEmpty
                        )
                    }
                }
                // Abort during the LLM await must not commit or deliver.
                guard canStoreTerminal(for: finalizeUtteranceId) else { return }
                // Flush the final draft when throttle skipped the last characters.
                publishStreamingAIAnswerIfNeeded(
                    answer,
                    sessionId: finalizeSessionId,
                    utteranceId: finalizeUtteranceId,
                    commandSeq: finalizeCommandSeq,
                    aiConversationID: aiConversationID,
                    force: true
                )
                await service.commitSuccessfulTurn(
                    question: question,
                    answer: answer,
                    conversationID: aiConversationID
                )
                claimAndStoreTerminal(utteranceId: finalizeUtteranceId) {
                    storeFinalizedResult(
                        answer,
                        warning: chunkNote,
                        sessionId: finalizeSessionId,
                        utteranceId: finalizeUtteranceId,
                        commandSeq: finalizeCommandSeq,
                        aiConversationID: aiConversationID
                    )
                }
            } catch {
                claimAndStoreTerminal(utteranceId: finalizeUtteranceId) {
                    storeFinalizedError(
                        Self.aiQuestionFailureMessage(for: error),
                        kind: .generic,
                        sessionId: finalizeSessionId,
                        utteranceId: finalizeUtteranceId,
                        commandSeq: finalizeCommandSeq,
                        aiConversationID: aiConversationID
                    )
                }
            }
            return
        }

        let polishContext = PolishContext(
            appContext: pipelineStore.detectedAppContext?.context ?? .unknown,
            precedingText: fieldContext?.precedingText,
            followingText: fieldContext?.followingText,
            fieldHints: fieldContext.map(FieldHints.init(from:)),
            maxPrecedingChars: 600,
            maxFollowingChars: 200
        )

        var delivered = text
        let polishStarted = Date()
        let isEditLastInput = utteranceMode == .editLastInput
        let isInstructionMode = isEditLastInput
        let polishMode: PolishingService.PolishMode = isInstructionMode
            ? .polish
            : pipelineStore.polishModeForPipeline
        let instructionPrompt: (system: String, user: String)? = {
            if isEditLastInput,
               let source = editSourceText,
               !source.isEmpty {
                let input = EditLastInputPromptComposer.Input(
                    sourceText: source,
                    spokenInstruction: textForPolish
                )
                return (
                    EditLastInputPromptComposer.systemPrompt(),
                    EditLastInputPromptComposer.userMessage(input)
                )
            }
            return nil
        }()

        if isInstructionMode, instructionPrompt == nil {
            FlowDiagnostics.log("instruction edit missing source — failing closed")
            guard claimTerminal(utteranceId: finalizeUtteranceId) else { return }
            storeFinalizedError(
                AppL10n.string("flow.error.editLastInputFailed"),
                kind: .generic,
                sessionId: finalizeSessionId,
                utteranceId: finalizeUtteranceId,
                commandSeq: finalizeCommandSeq
            )
            return
        }

        let modeLabel = isEditLastInput
            ? "editLastInput"
            : Self.polishModeLogLabel(polishMode)
        FlowDiagnostics.log(
            "finalize LLM mode=\(modeLabel) translationTarget=\(pipelineStore.translationTargetLocaleId)"
        )
        FlowTrace.transcript(
            "polish.input",
            textForPolish,
            "mode=\(modeLabel) engine=\(engineMode) "
                + "provider=\(pipelineStore.polishProviderIdOverride ?? "default") "
                + "recordedSeconds=\(String(format: "%.2f", recordingDuration))"
        )
        do {
            // If the finalize task was cancelled (cold-start churn / abort),
            // skip the LLM round-trip and deliver the raw transcript so the
            // keyboard is not left waiting on a result that never arrives.
            // Edit mode must never insert the instruction ASR.
            if Task.isCancelled {
                throw CancellationError()
            }
            let outcome = try await Self.polishWithHostTimeout(
                polisher: polisher,
                text: instructionPrompt?.user ?? textForPolish,
                mode: polishMode,
                systemPrompt: instructionPrompt?.system,
                providerIdOverride: pipelineStore.polishProviderIdOverride,
                taskKind: isEditLastInput ? .editLastInput : nil,
                requestPurpose: managedRequestPurpose,
                oobeFeature: managedOOBEFeature,
                context: isInstructionMode ? nil : polishContext,
                timeoutLimit: processingDeadlineAt.map {
                    max(0.1, $0 - Date().timeIntervalSince1970)
                }
            )
            let polished: String
            if isEditLastInput, let source = editSourceText {
                switch EditOutputValidator.validate(sourceText: source, output: outcome.text) {
                case .success(let validated):
                    polished = validated
                case .failure(let validationError):
                    throw validationError
                }
            } else {
                polished = outcome.text
            }
            delivered = polished
            FlowTrace.transcript(
                "polish.output",
                polished,
                "mode=\(modeLabel) inputLen=\(text.count) "
                    + "changed=\(polished == text ? 0 : 1) "
                    + "elapsed=\(FlowTrace.seconds(since: polishStarted))s"
            )
            guard claimTerminal(utteranceId: finalizeUtteranceId) else { return }
            let historyEntry = isEditLastInput
                ? nil
                : SpeechHistoryStore.shared.recordUtterance(
                    text: delivered,
                    prePolishText: text,
                    polishStyleID: outcome.polishStyleID
                        ?? pipelineStore.activePolishStyleId,
                    polishStylePrompt: outcome.polishStylePrompt,
                    engineMode: engineMode,
                    duration: recordingDuration,
                    wasTranslation: isInstructionMode
                        ? false
                        : pipelineStore.isTranslationEffective
                )
            storeFinalizedResult(
                polished,
                rawText: isInstructionMode ? nil : text,
                warning: Self.combinedWarning(
                    chunkNote,
                    outcome.qualityDegraded
                        ? AppL10n.string("flow.warning.polishDegradedQuality")
                        : nil
                ),
                sessionId: finalizeSessionId,
                utteranceId: finalizeUtteranceId,
                commandSeq: finalizeCommandSeq,
                historyEntryID: isEditLastInput
                    ? sourceHistoryEntryID
                    : historyEntry?.id,
                historyEntryRevision: isEditLastInput
                    ? sourceHistoryEntryRevision
                    : historyEntry?.revision
            )
            FlowDiagnostics.log(
                "polish done in \(String(format: "%.1f", Date().timeIntervalSince(polishStarted)))s " +
                "total=\(String(format: "%.1f", Date().timeIntervalSince(pipelineStarted)))s"
            )
        } catch {
            if isInstructionMode {
                FlowDiagnostics.log(
                    "instruction edit failed after \(String(format: "%.1f", Date().timeIntervalSince(polishStarted)))s: " +
                    "\(Self.safeErrorLogMetadata(error))"
                )
                FlowTrace.warn(
                    "editLastInput.failed",
                    "elapsed=\(FlowTrace.seconds(since: polishStarted))s "
                        + "cancelled=\(error is CancellationError ? 1 : 0) "
                        + "\(Self.safeErrorLogMetadata(error))"
                )
                guard claimTerminal(utteranceId: finalizeUtteranceId) else { return }
                storeFinalizedError(
                    AppL10n.string("flow.error.editLastInputFailed"),
                    kind: .generic,
                    sessionId: finalizeSessionId,
                    utteranceId: finalizeUtteranceId,
                    commandSeq: finalizeCommandSeq
                )
            } else {
                // CancellationError is common when the user jumps back via
                // startflow mid-polish; still deliver raw text. Other errors
                // keep the existing polish-warning fallback.
                let fallback = Self.makeFallbackDelivery(
                    rawText: text,
                    error: error,
                    engineMode: engineMode,
                    chunkWarning: chunkNote
                )
                FlowDiagnostics.log(
                    "polish failed after \(String(format: "%.1f", Date().timeIntervalSince(polishStarted)))s: " +
                    "\(Self.safeErrorLogMetadata(error))"
                )
                FlowTrace.warn(
                    "polish.failed",
                    "mode=\(Self.polishModeLogLabel(polishMode)) engine=\(engineMode) "
                        + "elapsed=\(FlowTrace.seconds(since: polishStarted))s "
                        + "cancelled=\(error is CancellationError ? 1 : 0) "
                        + "\(Self.safeErrorLogMetadata(error))"
                )
                FlowTrace.transcript(
                    "polish.fallback",
                    fallback.text,
                    "reason=polishFailed rawLen=\(text.count)"
                )
                delivered = fallback.text
                guard claimTerminal(utteranceId: finalizeUtteranceId) else { return }
                let historyEntry = SpeechHistoryStore.shared.recordUtterance(
                    text: delivered,
                    prePolishText: text,
                    polishStyleID: pipelineStore.activePolishStyleId,
                    // The provider failed, so this fallback was not produced by
                    // the style Prompt and must not become Prompt-learning evidence.
                    polishStylePrompt: nil,
                    engineMode: engineMode,
                    duration: recordingDuration,
                    wasTranslation: pipelineStore.isTranslationEffective
                )
                storeFinalizedResult(
                    fallback.text,
                    rawText: text,
                    warning: fallback.polishWarning,
                    sessionId: finalizeSessionId,
                    utteranceId: finalizeUtteranceId,
                    commandSeq: finalizeCommandSeq,
                    historyEntryID: historyEntry?.id,
                    historyEntryRevision: historyEntry?.revision
                )
            }
        }

        currentPartial = ""
        lastFinal = ""
        lastFinalWithPauseMarks = ""
        bestPartialSnapshot = ""
        utterancePCMSamples = []
        chunkWarnings = []
        chunkedPipeline = nil
        debug("utterance finalized length=\(text.count)")
    }

    /// Drop the processing gate and republish hostReady after finalize.
    /// Must not depend on a perfect id match — a racing fail/abort/cancel
    /// used to skip this block and leave the keyboard stuck on white「识别中」
    /// even after "utterance finalized" was logged.
    private func completeFinalizeCleanup(sessionId: UUID?, utteranceId: UUID?) {
        // A newer utterance may have started; never clobber its gate.
        if let current = currentUtteranceId,
           let finished = utteranceId,
           current != finished {
            debug(
                "finalize cleanup skipped — newer utterance live " +
                "finished=\(finished.uuidString.prefix(8)) current=\(current.uuidString.prefix(8))"
            )
            return
        }

        let wasProcessing = isUtteranceProcessing
        isUtteranceProcessing = false
        if currentUtteranceId == utteranceId || currentUtteranceId == nil {
            currentUtteranceId = nil
            currentCommandSeq = 0
        }
        refreshHostReady()
        debug(
            "finalize cleanup done wasProcessing=\(wasProcessing ? 1 : 0) " +
            "utterance=\(utteranceId?.uuidString.prefix(8) ?? "nil") " +
            "session=\(sessionId?.uuidString.prefix(8) ?? "nil")"
        )
    }

    private func publishStreamingAIAnswerIfNeeded(
        _ text: String,
        sessionId: UUID?,
        utteranceId: UUID?,
        commandSeq: Int64,
        aiConversationID: UUID?,
        force: Bool
    ) {
        guard aiAnswerStreamThrottle.shouldPublish(
            accumulatedCount: text.count,
            force: force
        ) else { return }
        storeStreamingAIAnswer(
            text,
            sessionId: sessionId,
            utteranceId: utteranceId,
            commandSeq: commandSeq,
            aiConversationID: aiConversationID
        )
    }

    private func storeStreamingAIAnswer(
        _ text: String,
        sessionId: UUID?,
        utteranceId: UUID?,
        commandSeq: Int64,
        aiConversationID: UUID?
    ) {
        guard let sessionId, let utteranceId else { return }
        guard currentUtteranceId == utteranceId,
              !terminalUtteranceIds.contains(utteranceId) else { return }
        FlowSessionBridge.writeResult(
            FlowResult(
                sessionId: sessionId,
                utteranceId: utteranceId,
                commandSeq: commandSeq,
                status: .streaming,
                text: text,
                hostGeneration: FlowSessionBridge.currentHostGeneration(),
                revision: Self.resultRevision(),
                fieldFingerprint: Self.fieldFingerprint(pendingFieldContext),
                utteranceMode: .aiQuestion,
                aiConversationID: aiConversationID
            )
        )
    }

    private func storeFinalizedResult(
        _ text: String,
        rawText: String? = nil,
        warning: String?,
        sessionId: UUID?,
        utteranceId: UUID?,
        commandSeq: Int64,
        historyEntryID: UUID? = nil,
        historyEntryRevision: Int64? = nil,
        aiConversationID: UUID? = nil
    ) {
        let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !trimmed.isEmpty else {
            storeFinalizedError(
                AppL10n.string("flow.error.noSpeech"),
                kind: .noSpeech,
                sessionId: sessionId,
                utteranceId: utteranceId,
                commandSeq: commandSeq
            )
            return
        }
        guard let sessionId, let utteranceId else { return }
        if currentUtteranceMode == .aiQuestion {
            FlowTrace.pipeline(
                "aiQuestion.delivered",
                "answerLen=\(trimmed.count) "
                    + "utterance=\(utteranceId.uuidString.prefix(8)) "
                    + "commandSeq=\(commandSeq)"
            )
        } else {
            FlowTrace.transcript(
                "host.delivered",
                trimmed,
                "utterance=\(utteranceId.uuidString.prefix(8)) commandSeq=\(commandSeq) "
                    + "warning=\(warning == nil ? 0 : 1)"
            )
        }
        FlowSessionBridge.writeResult(
            FlowResult(
                sessionId: sessionId,
                utteranceId: utteranceId,
                commandSeq: commandSeq,
                status: .final,
                text: trimmed,
                warning: warning,
                rawText: rawText,
                hostGeneration: FlowSessionBridge.currentHostGeneration(),
                revision: Self.resultRevision(),
                fieldFingerprint: Self.fieldFingerprint(pendingFieldContext),
                utteranceMode: currentUtteranceMode,
                historyEntryID: historyEntryID,
                historyEntryRevision: historyEntryRevision,
                aiConversationID: aiConversationID
            )
        )
    }

    private func storeRawCandidate(
        _ text: String,
        sessionId: UUID?,
        utteranceId: UUID?,
        commandSeq: Int64
    ) {
        let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !trimmed.isEmpty, let sessionId, let utteranceId else { return }
        FlowSessionBridge.writeResult(
            FlowResult(
                sessionId: sessionId,
                utteranceId: utteranceId,
                commandSeq: commandSeq,
                status: .rawReady,
                text: trimmed,
                rawText: trimmed,
                hostGeneration: FlowSessionBridge.currentHostGeneration(),
                revision: Self.resultRevision(),
                fieldFingerprint: Self.fieldFingerprint(pendingFieldContext),
                utteranceMode: currentUtteranceMode,
                aiConversationID: pendingAIConversationID
            )
        )
    }

    private func storeFinalizedError(
        _ message: String,
        kind: FlowSessionKeys.TranscriptionErrorKind,
        sessionId: UUID?,
        utteranceId: UUID?,
        commandSeq: Int64,
        status: FlowResult.Status = .error,
        aiConversationID: UUID? = nil
    ) {
        finishUtteranceAnalytics(
            utteranceID: utteranceId,
            success: false,
            failureCategory: Self.transcriptionFailureCategory(
                kind: kind,
                status: status
            )
        )
        guard let sessionId, let utteranceId else { return }
        FlowTrace.warn(
            "host.deliveredError",
            "kind=\(kind.rawValue) status=\(status.rawValue) "
                + "utterance=\(utteranceId.uuidString.prefix(8)) commandSeq=\(commandSeq) "
                + "messageBytes=\(message.utf8.count)"
        )
        FlowSessionBridge.writeResult(
            FlowResult(
                sessionId: sessionId,
                utteranceId: utteranceId,
                commandSeq: commandSeq,
                status: status,
                text: message,
                errorKind: kind,
                hostGeneration: FlowSessionBridge.currentHostGeneration(),
                revision: Self.resultRevision(),
                fieldFingerprint: Self.fieldFingerprint(pendingFieldContext),
                utteranceMode: currentUtteranceMode,
                aiConversationID: aiConversationID ?? pendingAIConversationID
            )
        )
    }

    private static func analyticsFeature(
        for taskKind: ManagedGatewayTaskKind
    ) -> AnalyticsFeature {
        FlowAnalyticsFeatureMapping.feature(for: taskKind)
    }

    private var analyticsExecutionMode: AnalyticsExecutionMode {
        if store.engineMode == "local" {
            return .local
        }
        return store.credentialSource == .managed ? .managed : .byok
    }

    private func finishUtteranceAnalytics(
        utteranceID: UUID?,
        success: Bool,
        failureCategory: AnalyticsFailureCategory = .unknown
    ) {
        if success {
            analyticsOperations.succeed(utteranceID: utteranceID)
        } else {
            analyticsOperations.fail(
                utteranceID: utteranceID,
                category: failureCategory
            )
        }
    }

    private static func transcriptionFailureCategory(
        kind: FlowSessionKeys.TranscriptionErrorKind,
        status: FlowResult.Status
    ) -> AnalyticsFailureCategory {
        if status == .aborted {
            return .cancelled
        }
        if status == .timeout {
            return .timeout
        }
        switch kind {
        case .noSpeech, .discardedEmpty:
            return .validation
        case .recognitionInterrupted:
            return .cancelled
        case .asrFailed, .audioUnavailable, .generic:
            return .provider
        }
    }

    private static var lastResultRevision: Int64 = 0

    private static func resultRevision() -> Int64 {
        let millis = Int64(Date().timeIntervalSince1970 * 1_000)
        lastResultRevision = max(lastResultRevision + 1, millis)
        return lastResultRevision
    }

    private static func fieldFingerprint(_ context: FlowFieldContext?) -> String? {
        context?.deliveryFingerprint
    }

    private static func polishModeLogLabel(_ mode: PolishingService.PolishMode) -> String {
        switch mode {
        case .polish:
            return "polish"
        case .translate(let targetLocaleId):
            return "translate(\(targetLocaleId))"
        }
    }

    private static func chunkWarningMessage(_ warnings: [String]) -> String? {
        guard !warnings.isEmpty else { return nil }
        return warnings.joined(separator: "\n")
    }

    private static func combinedWarning(_ values: String?...) -> String? {
        let present = values.compactMap { value -> String? in
            guard let value, !value.isEmpty else { return nil }
            return value
        }
        return present.isEmpty ? nil : present.joined(separator: "\n")
    }

    private func consumeRecordingDuration() -> TimeInterval {
        defer { utteranceRecordingStartedAt = nil }
        guard let start = utteranceRecordingStartedAt else { return 0 }
        return max(0, Date().timeIntervalSince(start))
    }

    /// Surface polish failures with a localised hint (e.g. missing API key)
    /// rather than letting the keyboard show a generic network error.
    static func makeFallbackDelivery(
        rawText: String,
        error: Error,
        engineMode: String,
        chunkWarning: String?
    ) -> TranscriptionDelivery {
        TranscriptionPolishFallback.makeDelivery(
            rawText: rawText,
            error: error,
            engineMode: engineMode,
            chunkWarning: chunkWarning
        )
    }

    /// Re-transcribe the full utterance PCM when pipelined chunking likely dropped tail text.
    private func runBatchASRFallback(currentText: String) async -> String {
        let samples = utterancePCMSamples
        guard !samples.isEmpty else {
            FlowTrace.warn("pipeline.batchFallback.noPCM", "currentLen=\(currentText.count)")
            return currentText
        }

        let locale = SpeechLocaleResolver.resolve(store.localeId)
        let stitched = lastFinal.trimmingCharacters(in: .whitespacesAndNewlines)
        let partial = bestPartialSnapshot.trimmingCharacters(in: .whitespacesAndNewlines)

        FlowDiagnostics.log(
            "batch fallback start samples=\(samples.count) stitchedLen=\(stitched.count) partialLen=\(partial.count)"
        )

        let asrService = asr
        let result = await HardTimeout.value(
            seconds: FlowSessionKeys.batchASRFallbackTimeout
        ) { [asrService] in
            await asrService.transcribeChunk(samples: samples, locale: locale)
        } onTimeout: {
            .cancelled
        }

        switch result {
        case .success(let batchText):
            let trimmedBatch = batchText.trimmingCharacters(in: .whitespacesAndNewlines)
            FlowTrace.transcript(
                "asr.batchFallback",
                trimmedBatch,
                "samples=\(samples.count) seconds=\(FlowTrace.seconds(samples: samples.count)) "
                    + "rms=\(FlowTrace.rms(samples)) currentLen=\(currentText.count)"
            )
            guard !trimmedBatch.isEmpty else { return currentText }
            let resolved = UtteranceBatchFallbackPolicy.preferredTranscript(
                batch: trimmedBatch,
                stitchedFinal: stitched,
                partialSnapshot: partial,
                current: currentText
            )
            FlowPipelineDiagnostics.logBatchFallback(
                sampleCount: samples.count,
                stitchedLength: stitched.count,
                partialLength: partial.count,
                batchLength: trimmedBatch.count
            )
            return resolved
        case .failure(let message):
            FlowDiagnostics.log(
                "batch fallback failed category=asrFailure errorBytes=\(message.utf8.count)"
            )
            FlowTrace.warn(
                "asr.batchFallback.failed",
                "samples=\(samples.count) rms=\(FlowTrace.rms(samples)) "
                    + "errorCategory=asrFailure errorBytes=\(message.utf8.count)"
            )
            return currentText
        case .cancelled:
            FlowTrace.asr("batchFallback.cancelled", "samples=\(samples.count)")
            return currentText
        }
    }

    private func asrWaitTimeout() -> TimeInterval {
        // Local engine is iOS `SpeechAnalyzer` only, so the
        // previous Qwen3-specific timeout collapses into the shared
        // local path.
        if store.engineMode == "local" {
            return FlowSessionKeys.localASRWaitTimeout
        }
        return FlowSessionKeys.cloudASRWaitTimeout
    }

    /// Host-level polish cap — does not wait for a cancelled LLM task to unwind.
    private static func polishWithHostTimeout(
        polisher: PolishingService,
        text: String,
        mode: PolishingService.PolishMode,
        systemPrompt: String? = nil,
        providerIdOverride: String?,
        taskKind: ManagedGatewayTaskKind? = nil,
        requestPurpose: ManagedGatewayRequestPurpose? = nil,
        oobeFeature: ManagedGatewayOOBEFeature? = nil,
        context: PolishContext?,
        timeoutLimit: TimeInterval? = nil
    ) async throws -> PolishingService.PolishOutcome {
        let scaled = FlowSessionKeys.polishTimeout(forCharacterCount: text.count)
        let timeout = timeoutLimit.map { min(scaled, $0) } ?? scaled
        return try await HardTimeout.run(seconds: timeout) {
            try await polisher.polishWithOutcome(
                text,
                mode: mode,
                systemPrompt: systemPrompt,
                providerIdOverride: providerIdOverride,
                taskKind: taskKind,
                requestPurpose: requestPurpose,
                oobeFeature: oobeFeature,
                context: context
            )
        }
    }

    // MARK: - Level publishing (main thread only)

    private func startLevelPublishing() {
        levelTask?.cancel()
        levelTask = Task { @MainActor [weak self] in
            while !Task.isCancelled {
                guard let self, self.isActive else { break }
                guard self.isUtteranceRecording || self.capture.engineIsLive else {
                    try? await Task.sleep(nanoseconds: 1_000_000_000)
                    continue
                }
                let levels = self.capture.currentAudioLevels()
                self.pipController.updateWaveformLevels(levels)
                if levels.contains(where: { $0 > 0 }) {
                    FlowSessionBridge.storeAudioLevels(levels)
                }
                try? await Task.sleep(nanoseconds: 100_000_000)
            }
        }
    }

    // MARK: - Timers

    private func startHeartbeat() {
        heartbeatTask?.cancel()
        FlowSessionBridge.writeHeartbeat()
        heartbeatTask = Task { @MainActor [weak self] in
            while !Task.isCancelled {
                guard let self else { break }
                if self.isActive,
                   !self.capture.engineIsLive,
                   self.isUtteranceRecording || self.isUtteranceProcessing {
                    await self.reactivateCaptureIfNeeded()
                }
                if self.wantsActiveSession,
                   !self.pipController.isPictureInPictureActive {
                    if self.isAppForeground {
                        self.ensurePiPActive(trigger: .healthCheck, duration: nil)
                    } else {
                        self.pipLifecycleState = .waitingForForeground
                        // Deduped to one report per episode — this branch runs
                        // once a second for as long as the loss lasts.
                        self.reportPiPLossIfNeeded(origin: "heartbeatHealthCheck")
                    }
                }
                FlowSessionBridge.writeHeartbeat()
                self.refreshHostReady()
                try? await Task.sleep(nanoseconds: 1_000_000_000)
                guard self.isActive else { break }
            }
        }
    }

    private func debug(_ message: String) {
        FlowDiagnostics.log(message)
    }

    /// Why a bounded PiP recovery gave up mid-loop. These aborts are not
    /// failures from the host's point of view — but they *are* the user-visible
    /// failure on the keyboard side, which sits there until its start budget
    /// expires. Naming the cause is what makes the two timelines line up.
    private func pipRecoveryAbortReason(operation: UInt64) -> String {
        var causes: [String] = []
        if operation != pipRecoveryOperation { causes.append("supersededOperation") }
        if !wantsActiveSession { causes.append("sessionIntentCleared") }
        if !isAppForeground { causes.append("appBackgrounded") }
        if Task.isCancelled { causes.append("taskCancelled") }
        return causes.isEmpty ? "unknown" : causes.joined(separator: "+")
    }

    private func startupFailureContext(
        trigger: String,
        attemptCount: Int,
        elapsed: TimeInterval
    ) -> [String: String] {
        let memory = OSGDiag.memorySnapshot()
        return [
            "trigger": trigger,
            "attemptCount": "\(attemptCount)",
            "elapsedSeconds": String(format: "%.3f", elapsed),
            "appForeground": isAppForeground ? "true" : "false",
            "sessionActive": isActive ? "true" : "false",
            "sessionIntent": wantsActiveSession ? "true" : "false",
            "pipLifecycleState": String(describing: pipLifecycleState),
            "pipGeneration": "\(pipController.generation)",
            "pipActive": pipController.isPictureInPictureActive ? "true" : "false",
            "hostReady": FlowSessionBridge.isHostReady() ? "true" : "false",
            "bridgeSessionActive": FlowSessionBridge.isSessionActive() ? "true" : "false",
            "heartbeatStalenessSeconds": FlowSessionBridge.heartbeatStaleness()
                .map { String(format: "%.3f", $0) } ?? "nil",
            "microphonePermission": String(describing: AppPermissions.micStatus),
            "speechPermission": String(describing: AppPermissions.speechStatus),
            "rssMB": String(format: "%.1f", memory.rssMB),
            "physicalFootprintMB": String(format: "%.1f", memory.physFootprintMB)
        ]
    }

    private static func safeErrorLogMetadata(_ error: Error) -> String {
        "errorCategory=\(String(reflecting: type(of: error))) "
            + "errorBytes=\(error.localizedDescription.utf8.count)"
    }

    private static func aiQuestionFailureMessage(for error: Error) -> String {
        if let webPageError = error as? WebPageContentError {
            switch webPageError {
            case .invalidURL:
                return AppL10n.string("flow.error.webPageInvalid")
            case .unsupportedContentType, .undecodableText, .emptyContent:
                return AppL10n.string("flow.error.webPageEmpty")
            case .tooManyRedirects, .invalidResponse, .httpStatus,
                 .responseTooLarge, .transport:
                return AppL10n.string("flow.error.webPageUnavailable")
            }
        }
        if let managedError = error as? ManagedGatewayError {
            return managedError.localizedDescription
        }
        return AppL10n.string("flow.error.aiQuestionFailed")
    }

    private func traceIgnoredCommand(reason: String, command: FlowCommand, detail: String) {
        let signature = "\(reason)|\(command.action.rawValue)|\(command.commandSeq)|\(command.sessionId.uuidString)|\(command.utteranceId.uuidString)|\(detail)"
        guard signature != lastIgnoredCommandSignature else { return }
        lastIgnoredCommandSignature = signature
        traceState(
            "command.ignored",
            extra: "reason=\(reason) action=\(command.action.rawValue) seq=\(command.commandSeq) \(detail)"
        )
    }

    private func traceState(_ event: String, extra: String? = nil) {
        let staleness = FlowSessionBridge.heartbeatStaleness().map { String(format: "%.1f", $0) } ?? "nil"
        let sessionId = activeSessionId?.uuidString ?? "nil"
        let utteranceId = currentUtteranceId?.uuidString ?? "nil"
        let summary = [
            "event=\(event)",
            "active=\(isActive)",
            "starting=\(isStarting)",
            "wantsActive=\(wantsActiveSession)",
            "pipState=\(String(describing: pipLifecycleState))",
            "pipGeneration=\(pipController.generation)",
            "coldStart=\(isColdStartHandoff)",
            "sessionId=\(sessionId)",
            "utteranceId=\(utteranceId)",
            "cmdSeq=\(currentCommandSeq)",
            "lastCmd=\(lastHandledCommandSeq)",
            "recording=\(isUtteranceRecording)",
            "processing=\(isUtteranceProcessing)",
            "storeEngine=\(store.engineMode)",
            "boundEngine=\(sessionASREngineMode ?? "nil")",
            "engineLive=\(capture.engineIsLive)",
            "hostReady=\(FlowSessionBridge.isHostReady())",
            "sessionActive=\(FlowSessionBridge.isSessionActive())",
            "heartbeatStaleness=\(staleness)"
        ].joined(separator: " ")
        if let extra, !extra.isEmpty {
            debug("[trace] \(summary) \(extra)")
        } else {
            debug("[trace] \(summary)")
        }
    }
}