/**
 * Bridge Manager (Layer 2) — Central orchestrator.
 *
 * Manages adapter lifecycles, runs message polling loops, implements
 * concurrency control, streaming preview state machine, and reconnection.
 *
 * Full message processing flow:
 *   1. adapter.consumeOne() → IncomingMessage
 *   2. policy.checkPolicy() → allowed/denied
 *   3. commands.parseImCommand() → if command, execute and return
 *   4. channelRouter.resolveSession() → sessionId, workspace
 *   5. sessionLock.acquire(key)
 *   6. conversationEngine.processMessage() with streaming callbacks
 *   7. sessionLock.release(key)
 */

import fs from 'fs'
import path from 'path'
import crypto from 'crypto'
import { imChannelRepo } from '@/lib/database/repositories/im-channel.repo'
import {getForgeDataDir, getPublicUploadUrl, getUploadsDir} from '@/lib/forge-data'
import type {ChannelAdapter} from '../adapters/base'
import {createAdapter} from '../adapters/registry'
// Ensure all adapters are registered
import '../adapters/telegram/adapter'
import '../adapters/feishu/adapter'
import '../adapters/discord/adapter'
import '../adapters/weixin/adapter'
import '../adapters/dingtalk/adapter'
import '../adapters/wecom/adapter'

import {ChannelRouter} from './channel-router'
import {SessionLockManager} from '../shared/concurrency'
import {ConversationEngine} from './conversation-engine'
import type {PlatformContext} from '@/lib/engine'
import {parseImCommand, executeImCommand, checkUnknownCommand} from './commands'
import {DeliveryLayer} from './delivery'
import {emitImEvent} from '../shared/im-events'
import {PermissionBroker} from '../shared/permission-broker'
import {checkPolicy, createPairingRequest, verifyPairingCode} from '../shared/policy'
import type {BridgeState, ChannelType, DingTalkCardInstance, IncomingMessage} from '../types'

declare const globalThis: {
    __forgeBridgeManager?: BridgeManager
} & typeof global

// ---------------------------------------------------------------------------
// Bridge Manager
// ---------------------------------------------------------------------------

class BridgeManager {
    private adapters = new Map<string, ChannelAdapter>()
    private pollControllers = new Map<string, AbortController>()
    private states = new Map<string, BridgeState>()
    private channelTypes = new Map<string, ChannelType>() // channelId → ChannelType lookup

    // Shared components
    private router = new ChannelRouter()
    private locks = new SessionLockManager()
    private engine = new ConversationEngine()
    private delivery = new DeliveryLayer()
    private permissionBroker = new PermissionBroker()

    // Reconnection state
    private reconnectAttempts = new Map<string, number>()
    private circuitBroken = new Map<string, number>() // channelId → timestamp when circuit opened

    // Active task abort controllers: sessionKey → AbortController
    // Used by /stop command to cancel running queries
    private activeTasks = new Map<string, AbortController>()

    // Backpressure: limit concurrent SDK invocations
    private activeSdkCalls = 0
    private readonly maxConcurrentSdkCalls = 3
    private sdkCallWaiters: Array<() => void> = []

    // Reaction typing indicator tracking: sessionKey → { messageId, emojiType }
    private activeReactions = new Map<string, { messageId: string; emojiType: string }>()

    // Per-channel startup lock to prevent concurrent startAdapter calls (race condition fix)
    private startingChannels = new Map<string, Promise<void>>()

    /** Queue-busy ACK phrases (randomly selected) */
    private static readonly QUEUE_BUSY_ACK_PHRASES = [
        '上一条还没结束，这条我已经记下，稍后按顺序继续处理。',
        '当前还在忙，你的新消息已经排队，上一条完成后我马上继续。',
        '我这边还在处理上一条，这条已加入队列，完成后继续处理。',
    ]

    // ---------------------------------------------------------------------------
    // Adapter lifecycle
    // ---------------------------------------------------------------------------

