openclaw-openclaw/apps/shared/OpenClawKit/Sources/OpenClawChatUI/ChatViewModel+Outbox.swift

import Foundation
import OpenClawKit
import OSLog

private let outboxLogger = Logger(subsystem: "ai.openclaw", category: "OpenClawChatOutbox")

/// Display state for a transcript row backed by a durable outbox command.
public enum OpenClawChatOutboxMessageState: Equatable, Sendable {
    case queued
    case sending
    case confirming
    case failed(reason: String?)

    public var isFailed: Bool {
        if case .failed = self { return true }
        return false
    }

    var preventsDeletion: Bool {
        self == .sending || self == .confirming
    }
}

/// Durable offline command outbox. Sends made while the gateway is unhealthy
/// are persisted (per gateway, alongside the transcript cache) and flushed
/// strictly in createdAt order when health recovers. A gateway ACK only moves
/// a row to awaiting-confirmation; canonical history owns durable completion.
extension OpenClawChatViewModel {
    struct OutboxDeliveryTarget: Hashable {
        let presentationSessionKey: String
        let deliverySessionKey: String
        let agentID: String?
    }

    public func outboxState(for messageID: UUID) -> OpenClawChatOutboxMessageState? {
        self.outboxStatesByMessageID[messageID]
    }

    /// Tap-to-retry for a failed command: reset attempts, refresh createdAt
    /// (so even an expired row can send again), and flush if healthy.
    public func retryOutboxMessage(_ messageID: UUID) {
        guard let outbox, let commandID = self.outboxCommandIDsByMessageID[messageID] else { return }
        let session = self.currentSessionSnapshot()
        Task { [weak self] in
            guard let self else { return }
            let agentID = self.outboxAgentID(for: session)
            if self.outboxRequiresAgentID(for: session), agentID == nil {
                self.errorText = "Select an agent before retrying this message."
                return
            }
            guard
                let deliverySessionKey = self.outboxDeliverySessionKey(for: session, agentID: agentID),
                let routingContract = self.outboxRoutingContract(for: session)
            else {
                self.errorText = "Reconnect to verify this message's delivery target before retrying."
                return
            }
            let result = await outbox.markCommandRetriedIfPresent(
                id: commandID,
                agentID: agentID,
                deliverySessionKey: deliverySessionKey,
                routingContract: routingContract)
            if result == .updated {
                // Durable work is gateway-global. Flush even when the visible
                // session changed while the SQLite update was suspended.
                self.flushOutboxIfNeeded()
            }
            guard self.isCurrentSession(session) else { return }
            switch result {
            case .updated:
                self.outboxStatesByMessageID[messageID] = .queued
            case .missing, .confirmed:
                self.clearOutboxState(forCommandID: commandID)
            case .unavailable:
                self.errorText = "Could not retry the queued message. Try again."
            }
        }
    }

    public func deleteOutboxMessage(_ messageID: UUID) {
        guard let outbox, let commandID = self.outboxCommandIDsByMessageID[messageID] else { return }
        self.cancelingOutboxCommandIDs.insert(commandID)
        self.outboxPresentationGeneration &+= 1
        Task { [weak self] in
            guard let self else { return }
            let result = await outbox.cancelCommand(id: commandID)
            guard result != .unavailable else {
                self.finishOutboxCancellation(commandID)
                self.errorText = "Could not delete the queued message. Try again."
                return
            }
            if result == .confirmed {
                self.finishOutboxCancellation(commandID)
                self.clearOutboxState(forCommandID: commandID)
                return
            }
            if result == .missing {
                let presentationGeneration = self.outboxPresentationGeneration
                let current = await outbox.loadCommands().first(where: { $0.id == commandID })
                guard presentationGeneration == self.outboxPresentationGeneration else {
                    self.finishOutboxCancellation(commandID)
                    return
                }
                if let current {
                    // Another view model claimed the row first. Keep the bubble
                    // and show its authoritative state; cancellation after claim
                    // would promise deletion while the request can still land.
                    self.finishOutboxCancellation(commandID)
                    self.presentOutboxCommands([current])
                    return
                }
            }
            if self.canonicalOutboxMessageKeys.contains("\(commandID):user") {
                self.finishOutboxCancellation(commandID)
                self.clearOutboxState(forCommandID: commandID)
                return
            }
            self.finishOutboxCancellation(commandID)
            self.outboxCommandIDsByMessageID.removeValue(forKey: messageID)
            self.outboxMessageIDsByCommandID.removeValue(forKey: commandID)
            self.outboxStatesByMessageID.removeValue(forKey: messageID)
            // `.missing` with no current row means another view already
            // canceled (or canonical history completed) this mapping. Never
            // leave its stale bubble looking like an ordinary sent message;
            // canonical history can re-add a genuinely delivered row.
            self.replaceMessages(self.messages.filter { $0.id != messageID })
        }
    }

