use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use futures_util::{SinkExt, StreamExt};
use prost::{Message as ProstMessage, bytes::Bytes};
use serde::Deserialize;
use tokio::sync::{Mutex, RwLock};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

use crate::bus::{MediaItem, MessageBus, OutboundMessage};
use crate::channels::base::{Channel, ChannelError, LivePolicy, TurnSink, TurnTarget};
use crate::config::FeishuChannelConfig;
use crate::session::{ToolStatus, TurnBlock, TurnSnapshot, TurnStatus};

const FEISHU_API_BASE: &str = "https://open.feishu.cn/open-apis";
const FEISHU_WS_BASE: &str = "https://open.feishu.cn";

/// Heartbeat timeout for WS connection — must be larger than ping_interval (default 120 s).
/// If no binary frame (pong or event) is received within this window, reconnect.
const WS_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(300);
const CHANNEL_STOP_GRACE: Duration = if cfg!(test) {
    Duration::from_millis(100)
} else {
    Duration::from_secs(5)
};
/// Refresh tenant token this many seconds before the announced expiry.
const TOKEN_REFRESH_SKEW: Duration = Duration::from_secs(120);
/// Default tenant token TTL when `expire`/`expires_in` is absent.
const DEFAULT_TOKEN_TTL: Duration = Duration::from_secs(7200);
/// Dedup cache TTL (30 minutes).
const DEDUP_CACHE_TTL: Duration = Duration::from_secs(30 * 60);
const WS_EVENT_QUEUE_CAPACITY: usize = 32;
const WS_ACK_TIMEOUT: Duration = Duration::from_secs(2);
const WS_FRAGMENT_TTL: Duration = Duration::from_secs(5 * 60);
const MAX_WS_FRAGMENTS: usize = 256;
const MAX_WS_EVENT_BYTES: usize = 8 * 1024 * 1024;
const STABLE_CONNECTION_WINDOW: Duration = Duration::from_secs(30);
const FEISHU_API_ATTEMPTS: usize = 3;
const FEISHU_INVALID_TOKEN_CODE: i32 = 99_991_663;

// ─────────────────────────────────────────────────────────────────────────────
// Protobuf types for Feishu WebSocket protocol (pbbp2.proto)
// ─────────────────────────────────────────────────────────────────────────────

#[derive(Clone, PartialEq, prost::Message)]
struct PbHeader {
    #[prost(string, tag = "1")]
    pub key: String,
    #[prost(string, tag = "2")]
    pub value: String,
}

/// Feishu WS frame.
/// method=0 → CONTROL (ping/pong)  method=1 → DATA (events)
#[derive(Clone, PartialEq, prost::Message)]
struct PbFrame {
    #[prost(uint64, tag = "1")]
    pub seq_id: u64,
    #[prost(uint64, tag = "2")]
    pub log_id: u64,
    #[prost(int32, tag = "3")]
    pub service: i32,
    #[prost(int32, tag = "4")]
    pub method: i32,
    #[prost(message, repeated, tag = "5")]
    pub headers: Vec<PbHeader>,
    #[prost(bytes = "vec", optional, tag = "8")]
    pub payload: Option<Vec<u8>>,
}

impl PbFrame {
    fn header_value(&self, key: &str) -> Option<&str> {
        self.headers
            .iter()
            .find(|header| header.key == key)
            .map(|header| header.value.as_str())
    }
}

struct FragmentAssembly {
    parts: Vec<Option<Vec<u8>>>,
    total_bytes: usize,
    created_at: Instant,
}

/// POST /callback/ws/endpoint response
#[derive(Deserialize)]
struct WsEndpointResp {
    code: i32,
    msg: Option<String>,
    data: Option<WsEndpoint>,
}

#[derive(Deserialize)]
struct WsEndpoint {
    #[serde(rename = "URL")]
    url: String,
    #[serde(default)]
    client_config: Option<WsClientConfig>,
}

#[derive(Deserialize, Default)]
struct WsClientConfig {
    #[serde(rename = "PingInterval")]
    ping_interval: Option<u64>,
}

/// Lark event envelope (method=1 / type=event payload)
#[derive(Deserialize)]
struct LarkEvent {
    header: LarkEventHeader,
    event: serde_json::Value,
}

#[derive(Deserialize)]
struct LarkEventHeader {
    event_type: String,
    event_id: String,
}

impl std::fmt::Debug for LarkEventHeader {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("LarkEventHeader")
            .field("event_type", &self.event_type)
            .field("event_id", &self.event_id)
            .finish()
    }
}

#[derive(Deserialize)]
struct MsgReceivePayload {
    sender: LarkSender,
    message: LarkMessage,
}

#[derive(Deserialize)]
struct LarkSender {
    sender_id: LarkSenderId,
    #[serde(default)]
    sender_type: String,
}

#[derive(Deserialize, Default)]
struct LarkSenderId {
    open_id: Option<String>,
}

#[derive(Deserialize)]
struct LarkMessage {
    message_id: String,
    chat_id: String,
    #[serde(default)]
    chat_type: String,
    message_type: String,
    #[serde(default)]
    content: String,
    #[serde(default)]
    parent_id: Option<String>,
    #[serde(default)]
    root_id: Option<String>,
    #[serde(default)]
    thread_id: Option<String>,
    #[serde(default)]
    mentions: Vec<serde_json::Value>,
}

// ─────────────────────────────────────────────────────────────────────────────

/// Cached tenant token with proactive refresh metadata.
#[derive(Clone)]
struct CachedTenantToken {
    value: String,
    refresh_after: Instant,
}

#[derive(Clone)]
pub struct FeishuChannel {
    config: FeishuChannelConfig,
    http_client: reqwest::Client,
    running: Arc<RwLock<bool>>,
    shutdown: Arc<RwLock<Option<CancellationToken>>>,
    run_task: Arc<Mutex<Option<JoinHandle<()>>>>,
    connected: Arc<RwLock<bool>>,
    /// Cached tenant access token with proactive refresh.
    tenant_token: Arc<RwLock<Option<CachedTenantToken>>>,
    /// Dedup cache: WS message_ids seen in the last ~30 min.
    seen_message_ids: Arc<RwLock<HashMap<String, Instant>>>,
    /// Bot identity used to enforce group-chat @mention admission.
    bot_open_id: Arc<RwLock<Option<String>>>,
}

/// Parsed message data from a Feishu frame
struct ParsedMessage {
    message_id: String,
    open_id: String,
    chat_id: String,
    content: String,
    media: Vec<MediaItem>,
    /// ID of the message this message is replying to (if any).
    /// Used to fetch quoted message content for display.
    parent_id: Option<String>,
    root_id: Option<String>,
    thread_id: Option<String>,
    chat_type: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct FeishuSendTarget {
    receive_id: String,
    receive_id_type: &'static str,
    reply_to: Option<String>,
    reply_in_thread: bool,
}

#[derive(Deserialize)]
struct FeishuMessageResponse {
    code: i32,
    msg: String,
    data: Option<FeishuMessageData>,
}

#[derive(Deserialize)]
struct FeishuMessageData {
    message_id: String,
}

impl FeishuSendTarget {
    fn from_message(
        chat_id: String,
        reply_to: Option<String>,
        metadata: &HashMap<String, String>,
    ) -> Self {
        let thread_id = metadata
            .get("feishu.thread_id")
            .filter(|value| !value.is_empty());
        if reply_to.is_none()
            && let Some(thread_id) = thread_id
        {
            return Self {
                receive_id: thread_id.clone(),
                receive_id_type: "thread_id",
                reply_to: None,
                reply_in_thread: false,
            };
        }
        let receive_id_type = if chat_id.starts_with("oc_") {
            "chat_id"
        } else {
            "open_id"
        };
        Self {
            receive_id: chat_id,
            receive_id_type,
            reply_to,
            reply_in_thread: thread_id.is_some(),
        }
    }

    fn request(&self, msg_type: &str, content: &str) -> (String, serde_json::Value) {
        let uuid = uuid::Uuid::new_v4().to_string();
        if let Some(message_id) = &self.reply_to {
            return (
                format!("{}/im/v1/messages/{message_id}/reply", FEISHU_API_BASE),
                serde_json::json!({
                    "msg_type": msg_type,
                    "content": content,
                    "reply_in_thread": self.reply_in_thread,
                    "uuid": uuid
                }),
            );
        }
        (
            format!(
                "{}/im/v1/messages?receive_id_type={}",
                FEISHU_API_BASE, self.receive_id_type
            ),
            serde_json::json!({
                "receive_id": self.receive_id,
                "msg_type": msg_type,
                "content": content,
                "uuid": uuid
            }),
        )
    }
}

impl FeishuChannel {
    pub fn new(
        mut config: FeishuChannelConfig,
        workspace_dir: &Path,
    ) -> Result<Self, ChannelError> {
        // Override media_dir to use workspace_dir/media/feishu
        let media_dir = workspace_dir.join("media").join("feishu");
        config.media_dir = media_dir.to_string_lossy().to_string();

        let request_timeout = Duration::from_secs(config.request_timeout_secs.clamp(5, 120));
        let http_client = reqwest::Client::builder()
            .timeout(request_timeout)
            .build()
            .map_err(|error| {
                ChannelError::ConfigError(format!("Failed to build Feishu HTTP client: {error}"))
            })?;

        Ok(Self {
            config,
            http_client,
            running: Arc::new(RwLock::new(false)),
            shutdown: Arc::new(RwLock::new(None)),
            run_task: Arc::new(Mutex::new(None)),
            connected: Arc::new(RwLock::new(false)),
            tenant_token: Arc::new(RwLock::new(None)),
            seen_message_ids: Arc::new(RwLock::new(HashMap::new())),
            bot_open_id: Arc::new(RwLock::new(None)),
        })
    }

    async fn invalidate_tenant_token(&self) {
        *self.tenant_token.write().await = None;
    }

    fn request_timeout(&self) -> Duration {
        Duration::from_secs(self.config.request_timeout_secs.clamp(5, 120))
    }

    async fn send_authenticated<F>(
        &self,
        operation: &str,
        build: F,
    ) -> Result<reqwest::Response, ChannelError>
    where
        F: Fn(&reqwest::Client, &str) -> reqwest::RequestBuilder,
    {
        for attempt in 0..FEISHU_API_ATTEMPTS {
            let token = self.get_tenant_access_token().await?;
            match build(&self.http_client, &token).send().await {
                Ok(response) => {
                    let status = response.status();
                    if status == reqwest::StatusCode::UNAUTHORIZED {
                        self.invalidate_tenant_token().await;
                    }
                    if status == reqwest::StatusCode::TOO_MANY_REQUESTS
                        || status.is_server_error()
                        || status == reqwest::StatusCode::UNAUTHORIZED
                    {
                        if attempt + 1 < FEISHU_API_ATTEMPTS {
                            let delay = retry_after_delay(response.headers(), attempt);
                            tokio::time::sleep(delay).await;
                            continue;
                        }
                        return Err(ChannelError::ConnectionError(format!(
                            "{operation} failed after {} attempts: HTTP {status}",
                            attempt + 1
                        )));
                    }
                    return Ok(response);
                }
                Err(error) => {
                    if attempt + 1 == FEISHU_API_ATTEMPTS {
                        return Err(ChannelError::ConnectionError(format!(
                            "{operation} failed after {} attempts: {error}",
                            attempt + 1
                        )));
                    }
                    tokio::time::sleep(api_retry_delay(attempt)).await;
                }
            }
        }
        unreachable!("Feishu API attempt loop always returns")
    }

    async fn refresh_bot_open_id(&self) -> Result<String, ChannelError> {
        let response = self
            .send_authenticated("fetch Feishu bot info", |client, token| {
                client
                    .get(format!("{}/bot/v3/info", FEISHU_API_BASE))
                    .bearer_auth(token)
            })
            .await?;
        let status = response.status();
        let body: serde_json::Value = response.json().await.map_err(|error| {
            ChannelError::Other(format!("Failed to parse bot info response: {error}"))
        })?;
        if !status.is_success() || body.get("code").and_then(|value| value.as_i64()) != Some(0) {
            return Err(ChannelError::Other(format!(
                "Bot info request failed: status={status}, code={}",
                body.get("code")
                    .and_then(|value| value.as_i64())
                    .unwrap_or(-1)
            )));
        }
        let open_id = body
            .pointer("/bot/open_id")
            .or_else(|| body.pointer("/data/bot/open_id"))
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| ChannelError::Other("Bot info response has no open_id".to_string()))?
            .to_string();
        *self.bot_open_id.write().await = Some(open_id.clone());
        Ok(open_id)
    }

    /// Get WebSocket endpoint URL from Feishu API
    async fn get_ws_endpoint(
        &self,
        client: &reqwest::Client,
    ) -> Result<(String, WsClientConfig), ChannelError> {
        let resp = client
            .post(format!("{}/callback/ws/endpoint", FEISHU_WS_BASE))
            .header("locale", "zh")
            .json(&serde_json::json!({
                "AppID": self.config.app_id,
                "AppSecret": self.config.app_secret,
            }))
            .send()
            .await
            .map_err(|e| ChannelError::ConnectionError(format!("HTTP error: {}", e)))?;

        let endpoint_resp: WsEndpointResp = resp.json().await.map_err(|e| {
            ChannelError::ConnectionError(format!("Failed to parse endpoint response: {}", e))
        })?;

        if endpoint_resp.code != 0 {
            return Err(ChannelError::ConnectionError(format!(
                "WS endpoint failed: code={} msg={}",
                endpoint_resp.code,
                endpoint_resp.msg.as_deref().unwrap_or("unknown")
            )));
        }

        let ep = endpoint_resp
            .data
            .ok_or_else(|| ChannelError::ConnectionError("Empty endpoint data".to_string()))?;

        let client_config = ep.client_config.unwrap_or_default();
        Ok((ep.url, client_config))
    }