    async startAdapter(channelId: string): Promise<void> {
        // Serialize concurrent startAdapter calls for the same channelId
        const existing = this.startingChannels.get(channelId)
        if (existing) {
            console.log(`[BridgeManager] startAdapter already in progress for ${channelId}, waiting...`)
            try {
                await existing
            } catch { /* ignore previous error */
            }
        }

        const promise = this._doStartAdapter(channelId)
        this.startingChannels.set(channelId, promise)
        try {
            await promise
        } finally {
            this.startingChannels.delete(channelId)
        }
    }

    private async _doStartAdapter(channelId: string): Promise<void> {
        // Stop existing adapter if running
        if (this.adapters.has(channelId)) {
            await this.stopAdapter(channelId)
        }

        this.setState(channelId, 'connecting')

        const channel = imChannelRepo.findTypeAndCredentials(channelId)
        if (!channel) throw new Error(`Channel not found: ${channelId}`)

        const config = {...(JSON.parse(channel.credentials) as Record<string, string>), channel_id: channelId}
        const adapter = createAdapter(channel.type)

        // Validate config before attempting connection
        const validation = adapter.validateConfig(config)
        if (!validation.valid) {
            this.setState(channelId, 'error', validation.error)
            throw new Error(validation.error)
        }

        // Register permission response handler
        adapter.onPermissionResponse((requestId, decision) => {
            this.permissionBroker.resolvePermission(requestId, decision)
        })

        try {
            await adapter.start(config)
            this.adapters.set(channelId, adapter)
            this.channelTypes.set(channelId, channel.type)
            this.setState(channelId, 'connected')
            this.reconnectAttempts.delete(channelId)
            this.circuitBroken.delete(channelId)

            // Update DB status
            imChannelRepo.updateStatus(channelId, 'connected')

            // Start poll loop
            this.startPollLoop(channelId, adapter)
        } catch (err) {
            const errorMsg = err instanceof Error ? err.message : 'Unknown error'
            this.setState(channelId, 'error', errorMsg)
            imChannelRepo.updateStatus(channelId, 'error')
            throw err
        }
    }

    async stopAdapter(channelId: string): Promise<void> {
        // Stop poll loop
        const controller = this.pollControllers.get(channelId)
        if (controller) {
            controller.abort()
            this.pollControllers.delete(channelId)
        }

        // Stop adapter
        const adapter = this.adapters.get(channelId)
        if (adapter) {
            try {
                await adapter.stop()
            } catch (err) {
                console.warn(`[BridgeManager] adapter.stop() error:`, err instanceof Error ? err.message : err)
            }
            this.adapters.delete(channelId)
        }

        // Cleanup only permission requests associated with this channel
        this.permissionBroker.cleanupChannel(channelId)

        // Clean up reconnection state for this channel
        this.reconnectAttempts.delete(channelId)
        this.circuitBroken.delete(channelId)

        this.setState(channelId, 'disconnected')

        // Update DB
        try {
            imChannelRepo.updateStatus(channelId, 'disconnected')
        } catch { /* ignore */
        }
    }

    async stopAll(): Promise<void> {
        const ids = [...this.adapters.keys()]
        await Promise.allSettled(ids.map(id => this.stopAdapter(id)))
        // Final cleanup of any remaining permission requests
        this.permissionBroker.cleanup()
    }

    /**
     * Auto-reconnect previously enabled channels on app startup.
     * Called once when the BridgeManager singleton is first created.
     */
    async autoReconnect(): Promise<void> {
        const channels = imChannelRepo.listEnabledConfigured()

        if (channels.length === 0) return

        // Reconnect all channels in parallel (P34 fix)
        console.log(`[BridgeManager] Auto-reconnecting ${channels.length} channel(s) in parallel...`)
        const results = await Promise.allSettled(
            channels.map(async (ch) => {
                await this.startAdapter(ch.id)
                console.log(`[BridgeManager] Auto-reconnected: ${ch.type} (${ch.id})`)
            }),
        )
        for (let i = 0; i < results.length; i++) {
            if (results[i].status === 'rejected') {
                const err = (results[i] as PromiseRejectedResult).reason
                console.error(`[BridgeManager] Auto-reconnect failed for ${channels[i].type}:`, err instanceof Error ? err.message : err)
            }
        }
    }