    // MARK: - Capture

    /// Durable capture path used before text or attachment delivery: persist
    /// first, then render the queued bubble. A full queue refuses the enqueue
    /// and keeps the exact draft intact.
    func enqueueOutboxCommand(
        text: String,
        draftInput: String,
        draftAttachments: [OpenClawPendingAttachment] = [],
        session: SessionSnapshot) async
    {
        guard let outbox else { return }
        let agentID = self.outboxAgentID(for: session)
        if self.outboxRequiresAgentID(for: session), agentID == nil {
            self.errorText = "Select an agent before queueing this message."
            return
        }
        guard
            let deliverySessionKey = self.outboxDeliverySessionKey(for: session, agentID: agentID),
            let routingContract = self.outboxRoutingContract(for: session)
        else {
            self.errorText = "Reconnect to verify this message's delivery target before queueing."
            return
        }
        let command = OpenClawChatOutboxCommand(
            id: UUID().uuidString,
            sessionKey: session.key,
            deliverySessionKey: deliverySessionKey,
            routingContract: routingContract,
            agentID: agentID,
            text: text,
            attachments: draftAttachments.map {
                OpenClawChatOutboxAttachment(
                    type: $0.type,
                    mimeType: $0.mimeType,
                    fileName: $0.fileName,
                    data: $0.data,
                    durationSeconds: $0.durationSeconds)
            },
            thinking: self.effectiveThinkingLevelForSend,
            createdAt: Date().timeIntervalSince1970,
            status: .queued,
            retryCount: 0,
            lastError: nil)
        let accepted = await outbox.enqueueCommand(command)
        guard self.isCurrentSession(session) else { return }
        guard accepted else {
            self.errorText = "Offline queue is full. Delete a queued message or reconnect to send."
            return
        }
        if self.input == draftInput { self.input = "" }
        let capturedAttachmentIDs = Set(draftAttachments.map(\.id))
        self.attachments.removeAll { capturedAttachmentIDs.contains($0.id) }
        self.errorText = nil
        self.presentOutboxCommands([command])
        // Health can recover between the send-gate check and the enqueue;
        // flushing here closes that gap instead of waiting for the next event.
        if self.healthOK {
            self.flushOutboxIfNeeded()
        }
    }

    /// Durable path for a failed live text send. Known pre-dispatch route
    /// changes remain queued; ambiguous results fail closed until canonical
    /// history or explicit retry resolves them.
    /// Returns false when the queue refuses (caller keeps the failure path).
    func preserveFailedLiveSend(
        runId: String,
        text: String,
        thinking: String,
        messageID: UUID,
        session: SessionSnapshot,
        deliveryIsAmbiguous: Bool) async -> Bool
    {
        guard let outbox else { return false }
        let agentID = self.outboxAgentID(for: session)
        guard !self.outboxRequiresAgentID(for: session) || agentID != nil,
              let deliverySessionKey = self.outboxDeliverySessionKey(for: session, agentID: agentID),
              let routingContract = self.outboxRoutingContract(for: session)
        else { return false }
        let command = OpenClawChatOutboxCommand(
            id: runId,
            sessionKey: session.key,
            deliverySessionKey: deliverySessionKey,
            routingContract: routingContract,
            agentID: agentID,
            text: text,
            thinking: thinking,
            createdAt: Date().timeIntervalSince1970,
            status: deliveryIsAmbiguous ? .failed : .queued,
            retryCount: 0,
            lastError: deliveryIsAmbiguous
                ? OpenClawChatSQLiteTranscriptCache.outboxUnconfirmedError
                : nil)
        guard await outbox.enqueueCommand(command) else { return false }
        guard self.isCurrentSession(session) else { return true }
        self.mapOutboxCommand(command, to: messageID)
        self.errorText = nil
        // Ambiguous rows only reconcile history; known-unsent rows may flush
        // automatically once the replacement route becomes healthy.
        self.flushOutboxIfNeeded()
        return true
    }

    // MARK: - Restore

