//! AI 消息 Repo — ai_messages 表(F-260619-03 消息拆分存储)
//!
//! 每条 ChatMessage 一行的独立表,替代 `ai_conversations.messages` 整对话 JSON 列存。
//! 全专用方法(insert_batch / list_by_conversation / delete_range / update_status /
//! update_content_by_tool_call_id),不走 `impl_repo!` 宏 —— 原因:
//! ① 无标准 created_at 通用排序(query 宏硬编码 ORDER BY created_at,本表用 seq);
//! ② insert_batch 批量语义特殊(单事务多条,非逐条 insert);
//! ③ update_status / update_content_by_tool_call_id 是单列定点更新,非全行 update_full。
//!
//! P0 阶段(本文件):仅建 Repo + CRUD 方法,不接 ContextManager 读写路径
//! (切读策略待用户决策,P2)。ContextManager 仍走旧 messages JSON 列双写期未启动。

use std::sync::Arc;

use rusqlite::{params, Connection, Row};
use tokio::sync::Mutex;

use df_types::error::Result;

use crate::db::Database;
use crate::models::AiMessageRecord;

use super::storage_err;

// ============================================================
// from_row 辅助
// ============================================================

fn ai_message_from_row(row: &Row<'_>) -> std::result::Result<AiMessageRecord, rusqlite::Error> {
    Ok(AiMessageRecord {
        id: row.get("id")?,
        conversation_id: row.get("conversation_id")?,
        seq: row.get("seq")?,
        role: row.get("role")?,
        content: row.get("content")?,
        parts: row.get("parts")?,
        tool_call_id: row.get("tool_call_id")?,
        tool_calls: row.get("tool_calls")?,
        model: row.get("model")?,
        status: row.get("status")?,
        reasoning_content: row.get("reasoning_content")?,
        timestamp: row.get("timestamp")?,
        created_at: row.get("created_at")?,
    })
}

// ============================================================
// AiMessageRepo
// ============================================================

/// ai_messages 表 CRUD Repo。
///
/// 方法语义对照消息拆分存储设计 §三/§Phase3:
/// - `insert_batch`:批量插入一批消息(单事务,保证对话内 seq 连续原子写入)
/// - `list_by_conversation`:按对话加载,ORDER BY seq(游标分页基础)
/// - `delete_range`:删除对话内 [min_seq, max_seq) 范围(compress 压缩/编辑重生成用)
/// - `update_status`:单条状态更新(compress → 'compressed' / 编辑 → 'truncated')
/// - `update_content_by_tool_call_id`:按工具调用 ID 定点改 content(replace_tool_result_content)
pub struct AiMessageRepo {
    conn: Arc<Mutex<Connection>>,
}

impl AiMessageRepo {
    pub fn new(db: &Database) -> Self {
        Self { conn: db.conn() }
    }