    /// Get tenant access token (cached with proactive refresh).
    async fn get_tenant_access_token(&self) -> Result<String, ChannelError> {
        // 1. Check cache
        {
            let cached = self.tenant_token.read().await;
            if let Some(ref token) = *cached
                && Instant::now() < token.refresh_after
            {
                return Ok(token.value.clone());
            }
        }

        // 2. Fetch new token
        let (token, ttl) = self.fetch_new_token().await?;

        // 3. Cache with proactive refresh time (提前 120 秒)
        let refresh_after = Instant::now() + ttl.saturating_sub(TOKEN_REFRESH_SKEW);
        {
            let mut cached = self.tenant_token.write().await;
            *cached = Some(CachedTenantToken {
                value: token.clone(),
                refresh_after,
            });
        }

        Ok(token)
    }

    /// Fetch a new tenant access token from Feishu.
    async fn fetch_new_token(&self) -> Result<(String, Duration), ChannelError> {
        let resp = self
            .http_client
            .post(format!(
                "{}/auth/v3/tenant_access_token/internal",
                FEISHU_API_BASE
            ))
            .header("Content-Type", "application/json")
            .json(&serde_json::json!({
                "app_id": self.config.app_id,
                "app_secret": self.config.app_secret,
            }))
            .send()
            .await
            .map_err(|e| ChannelError::ConnectionError(format!("HTTP error: {}", e)))?;

        #[derive(Deserialize)]
        struct TokenResponse {
            code: i32,
            tenant_access_token: Option<String>,
            expire: Option<i64>,
        }

        let token_resp: TokenResponse = resp
            .json()
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to parse token response: {}", e)))?;

        if token_resp.code != 0 {
            return Err(ChannelError::Other("Auth failed".to_string()));
        }

        let token = token_resp
            .tenant_access_token
            .ok_or_else(|| ChannelError::Other("No token in response".to_string()))?;

        let ttl = token_resp
            .expire
            .and_then(|v| u64::try_from(v).ok())
            .map(Duration::from_secs)
            .unwrap_or(DEFAULT_TOKEN_TTL);

        Ok((token, ttl))
    }

    /// Check if message_id has been seen (dedup), and mark it as seen if not.
    /// Returns true if the message was already processed.
    /// Note: GC of stale entries is handled in the heartbeat timeout_check loop.
    async fn is_message_seen(&self, message_id: &str) -> bool {
        let mut seen = self.seen_message_ids.write().await;
        let now = Instant::now();

        if seen.contains_key(message_id) {
            true
        } else {
            seen.insert(message_id.to_string(), now);
            false
        }
    }

    /// Download media and save locally, return (description, media_item)
    async fn download_media(
        &self,
        msg_type: &str,
        content_json: &serde_json::Value,
        message_id: &str,
    ) -> Result<(String, Option<MediaItem>), ChannelError> {
        let media_dir = Path::new(&self.config.media_dir);
        tokio::fs::create_dir_all(media_dir)
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to create media dir: {}", e)))?;

        match msg_type {
            "image" => {
                self.download_image(content_json, message_id, media_dir)
                    .await
            }
            "audio" | "file" | "media" => {
                self.download_file(content_json, message_id, media_dir, msg_type)
                    .await
            }
            _ => Ok((format!("[unsupported media type: {}]", msg_type), None)),
        }
    }

    /// Download image from Feishu
    async fn download_image(
        &self,
        content_json: &serde_json::Value,
        message_id: &str,
        media_dir: &Path,
    ) -> Result<(String, Option<MediaItem>), ChannelError> {
        let image_key = content_json
            .get("image_key")
            .and_then(|v| v.as_str())
            .ok_or_else(|| ChannelError::Other("No image_key in message".to_string()))?;

        // Use message resource API for downloading message images
        let url = format!(
            "{}/im/v1/messages/{}/resources/{}?type=image",
            FEISHU_API_BASE, message_id, image_key
        );

        #[cfg(debug_assertions)]
        tracing::debug!(url = %url, image_key = %image_key, message_id = %message_id, "Downloading image from Feishu via message resource API");

        let resp = self
            .send_authenticated("download Feishu image", |client, token| {
                client.get(&url).bearer_auth(token)
            })
            .await?;

        let status = resp.status();
        #[cfg(debug_assertions)]
        tracing::debug!(status = %status, "Image download response status");

        if !status.is_success() {
            let error_body =
                read_response_limited(resp, 64 * 1024, self.request_timeout(), "error response")
                    .await
                    .unwrap_or_default();
            return Err(ChannelError::Other(format!(
                "Image download failed {}: {}",
                status,
                String::from_utf8_lossy(&error_body)
            )));
        }

        let content_type = resp
            .headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("image/jpeg")
            .to_string();

        let ext = resolve_image_ext(&content_type);

        let data = read_response_limited(
            resp,
            self.config.max_image_bytes,
            self.request_timeout(),
            "image",
        )
        .await?;

        #[cfg(debug_assertions)]
        tracing::debug!(data_len = %data.len(), content_type = %content_type, "Downloaded image data");

        let filename = format!(
            "{}_{}.{}",
            message_id,
            &image_key[..8.min(image_key.len())],
            ext
        );
        let file_path = resolve_unique_path(media_dir, &filename).await;

        ensure_media_capacity(
            media_dir,
            data.len() as u64,
            self.config.media_dir_max_bytes,
        )
        .await?;

        tokio::fs::write(&file_path, &data)
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to write image: {}", e)))?;

        let media_item = MediaItem::new(file_path.to_string_lossy().to_string(), "image");

        tracing::info!(message_id = %message_id, filename = %filename, "Downloaded image");

        Ok((String::new(), Some(media_item)))
    }

    /// Download file/audio from Feishu
    async fn download_file(
        &self,
        content_json: &serde_json::Value,
        message_id: &str,
        media_dir: &Path,
        file_type: &str,
    ) -> Result<(String, Option<MediaItem>), ChannelError> {
        let file_key = content_json
            .get("file_key")
            .and_then(|v| v.as_str())
            .ok_or_else(|| ChannelError::Other("No file_key in message".to_string()))?;

        // Use message resource API for downloading message files
        let url = format!(
            "{}/im/v1/messages/{}/resources/{}?type=file",
            FEISHU_API_BASE, message_id, file_key
        );

        #[cfg(debug_assertions)]
        tracing::debug!(url = %url, file_key = %file_key, message_id = %message_id, "Downloading file from Feishu via message resource API");

        let resp = self
            .send_authenticated("download Feishu file", |client, token| {
                client.get(&url).bearer_auth(token)
            })
            .await?;

        let status = resp.status();
        if !status.is_success() {
            let error_body =
                read_response_limited(resp, 64 * 1024, self.request_timeout(), "error response")
                    .await
                    .unwrap_or_default();
            return Err(ChannelError::Other(format!(
                "File download failed {}: {}",
                status,
                String::from_utf8_lossy(&error_body)
            )));
        }

        let data = read_response_limited(
            resp,
            self.config.max_file_bytes,
            self.request_timeout(),
            file_type,
        )
        .await?;

        let filename = content_json
            .get("file_name")
            .and_then(|v| v.as_str())
            .map(sanitize_filename)
            .unwrap_or_else(|| {
                let ext = resolve_file_ext(content_json);
                if ext.is_empty() {
                    format!("{}_{}", message_id, &file_key[..8.min(file_key.len())])
                } else {
                    format!(
                        "{}_{}.{}",
                        message_id,
                        &file_key[..8.min(file_key.len())],
                        ext
                    )
                }
            });
        let file_path = resolve_unique_path(media_dir, &filename).await;

        ensure_media_capacity(
            media_dir,
            data.len() as u64,
            self.config.media_dir_max_bytes,
        )
        .await?;

        tokio::fs::write(&file_path, &data)
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to write file: {}", e)))?;

        let media_item = MediaItem::new(file_path.to_string_lossy().to_string(), file_type);

        tracing::info!(message_id = %message_id, filename = %filename, file_type = %file_type, "Downloaded file");

        Ok((String::new(), Some(media_item)))
    }

    /// Upload image to Feishu and return the image_key
    async fn upload_image(&self, file_path: &str) -> Result<String, ChannelError> {
        ensure_local_media_size(file_path, self.config.max_image_bytes, "image").await?;
        let token = self.get_tenant_access_token().await?;

        let mime = mime_guess::from_path(file_path)
            .first_or_octet_stream()
            .to_string();

        let file_name = std::path::Path::new(file_path)
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("image.jpg");

        let file_data = tokio::fs::read(file_path)
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to read file: {}", e)))?;

        let part = reqwest::multipart::Part::bytes(file_data)
            .file_name(file_name.to_string())
            .mime_str(&mime)
            .map_err(|e| ChannelError::Other(format!("Invalid mime type: {}", e)))?;

        let form = reqwest::multipart::Form::new()
            .text("image_type", "message".to_string())
            .part("image", part);

        let resp = self
            .http_client
            .post(format!("{}/im/v1/images", FEISHU_API_BASE))
            .header("Authorization", format!("Bearer {}", token))
            .multipart(form)
            .send()
            .await
            .map_err(|e| {
                ChannelError::ConnectionError(format!("Upload image HTTP error: {}", e))
            })?;

        let status = resp.status();
        let body_text = resp
            .text()
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to read upload response: {}", e)))?;
        tracing::debug!(status = %status, body = %body_text, "Feishu upload image");

        #[derive(Deserialize)]
        struct UploadResp {
            code: i32,
            msg: Option<String>,
            data: Option<UploadData>,
        }
        #[derive(Deserialize)]
        struct UploadData {
            image_key: String,
        }

        let result: UploadResp = serde_json::from_str(&body_text).map_err(|e| {
            ChannelError::Other(format!(
                "Parse upload response error: {} | body: {}",
                e, &body_text
            ))
        })?;

        if result.code != 0 {
            return Err(ChannelError::Other(format!(
                "Upload image failed: code={} msg={}",
                result.code,
                result.msg.as_deref().unwrap_or("unknown")
            )));
        }

        result
            .data
            .map(|d| d.image_key)
            .ok_or_else(|| ChannelError::Other("No image_key in response".to_string()))
    }

    /// Upload file to Feishu and return the file_key
    async fn upload_file(&self, file_path: &str) -> Result<String, ChannelError> {
        ensure_local_media_size(file_path, self.config.max_file_bytes, "file").await?;
        let token = self.get_tenant_access_token().await?;

        let file_name = std::path::Path::new(file_path)
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("file.bin");

        let extension = std::path::Path::new(file_path)
            .extension()
            .and_then(|e| e.to_str())
            .unwrap_or("")
            .to_lowercase();

        let file_type = match extension.as_str() {
            "opus" => "opus",
            "mp4" | "mov" | "avi" | "mkv" => "mp4",
            "pdf" => "pdf",
            "doc" | "docx" => "doc",
            "xls" | "xlsx" => "xls",
            "ppt" | "pptx" => "ppt",
            _ => "stream",
        };

        let file_data = tokio::fs::read(file_path)
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to read file: {}", e)))?;

        let part = reqwest::multipart::Part::bytes(file_data)
            .file_name(file_name.to_string())
            .mime_str("application/octet-stream")
            .map_err(|e| ChannelError::Other(format!("Invalid mime type: {}", e)))?;

        let form = reqwest::multipart::Form::new()
            .text("file_type", file_type.to_string())
            .text("file_name", file_name.to_string())
            .part("file", part);

        let resp = self
            .http_client
            .post(format!("{}/im/v1/files", FEISHU_API_BASE))
            .header("Authorization", format!("Bearer {}", token))
            .multipart(form)
            .send()
            .await
            .map_err(|e| ChannelError::ConnectionError(format!("Upload file HTTP error: {}", e)))?;

        let status = resp.status();
        let body_text = resp
            .text()
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to read upload response: {}", e)))?;
        tracing::debug!(status = %status, body = %body_text, "Feishu upload file");

        #[derive(Deserialize)]
        struct UploadResp {
            code: i32,
            msg: Option<String>,
            data: Option<UploadData>,
        }
        #[derive(Deserialize)]
        struct UploadData {
            file_key: String,
        }

        let result: UploadResp = serde_json::from_str(&body_text).map_err(|e| {
            ChannelError::Other(format!(
                "Parse upload response error: {} | body: {}",
                e, &body_text
            ))
        })?;

        if result.code != 0 {
            return Err(ChannelError::Other(format!(
                "Upload file failed: code={} msg={}",
                result.code,
                result.msg.as_deref().unwrap_or("unknown")
            )));
        }

        result
            .data
            .map(|d| d.file_key)
            .ok_or_else(|| ChannelError::Other("No file_key in response".to_string()))
    }