    // ---------------------------------------------------------------------------
    // State queries (backward compat for API/frontend)
    // ---------------------------------------------------------------------------

    getState(channelId: string): BridgeState | undefined {
        return this.states.get(channelId)
    }

    getAllStates(): BridgeState[] {
        return [...this.states.values()]
    }

    isConnected(channelId: string): boolean {
        return this.adapters.get(channelId)?.isRunning() || false
    }

    /** Expose the router for IM commands that need it */
    getRouter(): ChannelRouter {
        return this.router
    }

    /** Expose adapters map for cron notification delivery */
    getAdapters(): Map<string, ChannelAdapter> {
        return this.adapters
    }

    /**
     * Abort the active task for a given session key.
     * Used by the /stop command.
     * @returns true if a task was aborted, false if no task was running.
     */
    stopActiveTask(sessionKey: string): boolean {
        const controller = this.activeTasks.get(sessionKey)
        if (controller) {
            controller.abort()
            this.activeTasks.delete(sessionKey)
            return true
        }
        return false
    }

    // ---------------------------------------------------------------------------
    // Poll loop
    // ---------------------------------------------------------------------------

    /** Recently processed message IDs — prevents duplicate processing when SDK delivers same event twice */
    private processedMessageIds = new Set<string>()
    private processedMessageIdCleanupTimer: ReturnType<typeof setTimeout> | null = null

    private isMessageDuplicate(msg: IncomingMessage): boolean {
        if (!msg.messageId) return false // No ID, can't dedup
        if (this.processedMessageIds.has(msg.messageId)) {
            console.log(`[BridgeManager] Duplicate message skipped: ${msg.messageId}`)
            return true
        }
        this.processedMessageIds.add(msg.messageId)
        // Periodic cleanup: keep only last 5 minutes of IDs
        if (!this.processedMessageIdCleanupTimer) {
            this.processedMessageIdCleanupTimer = setTimeout(() => {
                this.processedMessageIds.clear()
                this.processedMessageIdCleanupTimer = null
            }, 300_000)
        }
        return false
    }

    private startPollLoop(channelId: string, adapter: ChannelAdapter): void {
        const controller = new AbortController()
        this.pollControllers.set(channelId, controller)

        // Run poll loop in background (fire and forget)
        this.runPollLoop(channelId, adapter, controller.signal).catch(err => {
            console.error(`[BridgeManager] Poll loop error for ${channelId}:`, err)
        })
    }

    private async runPollLoop(
        channelId: string,
        adapter: ChannelAdapter,
        signal: AbortSignal,
    ): Promise<void> {
        while (!signal.aborted) {
            try {
                const msg = await adapter.consumeOne(signal)
                if (!msg) {
                    if (!adapter.isRunning() && !signal.aborted) {
                        // Adapter stopped (connection lost) — attempt reconnect
                        await this.handleReconnect(channelId)
                        return // reconnect starts a new poll loop
                    }
                    continue
                }

                // Skip duplicate messages (SDK may redeliver on reconnect)
                if (this.isMessageDuplicate(msg)) continue

                // Process message in background (don't block poll loop)
                // NOTE: Backpressure is applied INSIDE processMessage, after the session lock,
                // so that messages waiting for the lock don't consume backpressure slots.
                this.processMessage(channelId, msg, adapter).catch(err => {
                    console.error(`[BridgeManager] Message processing error:`, err)
                })
            } catch (err) {
                if (signal.aborted) break

                console.error(`[BridgeManager] Poll error for ${channelId}:`, err)

                if (!adapter.isRunning()) {
                    await this.handleReconnect(channelId)
                    return
                }

                // Brief pause before retry
                await sleep(1000)
            }
        }
    }