    /// 批量插入消息(单事务原子提交)。
    ///
    /// INSERT OR IGNORE 幂等:id 主键冲突(同消息重复写)跳过,不报错。
    /// 适配双写场景(save_conversation 重写 dirty 范围时,旧消息先 delete_range 再 insert)。
    pub async fn insert_batch(&self, records: Vec<AiMessageRecord>) -> Result<()> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || -> Result<()> {
            let mut guard = conn.blocking_lock();
            // transaction() 返回 rusqlite::Error,需 map_err 转为 df_types::error::Error
            let tx = guard.transaction().map_err(storage_err)?;
            {
                let mut stmt = tx.prepare(
                    "INSERT OR IGNORE INTO ai_messages
                     (id, conversation_id, seq, role, content, parts, tool_call_id,
                      tool_calls, model, status, reasoning_content, timestamp, created_at)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
                )
                .map_err(storage_err)?;
                for rec in &records {
                    stmt.execute(params![
                        rec.id, rec.conversation_id, rec.seq, rec.role, rec.content,
                        rec.parts, rec.tool_call_id, rec.tool_calls, rec.model, rec.status,
                        rec.reasoning_content, rec.timestamp, rec.created_at
                    ])
                    .map_err(storage_err)?;
                }
            }
            tx.commit().map_err(storage_err)?;
            Ok(())
        })
        .await
        .map_err(storage_err)?
    }

    /// 按对话加载全部消息,ORDER BY seq ASC(对话内时间序)。
    ///
    /// 切读路径(restore_from_messages)用此替代反序列化 messages JSON。
    /// P0 阶段未接切读(P2),此方法先就绪供未来调用 + 测试验证。
    pub async fn list_by_conversation(&self, conversation_id: &str) -> Result<Vec<AiMessageRecord>> {
        let conn = self.conn.clone();
        let conv_id = conversation_id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT id, conversation_id, seq, role, content, parts, tool_call_id,
                            tool_calls, model, status, reasoning_content, timestamp, created_at
                     FROM ai_messages WHERE conversation_id = ?1 ORDER BY seq ASC",
                )
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![conv_id], ai_message_from_row)
                .map_err(storage_err)?;
            let mut results = Vec::new();
            for r in rows {
                results.push(r.map_err(storage_err)?);
            }
            Ok(results)
        })
        .await
        .map_err(storage_err)?
    }

    /// 按对话加载最近 N 条消息(分页懒加载,治长对话渲染卡顿)。
    ///
    /// 从尾部取最近 `limit` 条(ORDER BY seq DESC LIMIT),返回时反转为 ASC 顺序(与
    /// list_by_conversation 一致的 seq 升序)。`before_seq` 可选:指定后只取 seq < before_seq
    /// 的消息(滚顶加载更多时的游标,取下一页更早的历史)。
    ///
    /// 典型用法:
    /// - 首次切入对话:list_recent(conv_id, 50, None) → 最近 50 条
    /// - 滚顶加载更多:list_recent(conv_id, 50, Some(最早已加载消息的 seq)) → 再加载 50 条更早的
    pub async fn list_recent(
        &self,
        conversation_id: &str,
        limit: usize,
        before_seq: Option<i64>,
    ) -> Result<Vec<AiMessageRecord>> {
        let conn = self.conn.clone();
        let conv_id = conversation_id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            // before_seq 有无分两个 SQL(参数化 LIMIT 必须用固定占位,Rust 侧 clamp 防 0)
            let limit = limit.max(1) as i64;
            let sql = if before_seq.is_some() {
                "SELECT id, conversation_id, seq, role, content, parts, tool_call_id,
                        tool_calls, model, status, reasoning_content, timestamp, created_at
                 FROM ai_messages WHERE conversation_id = ?1 AND seq < ?2 ORDER BY seq DESC LIMIT ?3"
            } else {
                "SELECT id, conversation_id, seq, role, content, parts, tool_call_id,
                        tool_calls, model, status, reasoning_content, timestamp, created_at
                 FROM ai_messages WHERE conversation_id = ?1 ORDER BY seq DESC LIMIT ?2"
            };
            let mut stmt = guard.prepare(sql).map_err(storage_err)?;
            let rows = match before_seq {
                Some(seq) => stmt
                    .query_map(params![conv_id, seq, limit], ai_message_from_row)
                    .map_err(storage_err)?,
                None => stmt
                    .query_map(params![conv_id, limit], ai_message_from_row)
                    .map_err(storage_err)?,
            };
            let mut results: Vec<AiMessageRecord> = Vec::new();
            for r in rows {
                results.push(r.map_err(storage_err)?);
            }
            // DESC → 反转为 ASC(与 list_by_conversation 一致顺序)
            results.reverse();
            Ok(results)
        })
        .await
        .map_err(storage_err)?
    }

    /// 删除对话内 seq ∈ [min_seq, max_seq) 的消息(左闭右开)。
    ///
    /// compress 压缩 / 编辑重生成 dirty 范围重写用:delete_range → insert_batch 原子覆盖。
    /// 返回删除条数(0 表示无匹配,合法)。max_seq=None 表示删到末尾。
    pub async fn delete_range(
        &self,
        conversation_id: &str,
        min_seq: i64,
        max_seq: Option<i64>,
    ) -> Result<usize> {
        let conn = self.conn.clone();
        let conv_id = conversation_id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = match max_seq {
                Some(max) => guard.execute(
                    "DELETE FROM ai_messages WHERE conversation_id = ?1 AND seq >= ?2 AND seq < ?3",
                    params![conv_id, min_seq, max],
                ),
                None => guard.execute(
                    "DELETE FROM ai_messages WHERE conversation_id = ?1 AND seq >= ?2",
                    params![conv_id, min_seq],
                ),
            }
            .map_err(storage_err)?;
            Ok(affected)
        })
        .await
        .map_err(storage_err)?
    }

    /// 更新单条消息状态(compress → 'compressed' / 编辑 → 'truncated' / 恢复 → 'active')。
    ///
    /// 返回是否实际更新(0 = 消息不存在/已是该状态)。同步刷新 created_at 不需要
    /// (状态变更不改创建时间)。
    pub async fn update_status(
        &self,
        conversation_id: &str,
        seq: i64,
        status: &str,
    ) -> Result<bool> {
        let conn = self.conn.clone();
        let conv_id = conversation_id.to_owned();
        let status = status.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_messages SET status = ?1 WHERE conversation_id = ?2 AND seq = ?3",
                    params![status, conv_id, seq],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 全量重写对话的消息(单事务 DELETE + INSERT OR IGNORE,原子)。
    ///
    /// F-260619-03 批次 B(save_conversation 写路径切 ai_messages)的核心方法:
    /// 全量重写语义——以入参 records 为该对话的**唯一真相**,先删该 conv 全部旧行再批量插。
    /// 单事务保证「删 + 插」原子,无中间空窗(reload 不会读到半删半插的中间态)。
    ///
    /// 设计权衡(非 dirty 增量,留 P2.1 优化):
    /// - 内存 ContextManager 是运行时真相源,save 是同步点,每轮 save 全量回写简单可靠;
    /// - INSERT OR IGNORE 幂等:records 内 id 重复或与残留行(理论不应有,事务已 DELETE)冲突跳过;
    /// - 入参 records 的 conversation_id 应一致(调用方 save_conversation 保证),本方法不校验。
    ///
    /// 返回 Ok(()) —— 不返回受影响行数(DELETE + INSERT 两条计数语义混乱,调用方只关心成功)。
    pub async fn replace_conversation(
        &self,
        conv_id: &str,
        records: Vec<AiMessageRecord>,
    ) -> Result<()> {
        let conn = self.conn.clone();
        let conv_id = conv_id.to_owned();
        tokio::task::spawn_blocking(move || -> Result<()> {
            let mut guard = conn.blocking_lock();
            let tx = guard.transaction().map_err(storage_err)?;
            {
                // 先删该 conv 全部旧行(全量重写语义)
                tx.execute(
                    "DELETE FROM ai_messages WHERE conversation_id = ?1",
                    params![conv_id],
                )
                .map_err(storage_err)?;
                // 再批量插新行(INSERT OR IGNORE 幂等,id 冲突跳过)
                if !records.is_empty() {
                    let mut stmt = tx.prepare(
                        "INSERT OR IGNORE INTO ai_messages
                         (id, conversation_id, seq, role, content, parts, tool_call_id,
                          tool_calls, model, status, reasoning_content, timestamp, created_at)
                         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
                    )
                    .map_err(storage_err)?;
                    for rec in &records {
                        stmt.execute(params![
                            rec.id, rec.conversation_id, rec.seq, rec.role, rec.content,
                            rec.parts, rec.tool_call_id, rec.tool_calls, rec.model, rec.status,
                            rec.reasoning_content, rec.timestamp, rec.created_at
                        ])
                        .map_err(storage_err)?;
                    }
                }
            }
            tx.commit().map_err(storage_err)?;
            Ok(())
        })
        .await
        .map_err(storage_err)?
    }

    /// 按 tool_call_id 定点更新消息 content(replace_tool_result_content 用)。
    ///
    /// 返回是否实际更新(0 = 该 tool_call_id 在此对话无对应消息)。
    /// 仅改 content,不动其他字段(状态/时间戳不变)。
    pub async fn update_content_by_tool_call_id(
        &self,
        conversation_id: &str,
        tool_call_id: &str,
        content: &str,
    ) -> Result<bool> {
        let conn = self.conn.clone();
        let conv_id = conversation_id.to_owned();
        let tcid = tool_call_id.to_owned();
        let content = content.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_messages SET content = ?1 WHERE conversation_id = ?2 AND tool_call_id = ?3",
                    params![content, conv_id, tcid],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }
}