    /// Add a reaction emoji to a message and store the reaction_id for later removal.
    /// Returns the reaction_id if successful, None otherwise.
    async fn add_reaction(&self, message_id: &str) -> Result<Option<String>, ChannelError> {
        let emoji = self.config.reaction_emoji.trim();
        if emoji.is_empty() {
            return Ok(None);
        }
        let resp = self
            .send_authenticated("add Feishu reaction", |client, token| {
                client
                    .post(format!(
                        "{}/im/v1/messages/{}/reactions",
                        FEISHU_API_BASE, message_id
                    ))
                    .bearer_auth(token)
                    .json(&serde_json::json!({
                        "reaction_type": { "emoji_type": emoji }
                    }))
            })
            .await?;

        #[derive(Deserialize)]
        struct ReactionResp {
            code: i32,
            msg: Option<String>,
            data: Option<ReactionData>,
        }
        #[derive(Deserialize)]
        struct ReactionData {
            reaction_id: Option<String>,
        }

        let result: ReactionResp = resp
            .json()
            .await
            .map_err(|e| ChannelError::Other(format!("Parse reaction response error: {}", e)))?;

        if result.code != 0 {
            tracing::warn!(
                "Failed to add reaction to message {}: code={} msg={}",
                message_id,
                result.code,
                result.msg.as_deref().unwrap_or("unknown")
            );
            return Ok(None);
        }

        let reaction_id = result.data.and_then(|d| d.reaction_id);
        Ok(reaction_id)
    }

    /// Remove reaction using feishu metadata propagated through OutboundMessage.
    /// Reads feishu.message_id and feishu.reaction_id from metadata.
    async fn remove_reaction_from_metadata(
        &self,
        metadata: &std::collections::HashMap<String, String>,
    ) {
        let (message_id, reaction_id) = match (
            metadata.get("feishu.message_id"),
            metadata.get("feishu.reaction_id"),
        ) {
            (Some(msg_id), Some(rid)) => (msg_id.clone(), rid.clone()),
            _ => return,
        };
        if let Err(e) = self.remove_reaction(&message_id, &reaction_id).await {
            tracing::debug!(error = %e, message_id = %message_id, "Failed to remove reaction");
        }
    }

    /// Remove a reaction emoji from a message.
    async fn remove_reaction(
        &self,
        message_id: &str,
        reaction_id: &str,
    ) -> Result<(), ChannelError> {
        let resp = self
            .send_authenticated("remove Feishu reaction", |client, token| {
                client
                    .delete(format!(
                        "{}/im/v1/messages/{}/reactions/{}",
                        FEISHU_API_BASE, message_id, reaction_id
                    ))
                    .bearer_auth(token)
            })
            .await?;

        #[derive(Deserialize)]
        struct ReactionResp {
            code: i32,
            msg: Option<String>,
        }

        let result: ReactionResp = resp.json().await.map_err(|e| {
            ChannelError::Other(format!("Parse remove reaction response error: {}", e))
        })?;

        if result.code != 0 {
            tracing::debug!(
                "Failed to remove reaction {} from message {}: code={} msg={}",
                reaction_id,
                message_id,
                result.code,
                result.msg.as_deref().unwrap_or("unknown")
            );
        }
        Ok(())
    }

    const REPLY_CONTEXT_MAX_LEN: usize = 500;

    /// Fetch the text content of a Feishu message by ID.
    /// Returns a "[Reply to: ...]" context string, or None on failure.
    async fn get_message_content(&self, message_id: &str) -> Option<String> {
        let resp = self
            .send_authenticated("fetch Feishu parent message", |client, token| {
                client
                    .get(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id))
                    .bearer_auth(token)
            })
            .await
            .ok()?;

        #[derive(Deserialize)]
        struct MessageResp {
            code: i32,
            msg: Option<String>,
            data: Option<MessageData>,
        }
        #[derive(Deserialize)]
        struct MessageData {
            items: Option<Vec<MessageItem>>,
        }
        #[derive(Deserialize)]
        struct MessageItem {
            msg_type: String,
            body: Option<MessageBody>,
        }
        #[derive(Deserialize)]
        struct MessageBody {
            content: Option<String>,
        }

        let result: MessageResp = match resp.json().await {
            Ok(r) => r,
            Err(e) => {
                tracing::debug!(error = %e, message_id = %message_id, "Feishu: failed to parse parent message response");
                return None;
            }
        };

        if result.code != 0 {
            tracing::debug!(
                message_id = %message_id,
                code = %result.code,
                msg = ?result.msg,
                "Feishu: failed to fetch parent message"
            );
            return None;
        }

        let items = result.data?.items?;
        let msg_obj = items.first()?;

        let raw_content = msg_obj.body.as_ref()?.content.as_ref()?;
        let msg_type = msg_obj.msg_type.as_str();

        let text = match msg_type {
            "text" => serde_json::from_str::<serde_json::Value>(raw_content)
                .ok()?
                .get("text")?
                .as_str()?
                .to_string(),
            "post" => parse_post_content(raw_content),
            _ => String::new(),
        };

        if text.is_empty() {
            return None;
        }

        let text = if text.len() > Self::REPLY_CONTEXT_MAX_LEN {
            format!(
                "{}...",
                &text[..text.floor_char_boundary(Self::REPLY_CONTEXT_MAX_LEN)]
            )
        } else {
            text
        };