    /// Session switches drop the visible bubbles, so per-message outbox
    /// state must go with them, and the FIFO send gate must assume a backlog
    /// again until restore adopts the new session's durable rows. Without
    /// this reset a send issued right after a switch could go live ahead of
    /// that session's persisted queue.
    func resetOutboxPresentationForSessionSwitch() {
        self.hasRestoredOutboxMessages = false
        self.outboxCommandIDsByMessageID.removeAll()
        self.outboxMessageIDsByCommandID.removeAll()
        self.outboxStatesByMessageID.removeAll()
    }

    /// Re-adopts or re-appends queued bubbles for the visible session after
    /// cold open, session switches, and wholesale history replacement.
    func restoreOutboxMessages(session: SessionSnapshot) {
        guard let outbox else { return }
        Task { [weak self] in
            guard let self else { return }
            guard await self.recoverInterruptedOutboxSendsIfNeeded() else { return }
            while self.isCurrentSession(session) {
                let presentationGeneration = self.outboxPresentationGeneration
                guard let commands = await outbox.loadCommandsIfAvailable() else { return }
                guard self.isCurrentSession(session) else { return }
                if presentationGeneration != self.outboxPresentationGeneration {
                    // A cross-view cancellation/confirmation invalidated this
                    // snapshot. Reload so unrelated surviving rows still paint.
                    continue
                }
                self.presentOutboxCommands(commands.filter { self.commandMatchesTarget($0, session: session) })
                // The FIFO send gate assumes a backlog until this point.
                self.hasRestoredOutboxMessages = true
                // Relaunching while already healthy never sees an unhealthy ->
                // healthy transition, so kick the flush here as well.
                if self.healthOK, commands.contains(where: { $0.status == .queued }) {
                    self.flushOutboxIfNeeded()
                }
                return
            }
        }
    }

    /// Canonical history is the durable acceptance boundary. Any matching
    /// outbox row—including a delivery-unconfirmed row—is now safe to remove
    /// without replaying the user turn.
    func confirmOutboxCommands(in messages: [OpenClawChatMessage]) {
        self.observeCanonicalOutboxMessageKeys(in: messages)
        Task { [weak self] in
            await self?.confirmOutboxCommandsNow(in: messages)
        }
    }

    func confirmOutboxCommandsNow(in messages: [OpenClawChatMessage]) async {
        self.observeCanonicalOutboxMessageKeys(in: messages)
        guard let outbox else { return }
        let confirmedKeys = Set(messages.compactMap { Self.normalizedIdempotencyKey($0.idempotencyKey) })
        guard !confirmedKeys.isEmpty else { return }
        let commands = await outbox.loadCommands().filter { command in
            // Command UUIDs are gateway-global. Match the durable identity,
            // not a presentation alias (`main` vs `agent:<id>:main`).
            confirmedKeys.contains(Self.outboxUserIdempotencyKey(command.id))
        }
        for command in commands {
            if let canonicalMessage = messages.first(where: {
                Self.normalizedIdempotencyKey($0.idempotencyKey) ==
                    Self.outboxUserIdempotencyKey(command.id)
            }) {
                await self.persistCanonicalOutboxEvidence(canonicalMessage, for: command)
            }
            let result = await outbox.confirmCommand(id: command.id)
            if result != .unavailable {
                self.clearOutboxState(forCommandID: command.id)
            }
        }
    }

    private func persistCanonicalOutboxEvidence(
        _ message: OpenClawChatMessage,
        for command: OpenClawChatOutboxCommand) async
    {
        guard let transcriptCache = transcriptCache as? any OpenClawChatCanonicalTranscriptMerging else { return }
        let sessionKey = command.sessionKey
        let cacheAgentID = Self.transcriptCacheAgentID(
            sessionKey: sessionKey,
            agentID: command.agentID)
        let messageKey = Self.outboxUserIdempotencyKey(command.id)
        let canonicalMessage = Self.adoptingCanonicalMessage(
            message,
            over: Self.outboxUserMessage(for: command))
        let previous = self.pendingCacheWriteTask
        let task = Task.detached {
            await previous?.value
            await transcriptCache.mergeCanonicalTranscriptMessage(
                sessionKey: sessionKey,
                agentID: cacheAgentID,
                message: canonicalMessage,
                canonicalMessageIdempotencyKey: messageKey)
        }
        self.pendingCacheWriteTask = task
        await task.value
    }