    // ---------------------------------------------------------------------------
    // Message processing (full pipeline)
    // ---------------------------------------------------------------------------

    /** Overall timeout for a single SDK query (5 minutes) */
    private readonly sdkQueryTimeoutMs = 300_000

    private async processMessage(
        channelId: string,
        msg: IncomingMessage,
        adapter: ChannelAdapter,
    ): Promise<void> {
        // Inject the correct DB channel ID (adapters only know their platform type, not the DB ID)
        msg.channelId = channelId

        const t0 = Date.now()
        console.log(`[BridgeManager] Processing message: channel=${channelId}, chat=${msg.chatId}, isDm=${msg.isDm}, isGroupMention=${msg.isGroupMention}, text="${msg.text.slice(0, 40)}"`)

        // L3: Policy check
        const policyResult = checkPolicy(msg)
        if (!policyResult.allowed) {
            console.log(`[BridgeManager] Message BLOCKED by policy: ${policyResult.reason} (channel=${channelId}, chat=${msg.chatId})`)

            // Pairing flow: send verification code for unpaired DMs
            if (policyResult.reason === 'pairing_pending' && msg.isDm) {
                // Check if user is replying with a pairing code (6-digit number)
                const codeMatch = msg.text.trim().match(/^\d{6}$/)
                if (codeMatch) {
                    const verified = verifyPairingCode(channelId, msg.senderId, msg.text.trim())
                    if (verified) {
                        await this.delivery.deliver(adapter, msg.chatId, '✅ 配对成功！请重新发送你的消息。')
                    } else {
                        await this.delivery.deliver(adapter, msg.chatId, '❌ 验证码错误或已过期，请检查后再试。')
                    }
                    return
                }

                // Generate and send a new pairing code
                const code = createPairingRequest(channelId, msg.senderId)
                await this.delivery.deliver(
                    adapter,
                    msg.chatId,
                    `🔐 你还没有被授权使用此 Bot。\n请输入以下验证码完成配对：\n\n**${code}**\n\n（验证码 10 分钟内有效）`,
                )
            }
            return
        }
        console.log(`[BridgeManager] Policy check PASSED for channel=${channelId}`)

        // L3: Check for IM command
        const command = parseImCommand(msg.text)
        if (command) {
            console.log(`[BridgeManager] Executing IM command: /${command.name}`)
            // Send typing indicator for commands that may take time (P21 fix)
            try {
                await adapter.sendTypingIndicator(msg.chatId)
            } catch { /* best effort */
            }
            const response = await executeImCommand(command, msg, this.router, adapter, {
                stopActiveTask: (key) => this.stopActiveTask(key),
            })
            await this.delivery.deliver(adapter, msg.chatId, response, {skipDedup: true})

            // Emit command event for desktop sync
            const bindingInfo = this.router.getBindingInfo(msg.channelId, msg.channelType, msg.chatId)
            emitImEvent('im:command', {
                command: command.name,
                sessionId: bindingInfo?.sessionId || undefined,
                workspaceId: bindingInfo?.workspace || undefined,
            })
            return
        }

        // Check for unknown command attempts (e.g. /modle, /swtich)
        const unknownHint = checkUnknownCommand(msg.text)
        if (unknownHint) {
            console.log(`[BridgeManager] Unknown command attempt: ${msg.text.split(/\s+/)[0]}`)
            await this.delivery.deliver(adapter, msg.chatId, unknownHint, {skipDedup: true})
            return
        }

        // L3: Resolve session
        const {sessionId, workspace, userId} = this.router.resolveSession(msg)
        console.log(`[BridgeManager] Session resolved: sessionId=${sessionId.slice(0, 8)}, workspace=${workspace.slice(0, 8)}`)

        // Channel auth (C-1): delegate entirely to the adapter. Bridge no longer
        // knows any platform's auth internals — Feishu OAuth, WeCom CLI dir,
        // DingTalk credentials all live in adapter.ensureChannelAuth().
        const authContext = (await adapter.ensureChannelAuth(msg)) ?? {}
        const isFeishuChannel = msg.channelType === 'feishu'
        const isWeComChannel = msg.channelType === 'wecom'
        const isDingtalkChannel = msg.channelType === 'dingtalk'
        if (authContext.feishu) {
            console.log(`[BridgeManager] Feishu user token ready for ${msg.senderId.slice(0, 12)}... (brand=${authContext.feishu.brand})`)
        }
        if (authContext.dingtalk) {
            console.log(`[BridgeManager] DingTalk credentials ready for dws CLI (clientId=${authContext.dingtalk.clientId.slice(0, 12)}...)`)
        }
        if (authContext.wecom) {
            console.log(`[BridgeManager] WeCom CLI config dir: ${authContext.wecom.cliConfigDir} (adapter-cached)`)
        }
        const sessionKey = `${msg.channelId}:${msg.chatId}`
        // 钉钉 queue-busy 预建卡（C-2：传给 renderer，不再用 activeCards Map）
        let preCreatedCard: DingTalkCardInstance | undefined

        // L2: Acquire session lock (ensures only one SDK call per chat at a time)
        // NOTE: This does NOT consume a backpressure slot — only the actual SDK call does.
        console.log(`[BridgeManager] Waiting for session lock: ${sessionKey} (+${Date.now() - t0}ms)`)

        // Queue-busy ACK: if the lock is held, send immediate feedback so the user
        // knows their message was received (instead of silent wait).
        if (this.locks.isLocked(sessionKey)) {
            const ackPhrases = BridgeManager.QUEUE_BUSY_ACK_PHRASES
            const ackText = ackPhrases[Math.floor(Math.random() * ackPhrases.length)]
            console.log(`[BridgeManager] Queue busy for ${sessionKey}, sending ACK`)

            // 平台预建专属载体（如 DingTalk AI Card）；不支持时直接发 ACK 文本。
            if (adapter.preCreateBusyCard) {
                // 运行时只有 DingTalk 实现该方法，返回 DingTalkCardInstance
                const preCard = await adapter.preCreateBusyCard(msg.chatId, ackText) as DingTalkCardInstance | undefined
                if (preCard) {
                    preCreatedCard = preCard
                } else {
                    await this.delivery.deliver(adapter, msg.chatId, ackText, {skipDedup: true})
                }
            } else {
                await this.delivery.deliver(adapter, msg.chatId, ackText, {skipDedup: true})
            }
        }

        const release = await this.locks.acquire(sessionKey)
        console.log(`[BridgeManager] Session lock acquired: ${sessionKey} (+${Date.now() - t0}ms)`)

        // Create abort controller: combines /stop command abort + overall timeout
        const taskAbort = new AbortController()
        const timeoutId = setTimeout(() => {
            console.log(`[BridgeManager] SDK query timeout (${this.sdkQueryTimeoutMs}ms) for ${sessionKey}`)
            taskAbort.abort()
        }, this.sdkQueryTimeoutMs)
        this.activeTasks.set(sessionKey, taskAbort)

        // Reaction typing indicator: 平台声明式支持（如 Feishu 👍）
        let reactionAdded = false
        if (adapter.supportsReactionIndicator?.() && msg.messageId) {
            try {
                const emoji = 'THUMBSUP'
                await adapter.addReaction(msg.messageId, emoji)
                this.activeReactions.set(sessionKey, { messageId: msg.messageId, emojiType: emoji })
                reactionAdded = true
                console.log(`[BridgeManager] Added reaction indicator for ${sessionKey}`)
            } catch (err) {
                console.warn('[BridgeManager] Failed to add reaction indicator:', err instanceof Error ? err.message : err)
            }
        }

        // C-2: 平台流式渲染委托给 adapter 提供的 renderer（catch/finally 也要用，故在 try 外创建）
        const renderer = adapter.createStreamRenderer({
            adapter,
            chatId: msg.chatId,
            channelId,
            delivery: this.delivery,
            preCreatedCard,
            deliverPlatformFinal: (text) => this.deliverPlatformFinal(msg.channelId, msg.chatId, text),
        })

        try {
            // L4+L5: Process message with backpressure + streaming callbacks
            // Backpressure wraps ONLY the SDK call (after lock acquisition) so that
            // messages waiting for the session lock don't consume backpressure slots.
            // 将各渠道凭证与标记打包成 platformContext（消除 processMessage 参数泄漏）
            const platformContext: PlatformContext = {
                ...authContext,
                channelFlags: { feishu: isFeishuChannel, wecom: isWeComChannel, dingtalk: isDingtalkChannel },
            }
            const result = await this.withSdkBackpressure(() =>
                this.engine.processMessage(msg, sessionId, workspace, {
                    onTyping: async () => {
                        // Emit SSE event early so desktop UI updates in real-time (P13 fix)
                        emitImEvent('im:message', {sessionId, workspaceId: workspace})
                        await renderer.onTyping()
                    },

                    onReasoning: async (text: string) => { await renderer.onReasoning?.(text) },

                    onToolUseStart: async (tool) => { await renderer.onToolUseStart?.(tool) },

                    onToolUseEnd: async (tool) => { await renderer.onToolUseEnd?.(tool) },

                    onDraft: async (partialText: string) => { await renderer.onDraft(partialText) },

                    onFinal: async (text: string) => { await renderer.onFinal(text) },

                    onAttachments: async (attachments) => {
                        for (const att of attachments) {
                            try {
                                const prepared = this.prepareOutboundAttachmentForUpload(att, msg)
                                const buffer = fs.readFileSync(prepared.filePath)
                                // IM 渠道已经把文件本体上传到各自平台的云端，IM 端用户能直接打开。
                                // 这里不再附 `下载链接：${publicUrl}` 文案，避免对 IM 端展示无意义的 localhost URL。
                                const caption: string | undefined = att.name
                                if (att.isImage) {
                                    await this.delivery.deliverImage(adapter, msg.chatId, buffer, caption)
                                } else {
                                    await this.delivery.deliverFile(adapter, msg.chatId, buffer, att.name, caption)
                                }
                                console.log(`[BridgeManager] Sent attachment: ${prepared.name} (${att.isImage ? 'image' : 'file'}, ${att.size} bytes${prepared.publicUrl ? ', publicUrl=yes' : ''})`)
                            } catch (err) {
                                console.warn(`[BridgeManager] Failed to send attachment ${att.name}:`, err instanceof Error ? err.message : err)
                                await this.delivery.deliver(adapter, msg.chatId, `📎 ${att.name} (${att.size} bytes)`, {skipDedup: true})
                            }
                        }
                    },

                    onPermissionRequest: async (req) => {
                        return this.permissionBroker.requestPermission(req, adapter, channelId)
                    },

                    abortSignal: taskAbort.signal,
                }, userId, platformContext),
            )

            // If engine produced no text response and the renderer delivered nothing
            if (!result.text && !renderer.delivered) {
                await this.delivery.deliver(adapter, msg.chatId, '_(No text response)_')
            }
        } catch (err) {
            const errorMsg = err instanceof Error ? err.message : 'Unknown error'
            const isAbort = err instanceof Error && err.name === 'AbortError'
            if (isAbort) {
                console.log(`[BridgeManager] SDK query aborted for ${sessionKey}`)
            } else {
                console.error(`[${channelId}] Agent error:`, errorMsg)
            }

            // C-2: renderer 处理各平台渲染层错误清理（删 draft / 钉钉卡片错误文本 / 企微 replyStream 错误 / 飞书 abortCard）
            await renderer.abort(isAbort ? 'abort' : 'error', errorMsg)

            // 平台声明 renderer.abort 已自处理错误（如企微 replyStream），其余平台发错误文本
            if (!adapter.handlesErrorDeliveryInRenderer?.()) {
                if (isAbort) {
                    await this.delivery.deliver(adapter, msg.chatId, '⏱ 请求超时或已取消')
                } else {
                    await this.delivery.deliver(adapter, msg.chatId, `❌ Error: ${errorMsg}`)
                }
            }
        } finally {
            clearTimeout(timeoutId)
            this.activeTasks.delete(sessionKey)
            // C-2: renderer 清理（destroy controller / cleanup WeCom frame）
            renderer.destroy?.()

            // Remove reaction typing indicator (platform-declared)
            const reaction = this.activeReactions.get(sessionKey)
            if (reaction && adapter.supportsReactionIndicator?.()) {
                try {
                    await adapter.removeReaction(reaction.messageId, reaction.emojiType)
                    console.log(`[BridgeManager] Removed reaction indicator for ${sessionKey}`)
                } catch (err) {
                    console.warn('[BridgeManager] Failed to remove reaction indicator:', err instanceof Error ? err.message : err)
                }
                this.activeReactions.delete(sessionKey)
            }

            release()
            console.log(`[BridgeManager] Session lock released: ${sessionKey} (total: ${Date.now() - t0}ms)`)

            // Emit message event for desktop sync (after processing completes)
            emitImEvent('im:message', {sessionId, workspaceId: workspace})
        }
    }