        Some(format!("[Reply to: {}]", text))
    }

    /// Deliver one native Feishu message using create or reply semantics.
    async fn send_raw_message(
        &self,
        target: &FeishuSendTarget,
        msg_type: &str,
        content: &str,
    ) -> Result<String, ChannelError> {
        let mut active_target = target.clone();
        let mut token_refreshes = 0;
        loop {
            let (url, body) = active_target.request(msg_type, content);
            let response = self
                .send_authenticated("send Feishu message", |client, token| {
                    client.post(url.clone()).bearer_auth(token).json(&body)
                })
                .await?;
            let result: FeishuMessageResponse = response.json().await.map_err(|error| {
                ChannelError::Other(format!("Parse send response error: {error}"))
            })?;

            if result.code == FEISHU_INVALID_TOKEN_CODE && token_refreshes == 0 {
                token_refreshes += 1;
                self.invalidate_tenant_token().await;
                continue;
            }
            if matches!(result.code, 230011 | 231003)
                && active_target.reply_to.is_some()
                && !active_target.reply_in_thread
            {
                tracing::warn!(
                    code = result.code,
                    "Feishu reply target is unavailable; falling back to a new chat message"
                );
                active_target.reply_to = None;
                continue;
            }
            if result.code != 0 {
                return Err(ChannelError::Other(format!(
                    "Send message failed: code={} msg={}",
                    result.code, result.msg
                )));
            }
            return Ok(result.data.map_or_else(String::new, |data| data.message_id));
        }
    }

    /// Extract service_id from WebSocket URL query params
    fn extract_service_id(url: &str) -> i32 {
        url.split('?')
            .nth(1)
            .and_then(|qs| {
                qs.split('&')
                    .find(|kv| kv.starts_with("service_id="))
                    .and_then(|kv| kv.split('=').nth(1))
                    .and_then(|v| v.parse::<i32>().ok())
            })
            .unwrap_or(0)
    }

    /// Parse one complete event frame. Transport ACK and fragment handling happen earlier.
    async fn handle_frame(&self, frame: &PbFrame) -> Result<Option<ParsedMessage>, ChannelError> {
        // method 0 = CONTROL (ping/pong)
        if frame.method == 0 {
            return Ok(None);
        }

        // method 1 = DATA (events)
        if frame.method != 1 {
            return Ok(None);
        }

        let payload = frame
            .payload
            .as_deref()
            .ok_or_else(|| ChannelError::Other("No payload in frame".to_string()))?;

        #[cfg(debug_assertions)]
        tracing::debug!(payload_len = %payload.len(), "Received frame payload");

        let event: LarkEvent = serde_json::from_slice(payload)
            .map_err(|e| ChannelError::Other(format!("Parse event error: {}", e)))?;

        let event_type = event.header.event_type.as_str();
        #[cfg(debug_assertions)]
        tracing::debug!(event_type = %event_type, "Received event type");
        if event_type != "im.message.receive_v1" {
            return Ok(None);
        }

        let payload_data: MsgReceivePayload = serde_json::from_value(event.event.clone())
            .map_err(|e| ChannelError::Other(format!("Parse payload error: {}", e)))?;

        // Never let bot/app traffic trigger another model turn.
        if matches!(payload_data.sender.sender_type.as_str(), "bot" | "app") {
            return Ok(None);
        }

        let open_id = payload_data
            .sender
            .sender_id
            .open_id
            .ok_or_else(|| ChannelError::Other("No open_id".to_string()))?;

        if !self.is_allowed(&open_id) {
            tracing::warn!(sender = %open_id, "Rejected unauthorized Feishu sender");
            return Ok(None);
        }

        let message_id = payload_data.message.message_id.clone();
        let msg = payload_data.message;
        if msg.chat_type == "group" && self.config.require_mention {
            let bot_open_id = self.bot_open_id.read().await.clone();
            if !message_mentions_bot(&msg, bot_open_id.as_deref()) {
                #[cfg(debug_assertions)]
                tracing::debug!(message_id = %message_id, "Ignoring group message without bot mention");
                return Ok(None);
            }
        }

        // Deduplicate only after admission so rejected traffic does not consume cache capacity.
        if self.is_message_seen(&message_id).await {
            #[cfg(debug_assertions)]
            tracing::debug!(message_id = %message_id, "Duplicate message, skipping");
            return Ok(None);
        }

        #[cfg(debug_assertions)]
        tracing::debug!(message_id = %message_id, "Received Feishu message");

        let chat_id = msg.chat_id.clone();
        let msg_type = msg.message_type.as_str();
        let raw_content = msg.content.clone();
        let parent_id = msg.parent_id.clone();
        let root_id = msg.root_id.clone();
        let thread_id = msg.thread_id.clone().or_else(|| root_id.clone());
        let chat_type = msg.chat_type.clone();

        #[cfg(debug_assertions)]
        tracing::debug!(msg_type = %msg_type, chat_id = %chat_id, open_id = %open_id, "Parsing message content");

        let (mut content, media) = self
            .parse_and_download_message(msg_type, &raw_content, &message_id)
            .await?;
        content = normalize_mentions(
            &content,
            &msg.mentions,
            self.bot_open_id.read().await.as_deref(),
        );

        // Fetch and prepend quoted message content if this is a reply
        if let Some(ref pid) = parent_id
            && let Some(reply_ctx) = self.get_message_content(pid).await
        {
            content = format!("{}\n{}", reply_ctx, content);
        }

        #[cfg(debug_assertions)]
        for m in &media {
            tracing::debug!(media_type = %m.media_type, media_path = %m.path, "Media downloaded successfully");
        }

        Ok(Some(ParsedMessage {
            message_id,
            open_id,
            chat_id,
            content,
            media,
            parent_id,
            root_id,
            thread_id,
            chat_type,
        }))
    }

    /// Parse message content and download media if needed
    async fn parse_and_download_message(
        &self,
        msg_type: &str,
        content: &str,
        message_id: &str,
    ) -> Result<(String, Vec<MediaItem>), ChannelError> {
        let (text, media) = match msg_type {
            "text" => {
                let text = if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
                    parsed
                        .get("text")
                        .and_then(|v| v.as_str())
                        .unwrap_or(content)
                        .to_string()
                } else {
                    content.to_string()
                };
                (text, Vec::new())
            }
            "post" => {
                let text = parse_post_content(content);
                let mut media = Vec::new();

                for image_key in collect_post_image_keys(content) {
                    let content_json = serde_json::json!({ "image_key": image_key });
                    match self
                        .download_media("image", &content_json, message_id)
                        .await
                    {
                        Ok((_text, Some(item))) => media.push(item),
                        Ok((_text, None)) => {}
                        Err(e) => {
                            tracing::warn!(error = %e, "Failed to download image from Feishu post message");
                        }
                    }
                }

                (text, media)
            }
            "image" | "audio" | "file" | "media" => {
                if let Ok(content_json) = serde_json::from_str::<serde_json::Value>(content) {
                    match self
                        .download_media(msg_type, &content_json, message_id)
                        .await
                    {
                        Ok((text, Some(media))) => (text, vec![media]),
                        Ok((text, None)) => (text, Vec::new()),
                        Err(_) => (format!("[{}: content unavailable]", msg_type), Vec::new()),
                    }
                } else {
                    (format!("[{}: content unavailable]", msg_type), Vec::new())
                }
            }
            "share_chat" => {
                // Shared chat/cannel messages
                if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
                    let chat_id = parsed
                        .get("chat_id")
                        .and_then(|v| v.as_str())
                        .unwrap_or("unknown");
                    (format!("[shared chat: {}]", chat_id), Vec::new())
                } else {
                    ("[shared chat]".to_string(), Vec::new())
                }
            }
            "share_user" => {
                // Shared user messages
                if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
                    let user_id = parsed
                        .get("user_id")
                        .and_then(|v| v.as_str())
                        .unwrap_or("unknown");
                    (format!("[shared user: {}]", user_id), Vec::new())
                } else {
                    ("[shared user]".to_string(), Vec::new())
                }
            }
            "interactive" => {
                // Interactive card messages - extract text content
                match extract_interactive_content(content) {
                    Ok((text, Some(media))) => (text, vec![media]),
                    Ok((text, None)) => (text, Vec::new()),
                    Err(e) => {
                        tracing::warn!(error = %e, "Failed to extract interactive content");
                        (content.to_string(), Vec::new())
                    }
                }
            }
            "list" => {
                // List/bullet messages
                match parse_list_content(content) {
                    Ok((text, Some(media))) => (text, vec![media]),
                    Ok((text, None)) => (text, Vec::new()),
                    Err(_) => (content.to_string(), Vec::new()),
                }
            }
            "merge_forward" => ("[merged forward messages]".to_string(), Vec::new()),
            "share_calendar_event" => {
                if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
                    let event_key = parsed
                        .get("event_key")
                        .and_then(|v| v.as_str())
                        .unwrap_or("unknown");
                    (
                        format!("[shared calendar event: {}]", event_key),
                        Vec::new(),
                    )
                } else {
                    ("[shared calendar event]".to_string(), Vec::new())
                }
            }
            "system" => ("[system message]".to_string(), Vec::new()),
            _ => (content.to_string(), Vec::new()),
        };

        Ok((text, media))
    }

    async fn process_event_frame(&self, bus: &Arc<MessageBus>, frame: &PbFrame) {
        let parsed = match self.handle_frame(frame).await {
            Ok(Some(parsed)) => parsed,
            Ok(None) => return,
            Err(error) => {
                tracing::warn!(error = %error, "Failed to parse Feishu frame");
                return;
            }
        };

        let message_id = parsed.message_id.clone();
        let reaction_id = match self.add_reaction(&message_id).await {
            Ok(Some(reaction_id)) => Some(reaction_id),
            Ok(None) => None,
            Err(error) => {
                tracing::debug!(error = %error, message_id = %message_id, "Failed to add reaction");
                None
            }
        };

        let mut private_context = HashMap::new();
        private_context.insert("feishu.message_id".to_string(), message_id.clone());
        private_context.insert("feishu.chat_type".to_string(), parsed.chat_type);
        if let Some(parent_id) = parsed.parent_id.as_ref() {
            private_context.insert("feishu.parent_id".to_string(), parent_id.clone());
        }
        if let Some(root_id) = parsed.root_id {
            private_context.insert("feishu.root_id".to_string(), root_id);
        }
        if let Some(thread_id) = parsed.thread_id {
            private_context.insert("feishu.thread_id".to_string(), thread_id);
        }
        if let Some(reaction_id) = reaction_id {
            private_context.insert("feishu.reaction_id".to_string(), reaction_id);
        }

        let msg = crate::bus::InboundMessage {
            channel: "feishu".to_string(),
            sender_id: parsed.open_id.clone(),
            chat_id: parsed.chat_id.clone(),
            content: parsed.content,
            received_at: crate::bus::message::current_timestamp(),
            media: parsed.media,
            channel_context: crate::bus::ChannelContext {
                reply_to: Some(message_id),
                private: private_context,
            },
        };
        if let Err(error) = self.handle_and_publish(bus, &msg).await {
            tracing::error!(
                error = %error,
                open_id = %parsed.open_id,
                chat_id = %parsed.chat_id,
                "Failed to publish Feishu message to bus"
            );
        }
    }

    /// Send acknowledgment for a message
    async fn send_ack(
        frame: &PbFrame,
        write: &mut futures_util::stream::SplitSink<
            tokio_tungstenite::WebSocketStream<
                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
            >,
            tokio_tungstenite::tungstenite::Message,
        >,
    ) -> Result<(), ChannelError> {
        let mut ack = frame.clone();
        ack.payload = Some(br#"{"code":200,"headers":{},"data":[]}"#.to_vec());
        ack.headers.push(PbHeader {
            key: "biz_rt".into(),
            value: "0".into(),
        });
        write
            .send(tokio_tungstenite::tungstenite::Message::Binary(
                ack.encode_to_vec().into(),
            ))
            .await
            .map_err(|e| ChannelError::Other(format!("Failed to send ack: {}", e)))?;
        Ok(())
    }

    async fn run_ws_loop(
        &self,
        bus: Arc<MessageBus>,
        shutdown: CancellationToken,
    ) -> Result<Duration, ChannelError> {
        let (wss_url, client_config) = tokio::select! {
            result = self.get_ws_endpoint(&self.http_client) => result?,
            _ = shutdown.cancelled() => return Ok(Duration::ZERO),
        };

        let service_id = Self::extract_service_id(&wss_url);
        tracing::info!(service_id, "Connecting to Feishu WebSocket");

        let (ws_stream, _) = tokio::select! {
            result = tokio_tungstenite::connect_async(&wss_url) => result.map_err(|e| {
                ChannelError::ConnectionError(format!("WebSocket connection failed: {}", e))
            })?,
            _ = shutdown.cancelled() => return Ok(Duration::ZERO),
        };
        let connected_at = Instant::now();

        *self.connected.write().await = true;
        tracing::info!("Feishu WebSocket connected");

        let (mut write, mut read) = ws_stream.split();

        // Send initial ping
        let ping_frame = PbFrame {
            seq_id: 1,
            log_id: 0,
            service: service_id,
            method: 0,
            headers: vec![PbHeader {
                key: "type".into(),
                value: "ping".into(),
            }],
            payload: None,
        };
        tokio::select! {
            result = write.send(tokio_tungstenite::tungstenite::Message::Binary(
                ping_frame.encode_to_vec().into(),
            )) => result.map_err(|e| {
                ChannelError::ConnectionError(format!("Failed to send initial ping: {}", e))
            })?,
            _ = shutdown.cancelled() => return Ok(connected_at.elapsed()),
        };

        let (event_tx, mut event_rx) = tokio::sync::mpsc::channel(WS_EVENT_QUEUE_CAPACITY);
        let worker_channel = self.clone();
        let worker_bus = bus.clone();
        let mut event_worker = tokio::spawn(async move {
            while let Some(frame) = event_rx.recv().await {
                worker_channel
                    .process_event_frame(&worker_bus, &frame)
                    .await;
            }
        });

        let ping_interval = client_config.ping_interval.unwrap_or(120).max(10);
        let mut ping_interval_tok =
            tokio::time::interval(tokio::time::Duration::from_secs(ping_interval));
        let mut timeout_check = tokio::time::interval(tokio::time::Duration::from_secs(10));
        let mut seq: u64 = 1;
        let mut last_recv = Instant::now();
        let mut fragment_cache: HashMap<String, FragmentAssembly> = HashMap::new();

        // Consume the immediate tick
        ping_interval_tok.tick().await;
        timeout_check.tick().await;

        loop {
            tokio::select! {
                msg = read.next() => {
                    match msg {
                        Some(Ok(tokio_tungstenite::tungstenite::Message::Binary(data))) => {
                            last_recv = Instant::now();
                            let bytes: Bytes = data;
                            if let Ok(frame) = PbFrame::decode(bytes.as_ref()) {
                                if frame.method != 1 {
                                    continue;
                                }
                                match tokio::time::timeout(
                                    WS_ACK_TIMEOUT,
                                    Self::send_ack(&frame, &mut write),
                                )
                                .await
                                {
                                    Ok(Ok(())) => {}
                                    Ok(Err(error)) => {
                                        tracing::warn!(error = %error, "Failed to ACK Feishu DATA frame");
                                        break;
                                    }
                                    Err(_) => {
                                        tracing::warn!("Timed out ACKing Feishu DATA frame");
                                        break;
                                    }
                                }

                                match reassemble_data_frame(&frame, &mut fragment_cache) {
                                    Ok(Some(complete_frame)) => {
                                        match event_tx.try_send(complete_frame) {
                                            Ok(()) => {}
                                            Err(tokio::sync::mpsc::error::TrySendError::Full(frame)) => {
                                                tracing::warn!(
                                                    capacity = WS_EVENT_QUEUE_CAPACITY,
                                                    "Feishu event queue is full; processing one event on the connection task"
                                                );
                                                self.process_event_frame(&bus, &frame).await;
                                            }
                                            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
                                                tracing::warn!("Feishu event worker stopped unexpectedly");
                                                break;
                                            }
                                        }
                                    }
                                    Ok(None) => {}
                                    Err(error) => {
                                        tracing::warn!(error = %error, "Rejected invalid Feishu fragment sequence");
                                    }
                                }
                            }
                        }
                        Some(Ok(tokio_tungstenite::tungstenite::Message::Ping(data))) => {
                            last_recv = Instant::now();
                            let pong = PbFrame {
                                seq_id: seq.wrapping_add(1),
                                log_id: 0,
                                service: service_id,
                                method: 0,
                                headers: vec![PbHeader {
                                    key: "type".into(),
                                    value: "pong".into(),
                                }],
                                payload: Some(data.to_vec()),
                            };
                            let _ = write.send(tokio_tungstenite::tungstenite::Message::Binary(pong.encode_to_vec().into())).await;
                        }
                        Some(Ok(tokio_tungstenite::tungstenite::Message::Pong(_))) => {
                            last_recv = Instant::now();
                        }
                        Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) | None => {
                            #[cfg(debug_assertions)]
                            tracing::debug!("Feishu WebSocket closed");
                            break;
                        }
                        Some(Err(e)) => {
                            tracing::warn!(error = %e, "Feishu WebSocket error");
                            break;
                        }
                        _ => {}
                    }
                }
                _ = ping_interval_tok.tick() => {
                    seq = seq.wrapping_add(1);
                    let ping = PbFrame {
                        seq_id: seq,
                        log_id: 0,
                        service: service_id,
                        method: 0,
                        headers: vec![PbHeader {
                            key: "type".into(),
                            value: "ping".into(),
                        }],
                        payload: None,
                    };
                    if write.send(tokio_tungstenite::tungstenite::Message::Binary(ping.encode_to_vec().into())).await.is_err() {
                        tracing::warn!("Feishu ping failed, reconnecting");
                        break;
                    }
                }
                _ = timeout_check.tick() => {
                    if last_recv.elapsed() > WS_HEARTBEAT_TIMEOUT {
                        tracing::warn!("Feishu WebSocket heartbeat timeout, reconnecting");
                        break;
                    }
                    // GC dedup cache: remove entries older than TTL (matches zeroclaw pattern)
                    let now = Instant::now();
                    let mut seen = self.seen_message_ids.write().await;
                    seen.retain(|_, ts| now.duration_since(*ts) < DEDUP_CACHE_TTL);
                    fragment_cache.retain(|_, assembly| {
                        now.duration_since(assembly.created_at) < WS_FRAGMENT_TTL
                    });
                }
                _ = shutdown.cancelled() => {
                    tracing::info!("Feishu channel shutdown signal received");
                    break;
                }
            }
        }

        *self.connected.write().await = false;
        drop(event_tx);
        if tokio::time::timeout(CHANNEL_STOP_GRACE, &mut event_worker)
            .await
            .is_err()
        {
            event_worker.abort();
            let _ = event_worker.await;
        }
        Ok(connected_at.elapsed())
    }
}

fn reassemble_data_frame(
    frame: &PbFrame,
    cache: &mut HashMap<String, FragmentAssembly>,
) -> Result<Option<PbFrame>, ChannelError> {
    let sum = frame
        .header_value("sum")
        .and_then(|value| value.parse::<usize>().ok())
        .unwrap_or(1)
        .max(1);
    if sum == 1 {
        if frame
            .payload
            .as_ref()
            .is_some_and(|payload| payload.len() > MAX_WS_EVENT_BYTES)
        {
            return Err(ChannelError::Other(format!(
                "Feishu event exceeds {MAX_WS_EVENT_BYTES} byte limit"
            )));
        }
        return Ok(Some(frame.clone()));
    }
    if sum > MAX_WS_FRAGMENTS {
        return Err(ChannelError::Other(format!(
            "Feishu event declares too many fragments: {sum}"
        )));
    }

    let message_id = frame.header_value("message_id").unwrap_or_default();
    if message_id.is_empty() {
        return Err(ChannelError::Other(
            "Fragmented Feishu event has no message_id".to_string(),
        ));
    }
    let seq = frame
        .header_value("seq")
        .and_then(|value| value.parse::<usize>().ok())
        .unwrap_or(0);
    if seq >= sum {
        return Err(ChannelError::Other(format!(
            "Feishu fragment index {seq} is outside declared count {sum}"
        )));
    }

    let assembly = cache
        .entry(message_id.to_string())
        .or_insert_with(|| FragmentAssembly {
            parts: vec![None; sum],
            total_bytes: 0,
            created_at: Instant::now(),
        });
    if assembly.parts.len() != sum {
        *assembly = FragmentAssembly {
            parts: vec![None; sum],
            total_bytes: 0,
            created_at: Instant::now(),
        };
    }
    let payload = frame.payload.clone().unwrap_or_default();
    let previous_len = assembly.parts[seq].as_ref().map_or(0, Vec::len);
    assembly.total_bytes = assembly.total_bytes - previous_len + payload.len();
    if assembly.total_bytes > MAX_WS_EVENT_BYTES {
        cache.remove(message_id);
        return Err(ChannelError::Other(format!(
            "Feishu fragmented event exceeds {MAX_WS_EVENT_BYTES} byte limit"
        )));
    }
    assembly.parts[seq] = Some(payload);
    if assembly.parts.iter().any(Option::is_none) {
        return Ok(None);
    }

    let complete_payload = assembly
        .parts
        .iter()
        .flat_map(|part| part.as_deref().unwrap_or_default())
        .copied()
        .collect();
    cache.remove(message_id);
    let mut complete = frame.clone();
    complete.payload = Some(complete_payload);
    Ok(Some(complete))
}

fn reconnect_delay(attempt: u32) -> Duration {
    let exponent = attempt.saturating_sub(1).min(6);
    Duration::from_secs(1_u64 << exponent)
}

fn api_retry_delay(attempt: usize) -> Duration {
    Duration::from_millis(250_u64.saturating_mul(1_u64 << attempt.min(4)))
}

fn retry_after_delay(headers: &reqwest::header::HeaderMap, attempt: usize) -> Duration {
    headers
        .get(reqwest::header::RETRY_AFTER)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse::<u64>().ok())
        .map(|seconds| Duration::from_secs(seconds.min(5)))
        .unwrap_or_else(|| api_retry_delay(attempt))
}