    private func observeCanonicalOutboxMessageKeys(in messages: [OpenClawChatMessage]) {
        let keys = Set(messages.compactMap(\.idempotencyKey))
        for key in keys.sorted() {
            self.canonicalOutboxMessageKeys.removeAll(where: { $0 == key })
            self.canonicalOutboxMessageKeys.append(key)
        }
        if self.canonicalOutboxMessageKeys.count > 512 {
            self.canonicalOutboxMessageKeys.removeFirst(self.canonicalOutboxMessageKeys.count - 512)
        }
        self.transcriptCache?.observeCanonicalMessageIdempotencyKeys(keys)
    }

    /// Appends bubbles for commands in the current session, adopting rows
    /// that already carry the command's user idempotency key (cache pre-paint
    /// or an earlier restore), and refreshes their display states.
    private func presentOutboxCommands(_ commands: [OpenClawChatOutboxCommand]) {
        self.pruneOutboxMappings()
        guard !commands.isEmpty else { return }
        var next = self.messages
        for command in commands.sorted(by: { $0.createdAt < $1.createdAt }) {
            if self.cancelingOutboxCommandIDs.contains(command.id) { continue }
            let key = Self.outboxUserIdempotencyKey(command.id)
            if let existing = next.first(where: { $0.idempotencyKey == key }) {
                self.mapOutboxCommand(command, to: existing.id)
                continue
            }
            let message = Self.outboxUserMessage(for: command)
            next.append(message)
            self.mapOutboxCommand(command, to: message.id)
        }
        self.replaceMessages(next)
    }

    private static func outboxUserMessage(for command: OpenClawChatOutboxCommand) -> OpenClawChatMessage {
        var content = [
            OpenClawChatMessageContent(
                type: "text",
                text: command.text,
                mimeType: nil,
                fileName: nil,
                content: nil),
        ]
        content.append(contentsOf: command.attachments.map { attachment in
            OpenClawChatMessageContent(
                type: attachment.type,
                text: nil,
                mimeType: attachment.mimeType,
                fileName: attachment.fileName,
                durationSeconds: attachment.durationSeconds,
                content: AnyCodable(attachment.data.base64EncodedString()))
        })
        return OpenClawChatMessage(
            role: "user",
            content: content,
            // Message timestamps are milliseconds; outbox rows store seconds.
            timestamp: command.createdAt * 1000,
            idempotencyKey: self.outboxUserIdempotencyKey(command.id))
    }

    private func mapOutboxCommand(_ command: OpenClawChatOutboxCommand, to messageID: UUID) {
        self.outboxCommandIDsByMessageID[messageID] = command.id
        self.outboxMessageIDsByCommandID[command.id] = messageID
        self.outboxStatesByMessageID[messageID] = Self.outboxDisplayState(for: command)
    }

    private func pruneOutboxMappings() {
        let visibleMessageIDs = Set(self.messages.map(\.id))
        let staleMessageIDs = self.outboxCommandIDsByMessageID.keys.filter {
            !visibleMessageIDs.contains($0)
        }
        for messageID in staleMessageIDs {
            if let commandID = self.outboxCommandIDsByMessageID.removeValue(forKey: messageID) {
                self.outboxMessageIDsByCommandID.removeValue(forKey: commandID)
            }
            self.outboxStatesByMessageID.removeValue(forKey: messageID)
        }
    }

    // MARK: - Health

    func pollHealthIfNeeded(force: Bool, sessionSnapshot: SessionSnapshot? = nil) async {
        if !force, let last = lastHealthPollAt, Date().timeIntervalSince(last) < 10 {
            return
        }
        self.lastHealthPollAt = Date()
        do {
            let ok = try await self.transport.requestHealth(timeoutMs: 5000)
            if let sessionSnapshot, !self.isCurrentSession(sessionSnapshot) { return }
            self.applyTransportHealth(ok)
        } catch {
            if let sessionSnapshot, !self.isCurrentSession(sessionSnapshot) { return }
            self.applyTransportHealth(false)
        }
    }

    /// Single choke point for health updates so the offline outbox flushes
    /// exactly on the unhealthy -> healthy transition.
    func applyTransportHealth(_ ok: Bool) {
        let wasHealthy = self.healthOK
        self.healthOK = ok
        if ok, !wasHealthy {
            self.flushOutboxIfNeeded()
        }
    }

    // MARK: - Flush