    // ---------------------------------------------------------------------------
    // Backpressure: limit concurrent SDK invocations
    // ---------------------------------------------------------------------------

    private async withSdkBackpressure<T>(fn: () => Promise<T>): Promise<T> {
        // Wait if at capacity
        while (this.activeSdkCalls >= this.maxConcurrentSdkCalls) {
            await new Promise<void>(resolve => this.sdkCallWaiters.push(resolve))
        }
        this.activeSdkCalls++
        try {
            return await fn()
        } finally {
            this.activeSdkCalls--
            // Wake one waiter
            const waiter = this.sdkCallWaiters.shift()
            if (waiter) waiter()
        }
    }

    // ---------------------------------------------------------------------------
    // Reconnection with exponential backoff + circuit breaker
    // ---------------------------------------------------------------------------

    private async handleReconnect(channelId: string): Promise<void> {
        // Check circuit breaker
        const circuitOpenTime = this.circuitBroken.get(channelId)
        if (circuitOpenTime) {
            const elapsed = Date.now() - circuitOpenTime
            if (elapsed < 5 * 60 * 1000) {
                // Circuit still open — wait
                console.log(`[BridgeManager] Circuit breaker open for ${channelId}, waiting...`)
                return
            }
            // Circuit half-open — allow one attempt
            this.circuitBroken.delete(channelId)
        }

        const attempts = (this.reconnectAttempts.get(channelId) || 0) + 1
        this.reconnectAttempts.set(channelId, attempts)

        // Trip circuit breaker after 5 consecutive failures
        if (attempts > 5) {
            console.log(`[BridgeManager] Circuit breaker tripped for ${channelId}`)
            this.circuitBroken.set(channelId, Date.now())
            this.setState(channelId, 'error', 'Too many reconnection failures, pausing for 5 minutes')
            return
        }

        // Exponential backoff with jitter: 1s, 2s, 4s, 8s, 16s, max 60s
        const baseDelay = Math.min(1000 * Math.pow(2, attempts - 1), 60_000)
        const jitter = Math.random() * baseDelay * 0.3
        const delay = baseDelay + jitter

        console.log(`[BridgeManager] Reconnecting ${channelId} in ${Math.round(delay)}ms (attempt ${attempts})`)
        await sleep(delay)

        try {
            await this.startAdapter(channelId)
        } catch (err) {
            const errMsg = err instanceof Error ? err.message : String(err)
            console.error(`[BridgeManager] Reconnect failed for ${channelId}:`, errMsg)

            // Don't retry unrecoverable errors (auth failures, invalid config) — trip circuit immediately (P20 fix)
            if (/401|403|Unauthorized|Forbidden|invalid.*token|app_id|app_secret/i.test(errMsg)) {
                console.log(`[BridgeManager] Unrecoverable error for ${channelId}, stopping reconnection`)
                this.circuitBroken.set(channelId, Date.now())
                this.setState(channelId, 'error', `Authentication failed: ${errMsg}`)
            }
        }
    }