async fn read_response_limited(
    mut response: reqwest::Response,
    max_bytes: u64,
    timeout: Duration,
    media_type: &str,
) -> Result<Vec<u8>, ChannelError> {
    if response
        .content_length()
        .is_some_and(|length| length > max_bytes)
    {
        return Err(ChannelError::Other(format!(
            "Feishu {media_type} exceeds {max_bytes} byte limit"
        )));
    }
    tokio::time::timeout(timeout, async move {
        let mut data = Vec::new();
        while let Some(chunk) = response.chunk().await.map_err(|error| {
            ChannelError::ConnectionError(format!("Failed to read {media_type} data: {error}"))
        })? {
            if (data.len() as u64).saturating_add(chunk.len() as u64) > max_bytes {
                return Err(ChannelError::Other(format!(
                    "Feishu {media_type} exceeds {max_bytes} byte limit"
                )));
            }
            data.extend_from_slice(&chunk);
        }
        Ok(data)
    })
    .await
    .map_err(|_| ChannelError::ConnectionError(format!("Feishu {media_type} download timed out")))?
}

async fn ensure_media_capacity(
    media_dir: &Path,
    incoming_bytes: u64,
    max_bytes: u64,
) -> Result<(), ChannelError> {
    let mut total = 0_u64;
    let mut entries = tokio::fs::read_dir(media_dir)
        .await
        .map_err(|error| ChannelError::Other(format!("Failed to inspect media dir: {error}")))?;
    while let Some(entry) = entries
        .next_entry()
        .await
        .map_err(|error| ChannelError::Other(format!("Failed to inspect media dir: {error}")))?
    {
        if let Ok(metadata) = entry.metadata().await
            && metadata.is_file()
        {
            total = total.saturating_add(metadata.len());
        }
    }
    if total.saturating_add(incoming_bytes) > max_bytes {
        return Err(ChannelError::Other(format!(
            "Feishu media directory would exceed {max_bytes} byte limit"
        )));
    }
    Ok(())
}

async fn ensure_local_media_size(
    file_path: &str,
    max_bytes: u64,
    media_type: &str,
) -> Result<(), ChannelError> {
    let size = tokio::fs::metadata(file_path)
        .await
        .map_err(|error| ChannelError::Other(format!("Failed to inspect {media_type}: {error}")))?
        .len();
    if size > max_bytes {
        return Err(ChannelError::Other(format!(
            "Outbound {media_type} is {size} bytes, exceeding {max_bytes} byte limit"
        )));
    }
    Ok(())
}

fn sanitize_filename(filename: &str) -> String {
    let basename = Path::new(filename)
        .file_name()
        .and_then(|value| value.to_str())
        .unwrap_or("file.bin");
    let sanitized: String = basename
        .chars()
        .map(|character| {
            if character.is_control() || matches!(character, '/' | '\\' | ':' | '\0') {
                '_'
            } else {
                character
            }
        })
        .collect();
    let sanitized = sanitized.trim_matches(['.', ' ']);
    if sanitized.is_empty() {
        "file.bin".to_string()
    } else {
        let boundary = sanitized.floor_char_boundary(sanitized.len().min(255));
        sanitized[..boundary].to_string()
    }
}

fn parse_post_content(content: &str) -> String {
    /// Extract text from a single post element (text, link, at-mention).
    fn extract_element(el: &serde_json::Value, out: &mut Vec<String>) {
        match el.get("tag").and_then(|t| t.as_str()).unwrap_or("") {
            "text" => {
                if let Some(text) = el.get("text").and_then(|t| t.as_str()) {
                    out.push(text.to_string());
                }
            }
            "a" => {
                let link_text = el
                    .get("text")
                    .and_then(|t| t.as_str())
                    .filter(|s| !s.is_empty())
                    .or_else(|| el.get("href").and_then(|h| h.as_str()))
                    .unwrap_or("");
                out.push(link_text.to_string());
            }
            "at" => {
                let name = el
                    .get("user_name")
                    .and_then(|n| n.as_str())
                    .or_else(|| el.get("user_id").and_then(|i| i.as_str()))
                    .unwrap_or("user");
                out.push(format!("@{}", name));
            }
            "img" => {
                out.push("[image]".to_string());
            }
            "code_block" => {
                let lang = el.get("language").and_then(|l| l.as_str()).unwrap_or("");
                let code_text = el.get("text").and_then(|t| t.as_str()).unwrap_or("");
                out.push(format!("\n```{}\n{}\n```\n", lang, code_text));
            }
            _ => {
                if let Some(text) = el.get("text").and_then(|t| t.as_str()) {
                    out.push(text.to_string());
                }
            }
        }
    }

    /// Parse a single block {title, content: [[...]]} and append text to out.
    fn parse_block(block: &serde_json::Value, out: &mut Vec<String>) {
        let title = block
            .get("title")
            .and_then(|t| t.as_str())
            .filter(|s| !s.is_empty());
        if let Some(t) = title {
            out.push(t.to_string());
            out.push("\n\n".to_string());
        }

        if let Some(content_arr) = block.get("content").and_then(|c| c.as_array()) {
            for row in content_arr {
                if let Some(row_arr) = row.as_array() {
                    for el in row_arr {
                        extract_element(el, out);
                    }
                    out.push("\n".to_string());
                }
            }
        }
    }

    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) else {
        return content.to_string();
    };

    let mut texts = Vec::new();

    // Unwrap optional {"post": ...} envelope (nanobot: root = root["post"])
    let root = if parsed.get("post").and_then(|p| p.as_object()).is_some() {
        parsed.get("post").unwrap()
    } else {
        &parsed
    };

    // Try direct format: {"title": ..., "content": [[...]]}
    if root.get("content").and_then(|c| c.as_array()).is_some() {
        parse_block(root, &mut texts);
        let result = texts.join("");
        if !result.trim().is_empty() {
            return result.trim().to_string();
        }
        texts.clear();
    }

    // Try localized: {"zh_cn": {"title": ..., "content": [...]}}
    for key in ["zh_cn", "en_us", "ja_jp"] {
        if let Some(locale_data) = root.get(key).and_then(|l| l.as_object()) {
            parse_block(&serde_json::json!(locale_data), &mut texts);
            let result = texts.join("");
            if !result.trim().is_empty() {
                return result.trim().to_string();
            }
            texts.clear();
        }
    }

    // Fall back: try any dict child
    if let Some(root_obj) = root.as_object() {
        for (_key, val) in root_obj {
            if let Some(obj) = val.as_object()
                && obj.get("content").and_then(|c| c.as_array()).is_some()
            {
                parse_block(val, &mut texts);
                let result = texts.join("");
                if !result.trim().is_empty() {
                    return result.trim().to_string();
                }
                texts.clear();
            }
        }
    }

    content.to_string()
}

fn collect_post_image_keys(content: &str) -> Vec<String> {
    fn visit(value: &serde_json::Value, keys: &mut Vec<String>) {
        match value {
            serde_json::Value::Object(map) => {
                if let Some(image_key) = map.get("image_key").and_then(|v| v.as_str())
                    && !keys.iter().any(|k| k == image_key)
                {
                    keys.push(image_key.to_string());
                }

                for child in map.values() {
                    visit(child, keys);
                }
            }
            serde_json::Value::Array(items) => {
                for item in items {
                    visit(item, keys);
                }
            }
            _ => {}
        }
    }

    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) else {
        return Vec::new();
    };

    let mut keys = Vec::new();
    visit(&parsed, &mut keys);
    keys
}

/// Extract text content from interactive card messages
fn extract_interactive_content(content: &str) -> Result<(String, Option<MediaItem>), ChannelError> {
    let parsed = match serde_json::from_str::<serde_json::Value>(content) {
        Ok(p) => p,
        Err(_) => return Ok((content.to_string(), None)),
    };

    let mut texts = Vec::new();

    // Extract from elements array
    if let Some(elements) = parsed.get("elements").and_then(|e| e.as_array()) {
        for el in elements {
            extract_element_content(el, &mut texts);
        }
    }

    // Extract from card object
    if let Some(card) = parsed.get("card").and_then(|c| c.as_object())
        && let Some(elements) = card.get("elements").and_then(|e| e.as_array())
    {
        for el in elements {
            extract_element_content(el, &mut texts);
        }
    }

    // Extract from header
    if let Some(header) = parsed.get("header").and_then(|h| h.as_object())
        && let Some(title) = header.get("title").and_then(|t| t.as_object())
        && let Some(text) = title.get("content").and_then(|c| c.as_str())
    {
        texts.push(format!("title: {}\n", text));
    }

    let result = texts.join("").trim().to_string();
    if result.is_empty() {
        Ok((content.to_string(), None))
    } else {
        Ok((result, None))
    }
}

/// Extract content from a single card element
fn extract_element_content(element: &serde_json::Value, texts: &mut Vec<String>) {
    let tag = element.get("tag").and_then(|t| t.as_str()).unwrap_or("");

    match tag {
        "markdown" | "lark_md" => {
            if let Some(content) = element.get("content").and_then(|c| c.as_str()) {
                texts.push(content.to_string());
                texts.push("\n".to_string());
            }
        }
        "div" => {
            if let Some(text_obj) = element.get("text").and_then(|t| t.as_object()) {
                let content = text_obj
                    .get("content")
                    .and_then(|c| c.as_str())
                    .unwrap_or("");
                texts.push(content.to_string());
            } else if let Some(content) = element.get("text").and_then(|t| t.as_str()) {
                texts.push(content.to_string());
            }
            texts.push("\n".to_string());
        }
        "a" => {
            let href = element.get("href").and_then(|h| h.as_str()).unwrap_or("");
            let text = element.get("text").and_then(|t| t.as_str()).unwrap_or("");
            if !text.is_empty() {
                texts.push(text.to_string());
            } else if !href.is_empty() {
                texts.push(format!("link: {}", href));
            }
        }
        "img" => {
            let alt = element.get("alt");
            let alt_text = alt
                .and_then(|a| a.as_str())
                .or_else(|| {
                    alt.and_then(|a| a.as_object())
                        .and_then(|o| o.get("content"))
                        .and_then(|c| c.as_str())
                })
                .unwrap_or("[image]");
            texts.push(format!("{}\n", alt_text));
        }
        "note" => {
            if let Some(elements) = element.get("elements").and_then(|e| e.as_array()) {
                for el in elements {
                    extract_element_content(el, texts);
                }
            }
        }
        "column_set" => {
            if let Some(columns) = element.get("columns").and_then(|c| c.as_array()) {
                for col in columns {
                    if let Some(elements) = col.get("elements").and_then(|e| e.as_array()) {
                        for el in elements {
                            extract_element_content(el, texts);
                        }
                    }
                }
            }
        }
        "table" => {
            // Tables are complex, just indicate presence
            texts.push("[table]\n".to_string());
        }
        _ => {
            // Recursively check for nested elements
            if let Some(elements) = element.get("elements").and_then(|e| e.as_array()) {
                for el in elements {
                    extract_element_content(el, texts);
                }
            }
        }
    }
}

/// Parse Feishu list/bullet message content into plain text
fn parse_list_content(content: &str) -> Result<(String, Option<MediaItem>), ChannelError> {
    let parsed = match serde_json::from_str::<serde_json::Value>(content) {
        Ok(p) => p,
        Err(_) => return Ok((content.to_string(), None)),
    };

    let items = parsed
        .get("items")
        .and_then(|i| i.as_array())
        .or_else(|| parsed.get("content").and_then(|c| c.as_array()));

    let Some(items) = items else {
        return Ok((content.to_string(), None));
    };

    let mut lines = Vec::new();
    collect_list_items(items, &mut lines, 0);

    let result = lines.join("\n").trim().to_string();
    if result.is_empty() {
        Ok((content.to_string(), None))
    } else {
        Ok((result, None))
    }
}

/// Recursively collect list item text with indentation
fn collect_list_items(items: &[serde_json::Value], lines: &mut Vec<String>, depth: usize) {
    let indent = "  ".repeat(depth);

    for item in items {
        // Items can be arrays of inline elements or objects with content/children
        let inline_elements = if let Some(arr) = item.as_array() {
            arr.as_slice()
        } else if let Some(obj) = item.as_object() {
            obj.get("content")
                .and_then(|c| c.as_array())
                .map(|a| a.as_slice())
                .unwrap_or(&[])
        } else {
            continue;
        };

        let mut text = String::new();
        for el in inline_elements {
            extract_inline_text(el, &mut text);
        }

        let trimmed = text.trim();
        if !trimmed.is_empty() {
            lines.push(format!("{}- {}", indent, trimmed));
        }

        // Handle nested children
        if let Some(obj) = item.as_object() {
            if let Some(children) = obj.get("children").and_then(|c| c.as_array()) {
                collect_list_items(children, lines, depth + 1);
            }
        } else if let Some(children_arr) = item.as_array().and_then(|arr| {
            arr.iter()
                .find(|child| child.as_object().and_then(|o| o.get("children")).is_some())
        }) && let Some(children) = children_arr
            .as_object()
            .and_then(|o| o.get("children"))
            .and_then(|c| c.as_array())
        {
            collect_list_items(children, lines, depth + 1);
        }
    }
}

/// Extract text from inline elements (text, link, at-mention)
fn extract_inline_text(el: &serde_json::Value, out: &mut String) {
    match el.get("tag").and_then(|t| t.as_str()).unwrap_or("") {
        "text" => {
            if let Some(text) = el.get("text").and_then(|t| t.as_str()) {
                out.push_str(text);
            }
        }
        "a" => {
            let text = el
                .get("text")
                .and_then(|t| t.as_str())
                .filter(|s| !s.is_empty())
                .or_else(|| el.get("href").and_then(|h| h.as_str()))
                .unwrap_or("");
            out.push_str(text);
        }
        "at" => {
            let name = el
                .get("user_name")
                .and_then(|n| n.as_str())
                .or_else(|| el.get("user_id").and_then(|i| i.as_str()))
                .unwrap_or("user");
            out.push_str(&format!("@{}", name));
        }
        _ => {}
    }
}