    func flushOutboxIfNeeded() {
        guard self.outbox != nil, self.healthOK else { return }
        guard !self.isFlushingOutbox else {
            // Coalesce triggers that land mid-pass (tap-to-retry, enqueue
            // race) so their commands are not stranded until the next
            // health transition.
            self.isOutboxFlushRequestedWhileActive = true
            return
        }
        self.isFlushingOutbox = true
        Task { [weak self] in
            await self?.performOutboxFlush()
            guard let self else { return }
            self.isFlushingOutbox = false
            if self.isOutboxFlushRequestedWhileActive {
                self.isOutboxFlushRequestedWhileActive = false
                self.flushOutboxIfNeeded()
            }
        }
    }

    private func performOutboxFlush() async {
        guard let outbox else { return }
        let presentationGeneration = self.outboxPresentationGeneration
        guard let initialCommands = await outbox.loadCommandsIfAvailable() else {
            self.applyTransportHealth(false)
            return
        }
        guard presentationGeneration == self.outboxPresentationGeneration else {
            self.isOutboxFlushRequestedWhileActive = true
            return
        }
        let visibleSession = self.currentSessionSnapshot()
        self.presentOutboxCommands(initialCommands.filter { self.commandMatchesTarget($0, session: visibleSession) })
        // Do not capability-gate ordinary live chat when no durable work
        // needs a replay lease (notably against older gateways).
        let hasRouteWork = initialCommands.contains { command in
            command.status == .queued ||
                command.status == .sending ||
                Self.needsOutboxDeliveryReconciliation(command)
        }
        guard hasRouteWork else { return }
        let routeResult = await self.transport.acquireOutboxRouteLease()
        guard case let .available(routeLease) = routeResult else {
            // The store owner no longer matches the active gateway route.
            // Leave every row queued; the replacement view model owns the
            // new gateway and a later matching reconnect can resume this one.
            if case let .unavailable(reason) = routeResult, let reason {
                self.errorText = reason
            }
            self.applyTransportHealth(false)
            return
        }
        guard await self.recoverInterruptedOutboxSendsIfNeeded() else {
            self.applyTransportHealth(false)
            return
        }
        var confirmationTargets: Set<OutboxDeliveryTarget> = []
        while self.healthOK {
            let presentationGeneration = self.outboxPresentationGeneration
            let commands = await outbox.loadCommands()
            if presentationGeneration != self.outboxPresentationGeneration {
                continue
            }
            confirmationTargets.formUnion(
                commands.lazy
                    .filter(Self.needsOutboxDeliveryReconciliation)
                    .map(Self.deliveryTarget))
            let visibleSession = self.currentSessionSnapshot()
            self.presentOutboxCommands(commands.filter { self.commandMatchesTarget($0, session: visibleSession) })
            guard let next = await outbox.claimNextCommand() else { break }
            if self.transport.outboxRequiresSessionRoutingContract,
               next.routingContract != routeLease.sessionRoutingContract
            {
                guard await self.parkOutboxCommandForChangedTarget(next, outbox: outbox) else { break }
                continue
            }
            // Same ordering contract as the live send path: a run must not
            // start on a stale model while a sessions.patch(model) for its
            // session is still in flight.
            await self.waitForPendingModelPatches(in: next.sessionKey)
            self.setOutboxState(.sending, forCommandID: next.id)
            do {
                let response = try await routeLease.sendMessage(
                    sessionKey: next.deliverySessionKey,
                    agentID: next.agentID,
                    message: next.text,
                    // Preserve the queued level when supported, but never send
                    // an explicit unsupported level after the gate changes.
                    thinking: self.effectiveThinkingLevelForSend(
                        next.thinking,
                        sessionKey: next.sessionKey),
                    idempotencyKey: next.id,
                    attachments: next.attachments.map {
                        OpenClawChatAttachmentPayload(
                            type: $0.type,
                            mimeType: $0.mimeType,
                            fileName: $0.fileName,
                            content: $0.data.base64EncodedString())
                    })
                if response.status == "error" || response.status == "timeout" {
                    // Gateway rejected the run: this burns a retry attempt,
                    // unlike transport-level failures handled in catch.
                    let handled = await self.recordOutboxRejection(
                        of: next,
                        outbox: outbox,
                        reason: "Run failed to start (\(response.status)).")
                    if handled { continue } else { break }
                }
                // Deliberately no pendingRuns adoption for background flushes:
                // the reply still lands via handleChatEvent's external-run
                // final branch (session-scoped, run-id independent),
                // handleSessionMessageEvent, and the post-drain history
                // refresh below. Run tracking (typing indicator, streaming,
                // timeouts) stays owned by interactive performSend.
                // chat.send ACK precedes durable user-turn persistence. Keep
                // the outbox row until history carries its idempotency key;
                // the cache splice makes the acknowledged turn visible while
                // canonical history catches up.
                await self.pendingCacheWriteTask?.value
                await self.spliceSentCommandIntoCachedTranscript(next)
                let confirmationUpdate = await outbox.markCommandAwaitingConfirmation(id: next.id)
                if confirmationUpdate == .unavailable {
                    self.applyTransportHealth(false)
                    break
                }
                if confirmationUpdate == .updated {
                    self.setOutboxState(.confirming, forCommandID: next.id)
                } else {
                    // A concurrent canonical history/session.message
                    // confirmation already removed the row.
                    self.clearOutboxState(forCommandID: next.id)
                }
                confirmationTargets.insert(Self.deliveryTarget(for: next))
            } catch is OpenClawChatTransportSendError {
                // The transport proved this payload never reached its request
                // channel, so it is safe to retry automatically.
                await outbox.markCommandQueued(
                    id: next.id,
                    retryCount: next.retryCount,
                    lastError: nil)
                self.setOutboxState(.queued, forCommandID: next.id)
                self.applyTransportHealth(false)
                break
            } catch is CancellationError {
                // Cancellation while the request is suspended does not prove
                // that the gateway rejected it. Never replay automatically.
                let update = await self.parkOutboxCommandWithUnconfirmedDelivery(next, outbox: outbox)
                if update == .updated {
                    confirmationTargets.insert(Self.deliveryTarget(for: next))
                }
                self.applyTransportHealth(false)
                break
            } catch let error as GatewayResponseError {
                if error.detailsReason == OpenClawChatSessionRoutingContract.changedErrorReason {
                    guard await self.parkOutboxCommandForChangedTarget(next, outbox: outbox) else { break }
                    continue
                }
                // A response error proves the gateway rejected the request;
                // unlike a socket/timeout failure, replay cannot duplicate an
                // accepted run and should consume the normal retry budget.
                let handled = await self.recordOutboxRejection(
                    of: next,
                    outbox: outbox,
                    reason: error.localizedDescription)
                if handled { continue } else { break }
            } catch {
                // A socket error or timeout is not a gateway rejection: the
                // request may have landed before its ACK was lost. Preserve
                // it for history reconciliation or explicit user retry.
                outboxLogger.error("outbox flush send failed \(error.localizedDescription, privacy: .public)")
                let update = await self.parkOutboxCommandWithUnconfirmedDelivery(next, outbox: outbox)
                if update == .updated {
                    confirmationTargets.insert(Self.deliveryTarget(for: next))
                }
                self.applyTransportHealth(false)
                break
            }
        }
        if !confirmationTargets.isEmpty {
            await self.refreshHistoriesAfterOutboxFlush(
                targets: confirmationTargets,
                routeLease: routeLease)
        }
    }