    // ---------------------------------------------------------------------------
    // Internal state management
    // ---------------------------------------------------------------------------

    private setState(channelId: string, status: BridgeState['status'], error?: string): void {
        const type = this.channelTypes.get(channelId) || (channelId as ChannelType)
        this.states.set(channelId, {channelId, type, status, error})
    }

    private prepareOutboundAttachmentForUpload(
        att: { filePath: string; name: string },
        msg: IncomingMessage,
    ): { filePath: string; name: string; publicUrl: string | null } {
        const uploadsDir = getUploadsDir()
        const channel = msg.channelType
        const chatType = msg.isDm ? 'dm' : 'group'
        const chatHash = crypto.createHash('sha256').update(msg.chatId).digest('hex').slice(0, 8)
        const stamp = new Date()
            .toISOString()
            .replaceAll('-', '')
            .replaceAll(':', '')
            .replaceAll('.', '')
            .slice(0, 15) + 'Z'
        const safeName = this.sanitizeUploadName(att.name, 100)
        const destName = `agent_${channel}_out_${chatType}_${chatHash}_${stamp}_${Math.random().toString(36).slice(2, 6)}_${safeName}`
        const destPath = path.join(uploadsDir, destName)
        fs.copyFileSync(att.filePath, destPath)
        return {filePath: destPath, name: destName, publicUrl: getPublicUploadUrl(destName)}
    }