fn mention_open_id(mention: &serde_json::Value) -> Option<&str> {
    mention
        .pointer("/id/open_id")
        .or_else(|| mention.get("open_id"))
        .and_then(|value| value.as_str())
}

fn message_mentions_bot(message: &LarkMessage, bot_open_id: Option<&str>) -> bool {
    let Some(bot_open_id) = bot_open_id.filter(|value| !value.is_empty()) else {
        return false;
    };
    if message
        .mentions
        .iter()
        .any(|mention| mention_open_id(mention) == Some(bot_open_id))
    {
        return true;
    }

    fn post_contains_bot(value: &serde_json::Value, bot_open_id: &str) -> bool {
        match value {
            serde_json::Value::Object(map) => {
                let is_bot_mention = map.get("tag").and_then(|value| value.as_str()) == Some("at")
                    && map
                        .get("user_id")
                        .or_else(|| map.get("open_id"))
                        .and_then(|value| value.as_str())
                        == Some(bot_open_id);
                is_bot_mention
                    || map
                        .values()
                        .any(|value| post_contains_bot(value, bot_open_id))
            }
            serde_json::Value::Array(values) => values
                .iter()
                .any(|value| post_contains_bot(value, bot_open_id)),
            _ => false,
        }
    }

    serde_json::from_str::<serde_json::Value>(&message.content)
        .is_ok_and(|content| post_contains_bot(&content, bot_open_id))
}

/// Remove the bot's own placeholder while preserving human mentions as readable names.
fn normalize_mentions(
    text: &str,
    mentions: &[serde_json::Value],
    bot_open_id: Option<&str>,
) -> String {
    let mut normalized = text.to_string();
    for mention in mentions {
        let Some(key) = mention.get("key").and_then(|value| value.as_str()) else {
            continue;
        };
        let replacement = if mention_open_id(mention) == bot_open_id {
            String::new()
        } else {
            mention
                .get("name")
                .and_then(|value| value.as_str())
                .filter(|value| !value.is_empty())
                .map(|name| format!("@{name}"))
                .unwrap_or_else(|| key.to_string())
        };
        normalized = normalized.replace(key, &replacement);
    }
    normalized.trim().to_string()
}

fn resolve_image_ext(content_type: &str) -> &str {
    match content_type {
        "image/png" => "png",
        "image/gif" => "gif",
        "image/webp" => "webp",
        "image/bmp" => "bmp",
        _ => "jpg",
    }
}

fn resolve_file_ext(content_json: &serde_json::Value) -> String {
    if let Some(name) = content_json.get("file_name").and_then(|v| v.as_str())
        && let Some(ext) = std::path::Path::new(name)
            .extension()
            .and_then(|e| e.to_str())
    {
        return ext.to_string();
    }
    String::new()
}

async fn resolve_unique_path(dir: &Path, filename: &str) -> std::path::PathBuf {
    let candidate = dir.join(filename);
    if !tokio::fs::try_exists(&candidate).await.unwrap_or(false) {
        return candidate;
    }
    let stem = std::path::Path::new(filename)
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or(filename);
    let ext = std::path::Path::new(filename)
        .extension()
        .and_then(|s| s.to_str())
        .unwrap_or("");
    let mut n = 1;
    loop {
        let candidate = if ext.is_empty() {
            dir.join(format!("{}({})", stem, n))
        } else {
            dir.join(format!("{}({}).{}", stem, n, ext))
        };
        if !tokio::fs::try_exists(&candidate).await.unwrap_or(false) {
            return candidate;
        }
        n += 1;
    }
}

impl FeishuChannel {
    /// Build a Card JSON 2.0 interactive card with a single markdown element.
    fn build_card_content(markdown: &str) -> String {
        serde_json::json!({
            "schema": "2.0",
            "body": {
                "elements": [{
                    "tag": "markdown",
                    "content": markdown
                }]
            }
        })
        .to_string()
    }

    /// Max byte-size for markdown content in a single card.
    /// Card payloads have a ~30 KB limit; leave margin for JSON envelope.
    const CARD_MARKDOWN_MAX_BYTES: usize = 28_000;

    /// Split markdown content into chunks that fit within the card size limit.
    /// Splits on line boundaries to avoid breaking markdown syntax.
    fn split_markdown_chunks(text: &str) -> Vec<String> {
        if text.len() <= Self::CARD_MARKDOWN_MAX_BYTES {
            return vec![text.to_string()];
        }

        let mut chunks: Vec<String> = Vec::new();
        let mut start = 0;

        while start < text.len() {
            if start + Self::CARD_MARKDOWN_MAX_BYTES >= text.len() {
                chunks.push(text[start..].to_string());
                break;
            }

            let end = text.floor_char_boundary(start + Self::CARD_MARKDOWN_MAX_BYTES);
            let search_region = &text[start..end];
            let split_at = search_region
                .rfind('\n')
                .map(|pos| start + pos + 1)
                .unwrap_or(end);

            let split_at = if text.is_char_boundary(split_at) {
                split_at
            } else {
                (start..split_at)
                    .rev()
                    .find(|&i| text.is_char_boundary(i))
                    .unwrap_or(start)
            };

            if split_at <= start {
                let forced = (end..=text.len())
                    .find(|&i| text.is_char_boundary(i))
                    .unwrap_or(text.len());
                chunks.push(text[start..forced].to_string());
                start = forced;
            } else {
                chunks.push(text[start..split_at].to_string());
                start = split_at;
            }
        }

        chunks
    }

    /// Send an interactive card message to Feishu.
    async fn send_interactive_card(
        &self,
        target: &FeishuSendTarget,
        card_content: &str,
    ) -> Result<String, ChannelError> {
        let message_id = self
            .send_raw_message(target, "interactive", card_content)
            .await?;
        if message_id.is_empty() {
            Err(ChannelError::Other(
                "Feishu send response has no message_id".into(),
            ))
        } else {
            Ok(message_id)
        }
    }

    async fn update_interactive_card(
        &self,
        message_id: &str,
        card_content: &str,
    ) -> Result<(), ChannelError> {
        let card: serde_json::Value = serde_json::from_str(card_content)
            .map_err(|error| ChannelError::Other(format!("Invalid card JSON: {error}")))?;
        let response = self
            .send_authenticated("update Feishu card", |client, token| {
                client
                    .patch(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id))
                    .bearer_auth(token)
                    .json(&serde_json::json!({ "card": card }))
            })
            .await?;

        #[derive(Deserialize)]
        struct UpdateResp {
            code: i32,
            msg: String,
        }
        let result: UpdateResp = response.json().await.map_err(|error| {
            ChannelError::Other(format!("Parse update card response error: {error}"))
        })?;
        if result.code != 0 {
            return Err(ChannelError::Other(format!(
                "Update card failed: code={} msg={}",
                result.code, result.msg
            )));
        }
        Ok(())
    }
}

#[async_trait]
trait FeishuTurnApi: Send {
    async fn create_card(&mut self, markdown: &str) -> Result<String, ChannelError>;
    async fn update_card(&mut self, message_id: &str, markdown: &str) -> Result<(), ChannelError>;
    async fn cleanup(&mut self);
}

struct FeishuTurnBackend {
    channel: FeishuChannel,
    target: FeishuSendTarget,
    metadata: HashMap<String, String>,
}

#[async_trait]
impl FeishuTurnApi for FeishuTurnBackend {
    async fn create_card(&mut self, markdown: &str) -> Result<String, ChannelError> {
        let card = FeishuChannel::build_card_content(markdown);
        self.channel
            .send_interactive_card(&self.target, &card)
            .await
    }

    async fn update_card(&mut self, message_id: &str, markdown: &str) -> Result<(), ChannelError> {
        let card = FeishuChannel::build_card_content(markdown);
        self.channel
            .update_interactive_card(message_id, &card)
            .await
    }

    async fn cleanup(&mut self) {
        self.channel
            .remove_reaction_from_metadata(&self.metadata)
            .await;
    }
}

struct FeishuTurnSink {
    api: Box<dyn FeishuTurnApi>,
    message_id: Option<String>,
    cleaned_up: bool,
}

impl FeishuTurnSink {
    fn new(api: Box<dyn FeishuTurnApi>) -> Self {
        Self {
            api,
            message_id: None,
            cleaned_up: false,
        }
    }

    async fn cleanup(&mut self) {
        if !self.cleaned_up {
            self.api.cleanup().await;
            self.cleaned_up = true;
        }
    }

    async fn send_chunks(&mut self, chunks: &[String]) -> Result<(), ChannelError> {
        for chunk in chunks {
            self.api.create_card(chunk).await?;
        }
        Ok(())
    }

    async fn finish_snapshot(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
        let markdown = render_feishu_turn(snapshot);
        let chunks = if markdown.is_empty() {
            Vec::new()
        } else {
            FeishuChannel::split_markdown_chunks(&markdown)
        };

        let result = if chunks.is_empty() {
            Ok(())
        } else if let Some(message_id) = self.message_id.clone() {
            match self.api.update_card(&message_id, &chunks[0]).await {
                Ok(()) => self.send_chunks(&chunks[1..]).await,
                Err(error) => {
                    tracing::warn!(error = %error, "Final Feishu card update failed; sending complete fallback");
                    self.send_chunks(&chunks).await
                }
            }
        } else {
            self.send_chunks(&chunks).await
        };
        self.cleanup().await;
        result
    }
}

#[async_trait]
impl TurnSink for FeishuTurnSink {
    async fn update(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
        let markdown = render_feishu_turn(snapshot);
        if markdown.is_empty() {
            return Ok(());
        }
        let live_markdown = truncate_feishu_live_markdown(&markdown);
        if let Some(message_id) = self.message_id.clone() {
            self.api.update_card(&message_id, &live_markdown).await
        } else {
            self.message_id = Some(self.api.create_card(&live_markdown).await?);
            Ok(())
        }
    }

    async fn finish(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
        self.finish_snapshot(snapshot).await
    }

    async fn abort(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
        self.finish_snapshot(snapshot).await
    }
}

fn render_feishu_turn(snapshot: &TurnSnapshot) -> String {
    let mut sections = Vec::new();
    for block in &snapshot.blocks {
        match block {
            TurnBlock::Reasoning { text, .. } if !text.trim().is_empty() => {
                sections.push(format!(
                    "> **思考过程**\n> {}",
                    text.trim().replace('\n', "\n> ")
                ));
            }
            TurnBlock::Assistant { text, .. } if !text.trim().is_empty() => {
                sections.push(text.trim().to_string());
            }
            TurnBlock::Tool {
                name,
                status,
                preview,
                ..
            } => {
                let status = match status {
                    ToolStatus::Running => "执行中",
                    ToolStatus::Completed => "已完成",
                    ToolStatus::Cancelled => "已停止",
                    ToolStatus::Failed => "失败",
                };
                let mut section = format!("> 🔧 **{name}** · {status}");
                if let Some(preview) = preview.as_deref().filter(|value| !value.trim().is_empty()) {
                    section.push_str("\n> ");
                    section.push_str(&preview.trim().replace('\n', "\n> "));
                }
                sections.push(section);
            }
            _ => {}
        }
    }

    if sections.is_empty() {
        if snapshot.status == TurnStatus::Failed {
            sections.push(format!(
                "⚠️ 回复失败：{}",
                snapshot.error.as_deref().unwrap_or("未知错误")
            ));
        } else if snapshot.status == TurnStatus::Cancelled {
            sections.push("已停止生成。".to_string());
        } else {
            return String::new();
        }
    }

    let status = match snapshot.status {
        TurnStatus::Running => Some(match snapshot.phase {
            crate::session::TurnPhase::Queued => "排队中",
            crate::session::TurnPhase::Reasoning => "思考中",
            crate::session::TurnPhase::Responding => "生成中",
            crate::session::TurnPhase::Acting => "调用工具中",
            crate::session::TurnPhase::Finalizing => "收尾中",
        }),
        TurnStatus::Cancelled => Some("已停止"),
        TurnStatus::Failed => Some("失败"),
        TurnStatus::Completed => None,
    };
    if let Some(status) = status {
        sections.push(format!("_{status}_"));
    }
    sections.join("\n\n")
}

fn truncate_feishu_live_markdown(markdown: &str) -> String {
    if markdown.len() <= FeishuChannel::CARD_MARKDOWN_MAX_BYTES {
        return markdown.to_string();
    }
    const SUFFIX: &str = "\n\n_内容仍在生成，已暂时截断…_";
    let limit = FeishuChannel::CARD_MARKDOWN_MAX_BYTES.saturating_sub(SUFFIX.len());
    let boundary = markdown.floor_char_boundary(limit);
    format!("{}{SUFFIX}", &markdown[..boundary])
}

#[async_trait]
impl Channel for FeishuChannel {
    fn name(&self) -> &str {
        "feishu"
    }

    fn is_allowed(&self, sender_id: &str) -> bool {
        self.config.allow_from.iter().any(|allowed| {
            let allowed = allowed.trim();
            allowed == "*" || allowed == sender_id
        })
    }

    /// Handle an inbound message: check for slash commands first, then publish to bus
    async fn handle_and_publish(
        &self,
        bus: &Arc<MessageBus>,
        msg: &crate::bus::InboundMessage,
    ) -> Result<(), ChannelError> {
        // All messages (including slash commands) go through the normal inbound flow
        // SessionManager handles session creation/reuse internally
        bus.publish_inbound(msg.clone()).await?;
        Ok(())
    }