    private static func needsOutboxDeliveryReconciliation(_ command: OpenClawChatOutboxCommand) -> Bool {
        command.status == .awaitingConfirmation ||
            (command.status == .failed &&
                command.lastError == OpenClawChatSQLiteTranscriptCache.outboxUnconfirmedError)
    }

    private static func deliveryTarget(for command: OpenClawChatOutboxCommand) -> OutboxDeliveryTarget {
        OutboxDeliveryTarget(
            presentationSessionKey: command.sessionKey,
            deliverySessionKey: command.deliverySessionKey,
            agentID: command.agentID)
    }

    private func parkOutboxCommandWithUnconfirmedDelivery(
        _ command: OpenClawChatOutboxCommand,
        outbox: any OpenClawChatCommandOutbox) async -> OpenClawChatOutboxUpdateResult
    {
        let update = await outbox.markCommandFailedIfPresent(
            id: command.id,
            retryCount: command.retryCount,
            lastError: OpenClawChatSQLiteTranscriptCache.outboxUnconfirmedError)
        switch update {
        case .updated:
            self.setOutboxState(
                .failed(reason: OpenClawChatSQLiteTranscriptCache.outboxUnconfirmedError),
                forCommandID: command.id)
        case .missing, .confirmed:
            self.clearOutboxState(forCommandID: command.id)
        case .unavailable:
            self.applyTransportHealth(false)
        }
        return update
    }