// ============================================================
// 单元测试 — V21 迁移幂等 + AiMessageRepo CRUD
// ============================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::Database;
    use crate::models::AiMessageRecord;
    use super::super::now_millis_str;

    /// 构造测试消息记录
    fn mk_msg(id: &str, conv: &str, seq: i64, role: &str, content: &str) -> AiMessageRecord {
        AiMessageRecord {
            id: id.into(),
            conversation_id: conv.into(),
            seq,
            role: role.into(),
            content: content.into(),
            parts: None,
            tool_call_id: None,
            tool_calls: None,
            model: None,
            status: "active".into(),
            reasoning_content: None,
            timestamp: None,
            created_at: now_millis_str(),
        }
    }

    /// insert_batch + list_by_conversation round-trip
    #[tokio::test]
    async fn insert_batch_and_list_roundtrip() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);

        let msgs = vec![
            mk_msg("msg_1", "conv_a", 0, "user", "你好"),
            mk_msg("msg_2", "conv_a", 1, "assistant", "你好,有什么可以帮你?"),
            mk_msg("msg_3", "conv_a", 2, "tool", "工具结果"),
        ];
        repo.insert_batch(msgs).await.expect("insert_batch");

        let got = repo.list_by_conversation("conv_a").await.expect("list");
        assert_eq!(got.len(), 3);
        // ORDER BY seq ASC
        assert_eq!(got[0].seq, 0);
        assert_eq!(got[0].content, "你好");
        assert_eq!(got[1].seq, 1);
        assert_eq!(got[2].seq, 2);
        assert_eq!(got[2].role, "tool");

        // 对话隔离:另一对话查不到
        let other = repo.list_by_conversation("conv_b").await.expect("list b");
        assert!(other.is_empty());
    }

    /// insert_batch 幂等:同 id 重复 INSERT OR IGNORE 不报错(双写场景)
    #[tokio::test]
    async fn insert_batch_idempotent() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);

        let m = mk_msg("msg_dup", "conv", 0, "user", "v1");
        repo.insert_batch(vec![m.clone()]).await.expect("insert 1");

        // 重复同 id 不同内容 → IGNORE,保留第一次
        let m2 = mk_msg("msg_dup", "conv", 0, "user", "v2");
        repo.insert_batch(vec![m2]).await.expect("insert 2 dup");

        let got = repo.list_by_conversation("conv").await.expect("list");
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].content, "v1", "IGNORE 应保留首次写入");
    }

    /// delete_range 左闭右开 + None 到末尾
    #[tokio::test]
    async fn delete_range_half_open_and_open_end() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);
        repo.insert_batch(vec![
            mk_msg("m0", "c", 0, "user", "0"),
            mk_msg("m1", "c", 1, "user", "1"),
            mk_msg("m2", "c", 2, "user", "2"),
            mk_msg("m3", "c", 3, "user", "3"),
        ])
        .await
        .expect("insert");

        // 左闭右开 [1, 3):删 seq 1, 2
        let n = repo.delete_range("c", 1, Some(3)).await.expect("delete");
        assert_eq!(n, 2);
        let got = repo.list_by_conversation("c").await.expect("list");
        assert_eq!(got.len(), 2);
        assert_eq!(got[0].seq, 0);
        assert_eq!(got[1].seq, 3);

        // None 到末尾:删 seq >= 3
        let n = repo.delete_range("c", 3, None).await.expect("delete end");
        assert_eq!(n, 1);
        let got = repo.list_by_conversation("c").await.expect("list");
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].seq, 0);
    }

    /// update_status 单条状态更新
    #[tokio::test]
    async fn update_status_single() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);
        repo.insert_batch(vec![mk_msg("m0", "c", 0, "user", "hi")])
            .await
            .expect("insert");

        let ok = repo.update_status("c", 0, "truncated").await.expect("update");
        assert!(ok, "应实际更新");
        let got = repo.list_by_conversation("c").await.expect("list");
        assert_eq!(got[0].status, "truncated");

        // 不存在的 seq → false
        let ok = repo.update_status("c", 99, "active").await.expect("update");
        assert!(!ok, "不存在 seq 应返回 false");
    }

    /// update_content_by_tool_call_id 定点改 content
    #[tokio::test]
    async fn update_content_by_tool_call_id_targeted() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);
        let mut m = mk_msg("m0", "c", 0, "tool", "原始结果");
        m.tool_call_id = Some("call_abc".into());
        repo.insert_batch(vec![m]).await.expect("insert");

        let ok = repo
            .update_content_by_tool_call_id("c", "call_abc", "替换后的结果")
            .await
            .expect("update");
        assert!(ok);
        let got = repo.list_by_conversation("c").await.expect("list");
        assert_eq!(got[0].content, "替换后的结果");

        // 不存在的 tool_call_id → false,不改其他
        let ok = repo
            .update_content_by_tool_call_id("c", "call_xxx", "nope")
            .await
            .expect("update");
        assert!(!ok);
        let got = repo.list_by_conversation("c").await.expect("list");
        assert_eq!(got[0].content, "替换后的结果", "其他消息不应被改");
    }

    // ---------- replace_conversation(F-260619-03 批次 B)----------

    /// replace_conversation 全量重写:删旧 + 插新原子,list 一致
    #[tokio::test]
    async fn replace_conversation_full_rewrite() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);

        // 预置旧数据(将被 replace 删除)
        repo.insert_batch(vec![
            mk_msg("old_0", "conv", 0, "user", "旧0"),
            mk_msg("old_1", "conv", 1, "user", "旧1"),
        ])
        .await
        .expect("insert old");

        // replace 成全新内容(完全不同的 id,旧 id 应被删)
        let now = now_millis_str();
        let records = vec![
            AiMessageRecord {
                id: "new_0".into(),
                conversation_id: "conv".into(),
                seq: 0,
                role: "user".into(),
                content: "新0".into(),
                parts: None,
                tool_call_id: None,
                tool_calls: None,
                model: Some("deepseek-chat".into()),
                status: "active".into(),
                reasoning_content: Some("思考".into()),
                timestamp: Some(1700000000000),
                created_at: now.clone(),
            },
            AiMessageRecord {
                id: "new_1".into(),
                conversation_id: "conv".into(),
                seq: 1,
                role: "assistant".into(),
                content: "新1".into(),
                parts: None,
                tool_call_id: None,
                tool_calls: Some(r#"[{"id":"c1","type":"function","function":{"name":"f","arguments":"{}"}}]"#.into()),
                model: None,
                status: "active".into(),
                reasoning_content: None,
                timestamp: None,
                created_at: now,
            },
        ];
        repo.replace_conversation("conv", records).await.expect("replace");

        let got = repo.list_by_conversation("conv").await.expect("list");
        assert_eq!(got.len(), 2, "旧 2 条应被删,新 2 条入");
        assert_eq!(got[0].id, "new_0");
        assert_eq!(got[0].content, "新0");
        assert_eq!(got[0].model.as_deref(), Some("deepseek-chat"));
        assert_eq!(got[0].reasoning_content.as_deref(), Some("思考"));
        assert_eq!(got[1].id, "new_1");
        assert!(got[1].tool_calls.is_some());

        // 确认旧 id 已删
        let ids: Vec<&str> = got.iter().map(|r| r.id.as_str()).collect();
        assert!(!ids.contains(&"old_0") && !ids.contains(&"old_1"));
    }

    /// replace_conversation 空列表 = 清空该对话消息(单事务 DELETE 不插)
    #[tokio::test]
    async fn replace_conversation_empty_clears() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);
        repo.insert_batch(vec![
            mk_msg("m0", "c", 0, "user", "0"),
            mk_msg("m1", "c", 1, "user", "1"),
        ])
        .await
        .expect("insert");

        repo.replace_conversation("c", vec![]).await.expect("replace empty");
        let got = repo.list_by_conversation("c").await.expect("list");
        assert!(got.is_empty(), "空 records 应清空对话");
    }

    /// replace_conversation 对话隔离:两 conv 并发写不串
    #[tokio::test]
    async fn replace_conversation_isolation_between_convs() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);

        // 预置 conv_a 与 conv_b
        repo.insert_batch(vec![mk_msg("a0", "conv_a", 0, "user", "a0")])
            .await
            .expect("insert a");
        repo.insert_batch(vec![mk_msg("b0", "conv_b", 0, "user", "b0")])
            .await
            .expect("insert b");

        // 只 replace conv_a,conv_b 不应被影响
        let now = now_millis_str();
        repo.replace_conversation(
            "conv_a",
            vec![AiMessageRecord {
                id: "a_new".into(),
                conversation_id: "conv_a".into(),
                seq: 0,
                role: "user".into(),
                content: "a_new".into(),
                parts: None,
                tool_call_id: None,
                tool_calls: None,
                model: None,
                status: "active".into(),
                reasoning_content: None,
                timestamp: None,
                created_at: now,
            }],
        )
        .await
        .expect("replace a");

        let got_a = repo.list_by_conversation("conv_a").await.expect("list a");
        assert_eq!(got_a.len(), 1);
        assert_eq!(got_a[0].id, "a_new", "conv_a 应被全量重写");

        let got_b = repo.list_by_conversation("conv_b").await.expect("list b");
        assert_eq!(got_b.len(), 1, "conv_b 不应被影响");
        assert_eq!(got_b[0].id, "b0");
        assert_eq!(got_b[0].content, "b0");
    }

    /// replace_conversation 幂等性:INSERT OR IGNORE 在同事务内 DELETE 后无残留,
    /// 重复 replace 同 id 不报错(DELETE 后表内该 conv 空,INSERT 必然成功)
    #[tokio::test]
    async fn replace_conversation_idempotent_rerun() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiMessageRepo::new(&db);
        let now = now_millis_str();
        let rec = || AiMessageRecord {
            id: "m0".into(),
            conversation_id: "c".into(),
            seq: 0,
            role: "user".into(),
            content: "0".into(),
            parts: None,
            tool_call_id: None,
            tool_calls: None,
            model: None,
            status: "active".into(),
            reasoning_content: None,
            timestamp: None,
            created_at: now.clone(),
        };
        repo.replace_conversation("c", vec![rec()]).await.expect("1st");
        repo.replace_conversation("c", vec![rec()]).await.expect("2nd");
        let got = repo.list_by_conversation("c").await.expect("list");
        assert_eq!(got.len(), 1, "重复 replace 不应叠加");
    }
}