    private sanitizeUploadName(name: string, maxLength: number): string {
        const sanitized = name.replace(/[^a-zA-Z0-9._-]/g, '_').replace(/^_+$/, '')
        return (sanitized || 'file').slice(0, maxLength)
    }

    /**
     * Deliver a final message using platform-native rendering.
     * Shared between interactive conversations (onFinal) and cron notifications (notifyIm).
     * Handles Feishu cards, WeCom template cards, DingTalk AI Cards per platform.
     */
    async deliverPlatformFinal(channelId: string, chatId: string, text: string): Promise<void> {
        const adapter = this.adapters.get(channelId)
        if (!adapter) return

        // 平台特化投递：adapter 声明能力则委托，已处理（true）则返回；否则走默认文本。
        if (adapter.deliverPlatformFinal) {
            const handled = await adapter.deliverPlatformFinal(chatId, text, this.delivery)
            if (handled) return
        }

        // Default: plain text with markdown rendering
        await this.delivery.deliver(adapter, chatId, text, { skipDedup: true })
    }

}

// ---------------------------------------------------------------------------
// Singleton accessor
// ---------------------------------------------------------------------------

export function getBridgeManager(): BridgeManager {
    if (!globalThis.__forgeBridgeManager) {
        globalThis.__forgeBridgeManager = new BridgeManager()
        // Auto-reconnect previously enabled channels (fire and forget)
        globalThis.__forgeBridgeManager.autoReconnect().catch(err => {
            console.error('[BridgeManager] Auto-reconnect error:', err)
        })
    }
    return globalThis.__forgeBridgeManager
}

// ---------------------------------------------------------------------------
// Utility
// ---------------------------------------------------------------------------

function sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms))
}