    private func parkOutboxCommandForChangedTarget(
        _ command: OpenClawChatOutboxCommand,
        outbox: any OpenClawChatCommandOutbox) async -> Bool
    {
        let update = await outbox.markCommandFailedIfPresent(
            id: command.id,
            retryCount: command.retryCount,
            lastError: OpenClawChatSQLiteTranscriptCache.outboxChangedTargetError)
        guard update != .unavailable else {
            self.applyTransportHealth(false)
            return false
        }
        if update == .updated {
            let reason = "Gateway session routing changed; review and retry this message."
            self.setOutboxState(.failed(reason: reason), forCommandID: command.id)
        } else {
            self.clearOutboxState(forCommandID: command.id)
        }
        return true
    }

    /// Gateway rejections ("error"/"timeout" send acks) burn a retry attempt
    /// and become terminally 'failed' after `maxOutboxSendAttempts`. Returns
    /// true when the flush pass may continue with younger commands.
    private func recordOutboxRejection(
        of command: OpenClawChatOutboxCommand,
        outbox: any OpenClawChatCommandOutbox,
        reason: String) async -> Bool
    {
        outboxLogger.error("outbox flush send rejected \(reason, privacy: .public)")
        let attempts = command.retryCount + 1
        if attempts >= Self.maxOutboxSendAttempts {
            let update = await outbox.markCommandFailedIfPresent(
                id: command.id,
                retryCount: attempts,
                lastError: reason)
            guard update != .unavailable else {
                self.applyTransportHealth(false)
                return false
            }
            if update == .updated {
                self.setOutboxState(.failed(reason: reason), forCommandID: command.id)
            } else {
                // Canonical history may have removed the claimed row while
                // the rejection was in flight.
                self.clearOutboxState(forCommandID: command.id)
            }
            // Terminal failure needs user action; let younger commands
            // flush instead of blocking behind it forever.
            return true
        }
        await outbox.markCommandQueued(id: command.id, retryCount: attempts, lastError: reason)
        self.setOutboxState(.queued, forCommandID: command.id)
        // Strict createdAt ordering: never skip ahead of a command that
        // still has retries left.
        self.scheduleOutboxRetry(afterAttempts: attempts)
        return false
    }

    private func scheduleOutboxRetry(afterAttempts attempts: Int) {
        let delays = self.outboxRetryDelaysMs
        guard !delays.isEmpty else {
            self.outboxRetryTask?.cancel()
            self.outboxRetryTask = Task { [weak self] in
                await Task.yield()
                self?.flushOutboxIfNeeded()
            }
            return
        }
        let delayMs = delays[min(max(attempts - 1, 0), delays.count - 1)]
        self.outboxRetryTask?.cancel()
        self.outboxRetryTask = Task { [weak self] in
            try? await Task.sleep(nanoseconds: delayMs * 1_000_000)
            guard !Task.isCancelled else { return }
            self?.flushOutboxIfNeeded()
        }
    }

    /// Appends a flushed background-session turn to that session's cached
    /// transcript so a cold offline reopen shows it before live history.
    private func spliceSentCommandIntoCachedTranscript(_ command: OpenClawChatOutboxCommand) async {
        guard let transcriptCache else { return }
        let key = Self.outboxUserIdempotencyKey(command.id)
        let cacheAgentID = Self.transcriptCacheAgentID(
            sessionKey: command.sessionKey,
            agentID: command.agentID)
        var cached = await transcriptCache.loadTranscript(
            sessionKey: command.sessionKey,
            agentID: cacheAgentID)
        guard !cached.contains(where: { $0.idempotencyKey == key }) else { return }
        cached.append(Self.outboxUserMessage(for: command))
        await transcriptCache.storeTranscript(
            sessionKey: command.sessionKey,
            agentID: cacheAgentID,
            messages: cached)
    }

    private func recoverInterruptedOutboxSendsIfNeeded() async -> Bool {
        guard let outbox else { return false }
        // The store owns the once-per-process gate so overlapping/replacement
        // view models cannot reset another active sender's claim.
        return await outbox.recoverInterruptedSends()
    }

    private func outboxAgentID(for session: SessionSnapshot) -> String? {
        guard self.transport.outboxRequiresSessionRoutingContract else { return nil }
        if session.key.trimmingCharacters(in: .whitespacesAndNewlines).lowercased() == "unknown" {
            return nil
        }
        let normalized = session.deliveryAgentID?.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()
        return normalized?.isEmpty == false ? normalized : nil
    }

    private func outboxRequiresAgentID(for session: SessionSnapshot) -> Bool {
        guard self.transport.outboxRequiresSessionRoutingContract else { return false }
        return session.key.trimmingCharacters(in: .whitespacesAndNewlines).lowercased() != "unknown"
    }