    async fn start(&self, bus: Arc<MessageBus>) -> Result<(), ChannelError> {
        if self.config.app_id.is_empty() || self.config.app_secret.is_empty() {
            return Err(ChannelError::ConfigError(
                "Feishu app_id or app_secret is not configured".to_string(),
            ));
        }

        if self.config.require_mention {
            match self.refresh_bot_open_id().await {
                Ok(open_id) => {
                    tracing::info!(bot_open_id = %open_id, "Resolved Feishu bot identity")
                }
                Err(error) => tracing::warn!(
                    error = %error,
                    "Failed to resolve Feishu bot identity; group messages will be ignored"
                ),
            }
        }

        let mut run_task = self.run_task.lock().await;
        if run_task.as_ref().is_some_and(|task| !task.is_finished()) {
            return Ok(());
        }
        if let Some(finished) = run_task.take() {
            let _ = finished.await;
        }

        *self.running.write().await = true;

        let shutdown = CancellationToken::new();
        *self.shutdown.write().await = Some(shutdown.clone());

        let channel = self.clone();
        let bus = bus.clone();
        *run_task = Some(tokio::spawn(async move {
            let mut retry_attempt = 0_u32;

            loop {
                if !*channel.running.read().await {
                    break;
                }

                match channel.run_ws_loop(bus.clone(), shutdown.clone()).await {
                    Ok(connected_for) => {
                        if connected_for >= STABLE_CONNECTION_WINDOW {
                            retry_attempt = 0;
                        } else {
                            retry_attempt = retry_attempt.saturating_add(1).min(7);
                        }
                        tracing::info!(
                            connected_secs = connected_for.as_secs(),
                            "Feishu WebSocket disconnected"
                        );
                    }
                    Err(e) => {
                        retry_attempt = retry_attempt.saturating_add(1).min(6);
                        tracing::error!(attempt = retry_attempt, error = %e, "Feishu WebSocket error");
                    }
                }

                if !*channel.running.read().await || shutdown.is_cancelled() {
                    break;
                }

                let retry_delay = reconnect_delay(retry_attempt);
                tracing::info!(
                    delay_secs = retry_delay.as_secs(),
                    "Feishu channel reconnect scheduled"
                );
                tokio::select! {
                    _ = tokio::time::sleep(retry_delay) => {}
                    _ = shutdown.cancelled() => break,
                }
            }

            *channel.running.write().await = false;
            tracing::info!("Feishu channel stopped");
        }));

        tracing::info!("Feishu channel started");
        Ok(())
    }

    async fn stop(&self) -> Result<(), ChannelError> {
        *self.running.write().await = false;
        *self.connected.write().await = false;

        if let Some(shutdown) = self.shutdown.write().await.take() {
            shutdown.cancel();
        }

        let task = { self.run_task.lock().await.take() };
        if let Some(mut task) = task {
            match tokio::time::timeout(CHANNEL_STOP_GRACE, &mut task).await {
                Ok(result) => result.map_err(|error| {
                    ChannelError::Other(format!("Feishu channel task failed to join: {error}"))
                })?,
                Err(_) => {
                    tracing::warn!(
                        grace_ms = CHANNEL_STOP_GRACE.as_millis(),
                        "Feishu channel did not stop in time; aborting connection task"
                    );
                    task.abort();
                    let _ = task.await;
                }
            }
        }

        Ok(())
    }

    fn is_running(&self) -> bool {
        self.running.try_read().map(|r| *r).unwrap_or(false)
    }

    fn live_policy(&self) -> LivePolicy {
        if self.config.live_updates {
            LivePolicy::Snapshot {
                min_interval: Duration::from_millis(
                    self.config.live_update_interval_ms.clamp(250, 5_000),
                ),
            }
        } else {
            LivePolicy::FinalOnly
        }
    }

    fn presentation_policy(&self) -> crate::delivery::PresentationPolicy {
        crate::delivery::PresentationPolicy::external(self.config.live_updates)
    }

    async fn open_turn(&self, target: TurnTarget) -> Result<Box<dyn TurnSink>, ChannelError> {
        let send_target =
            FeishuSendTarget::from_message(target.chat_id, target.reply_to, &target.metadata);
        Ok(Box::new(FeishuTurnSink::new(Box::new(FeishuTurnBackend {
            channel: self.clone(),
            target: send_target,
            metadata: target.metadata,
        }))))
    }

    async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> {
        let target = FeishuSendTarget::from_message(
            msg.chat_id.clone(),
            msg.reply_to.clone(),
            &msg.metadata,
        );

        // If no media, send as interactive card with raw markdown
        if msg.media.is_empty() {
            let content = msg.content.trim();

            // Empty content
            if content.is_empty() {
                self.remove_reaction_from_metadata(&msg.metadata).await;
                return Ok(());
            }

            let chunks = Self::split_markdown_chunks(content);
            for chunk in &chunks {
                let card = Self::build_card_content(chunk);
                if let Err(e) = self.send_interactive_card(&target, &card).await {
                    tracing::warn!(error = %e, "Failed to send interactive card, falling back to text");
                    let text_content = serde_json::json!({ "text": chunk }).to_string();
                    let result = self
                        .send_raw_message(&target, "text", &text_content)
                        .await
                        .map(|_| ());
                    self.remove_reaction_from_metadata(&msg.metadata).await;
                    return result;
                }
            }
            self.remove_reaction_from_metadata(&msg.metadata).await;
            return Ok(());
        }

        // Handle multimodal message - send with media
        // Separate images (can embed in post) from files (sent as separate messages)
        let mut image_items = Vec::new();
        let mut file_items = Vec::new();
        for media_item in &msg.media {
            match media_item.media_type.as_str() {
                "image" => image_items.push(media_item),
                "audio" | "video" | "file" => file_items.push(media_item),
                _ => {
                    tracing::warn!(media_type = %media_item.media_type, "Unsupported media type for sending");
                }
            }
        }

        // Upload and send files as separate messages (one per file)
        for item in &file_items {
            match self.upload_file(&item.path).await {
                Ok(file_key) => {
                    let file_msg_type = match item.media_type.as_str() {
                        "audio" => "audio",
                        "video" => "media",
                        _ => "file",
                    };
                    let file_content = serde_json::json!({"file_key": file_key}).to_string();
                    if let Err(e) = self
                        .send_raw_message(&target, file_msg_type, &file_content)
                        .await
                    {
                        tracing::warn!(error = %e, msg_type = file_msg_type, "Failed to send file message");
                    }
                }
                Err(e) => {
                    tracing::warn!(error = %e, path = %item.path, "Failed to upload file");
                }
            }
        }

        // Build content parts for post (text + images)
        let mut content_parts = Vec::new();

        if !msg.content.is_empty() {
            const MAX_TEXT_LENGTH: usize = 60_000;
            let truncated_text = if msg.content.len() > MAX_TEXT_LENGTH {
                format!(
                    "{}...\n\n[Content truncated due to length limit]",
                    &msg.content[..msg.content.ceil_char_boundary(MAX_TEXT_LENGTH)]
                )
            } else {
                msg.content.clone()
            };
            content_parts.push(serde_json::json!({
                "tag": "text",
                "text": truncated_text
            }));
        }

        for item in &image_items {
            match self.upload_image(&item.path).await {
                Ok(image_key) => {
                    content_parts.push(serde_json::json!({
                        "tag": "img",
                        "image_key": image_key
                    }));
                }
                Err(e) => {
                    tracing::warn!(error = %e, path = %item.path, "Failed to upload image");
                }
            }
        }

        // If no post content after processing (no text, no images), skip
        if content_parts.is_empty() {
            self.remove_reaction_from_metadata(&msg.metadata).await;
            return Ok(());
        }

        // Determine message type and build content
        let msg_type = if msg.content.is_empty() && image_items.len() == 1 {
            "image"
        } else {
            "post"
        };

        let content = if msg_type == "image" {
            // Image-only: content is just {"image_key": "..."}
            let image_key = content_parts[0]["image_key"].as_str().unwrap_or("");
            serde_json::json!({"image_key": image_key}).to_string()
        } else {
            // Post with media: zh_cn wrapped post structure
            let post_content: Vec<Vec<serde_json::Value>> =
                content_parts.into_iter().map(|part| vec![part]).collect();
            serde_json::json!({
                "zh_cn": {
                    "title": "",
                    "content": post_content
                }
            })
            .to_string()
        };

        self.send_raw_message(&target, msg_type, &content).await?;

        // Remove pending reaction after successfully sending
        self.remove_reaction_from_metadata(&msg.metadata).await;

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::agent::TurnEvent;
    use crate::delivery::{PresentationPolicy, project_snapshot};

    #[derive(Default)]
    struct MockTurnState {
        created: Vec<String>,
        updated: Vec<(String, String)>,
        cleanups: usize,
        fail_updates: usize,
    }

    struct MockTurnApi {
        state: Arc<Mutex<MockTurnState>>,
    }

    #[async_trait]
    impl FeishuTurnApi for MockTurnApi {
        async fn create_card(&mut self, markdown: &str) -> Result<String, ChannelError> {
            let mut state = self.state.lock().await;
            state.created.push(markdown.to_string());
            Ok(format!("card-{}", state.created.len()))
        }

        async fn update_card(
            &mut self,
            message_id: &str,
            markdown: &str,
        ) -> Result<(), ChannelError> {
            let mut state = self.state.lock().await;
            if state.fail_updates > 0 {
                state.fail_updates -= 1;
                return Err(ChannelError::Other("card can no longer be edited".into()));
            }
            state
                .updated
                .push((message_id.to_string(), markdown.to_string()));
            Ok(())
        }

        async fn cleanup(&mut self) {
            self.state.lock().await.cleanups += 1;
        }
    }

    fn mock_sink(state: Arc<Mutex<MockTurnState>>) -> FeishuTurnSink {
        FeishuTurnSink::new(Box::new(MockTurnApi { state }))
    }

    fn test_channel() -> FeishuChannel {
        FeishuChannel::new(
            FeishuChannelConfig {
                enabled: true,
                app_id: "test-app".to_string(),
                app_secret: "test-secret".to_string(),
                allow_from: vec!["*".to_string()],
                require_mention: true,
                agent: String::new(),
                media_dir: String::new(),
                reaction_emoji: "THUMBSUP".to_string(),
                live_updates: false,
                live_update_interval_ms: 500,
                max_image_bytes: 10 * 1024 * 1024,
                max_file_bytes: 25 * 1024 * 1024,
                media_dir_max_bytes: 512 * 1024 * 1024,
                request_timeout_secs: 30,
            },
            Path::new("/tmp"),
        )
        .expect("test channel should be valid")
    }

    fn inbound_frame(
        message_id: &str,
        sender_id: &str,
        chat_type: &str,
        text: &str,
        mentions: serde_json::Value,
    ) -> PbFrame {
        PbFrame {
            seq_id: 1,
            log_id: 1,
            service: 1,
            method: 1,
            headers: vec![],
            payload: Some(
                serde_json::json!({
                    "header": {
                        "event_type": "im.message.receive_v1",
                        "event_id": format!("event-{message_id}")
                    },
                    "event": {
                        "sender": {
                            "sender_id": { "open_id": sender_id },
                            "sender_type": "user"
                        },
                        "message": {
                            "message_id": message_id,
                            "chat_id": "oc_test",
                            "chat_type": chat_type,
                            "message_type": "text",
                            "content": serde_json::json!({ "text": text }).to_string(),
                            "mentions": mentions
                        }
                    }
                })
                .to_string()
                .into_bytes(),
            ),
        }
    }

    #[tokio::test]
    async fn inbound_admission_enforces_allowlist_before_parsing() {
        let mut channel = test_channel();
        channel.config.allow_from = vec!["ou_allowed".to_string()];
        let frame = inbound_frame(
            "om_denied",
            "ou_denied",
            "p2p",
            "hello",
            serde_json::json!([]),
        );

        assert!(channel.handle_frame(&frame).await.unwrap().is_none());
        assert!(channel.seen_message_ids.read().await.is_empty());
    }

    #[tokio::test]
    async fn group_message_requires_bot_mention_and_removes_only_self_mention() {
        let channel = test_channel();
        *channel.bot_open_id.write().await = Some("ou_bot".to_string());

        let ignored = inbound_frame(
            "om_ignored",
            "ou_user",
            "group",
            "hello",
            serde_json::json!([]),
        );
        assert!(channel.handle_frame(&ignored).await.unwrap().is_none());

        let admitted = inbound_frame(
            "om_admitted",
            "ou_user",
            "group",
            "@_user_1 ask @_user_2",
            serde_json::json!([
                {"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "PicoBot"},
                {"key": "@_user_2", "id": {"open_id": "ou_peer"}, "name": "Alice"}
            ]),
        );
        let parsed = channel
            .handle_frame(&admitted)
            .await
            .unwrap()
            .expect("mentioned group message should be admitted");
        assert_eq!(parsed.content, "ask @Alice");
    }

    #[tokio::test]
    async fn inbound_context_replies_to_current_message_and_preserves_thread_metadata() {
        let mut channel = test_channel();
        channel.config.reaction_emoji.clear();
        *channel.bot_open_id.write().await = Some("ou_bot".to_string());
        let mut frame = inbound_frame(
            "om_current",
            "ou_user",
            "group",
            "@_user_1 hello",
            serde_json::json!([
                {"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "PicoBot"}
            ]),
        );
        let mut payload: serde_json::Value =
            serde_json::from_slice(frame.payload.as_deref().unwrap()).unwrap();
        payload["event"]["message"]["root_id"] = serde_json::json!("om_root");
        payload["event"]["message"]["thread_id"] = serde_json::json!("omt_thread");
        frame.payload = Some(payload.to_string().into_bytes());
        let bus = Arc::new(MessageBus::new(4));

        channel.process_event_frame(&bus, &frame).await;
        let inbound = bus.consume_inbound().await.unwrap();

        assert_eq!(
            inbound.channel_context.reply_to.as_deref(),
            Some("om_current")
        );
        assert_eq!(
            inbound
                .channel_context
                .private
                .get("feishu.thread_id")
                .map(String::as_str),
            Some("omt_thread")
        );
        assert_eq!(
            inbound
                .channel_context
                .private
                .get("feishu.root_id")
                .map(String::as_str),
            Some("om_root")
        );
    }

    #[test]
    fn post_mentions_can_gate_group_messages_when_top_level_mentions_are_absent() {
        let message = LarkMessage {
            message_id: "om_post".to_string(),
            chat_id: "oc_test".to_string(),
            chat_type: "group".to_string(),
            message_type: "post".to_string(),
            content: serde_json::json!({
                "zh_cn": {"content": [[{"tag": "at", "user_id": "ou_bot"}]]}
            })
            .to_string(),
            parent_id: None,
            root_id: None,
            thread_id: None,
            mentions: vec![],
        };

        assert!(message_mentions_bot(&message, Some("ou_bot")));
        assert!(!message_mentions_bot(&message, Some("ou_other")));
    }

    fn fragment_frame(message_id: &str, seq: usize, sum: usize, payload: &[u8]) -> PbFrame {
        PbFrame {
            seq_id: seq as u64,
            log_id: 1,
            service: 1,
            method: 1,
            headers: vec![
                PbHeader {
                    key: "message_id".to_string(),
                    value: message_id.to_string(),
                },
                PbHeader {
                    key: "seq".to_string(),
                    value: seq.to_string(),
                },
                PbHeader {
                    key: "sum".to_string(),
                    value: sum.to_string(),
                },
            ],
            payload: Some(payload.to_vec()),
        }
    }

    #[test]
    fn fragmented_frames_are_reassembled_in_sequence_order() {
        let mut cache = HashMap::new();
        assert!(
            reassemble_data_frame(&fragment_frame("event", 1, 3, b"world"), &mut cache)
                .unwrap()
                .is_none()
        );
        assert!(
            reassemble_data_frame(&fragment_frame("event", 0, 3, b"hello "), &mut cache)
                .unwrap()
                .is_none()
        );
        let complete = reassemble_data_frame(&fragment_frame("event", 2, 3, b"!"), &mut cache)
            .unwrap()
            .expect("last fragment should complete the event");

        assert_eq!(
            complete.payload.as_deref(),
            Some(b"hello world!".as_slice())
        );
        assert!(cache.is_empty());
    }

    #[test]
    fn fragmented_frames_reject_invalid_shape_and_bound_total_size() {
        let mut cache = HashMap::new();
        assert!(
            reassemble_data_frame(
                &fragment_frame("event", MAX_WS_FRAGMENTS, MAX_WS_FRAGMENTS, b"x"),
                &mut cache,
            )
            .is_err()
        );
        assert!(
            reassemble_data_frame(
                &fragment_frame("event", 0, 2, &vec![0; MAX_WS_EVENT_BYTES + 1]),
                &mut cache,
            )
            .is_err()
        );
        assert!(cache.is_empty());
    }

    #[test]
    fn reconnect_backoff_is_bounded_and_resets_after_success() {
        assert_eq!(reconnect_delay(0), Duration::from_secs(1));
        assert_eq!(reconnect_delay(1), Duration::from_secs(1));
        assert_eq!(reconnect_delay(2), Duration::from_secs(2));
        assert_eq!(reconnect_delay(7), Duration::from_secs(64));
        assert_eq!(reconnect_delay(100), Duration::from_secs(64));
    }

    #[test]
    fn send_target_uses_reply_endpoint_and_thread_flag() {
        let metadata = HashMap::from([("feishu.thread_id".to_string(), "omt_thread".to_string())]);
        let target = FeishuSendTarget::from_message(
            "oc_chat".to_string(),
            Some("om_current".to_string()),
            &metadata,
        );
        let (url, body) = target.request("interactive", "{}");

        assert!(url.ends_with("/im/v1/messages/om_current/reply"));
        assert_eq!(body["reply_in_thread"], true);
        assert_eq!(body["msg_type"], "interactive");
        assert!(body.get("receive_id").is_none());
    }

    #[test]
    fn proactive_thread_target_creates_inside_thread() {
        let metadata = HashMap::from([("feishu.thread_id".to_string(), "omt_thread".to_string())]);
        let target = FeishuSendTarget::from_message("oc_chat".to_string(), None, &metadata);
        let (url, body) = target.request("text", "{}");

        assert!(url.ends_with("/im/v1/messages?receive_id_type=thread_id"));
        assert_eq!(body["receive_id"], "omt_thread");
    }

    #[test]
    fn inbound_filename_is_reduced_to_a_safe_basename() {
        assert_eq!(sanitize_filename("../../secret.txt"), "secret.txt");
        assert_eq!(sanitize_filename(".."), "file.bin");
        assert_eq!(sanitize_filename("report:\0.txt"), "report__.txt");
    }

    #[tokio::test]
    async fn media_capacity_and_file_size_are_checked_before_writing_or_uploading() {
        let dir = tempfile::tempdir().unwrap();
        let existing = dir.path().join("existing.bin");
        tokio::fs::write(&existing, vec![0_u8; 8]).await.unwrap();

        ensure_media_capacity(dir.path(), 2, 10).await.unwrap();
        assert!(ensure_media_capacity(dir.path(), 3, 10).await.is_err());
        ensure_local_media_size(existing.to_str().unwrap(), 8, "file")
            .await
            .unwrap();
        assert!(
            ensure_local_media_size(existing.to_str().unwrap(), 7, "file")
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn authenticated_requests_retry_transient_http_failures() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let attempts = Arc::new(AtomicUsize::new(0));
        let route_attempts = attempts.clone();
        let app = axum::Router::new().route(
            "/test",
            axum::routing::get(move || {
                let route_attempts = route_attempts.clone();
                async move {
                    if route_attempts.fetch_add(1, Ordering::SeqCst) < 2 {
                        axum::http::StatusCode::SERVICE_UNAVAILABLE
                    } else {
                        axum::http::StatusCode::OK
                    }
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        let channel = test_channel();
        *channel.tenant_token.write().await = Some(CachedTenantToken {
            value: "test-token".to_string(),
            refresh_after: Instant::now() + Duration::from_secs(60),
        });
        let url = format!("http://{address}/test");

        let response = channel
            .send_authenticated("test request", |client, token| {
                client.get(&url).bearer_auth(token)
            })
            .await
            .unwrap();

        assert_eq!(response.status(), reqwest::StatusCode::OK);
        assert_eq!(attempts.load(Ordering::SeqCst), 3);
        server.abort();
    }

    #[tokio::test]
    async fn turn_sink_creates_once_updates_same_card_and_cleans_up_at_finish() {
        let state = Arc::new(Mutex::new(MockTurnState::default()));
        let mut sink = mock_sink(state.clone());
        let (controller, emitter, _) =
            crate::session::TurnController::start("feishu:chat:dialog", "message");
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 0,
                delta: "hello".into(),
            })
            .unwrap();
        sink.update(&project_snapshot(
            &controller.snapshot(),
            PresentationPolicy::external(true),
        ))
        .await
        .unwrap();
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 0,
                delta: " world".into(),
            })
            .unwrap();
        sink.update(&project_snapshot(
            &controller.snapshot(),
            PresentationPolicy::external(true),
        ))
        .await
        .unwrap();
        controller.complete(None);
        sink.finish(&project_snapshot(
            &controller.snapshot(),
            PresentationPolicy::external(true),
        ))
        .await
        .unwrap();

        let state = state.lock().await;
        assert_eq!(state.created.len(), 1);
        assert_eq!(state.updated.len(), 2);
        assert!(state.updated.iter().all(|(id, _)| id == "card-1"));
        assert!(state.updated.last().unwrap().1.contains("hello world"));
        assert_eq!(state.cleanups, 1);
    }

    #[tokio::test]
    async fn final_update_failure_sends_complete_fallback_and_cleanup_is_idempotent() {
        let state = Arc::new(Mutex::new(MockTurnState::default()));
        let mut sink = mock_sink(state.clone());
        let (controller, emitter, _) =
            crate::session::TurnController::start("feishu:chat:dialog", "message");
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 0,
                delta: "partial".into(),
            })
            .unwrap();
        sink.update(&controller.snapshot()).await.unwrap();
        state.lock().await.fail_updates = 1;
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 0,
                delta: " final".into(),
            })
            .unwrap();
        controller.complete(None);
        sink.finish(&controller.snapshot()).await.unwrap();
        sink.finish(&controller.snapshot()).await.unwrap();

        let state = state.lock().await;
        assert_eq!(state.created.len(), 2);
        assert!(state.created[1].contains("partial final"));
        assert_eq!(state.cleanups, 1);
    }