    private func outboxRoutingContract(for session: SessionSnapshot) -> String? {
        if !self.transport.outboxRequiresSessionRoutingContract {
            return OpenClawChatOutboxCommand.legacyUnboundRoutingContract
        }
        let normalized = session.sessionRoutingContract?
            .trimmingCharacters(in: .whitespacesAndNewlines)
        return normalized?.isEmpty == false ? normalized : nil
    }

    /// Resolve once, before persistence. Re-resolving a presentation alias
    /// after reconnect could deliver to a newly selected/default agent.
    private func outboxDeliverySessionKey(
        for session: SessionSnapshot,
        agentID: String?) -> String?
    {
        let raw = session.key.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !raw.isEmpty else { return nil }
        guard self.transport.outboxRequiresSessionRoutingContract else { return raw }
        if raw.lowercased() == "unknown" { return raw }
        guard let agentID else { return nil }
        let normalized = raw.lowercased()
        if normalized == "global" { return "global" }
        if Self.agentID(fromSessionKey: raw) != nil { return raw }
        // A malformed ownership prefix must fail closed, not become a nested
        // key such as agent:<id>:agent::main.
        guard !normalized.hasPrefix("agent:") else { return nil }
        // The gateway owns structural normalization and preserves opaque
        // Matrix/Signal peer IDs. Keep the request key byte-for-byte here.
        return "agent:\(agentID):\(raw)"
    }

    private func commandMatchesTarget(
        _ command: OpenClawChatOutboxCommand,
        session: SessionSnapshot) -> Bool
    {
        guard command.sessionKey == session.key else { return false }
        // Failed rows never auto-send. Keep them reachable on their original
        // presentation alias after an owner change for explicit retry/delete.
        if command.status == .failed { return true }
        // Migrated v2 aliases have no owner and are parked as failed. Show
        // them so explicit retry can adopt the currently selected agent.
        guard let commandAgentID = command.agentID else { return true }
        guard let currentAgentID = self.outboxAgentID(for: session) else {
            // Cold offline launch has not recovered gateway ownership yet.
            // Keep the durable turn visible; the route lease verifies its
            // captured owner and contract before any later delivery.
            return true
        }
        return commandAgentID == currentAgentID
    }

    private func setOutboxState(_ state: OpenClawChatOutboxMessageState, forCommandID commandID: String) {
        guard let messageID = self.outboxMessageIDsByCommandID[commandID] else { return }
        self.outboxStatesByMessageID[messageID] = state
    }

    private func clearOutboxState(forCommandID commandID: String) {
        guard let messageID = self.outboxMessageIDsByCommandID.removeValue(forKey: commandID) else { return }
        self.outboxCommandIDsByMessageID.removeValue(forKey: messageID)
        self.outboxStatesByMessageID.removeValue(forKey: messageID)
    }

    func handleOutboxChange(_ change: OpenClawChatOutboxChange) {
        // Invalidates every command snapshot that started loading before the
        // store mutation, including snapshots owned by another view model.
        self.outboxPresentationGeneration &+= 1
        switch change {
        case let .canceled(commandID):
            // The initiating view owns its async result so canonical proof
            // observed before that continuation can still preserve the row.
            // Other views have no local cancellation task and apply the event.
            guard !self.cancelingOutboxCommandIDs.contains(commandID) else { return }
            guard let messageID = self.outboxMessageIDsByCommandID[commandID] else { return }
            self.clearOutboxState(forCommandID: commandID)
            self.replaceMessages(self.messages.filter { $0.id != messageID })
        case let .confirmed(commandID):
            // Canonical history owns the message row; only its outbox badge
            // and command mapping disappear.
            self.clearOutboxState(forCommandID: commandID)
        }
    }

    private func finishOutboxCancellation(_ commandID: String) {
        // Loads started while cancellation was in flight carry the previous
        // generation and cannot re-present their stale command snapshot.
        self.outboxPresentationGeneration &+= 1
        self.cancelingOutboxCommandIDs.remove(commandID)
    }

    private static func outboxDisplayState(for command: OpenClawChatOutboxCommand)
        -> OpenClawChatOutboxMessageState
    {
        switch command.status {
        case .queued:
            .queued
        case .sending:
            .sending
        case .awaitingConfirmation:
            .confirming
        case .failed:
            .failed(reason: command.lastError)
        }
    }

    /// Matches the optimistic-send convention (`"<runId>:user"`), which is
    /// also the key the gateway persists on the durable user row.
    static func outboxUserIdempotencyKey(_ commandID: String) -> String {
        "\(commandID):user"
    }
}