    #[tokio::test]
    async fn final_only_sink_sends_no_fragments_and_abort_without_text_is_visible() {
        let state = Arc::new(Mutex::new(MockTurnState::default()));
        let mut sink = mock_sink(state.clone());
        let (controller, _emitter, _) =
            crate::session::TurnController::start("feishu:chat:dialog", "message");
        controller.fail("provider unavailable");

        sink.abort(&controller.snapshot()).await.unwrap();

        let state = state.lock().await;
        assert_eq!(state.created.len(), 1);
        assert!(state.created[0].contains("provider unavailable"));
        assert_eq!(state.updated.len(), 0);
        assert_eq!(state.cleanups, 1);
    }

    #[test]
    fn external_projection_removes_reasoning_before_feishu_rendering() {
        let (controller, emitter, _) =
            crate::session::TurnController::start("feishu:chat:dialog", "message");
        emitter
            .emit(TurnEvent::ReasoningDelta {
                iteration: 0,
                delta: "private".into(),
            })
            .unwrap();
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 0,
                delta: "public".into(),
            })
            .unwrap();
        let projected =
            project_snapshot(&controller.snapshot(), PresentationPolicy::external(true));

        let markdown = render_feishu_turn(&projected);
        assert!(markdown.contains("public"));
        assert!(!markdown.contains("private"));
    }

    #[test]
    fn live_card_truncation_preserves_utf8_and_payload_limit() {
        let markdown = "你".repeat(FeishuChannel::CARD_MARKDOWN_MAX_BYTES);
        let truncated = truncate_feishu_live_markdown(&markdown);

        assert!(truncated.len() <= FeishuChannel::CARD_MARKDOWN_MAX_BYTES);
        assert!(truncated.ends_with("_内容仍在生成，已暂时截断…_"));
    }

    #[test]
    fn final_card_chunking_preserves_long_utf8_content() {
        let markdown = "你".repeat(FeishuChannel::CARD_MARKDOWN_MAX_BYTES);
        let chunks = FeishuChannel::split_markdown_chunks(&markdown);

        assert!(chunks.len() > 1);
        assert!(
            chunks
                .iter()
                .all(|chunk| chunk.len() <= FeishuChannel::CARD_MARKDOWN_MAX_BYTES)
        );
        assert_eq!(chunks.concat(), markdown);
    }

    #[test]
    fn live_policy_uses_configured_bounded_interval() {
        let mut channel = test_channel();
        assert_eq!(channel.live_policy(), LivePolicy::FinalOnly);
        channel.config.live_updates = true;
        channel.config.live_update_interval_ms = 10;

        assert_eq!(
            channel.live_policy(),
            LivePolicy::Snapshot {
                min_interval: Duration::from_millis(250)
            }
        );
    }

    #[tokio::test]
    async fn stop_aborts_connection_task_that_ignores_cancellation() {
        let channel = test_channel();
        let shutdown = CancellationToken::new();

        *channel.running.write().await = true;
        *channel.connected.write().await = true;
        *channel.shutdown.write().await = Some(shutdown.clone());
        *channel.run_task.lock().await = Some(tokio::spawn(std::future::pending()));

        tokio::time::timeout(Duration::from_secs(1), channel.stop())
            .await
            .expect("stop must have a hard deadline")
            .expect("stop should succeed after aborting the stuck task");

        assert!(shutdown.is_cancelled());
        assert!(!channel.is_running());
        assert!(!*channel.connected.read().await);
        assert!(channel.run_task.lock().await.is_none());
    }

    #[test]
    fn collect_post_image_keys_finds_nested_images() {
        let content = serde_json::json!({
            "zh_cn": {
                "title": "",
                "content": [[
                    {"tag": "img", "image_key": "img_v3_001"},
                    {"tag": "text", "text": "这是哪里？"},
                    {"tag": "img", "image_key": "img_v3_002"},
                    {"tag": "img", "image_key": "img_v3_001"}
                ]]
            }
        })
        .to_string();

        assert_eq!(
            collect_post_image_keys(&content),
            vec!["img_v3_001".to_string(), "img_v3_002".to_string()]
        );
    }

    #[test]
    fn parse_post_content_preserves_image_positions() {
        let content = serde_json::json!({
            "zh_cn": {
                "title": "",
                "content": [[
                    {"tag": "text", "text": "这是一张图："},
                    {"tag": "img", "image_key": "img_v3_001"},
                    {"tag": "text", "text": "看完继续说"}
                ]]
            }
        })
        .to_string();

        assert_eq!(
            parse_post_content(&content),
            "这是一张图：[image]看完继续说"
        );
    }
}