//! Repository 模式的 CRUD 操作
//!
//! 为 7 张表各提供一个 Repo 结构体，统一实现 insert / get_by_id / list_all / query / update_field / delete。

use std::sync::Arc;

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

use df_core::error::{Error, Result};

use crate::db::Database;
use crate::models::{
    AiConversationRecord, AiProviderRecord, AiToolExecutionRecord, BranchRecord, IdeaRecord,
    KnowledgeEventRecord, KnowledgeRecord, NodeExecutionRecord, ProjectRecord, ReleaseRecord,
    TaskRecord, WorkflowRecord,
};

/// rusqlite 错误 → df-core `Error::Storage` 统一包装(R-PD-10 DRY,
/// 替代散落的 `.map_err(storage_err)`)。
fn storage_err<E: std::string::ToString>(e: E) -> Error {
    Error::Storage(e.to_string())
}

/// 规范化路径用于比较:canonicalize 解析绝对规范路径(失败降级),
/// 统一正斜杠 + 小写。与 `df_project::scan::normalize_path` **同算法镜像**(df-storage
/// 不依赖 df-project,故独立实现;改动须同步)。防 `C:\a\b` vs `C:/a/b/` 绕过。
fn normalize_stored_path(p: &str) -> String {
    match std::path::Path::new(p).canonicalize() {
        Ok(abs) => abs.to_string_lossy().replace('\\', "/").to_lowercase(),
        Err(_) => p
            .trim_end_matches(['\\', '/'])
            .replace('\\', "/")
            .to_lowercase(),
    }
}

// ============================================================
// 知识库 SELECT 列清单(防 COLS 漂移)
// ============================================================

/// `knowledges` 表对应 `KnowledgeRecord` 14 个字段的列名(顺序与结构体一致)。
///
/// 多处 `search`/`search_vector` 内联 COLS 串的 DRY 收口(CR-260615-03):集中一处定义,
/// 配合下方 `KNOWLEDGE_COL_COUNT` 断言,任一处加列漏改会被测试 `test_knowledge_cols_matches_record`
/// 立即捕获(`knowledge_from_row` 按 name 取列,SELECT 漏列会运行时 rusqlite 报错,故提前断言)。
const KNOWLEDGE_COLS: &str = "id,kind,title,content,tags,status,confidence,reuse_count,verified,source_project,source_ref,reasoning,created_at,updated_at";
/// `KnowledgeRecord` 字段数(与上面列清单的逗号分隔项数一致,被测试断言)。
/// 仅测试期消费(列漂移断言);保留为非 `cfg(test)` 以便测试外的阅读者一眼看到字段数。
#[cfg_attr(not(test), allow(dead_code))]
const KNOWLEDGE_COL_COUNT: usize = 14;
/// `search_vector` 用的列清单:KNOWLEDGE_COLS + embedding(余弦计算用,不入 KnowledgeRecord)。
const KNOWLEDGE_COLS_WITH_EMBEDDING: &str = concat!(
    "id,kind,title,content,tags,status,confidence,reuse_count,verified,",
    "source_project,source_ref,reasoning,created_at,updated_at,embedding"
);

// ============================================================
// 辅助宏 — 消除 6 个 Repo 的重复样板
// ============================================================

/// 一次性为 7 张表生成完整的 Repo 结构体及其 CRUD 实现。
///
/// 用法: `impl_repo!(RepoName, ModelType, "table_name", from_row_body, insert_body, update_body);`
///
/// 约束：`$record` 须实现 `Clone`（`update_full` 内部 clone 后丢给 spawn_blocking）。
macro_rules! impl_repo {
    (
        $(#[$meta:meta])*
        $name:ident, $record:ident, $table:literal,
        from_row => |$row:ident| $from_body:expr,
        insert  => |$conn:ident, $rec:ident| $insert_body:expr,
        update  => |$u_conn:ident, $u_rec:ident| $update_body:expr
    ) => {
        $(#[$meta])*
        pub struct $name {
            conn: Arc<Mutex<Connection>>,
        }

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

            pub async fn insert(&self, record: $record) -> Result<String> {
                let conn = self.conn.clone();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let id = record.id.clone();
                    let $conn = &*guard;
                    let $rec = &record;
                    $insert_body.map_err(storage_err)?;
                    Ok(id)
                })
                .await
                .map_err(storage_err)?
            }

            pub async fn get_by_id(&self, id: &str) -> Result<Option<$record>> {
                let conn = self.conn.clone();
                let id = id.to_owned();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let mut stmt = guard
                        .prepare(&format!("SELECT * FROM {} WHERE id = ?1", $table))
                        .map_err(storage_err)?;
                    let row = stmt
                        .query_row(params![id], |$row| $from_body)
                        .optional()
                        .map_err(storage_err)?;
                    Ok(row)
                })
                .await
                .map_err(storage_err)?
            }

            pub async fn list_all(&self) -> Result<Vec<$record>> {
                let conn = self.conn.clone();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let mut stmt = guard
                        .prepare(&format!("SELECT * FROM {} ORDER BY created_at DESC", $table))
                        .map_err(storage_err)?;
                    let rows = stmt
                        .query_map([], |$row| $from_body)
                        .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)?
            }

            pub async fn query(&self, field: &str, value: &str) -> Result<Vec<$record>> {
                // 白名单校验，防止 SQL 注入
                validate_column_name(field, $table)?;
                let conn = self.conn.clone();
                let sql = format!("SELECT * FROM {} WHERE {} = ?1 ORDER BY created_at DESC", $table, field);
                let value = value.to_owned();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let mut stmt = guard
                        .prepare(&sql)
                        .map_err(storage_err)?;
                    let rows = stmt
                        .query_map(params![value], |$row| $from_body)
                        .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)?
            }

            pub async fn update_field(&self, id: &str, field: &str, value: &str) -> Result<bool> {
                validate_column_name(field, $table)?;
                let conn = self.conn.clone();
                let sql = format!("UPDATE {} SET {} = ?1, updated_at = ?2 WHERE id = ?3", $table, field);
                let id = id.to_owned();
                let value = value.to_owned();
                let now = now_millis_str();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let affected = guard
                        .execute(&sql, params![value, now, id])
                        .map_err(storage_err)?;
                    Ok(affected > 0)
                })
                .await
                .map_err(storage_err)?
            }

            /// 整体更新记录（全可变字段，保留 id 与 created_at），单次原子写
            pub async fn update_full(&self, record: &$record) -> Result<bool> {
                let conn = self.conn.clone();
                let rec = record.clone();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let $u_conn = &*guard;
                    let $u_rec = &rec;
                    let affected =
                        $update_body.map_err(storage_err)?;
                    Ok(affected > 0)
                })
                .await
                .map_err(storage_err)?
            }

            pub async fn delete(&self, id: &str) -> Result<bool> {
                let conn = self.conn.clone();
                let id = id.to_owned();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let affected = guard
                        .execute(&format!("DELETE FROM {} WHERE id = ?1", $table), params![id])
                        .map_err(storage_err)?;
                    Ok(affected > 0)
                })
                .await
                .map_err(storage_err)?
            }
        }
    };
}

// ============================================================
// 通用键值设置 — app_settings 表(KV,不适合 impl_repo! 宏)
// ============================================================

/// 通用应用设置 KV Repo(表 `app_settings`)
///
/// 前端 localStorage 迁移目标:存主题/语言/AI 偏好/连接配置等,`value` 为 JSON 字符串。
/// KV 无固定 schema 记录结构,故不走 `impl_repo!` 宏,手写更直白。
pub struct SettingsRepo {
    conn: Arc<Mutex<Connection>>,
}

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

    /// 取单个 key 的值(JSON 字符串),不存在返回 None
    pub async fn get(&self, key: &str) -> Result<Option<String>> {
        let conn = self.conn.clone();
        let key = key.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let v = guard
                .query_row(
                    "SELECT value FROM app_settings WHERE key = ?1",
                    params![key],
                    |row| row.get::<_, String>(0),
                )
                .optional()
                .map_err(storage_err)?;
            Ok(v)
        })
        .await
        .map_err(storage_err)?
    }

    /// 写 key/value(`INSERT OR REPLACE`),刷新 updated_at
    pub async fn set(&self, key: &str, value: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let key = key.to_owned();
        let value = value.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "INSERT OR REPLACE INTO app_settings (key, value, updated_at) VALUES (?1, ?2, ?3)",
                    params![key, value, now],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 取全部 key/value
    pub async fn get_all(&self) -> Result<Vec<(String, String)>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT key, value FROM app_settings")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))
                .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)?
    }

    /// 删除 key,返回是否实际删除
    pub async fn delete(&self, key: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let key = key.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute("DELETE FROM app_settings WHERE key = ?1", params![key])
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }
}

// ============================================================
// 列名白名单 — 防止 SQL 注入
// ============================================================

/// 按表返回允许用于 query / update_field 的列名。
///
/// 白名单两重作用:① 防注入(列名参数化前校验);② 按表隔离(update_task 误传
/// projects 的 "name" 会在校验阶段拒绝,而非靠 SQLite "no such column" 兜底报错)。
/// 专用更新路径的列不列入:knowledges.embedding(set_embedding)、projects.deleted_at
/// (soft_delete/restore)、tasks.deleted_at(soft_delete/restore)。未登记的表返回 None
/// → 放行(仅靠参数化防注入,向后兼容)。
pub fn allowed_columns_for(table: &str) -> Option<&'static [&'static str]> {
    Some(match table {
        "ideas" => &[
            "id", "title", "description", "status", "priority", "score", "tags", "source",
            "promoted_to", "ai_analysis", "scores", "created_at", "updated_at",
        ],
        "projects" => &[
            "id", "name", "description", "status", "idea_id", "path", "stack", "created_at",
            "updated_at",
        ],
        "tasks" => &[
            // id/created_at 不列入:主键与创建时间不可通过通用 update_field 改写
            // (防篡改主键/伪造创建时间/跨项目移动)。
            // status 不列入(D-260616-04 status 收口):所有 status 改动走
            // advance_status_atomic(CAS SQL `WHERE id AND status=expected`,唯一 status 写入路径),
            // 不经通用 update_field。防 update_task IPC / AI 工具旁路改 status 绕过状态机
            // (can_transition 校验)与 review_rounds 累加。后人勿把 status 补回白名单。
            // review_rounds 不列入:它仅 advance_status_atomic 退回转换
            // (in_review→in_progress / testing→in_review)时原子 +1(收口:仅该专用路径可改,
            // update_task/update_field 白名单均不含)。后人勿把 review_rounds 补进白名单,
            // 否则破坏「review_rounds 唯一写入路径」收口、引入旁路写导致计数错乱。
            "project_id", "title", "description", "priority", "branch_name",
            "assignee", "workflow_def_id", "base_branch",
            // output_json:ai_execute 写产出 / ai_self_review 读产出自审 / human_review 展示对象
            // (决策 a:task 中心,产出跟 task 走)。非状态机收口字段,合法可写。
            "output_json",
            "updated_at",
            // TODO(B-260616-16): project_id 跨表存在性校验待 commands/task.rs 层补。
            //   通用 CRUD 层(db repo)只懂表/列语义,不持有跨表业务约束(查 projects 表存在性)。
            //   project_id 当前可在白名单内改写,合法目标存在性由上层命令层校验。
        ],
        "releases" => &[
            "id", "project_id", "version", "status", "task_ids", "changelog", "created_at",
            "released_at",
        ],
        "branches" => &[
            "id", "project_id", "task_id", "name", "base", "status", "created_at", "updated_at",
            "merged_at",
        ],
        "workflow_executions" => &[
            "id", "name", "dag_json", "status", "triggered_by", "project_id", "task_id",
            "created_at", "completed_at",
        ],
        "node_executions" => &[
            "id", "workflow_id", "node_id", "node_type", "status", "input_json", "output_json",
            "error_message", "started_at", "completed_at",
        ],
        "ai_providers" => &[
            "id", "name", "provider_type", "api_key", "base_url", "default_model", "models",
            "is_default", "config", "created_at", "updated_at",
        ],
        "ai_conversations" => &[
            "id", "title", "messages", "provider_id", "model", "models", "archived",
            "prompt_tokens", "completion_tokens", "created_at", "updated_at",
        ],
        "ai_tool_executions" => &[
            "id", "conversation_id", "tool_call_id", "tool_name", "arguments", "result",
            "status", "risk_level", "requested_at", "executed_at", "decided_by",
        ],
        "knowledges" => &[
            "id", "kind", "title", "content", "tags", "status", "confidence", "reuse_count",
            "verified", "source_project", "source_ref", "reasoning", "created_at", "updated_at",
        ],
        "knowledge_events" => &[
            "id", "knowledge_id", "event_type", "source_ref", "context_json", "timestamp",
        ],
        _ => return None,
    })
}

fn validate_column_name(field: &str, table: &str) -> Result<()> {
    match allowed_columns_for(table) {
        Some(cols) if cols.contains(&field) => Ok(()),
        Some(_) => Err(Error::Storage(format!("表 {} 不允许的字段名: {}", table, field))),
        // 未登记表保守拒绝(FR-S6: 原放行 Ok，若未来未登记表走通用查询路径，列名直进字符串拼接即 SQL 注入；与 is_allowed_column 的 None=>false 对齐)
        None => Err(Error::Storage(format!("表 {} 未登记列白名单，拒绝防注入", table))),
    }
}

/// 是否允许更新某表的某列(按表隔离白名单)。未登记表返回 false(保守拒绝,防误传)。
///
/// 供 AI 工具层等外部调用方复用 CRUD 白名单,避免双份字段列表不同步。
pub fn is_allowed_column(table: &str, field: &str) -> bool {
    match allowed_columns_for(table) {
        Some(cols) => cols.contains(&field),
        None => false,
    }
}

// ============================================================
// 时间工具
// ============================================================

fn now_millis_str() -> String {
    // 转发 df_core::now_millis(DRY:时间获取统一入口在 df-core,避免本 crate 与 commands 层各写一份 SystemTime 调用)
    df_core::now_millis().to_string()
}

// ============================================================
// 向量工具 — embedding BLOB 序列化 + 余弦相似度
// ============================================================

/// Vec<f32> → 小端字节 BLOB
fn f32s_to_blob(v: &[f32]) -> Vec<u8> {
    v.iter().flat_map(|f| f.to_le_bytes()).collect()
}

/// BLOB → Vec<f32>(长度非 4 倍数时截断尾部残字节)
fn blob_to_f32s(blob: &[u8]) -> Vec<f32> {
    blob.chunks_exact(4)
        .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
        .collect()
}

/// 余弦相似度(确定性数学,与任何实现结果一致)
fn cosine_similarity(a: &[f32], b: &[f32]) -> f32 {
    let dot: f32 = a.iter().zip(b).map(|(x, y)| x * y).sum();
    let norm_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
    let norm_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
    dot / (norm_a * norm_b + 1e-8)
}

// ============================================================
// from_row 辅助函数
// ============================================================

fn idea_from_row(row: &Row<'_>) -> std::result::Result<IdeaRecord, rusqlite::Error> {
    Ok(IdeaRecord {
        id: row.get("id")?,
        title: row.get("title")?,
        description: row.get("description")?,
        status: row.get("status")?,
        priority: row.get("priority")?,
        score: row.get("score")?,
        tags: row.get("tags")?,
        source: row.get("source")?,
        promoted_to: row.get("promoted_to")?,
        ai_analysis: row.get("ai_analysis")?,
        scores: row.get("scores")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
    })
}

fn project_from_row(row: &Row<'_>) -> std::result::Result<ProjectRecord, rusqlite::Error> {
    Ok(ProjectRecord {
        id: row.get("id")?,
        name: row.get("name")?,
        description: row.get("description")?,
        status: row.get("status")?,
        idea_id: row.get("idea_id")?,
        path: row.get("path")?,
        stack: row.get("stack")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
    })
}

fn task_from_row(row: &Row<'_>) -> std::result::Result<TaskRecord, rusqlite::Error> {
    Ok(TaskRecord {
        id: row.get("id")?,
        project_id: row.get("project_id")?,
        title: row.get("title")?,
        description: row.get("description")?,
        status: row.get("status")?,
        priority: row.get("priority")?,
        branch_name: row.get("branch_name")?,
        assignee: row.get("assignee")?,
        workflow_def_id: row.get("workflow_def_id")?,
        base_branch: row.get("base_branch")?,
        review_rounds: row.get("review_rounds")?,
        output_json: row.get("output_json")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
    })
}

fn release_from_row(row: &Row<'_>) -> std::result::Result<ReleaseRecord, rusqlite::Error> {
    Ok(ReleaseRecord {
        id: row.get("id")?,
        project_id: row.get("project_id")?,
        version: row.get("version")?,
        status: row.get("status")?,
        task_ids: row.get("task_ids")?,
        changelog: row.get("changelog")?,
        created_at: row.get("created_at")?,
        released_at: row.get("released_at")?,
    })
}

fn workflow_from_row(row: &Row<'_>) -> std::result::Result<WorkflowRecord, rusqlite::Error> {
    Ok(WorkflowRecord {
        id: row.get("id")?,
        name: row.get("name")?,
        dag_json: row.get("dag_json")?,
        status: row.get("status")?,
        triggered_by: row.get("triggered_by")?,
        project_id: row.get("project_id")?,
        task_id: row.get("task_id")?,
        created_at: row.get("created_at")?,
        completed_at: row.get("completed_at")?,
    })
}

fn branch_from_row(row: &Row<'_>) -> std::result::Result<BranchRecord, rusqlite::Error> {
    Ok(BranchRecord {
        id: row.get("id")?,
        project_id: row.get("project_id")?,
        task_id: row.get("task_id")?,
        name: row.get("name")?,
        base: row.get("base")?,
        status: row.get("status")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
        merged_at: row.get("merged_at")?,
    })
}

fn node_execution_from_row(
    row: &Row<'_>,
) -> std::result::Result<NodeExecutionRecord, rusqlite::Error> {
    Ok(NodeExecutionRecord {
        id: row.get("id")?,
        workflow_id: row.get("workflow_id")?,
        node_id: row.get("node_id")?,
        node_type: row.get("node_type")?,
        status: row.get("status")?,
        input_json: row.get("input_json")?,
        output_json: row.get("output_json")?,
        error_message: row.get("error_message")?,
        started_at: row.get("started_at")?,
        completed_at: row.get("completed_at")?,
    })
}

// ============================================================
// 7 个 Repo 实现
// ============================================================

impl_repo!(
    /// 想法表 CRUD
    IdeaRepo,
    IdeaRecord,
    "ideas",
    from_row => |row| idea_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO ideas (id, title, description, status, priority, score, tags, source, promoted_to, ai_analysis, scores, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
            params![
                rec.id, rec.title, rec.description, rec.status, rec.priority,
                rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
                rec.scores, rec.created_at, rec.updated_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, updated_at = ?11 WHERE id = ?12",
            params![
                rec.title, rec.description, rec.status, rec.priority,
                rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
                rec.scores, rec.updated_at, rec.id
            ],
        )
    }
);

impl_repo!(
    /// 项目表 CRUD
    ProjectRepo,
    ProjectRecord,
    "projects",
    from_row => |row| project_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO projects (id, name, description, status, idea_id, path, stack, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                rec.id, rec.name, rec.description, rec.status, rec.idea_id,
                rec.path, rec.stack, rec.created_at, rec.updated_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE projects SET name = ?1, description = ?2, status = ?3, idea_id = ?4, path = ?5, stack = ?6, updated_at = ?7 WHERE id = ?8",
            params![
                rec.name, rec.description, rec.status, rec.idea_id,
                rec.path, rec.stack, rec.updated_at, rec.id
            ],
        )
    }
);

impl ProjectRepo {
    /// 列出未删除项目(deleted_at IS NULL)
    pub async fn list_active(&self) -> Result<Vec<ProjectRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT id, name, description, status, idea_id, path, stack, created_at, updated_at FROM projects WHERE deleted_at IS NULL ORDER BY created_at DESC")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| project_from_row(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)?
    }

    /// 软删:标记 deleted_at(进回收站,可恢复)。仅作用于未删项目,返回是否命中。
    pub async fn soft_delete(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE projects SET deleted_at = ?1, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NULL",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 恢复:清 deleted_at(从回收站还原)
    pub async fn restore(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE projects SET deleted_at = NULL, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NOT NULL",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 查找已绑定该规范化路径的项目(排除 exclude_id 自身)。无冲突返回 None。
    ///
    /// DRY(R-PD-11):统一 project.rs::find_binding_conflict 与 tool_registry.rs::bind_dir_to_project
    /// 的「列项目→排除自身→按规范化路径比较」逻辑。本 crate 不依赖 df-project,故 norm_path 须由
    /// 调用方先经 `df_project::scan::normalize_path`(内含 canonicalize,防 `C:\a\b` vs `C:/a/b/` 绕过)。
    pub async fn find_path_conflict(
        &self,
        norm_path: &str,
        exclude_id: Option<&str>,
    ) -> Result<Option<ProjectRecord>> {
        let projects = self.list_active().await?;
        Ok(projects.into_iter().find(|p| {
            let excluded = exclude_id.is_some_and(|eid| p.id == eid);
            !excluded && p.path.as_ref().is_some_and(|pp| {
                // 路径规范化由调用方完成目标侧;这里为已存项目路径补规范化
                // (历史路径写法可能不规范,统一规范化比较避免误判/漏判)
                normalize_stored_path(pp) == norm_path
            })
        }))
    }

    /// 列出回收站(deleted_at IS NOT NULL),按更新时间(≈删除时间)降序
    pub async fn list_deleted(&self) -> Result<Vec<ProjectRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT id, name, description, status, idea_id, path, stack, created_at, updated_at FROM projects WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| project_from_row(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)?
    }

    /// 彻底删除:事务级联删 branches→releases→tasks→projects(不可恢复)
    ///
    /// SQLite 已开 PRAGMA foreign_keys=ON 但表无 ON DELETE CASCADE,ALTER 改不了 FK 约束,
    /// 故应用层级联:子表先于父表删,单事务保证一致性。
    pub async fn purge_with_descendants(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        tokio::task::spawn_blocking(move || {
            let mut guard = conn.blocking_lock();
            let tx = guard.transaction().map_err(storage_err)?;
            tx.execute("DELETE FROM branches WHERE project_id = ?1", params![id])
                .map_err(storage_err)?;
            tx.execute("DELETE FROM releases WHERE project_id = ?1", params![id])
                .map_err(storage_err)?;
            tx.execute("DELETE FROM tasks WHERE project_id = ?1", params![id])
                .map_err(storage_err)?;
            let affected = tx
                .execute("DELETE FROM projects WHERE id = ?1", params![id])
                .map_err(storage_err)?;
            tx.commit().map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }
}

impl_repo!(
    /// 任务表 CRUD
    TaskRepo,
    TaskRecord,
    "tasks",
    from_row => |row| task_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO tasks (id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
            params![
                rec.id, rec.project_id, rec.title, rec.description, rec.status, rec.priority,
                rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
                rec.review_rounds, rec.output_json, rec.created_at, rec.updated_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, updated_at = ?12 WHERE id = ?13",
            params![
                rec.project_id, rec.title, rec.description, rec.status, rec.priority,
                rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
                rec.review_rounds, rec.output_json, rec.updated_at, rec.id
            ],
        )
    }
);

impl TaskRepo {
    /// 列出未删除任务(deleted_at IS NULL)— 对标 ProjectRepo::list_active
    ///
    /// 显式列出 14 个 TaskRecord 列名(同 ProjectRepo::list_active 写法),
    /// 不 SELECT deleted_at:TaskRecord 不带该字段,取了 from_row 会因未知列报错。
    pub async fn list_active(&self) -> Result<Vec<TaskRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| task_from_row(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)?
    }

    /// 软删:标记 deleted_at(进回收站,可恢复)。仅作用于未删任务,返回是否命中。
    /// 对标 ProjectRepo::soft_delete。
    pub async fn soft_delete(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE tasks SET deleted_at = ?1, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NULL",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 恢复:清 deleted_at(从回收站还原)。仅作用于已删任务,返回是否命中。
    /// 对标 ProjectRepo::restore。
    pub async fn restore(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE tasks SET deleted_at = NULL, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NOT NULL",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 原子推进任务状态(任务推进链 F-260616-02 唯一 status 写入路径)
    ///
    /// 下沉 SQL `WHERE id=? AND status=?expected` 做 CAS(Compare-And-Swap)防 TOCTOU:
    /// 并发推进/旁路修改若已改 status,affected_rows==0,本方法返回 None,调用方
    /// (task_advance_node)据此报「状态已变,推进中止」。`review_rounds` 不进通用
    /// update_field 白名单(收口:仅本方法可改 status 与 review_rounds)。
    ///
    /// - `expected`:调用方读取的当前 status(状态机校验时的 from),CAS 前置。
    /// - `new_status`:目标 status(状态机 can_transition 已校验合法)。
    /// - `bump_rounds`:退回转换(in_review→in_progress / testing→in_review)传 true,
    ///   一并 `review_rounds = review_rounds + 1`(同 UPDATE 原子,避免读改写竞争)。
    ///   前向推进 / 进出 blocked / 进 cancelled 传 false,不动 review_rounds。
    ///
    /// 返回:成功推进返回更新后的 TaskRecord;affected==0(状态已变/任务不存在)返回 None。
    pub async fn advance_status_atomic(
        &self,
        id: &str,
        expected: &str,
        new_status: &str,
        bump_rounds: bool,
    ) -> Result<Option<TaskRecord>> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let expected = expected.to_owned();
        let new_status = new_status.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            // CAS:WHERE id AND status=expected 锁定当前态;affected==0 即并发已改动。
            // deleted_at IS NULL:回收站任务(soft_delete 设了 deleted_at)CAS 必败→affected=0,
            // 返回 None,杜绝回收站任务被推进(D-02 软删语义收口,一处关闭)。
            let sql = if bump_rounds {
                "UPDATE tasks SET status = ?1, review_rounds = review_rounds + 1, updated_at = ?2 \
                 WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL"
            } else {
                "UPDATE tasks SET status = ?1, updated_at = ?2 \
                 WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL"
            };
            let affected = guard
                .execute(sql, params![new_status, now, id, expected])
                .map_err(storage_err)?;
            if affected == 0 {
                return Ok(None);
            }
            // 回读更新后的记录(含新 status / 累加后的 review_rounds / 新 updated_at)。
            let mut stmt = guard
                .prepare("SELECT id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at FROM tasks WHERE id = ?1")
                .map_err(storage_err)?;
            let row = stmt
                .query_row(params![id], |row| task_from_row(row))
                .optional()
                .map_err(storage_err)?;
            Ok(row)
        })
        .await
        .map_err(storage_err)?
    }

    /// 列出回收站(deleted_at IS NOT NULL),按更新时间(≈删除时间)降序。对标 ProjectRepo::list_deleted。
    ///
    /// 注:任务表无专用 list_active_by_project 方法,按项目列活跃任务由 commands/task.rs
    /// 的 list_tasks 用 list_active 后内存过滤 project_id 实现(任务量小,无需 SQL 下推)。
    pub async fn list_deleted(&self) -> Result<Vec<TaskRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at FROM tasks WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| task_from_row(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)?
    }
}

impl_repo!(
    /// 分支表 CRUD
    BranchRepo,
    BranchRecord,
    "branches",
    from_row => |row| branch_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO branches (id, project_id, task_id, name, base, status, created_at, updated_at, merged_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                rec.id, rec.project_id, rec.task_id, rec.name, rec.base,
                rec.status, rec.created_at, rec.updated_at, rec.merged_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE branches SET project_id = ?1, task_id = ?2, name = ?3, base = ?4, status = ?5, updated_at = ?6, merged_at = ?7 WHERE id = ?8",
            params![
                rec.project_id, rec.task_id, rec.name, rec.base,
                rec.status, rec.updated_at, rec.merged_at, rec.id
            ],
        )
    }
);

impl_repo!(
    /// 发布表 CRUD
    ReleaseRepo,
    ReleaseRecord,
    "releases",
    from_row => |row| release_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO releases (id, project_id, version, status, task_ids, changelog, created_at, released_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
            params![
                rec.id, rec.project_id, rec.version, rec.status, rec.task_ids,
                rec.changelog, rec.created_at, rec.released_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE releases SET project_id = ?1, version = ?2, status = ?3, task_ids = ?4, changelog = ?5, released_at = ?6 WHERE id = ?7",
            params![
                rec.project_id, rec.version, rec.status, rec.task_ids,
                rec.changelog, rec.released_at, rec.id
            ],
        )
    }
);

impl_repo!(
    /// 工作流执行表 CRUD
    WorkflowRepo,
    WorkflowRecord,
    "workflow_executions",
    from_row => |row| workflow_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO workflow_executions (id, name, dag_json, status, triggered_by, project_id, task_id, created_at, completed_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                rec.id, rec.name, rec.dag_json, rec.status, rec.triggered_by,
                rec.project_id, rec.task_id, rec.created_at, rec.completed_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE workflow_executions SET name = ?1, dag_json = ?2, status = ?3, triggered_by = ?4, project_id = ?5, task_id = ?6, completed_at = ?7 WHERE id = ?8",
            params![
                rec.name, rec.dag_json, rec.status, rec.triggered_by,
                rec.project_id, rec.task_id, rec.completed_at, rec.id
            ],
        )
    }
);

impl_repo!(
    /// 节点执行表 CRUD
    NodeExecutionRepo,
    NodeExecutionRecord,
    "node_executions",
    from_row => |row| node_execution_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO node_executions (id, workflow_id, node_id, node_type, status, input_json, output_json, error_message, started_at, completed_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
            params![
                rec.id, rec.workflow_id, rec.node_id, rec.node_type, rec.status,
                rec.input_json, rec.output_json, rec.error_message, rec.started_at, rec.completed_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE node_executions SET workflow_id = ?1, node_id = ?2, node_type = ?3, status = ?4, input_json = ?5, output_json = ?6, error_message = ?7, started_at = ?8, completed_at = ?9 WHERE id = ?10",
            params![
                rec.workflow_id, rec.node_id, rec.node_type, rec.status,
                rec.input_json, rec.output_json, rec.error_message, rec.started_at, rec.completed_at, rec.id
            ],
        )
    }
);

// ============================================================
// AI 相关 Repo (V3)
// ============================================================

fn ai_provider_from_row(row: &Row<'_>) -> std::result::Result<AiProviderRecord, rusqlite::Error> {
    // model_configs:DB TEXT 列存 JSON 字符串。读 Option<String> 兼容老库 NULL,
    // 再经 deserialize_model_configs 解析(老字符串数组/新对象数组/空 → Vec<ModelConfig>)。
    // 解析失败不致命:降级为空 Vec(防单行坏数据拖垮 list_all)。
    let model_configs: Vec<df_ai_core::model::ModelConfig> = {
        let raw: Option<String> = row.get("model_configs").ok();
        match raw {
            None => Vec::new(),
            Some(s) => serde_json::from_str::<ModelConfigsWrap>(&format!(
                r#"{{"v":{}}}"#,
                if s.trim().is_empty() {
                    "null".to_string()
                } else if s.trim_start().starts_with('[') || s.trim_start().starts_with('{') {
                    s
                } else {
                    // 非 JSON 字面文本(理论不会出现)→ 包装为 JSON 字符串让 deserialize 兜底
                    serde_json::to_string(&s).unwrap_or_else(|_| "null".into())
                }
            ))
            .map(|w| w.v)
            .unwrap_or_default(),
        }
    };
    Ok(AiProviderRecord {
        id: row.get("id")?,
        name: row.get("name")?,
        provider_type: row.get("provider_type")?,
        api_key: row.get("api_key")?,
        base_url: row.get("base_url")?,
        default_model: row.get("default_model")?,
        models: row.get("models")?,
        model_configs,
        is_default: row.get::<_, i32>("is_default")? != 0,
        config: row.get("config")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
    })
}

/// from_row 内部辅助:复用 deserialize_model_configs 解析 DB TEXT 列 JSON。
/// 包一层 { "v": <原始值> } 把任意 JSON 值送进 deserialize_model_configs。
#[derive(serde::Deserialize)]
struct ModelConfigsWrap {
    #[serde(default, deserialize_with = "df_ai_core::model::deserialize_model_configs")]
    v: Vec<df_ai_core::model::ModelConfig>,
}

fn ai_conversation_from_row(row: &Row<'_>) -> std::result::Result<AiConversationRecord, rusqlite::Error> {
    Ok(AiConversationRecord {
        id: row.get("id")?,
        title: row.get("title")?,
        messages: row.get("messages")?,
        provider_id: row.get("provider_id")?,
        model: row.get("model")?,
        models: row.get("models")?,
        archived: row.get::<_, i32>("archived")? != 0,
        pinned: row.get::<_, i32>("pinned")? != 0,
        prompt_tokens: row.get("prompt_tokens")?,
        completion_tokens: row.get("completion_tokens")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
    })
}

fn ai_tool_execution_from_row(row: &Row<'_>) -> std::result::Result<AiToolExecutionRecord, rusqlite::Error> {
    Ok(AiToolExecutionRecord {
        id: row.get("id")?,
        conversation_id: row.get("conversation_id")?,
        tool_call_id: row.get("tool_call_id")?,
        tool_name: row.get("tool_name")?,
        arguments: row.get("arguments")?,
        result: row.get("result")?,
        status: row.get("status")?,
        risk_level: row.get("risk_level")?,
        requested_at: row.get("requested_at")?,
        executed_at: row.get("executed_at")?,
        decided_by: row.get("decided_by")?,
    })
}

fn knowledge_from_row(row: &Row<'_>) -> std::result::Result<KnowledgeRecord, rusqlite::Error> {
    Ok(KnowledgeRecord {
        id: row.get("id")?,
        kind: row.get("kind")?,
        title: row.get("title")?,
        content: row.get("content")?,
        tags: row.get("tags")?,
        status: row.get("status")?,
        confidence: row.get("confidence")?,
        reuse_count: row.get("reuse_count")?,
        verified: row.get::<_, i32>("verified")? != 0,
        source_project: row.get("source_project")?,
        source_ref: row.get("source_ref")?,
        reasoning: row.get("reasoning")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
    })
}

fn knowledge_event_from_row(row: &Row<'_>) -> std::result::Result<KnowledgeEventRecord, rusqlite::Error> {
    Ok(KnowledgeEventRecord {
        id: row.get("id")?,
        knowledge_id: row.get("knowledge_id")?,
        event_type: row.get("event_type")?,
        source_ref: row.get("source_ref")?,
        context_json: row.get("context_json")?,
        timestamp: row.get("timestamp")?,
    })
}

impl_repo!(
    /// AI 提供商配置表 CRUD
    AiProviderRepo,
    AiProviderRecord,
    "ai_providers",
    from_row => |row| ai_provider_from_row(row),
    insert  => |conn, rec| {
        let is_default = if rec.is_default { 1i32 } else { 0i32 };
        // model_configs:Vec<ModelConfig> → JSON 字符串落 TEXT 列
        let model_configs_json = serde_json::to_string(&rec.model_configs).unwrap_or_else(|_| "[]".into());
        conn.execute(
            "INSERT OR REPLACE INTO ai_providers (id, name, provider_type, api_key, base_url, default_model, models, model_configs, is_default, config, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
            params![
                rec.id, rec.name, rec.provider_type, rec.api_key, rec.base_url,
                rec.default_model, rec.models, model_configs_json, is_default, rec.config, rec.created_at, rec.updated_at
            ],
        )
    },
    update  => |conn, rec| {
        let is_default = if rec.is_default { 1i32 } else { 0i32 };
        let model_configs_json = serde_json::to_string(&rec.model_configs).unwrap_or_else(|_| "[]".into());
        conn.execute(
            "UPDATE ai_providers SET name = ?1, provider_type = ?2, api_key = ?3, base_url = ?4, default_model = ?5, models = ?6, model_configs = ?7, is_default = ?8, config = ?9, updated_at = ?10 WHERE id = ?11",
            params![
                rec.name, rec.provider_type, rec.api_key, rec.base_url,
                rec.default_model, rec.models, model_configs_json, is_default, rec.config, rec.updated_at, rec.id
            ],
        )
    }
);

impl_repo!(
    /// AI 对话历史表 CRUD
    AiConversationRepo,
    AiConversationRecord,
    "ai_conversations",
    from_row => |row| ai_conversation_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
            params![
                rec.id, rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived,
                if rec.pinned { 1i32 } else { 0i32 },
                rec.prompt_tokens, rec.completion_tokens,
                rec.created_at, rec.updated_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE ai_conversations SET title = ?1, messages = ?2, provider_id = ?3, model = ?4, models = ?5, archived = ?6, pinned = ?7, prompt_tokens = ?8, completion_tokens = ?9, updated_at = ?10 WHERE id = ?11",
            params![
                rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived,
                if rec.pinned { 1i32 } else { 0i32 },
                rec.prompt_tokens, rec.completion_tokens,
                rec.updated_at, rec.id
            ],
        )
    }
);

impl_repo!(
    /// AI 工具执行审计表 CRUD
    AiToolExecutionRepo,
    AiToolExecutionRecord,
    "ai_tool_executions",
    from_row => |row| ai_tool_execution_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO ai_tool_executions (id, conversation_id, tool_call_id, tool_name, arguments, result, status, risk_level, requested_at, executed_at, decided_by)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
            params![
                rec.id, rec.conversation_id, rec.tool_call_id, rec.tool_name,
                rec.arguments, rec.result, rec.status, rec.risk_level,
                rec.requested_at, rec.executed_at, rec.decided_by
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE ai_tool_executions SET conversation_id = ?1, tool_call_id = ?2, tool_name = ?3, arguments = ?4, result = ?5, status = ?6, risk_level = ?7, requested_at = ?8, executed_at = ?9, decided_by = ?10 WHERE id = ?11",
            params![
                rec.conversation_id, rec.tool_call_id, rec.tool_name,
                rec.arguments, rec.result, rec.status, rec.risk_level,
                rec.requested_at, rec.executed_at, rec.decided_by, rec.id
            ],
        )
    }
);

// ai_tool_executions 无 created_at 列（用 requested_at/executed_at 计时），
// 通用 query 宏硬编码 ORDER BY created_at 会触发 "no such column" → 调用方 unwrap_or_default 吞错。
// 故为此表提供专用查询，绕过通用 query。详见 ai.rs audit_finalize。
impl AiToolExecutionRepo {
    /// 按 tool_call_id 查最新一条审计记录（审批回填定位用）。
    pub async fn find_by_tool_call_id(
        &self,
        tool_call_id: &str,
    ) -> Result<Option<AiToolExecutionRecord>> {
        let conn = self.conn.clone();
        let tid = tool_call_id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT * FROM ai_tool_executions WHERE tool_call_id = ?1 ORDER BY requested_at DESC LIMIT 1",
                )
                .map_err(storage_err)?;
            let row = stmt
                .query_row(params![tid], |row| ai_tool_execution_from_row(row))
                .optional()
                .map_err(storage_err)?;
            Ok(row)
        })
        .await
        .map_err(storage_err)?
    }

    /// 列出所有 status=pending 的审计行(启动重建 pending_approvals 用)
    ///
    /// 专用 SELECT(非 query 宏——后者硬编码 ORDER BY created_at,而本表无该列)。
    pub async fn list_pending(&self) -> Result<Vec<AiToolExecutionRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT * FROM ai_tool_executions WHERE status = 'pending' ORDER BY requested_at ASC")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| ai_tool_execution_from_row(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)?
    }

    /// 审批历史面板分页查询:按 requested_at 倒序(最新在前),limit 默认 50。
    ///
    /// 与 list_pending 同理走专用 SELECT,绕过通用 query 宏(后者硬编码
    /// ORDER BY created_at,本表无该列)。limit/offset 上限钳制(limit ≤ 200),
    /// 防前端恶意/失误传超大值。
    pub async fn list_recent(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AiToolExecutionRecord>> {
        let conn = self.conn.clone();
        // 钳制 limit 防滥用(默认 50,最大 200)
        let safe_limit = limit.min(200) as i64;
        let safe_offset = offset as i64;
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT * FROM ai_tool_executions ORDER BY requested_at DESC LIMIT ?1 OFFSET ?2",
                )
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![safe_limit, safe_offset], |row| {
                    ai_tool_execution_from_row(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)?
    }
}

impl_repo!(
    /// 知识库表 CRUD
    KnowledgeRepo,
    KnowledgeRecord,
    "knowledges",
    from_row => |row| knowledge_from_row(row),
    insert  => |conn, rec| {
        let verified = if rec.verified { 1i32 } else { 0i32 };
        conn.execute(
            "INSERT INTO knowledges (id, kind, title, content, tags, status, confidence, reuse_count, verified, source_project, source_ref, reasoning, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
            params![
                rec.id, rec.kind, rec.title, rec.content, rec.tags, rec.status, rec.confidence,
                rec.reuse_count, verified, rec.source_project, rec.source_ref, rec.reasoning,
                rec.created_at, rec.updated_at
            ],
        )
    },
    update  => |conn, rec| {
        let verified = if rec.verified { 1i32 } else { 0i32 };
        conn.execute(
            "UPDATE knowledges SET kind = ?1, title = ?2, content = ?3, tags = ?4, status = ?5, confidence = ?6, reuse_count = ?7, verified = ?8, source_project = ?9, source_ref = ?10, reasoning = ?11, updated_at = ?12 WHERE id = ?13",
            params![
                rec.kind, rec.title, rec.content, rec.tags, rec.status, rec.confidence,
                rec.reuse_count, verified, rec.source_project, rec.source_ref, rec.reasoning,
                rec.updated_at, rec.id
            ],
        )
    }
);

// KnowledgeRepo 的整体更新已由 impl_repo! 宏统一生成的 update_full 提供。

impl KnowledgeRepo {
    /// 检索知识: title/content LIKE 匹配,可选 kind 过滤,按 reuse_count 降序,top-N
    ///
    /// 克制检索: top-N≤3(由调用方 limit 控制),精确匹配优先(语义模糊后做)。
    pub async fn search(&self, query: &str, kind: Option<&str>, limit: usize) -> Result<Vec<KnowledgeRecord>> {
        let conn = self.conn.clone();
        let pattern = format!("%{}%", query);
        let kind = kind.map(|s| s.to_owned());
        let limit_i = limit as i64;
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut results = Vec::new();
            if let Some(k) = &kind {
                let mut stmt = guard
                    .prepare(&format!("SELECT {KNOWLEDGE_COLS} FROM knowledges WHERE status = 'published' AND (title LIKE ?1 OR content LIKE ?2) AND kind = ?3 ORDER BY reuse_count DESC LIMIT ?4"))
                    .map_err(storage_err)?;
                let rows = stmt
                    .query_map(params![pattern, pattern, k, limit_i], |row| knowledge_from_row(row))
                    .map_err(storage_err)?;
                for r in rows {
                    results.push(r.map_err(storage_err)?);
                }
            } else {
                let mut stmt = guard
                    .prepare(&format!("SELECT {KNOWLEDGE_COLS} FROM knowledges WHERE status = 'published' AND (title LIKE ?1 OR content LIKE ?2) ORDER BY reuse_count DESC LIMIT ?3"))
                    .map_err(storage_err)?;
                let rows = stmt
                    .query_map(params![pattern, pattern, limit_i], |row| knowledge_from_row(row))
                    .map_err(storage_err)?;
                for r in rows {
                    results.push(r.map_err(storage_err)?);
                }
            }
            Ok(results)
        })
        .await
        .map_err(storage_err)?
    }

    /// 按状态列出(审核收件箱用): 按 confidence 语义排序(high>medium>low),次按 created_at
    ///
    /// 用 CASE WHEN 替代纯 TEXT 排序(字典序 high>low>medium 非预期语义)。
    pub async fn list_by_status(&self, status: &str) -> Result<Vec<KnowledgeRecord>> {
        let conn = self.conn.clone();
        let status = status.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT id,kind,title,content,tags,status,confidence,reuse_count,verified,\
                     source_project,source_ref,reasoning,created_at,updated_at \
                     FROM knowledges WHERE status = ?1
                     ORDER BY CASE confidence
                         WHEN 'high' THEN 3
                         WHEN 'medium' THEN 2
                         WHEN 'low' THEN 1
                         ELSE 0
                     END DESC, created_at DESC",
                )
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![status], |row| knowledge_from_row(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)?
    }

    /// 复用计数 +1(SQL 行级原子操作,并发安全)
    pub async fn increment_reuse_count(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE knowledges SET reuse_count = reuse_count + 1, updated_at = ?1 WHERE id = ?2",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 写入向量嵌入(BLOB = Vec<f32> 小端字节序列化)
    ///
    /// embedding 列不进 KnowledgeRecord(IPC 不需要传向量给前端),专用方法读写。
    pub async fn set_embedding(&self, id: &str, embedding: &[f32]) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let blob = f32s_to_blob(embedding);
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE knowledges SET embedding = ?1 WHERE id = ?2",
                    params![blob, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 向量检索: 加载全部 published 且有 embedding 的记录,纯 Rust 余弦相似度取 top-N
    ///
    /// 返回 (记录, 相似度分数)。数据量 <50k 时暴力遍历 <50ms,够用;
    /// 更大规模再升 sqlite-vec HNSW(结果不变,只提速)。
    ///
    /// 列限定: 显式列出所需列(与 search/list_by_status/top_used 一致),避免 SELECT *
    /// 拉到未知新增列;embedding 单独取(不入 KnowledgeRecord)。
    ///
    /// TODO(性能,低优先): 调用方(hybrid_search→merge_hybrid_results→build_knowledge_context)
    /// 实际只消费 id/kind/title/content/reuse_count;reasoning(AI 生成大文本)、tags、
    /// source_project/source_ref 等元字段未被使用却仍随每行读出。真正省 IO 需返回精简结构
    /// (如 KnowledgeVectorHit { id, kind, title, content, reuse_count })替换返回类型,
    /// 但这会改变 search_vector 签名与 merge_hybrid_results 调用契约——当前保守不动,
    /// 待向量检索量级或 reasoning 文本体积成为瓶颈再单独立项。SELECT 列化本身不省字段,
    /// 仅消除 SELECT * 的隐式依赖与未知列风险。
    pub async fn search_vector(&self, query_vec: &[f32], limit: usize) -> Result<Vec<(KnowledgeRecord, f32)>> {
        let conn = self.conn.clone();
        let query_vec = query_vec.to_vec();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            // 显式列: 14 个 KnowledgeRecord 字段 + embedding(余弦计算用,不入 KnowledgeRecord)
            let mut stmt = guard
                .prepare(&format!(
                    "SELECT {KNOWLEDGE_COLS_WITH_EMBEDDING} FROM knowledges WHERE status = 'published' AND embedding IS NOT NULL"
                ))
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| {
                    let rec = knowledge_from_row(row)?;
                    let blob: Vec<u8> = row.get("embedding")?;
                    Ok((rec, blob))
                })
                .map_err(storage_err)?;

            let mut scored: Vec<(KnowledgeRecord, f32)> = Vec::new();
            for r in rows {
                let (rec, blob) = r.map_err(storage_err)?;
                let emb = blob_to_f32s(&blob);
                // 维度不匹配(换过 embedding 模型的旧向量)跳过
                if emb.len() != query_vec.len() {
                    continue;
                }
                let score = cosine_similarity(&query_vec, &emb);
                scored.push((rec, score));
            }
            scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
            scored.truncate(limit);
            Ok(scored)
        })
        .await
        .map_err(storage_err)?
    }

    /// 列出非归档知识(全部 status != 'archived'),按 confidence 语义排序
    pub async fn list_non_archived(&self) -> Result<Vec<KnowledgeRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT id,kind,title,content,tags,status,confidence,reuse_count,verified,\
                     source_project,source_ref,reasoning,created_at,updated_at \
                     FROM knowledges WHERE status != 'archived'
                     ORDER BY CASE confidence
                         WHEN 'high' THEN 3
                         WHEN 'medium' THEN 2
                         WHEN 'low' THEN 1
                         ELSE 0
                     END DESC, created_at DESC",
                )
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| knowledge_from_row(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)?
    }

    /// 热门知识(已发布,按复用次数降序)
    pub async fn top_used(&self, limit: usize) -> Result<Vec<KnowledgeRecord>> {
        let conn = self.conn.clone();
        let limit_i = limit as i64;
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT id,kind,title,content,tags,status,confidence,reuse_count,verified,source_project,source_ref,reasoning,created_at,updated_at FROM knowledges WHERE status = 'published' ORDER BY reuse_count DESC LIMIT ?1")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![limit_i], |row| knowledge_from_row(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)?
    }
}

impl_repo!(
    /// 知识生命线事件表 CRUD(追加型审计表)
    KnowledgeEventsRepo,
    KnowledgeEventRecord,
    "knowledge_events",
    from_row => |row| knowledge_event_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO knowledge_events (id, knowledge_id, event_type, source_ref, context_json, timestamp)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![rec.id, rec.knowledge_id, rec.event_type, rec.source_ref, rec.context_json, rec.timestamp],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE knowledge_events SET knowledge_id = ?1, event_type = ?2, source_ref = ?3, context_json = ?4, timestamp = ?5 WHERE id = ?6",
            params![rec.knowledge_id, rec.event_type, rec.source_ref, rec.context_json, rec.timestamp, rec.id],
        )
    }
);

impl KnowledgeEventsRepo {
    /// 按知识 ID 查询全部事件(时间正序,构建生命线视图用)
    pub async fn list_by_knowledge(&self, knowledge_id: &str) -> Result<Vec<KnowledgeEventRecord>> {
        let conn = self.conn.clone();
        let knowledge_id = knowledge_id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT id,knowledge_id,event_type,source_ref,context_json,timestamp FROM knowledge_events WHERE knowledge_id = ?1 ORDER BY timestamp ASC")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![knowledge_id], |row| knowledge_event_from_row(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)?
    }

    /// 按知识 ID + event_type 查询最近 N 条(如引用记录翻页: type=referenced)
    pub async fn list_by_knowledge_type(
        &self,
        knowledge_id: &str,
        event_type: &str,
        limit: usize,
    ) -> Result<Vec<KnowledgeEventRecord>> {
        let conn = self.conn.clone();
        let knowledge_id = knowledge_id.to_owned();
        let event_type = event_type.to_owned();
        let limit_i = limit as i64;
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT id,knowledge_id,event_type,source_ref,context_json,timestamp FROM knowledge_events WHERE knowledge_id = ?1 AND event_type = ?2 ORDER BY timestamp DESC LIMIT ?3")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![knowledge_id, event_type, limit_i], |row| knowledge_event_from_row(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)?
    }
}

// AiConversationRepo 的整体更新已由 impl_repo! 宏统一生成的 update_full 提供。

impl AiConversationRepo {
    /// 清空对话消息内容(保留 conversation 记录本身,只清 messages JSON + 清零 token 计数)
    ///
    /// "清空对话"语义:对话壳保留(侧栏仍可见,可继续在该对话内聊),仅清空历史消息。
    /// messages 是 ai_conversations 表内的 JSON 列而非独立行,故"删 messages"= 置空该列。
    pub async fn clear_messages(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_conversations SET messages = '[]', prompt_tokens = 0, completion_tokens = 0, updated_at = ?1 WHERE id = ?2",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 设置归档标记(仅改 archived,不动 updated_at)
    ///
    /// 区别于 update_field(后者强制 SET updated_at=now,会把归档/取消归档误判为内容更新,
    /// 导致侧栏相对时间跳变为"刚刚")。归档是纯元数据标记,应保持时间不变。
    pub async fn set_archived(&self, id: &str, archived: bool) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_conversations SET archived = ?1 WHERE id = ?2",
                    params![if archived { 1 } else { 0 }, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 设置置顶标记(仅改 pinned,不动 updated_at) — UX-17
    ///
    /// 同 set_archived:置顶是纯元数据标记,不应改变相对时间。前端排序读 pinned DESC, updated_at DESC。
    pub async fn set_pinned(&self, id: &str, pinned: bool) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_conversations SET pinned = ?1 WHERE id = ?2",
                    params![if pinned { 1 } else { 0 }, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }
}

// ============================================================
// 单元测试 — 向量纯函数 + KnowledgeRepo 内存 DB
// ============================================================

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

    // ---------- COLS 漂移防护(CR-260615-03) ----------

    /// model_configs DB roundtrip + 老库空兼容(F-01 阶段1)
    #[tokio::test]
    async fn ai_provider_model_configs_roundtrip_and_old_db_compat() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiProviderRepo::new(&db);
        use crate::models::AiProviderRecord;
        use df_ai_core::model::{Capability, IntelligenceTier, Modality, ModelConfig};

        // 新格式:带多模型 + 多维度配置
        let configs = vec![
            ModelConfig::with_defaults("glm-4-flash"),
            ModelConfig {
                model_id: "glm-4v".into(),
                modalities: vec![Modality::Text, Modality::Vision],
                capabilities: vec![Capability::ToolUse],
                intelligence: IntelligenceTier::Plus,
                weight: 70,
                context_window: 128_000,
                ..ModelConfig::with_defaults("glm-4v")
            },
        ];
        let rec = AiProviderRecord {
            id: "p1".into(),
            name: "测试".into(),
            provider_type: "openai_compat".into(),
            api_key: String::new(),
            base_url: "https://x".into(),
            default_model: "glm-4-flash".into(),
            models: None,
            model_configs: configs.clone(),
            is_default: false,
            config: None,
            created_at: "0".into(),
            updated_at: "0".into(),
        };
        repo.insert(rec).await.expect("insert");

        let got = repo.get_by_id("p1").await.expect("get").expect("row exists");
        assert_eq!(got.model_configs.len(), 2);
        assert_eq!(got.model_configs[0].model_id, "glm-4-flash");
        assert_eq!(got.model_configs[1].model_id, "glm-4v");
        assert_eq!(got.model_configs[1].intelligence, IntelligenceTier::Plus);
        assert_eq!(got.model_configs[1].context_window, 128_000);

        // 老库空兼容:直接写 model_configs=NULL 的行(模拟 V18 之前的老库行)
        // 然后 from_row 应得空 Vec
        {
            let conn = db.conn();
            let g = conn.lock().await;
            g.execute(
                "INSERT OR REPLACE INTO ai_providers \
                 (id,name,provider_type,api_key,base_url,default_model,models,model_configs,is_default,config,created_at,updated_at) \
                 VALUES ('old','','openai_compat','','','','{}',NULL,0,NULL,'0','0')",
                [],
            )
            .expect("raw insert old row");
        }
        let old = repo.get_by_id("old").await.expect("get").expect("old row");
        assert!(old.model_configs.is_empty(), "NULL 列应得空 Vec");

        // 老格式字符串数组 JSON(向后兼容 deserialize_model_configs)
        {
            let conn = db.conn();
            let g = conn.lock().await;
            g.execute(
                "INSERT OR REPLACE INTO ai_providers \
                 (id,name,provider_type,api_key,base_url,default_model,models,model_configs,is_default,config,created_at,updated_at) \
                 VALUES ('legacy','','openai_compat','','','','{}','[\"glm-4-flash\",\"glm-4v\"]',0,NULL,'0','0')",
                [],
            )
            .expect("raw insert legacy row");
        }
        let legacy = repo.get_by_id("legacy").await.expect("get").expect("legacy row");
        assert_eq!(legacy.model_configs.len(), 2, "老字符串数组应转 2 个默认 ModelConfig");
        assert_eq!(legacy.model_configs[0].model_id, "glm-4-flash");
    }

    /// KNOWLEDGE_COLS 列数须等于 KNOWLEDGE_COL_COUNT(任一处漂移:加列漏改 / 串错位 → 立即失败)。
    /// `knowledge_from_row` 按 name 取列,SELECT 漏列会在运行时被 rusqlite 报错;此断言提前到测试期捕获。
    #[test]
    fn test_knowledge_cols_matches_record() {
        let count = KNOWLEDGE_COLS.split(',').count();
        assert_eq!(
            count, KNOWLEDGE_COL_COUNT,
            "KNOWLEDGE_COLS({count}列) ≠ KNOWLEDGE_COL_COUNT({KNOWLEDGE_COL_COUNT}); \
             修改一处须同步另一处"
        );
        // search_vector 多一列 embedding
        let count_with_emb = KNOWLEDGE_COLS_WITH_EMBEDDING.split(',').count();
        assert_eq!(
            count_with_emb,
            KNOWLEDGE_COL_COUNT + 1,
            "KNOWLEDGE_COLS_WITH_EMBEDDING({count_with_emb}列) ≠ KNOWLEDGE_COL_COUNT+1({}); \
             embedding 列应单独追加",
            KNOWLEDGE_COL_COUNT + 1
        );
        // 每个列名须能被 split 出来(防末尾多逗号 / 空段)
        for col in KNOWLEDGE_COLS.split(',') {
            assert!(!col.is_empty(), "KNOWLEDGE_COLS 含空列名段");
        }
    }

    // ---------- 向量纯函数 ----------

    #[test]
    fn f32s_blob_roundtrip_nonempty() {
        let v = vec![0.0, 1.5, -2.25, 3.14159, -0.0001];
        let blob = f32s_to_blob(&v);
        assert_eq!(blob.len(), v.len() * 4);
        let back = blob_to_f32s(&blob);
        assert_eq!(back, v);
    }

    #[test]
    fn f32s_blob_roundtrip_empty() {
        let v: Vec<f32> = vec![];
        let blob = f32s_to_blob(&v);
        assert!(blob.is_empty());
        assert!(blob_to_f32s(&blob).is_empty());
    }

    #[test]
    fn blob_to_f32s_drops_trailing_partial_bytes() {
        // 1 个完整 f32 + 3 残字节 → chunks_exact 截断尾部
        let blob = f32s_to_blob(&[42.0]);
        let with_garbage: Vec<u8> = blob.into_iter().chain([0xff, 0xff, 0xff]).collect();
        assert_eq!(blob_to_f32s(&with_garbage), vec![42.0]);
    }

    #[test]
    fn cosine_identical_vectors_near_one() {
        let a = vec![1.0, 2.0, 3.0];
        let sim = cosine_similarity(&a, &a);
        assert!((sim - 1.0).abs() < 1e-5, "identical ≈ 1.0, got {}", sim);
    }

    #[test]
    fn cosine_orthogonal_vectors_near_zero() {
        let a = vec![1.0, 0.0];
        let b = vec![0.0, 1.0];
        let sim = cosine_similarity(&a, &b);
        assert!(sim.abs() < 1e-5, "orthogonal ≈ 0.0, got {}", sim);
    }

    #[test]
    fn cosine_opposite_vectors_near_neg_one() {
        let a = vec![1.0, 2.0, 3.0];
        let b = vec![-1.0, -2.0, -3.0];
        let sim = cosine_similarity(&a, &b);
        assert!((sim + 1.0).abs() < 1e-5, "opposite ≈ -1.0, got {}", sim);
    }

    #[test]
    fn cosine_dimension_mismatch_zips_to_shortest() {
        // 实现用 zip(a,b),维度不匹配按较短的那个对齐,不 panic
        let a = vec![1.0, 0.0, 0.0]; // 3 维
        let b = vec![1.0, 0.0]; // 2 维 → zip 取前 2 个
        let sim = cosine_similarity(&a, &b);
        assert!((sim - 1.0).abs() < 1e-5, "zip 截断后应 ≈ 1.0, got {}", sim);
    }

    #[test]
    fn cosine_zero_vector_does_not_panic() {
        // 分母有 +1e-8 保护,零向量返回有限值而非 NaN/inf
        let z = vec![0.0, 0.0, 0.0];
        let sim = cosine_similarity(&z, &z);
        assert!(sim.is_finite(), "零向量相似度应有限, got {}", sim);
    }

    // ---------- KnowledgeRepo 内存 DB ----------

    /// 构造一条 KnowledgeRecord fixture
    fn krec(
        id: &str,
        title: &str,
        content: &str,
        kind: &str,
        status: &str,
        confidence: Option<&str>,
        reuse: i32,
    ) -> KnowledgeRecord {
        KnowledgeRecord {
            id: id.to_string(),
            kind: kind.to_string(),
            title: title.to_string(),
            content: content.to_string(),
            tags: Some("[]".to_string()),
            status: status.to_string(),
            confidence: confidence.map(|s| s.to_string()),
            reuse_count: reuse,
            verified: false,
            source_project: Some("proj-1".to_string()),
            source_ref: Some("conv:c1".to_string()),
            reasoning: None,
            created_at: "1700000000000".to_string(),
            updated_at: "1700000000000".to_string(),
        }
    }

    async fn setup_repo() -> KnowledgeRepo {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        KnowledgeRepo::new(&db)
    }

    #[tokio::test]
    async fn search_matches_title_and_filters_non_published() {
        let repo = setup_repo().await;
        // 一条命中(title 含关键词)、一条 published 不命中、一条 status 非 published 但命中
        repo.insert(krec("k1", "Rust 异步并发模型", "tokio 运行时", "lesson", "published", Some("high"), 5))
            .await
            .unwrap();
        repo.insert(krec("k2", "无关标题", "无关内容", "lesson", "published", None, 0))
            .await
            .unwrap();
        repo.insert(krec("k3", "Rust 异步进阶", "...", "lesson", "candidate", Some("low"), 9))
            .await
            .unwrap();

        let hits = repo.search("异步", None, 10).await.unwrap();
        // 只有 k1 是 published 且命中;k3 命中但非 published 被过滤
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].id, "k1");
    }

    #[tokio::test]
    async fn search_matches_content_branch() {
        let repo = setup_repo().await;
        // title 不含关键词,content 含 → 命中 content LIKE 分支
        repo.insert(krec("k1", "标题", "深入理解 Rust 所有权与借用", "lesson", "published", None, 1))
            .await
            .unwrap();

        let hits = repo.search("所有权", None, 10).await.unwrap();
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].id, "k1");
    }

    #[tokio::test]
    async fn search_no_keyword_match_returns_empty() {
        let repo = setup_repo().await;
        repo.insert(krec("k1", "Rust", "tokio", "lesson", "published", None, 1))
            .await
            .unwrap();

        let hits = repo.search("不存在的关键词xyz", None, 10).await.unwrap();
        assert!(hits.is_empty());
    }

    #[tokio::test]
    async fn search_kind_filter_narrows_results() {
        let repo = setup_repo().await;
        repo.insert(krec("k1", "Rust 规范", "...", "lesson", "published", None, 1))
            .await
            .unwrap();
        repo.insert(krec("k2", "Rust 决策", "...", "decision", "published", None, 1))
            .await
            .unwrap();

        // 不过滤 kind → 两条都命中
        let all = repo.search("Rust", None, 10).await.unwrap();
        assert_eq!(all.len(), 2);
        // 只取 lesson → 仅 k1
        let only_lesson = repo.search("Rust", Some("lesson"), 10).await.unwrap();
        assert_eq!(only_lesson.len(), 1);
        assert_eq!(only_lesson[0].id, "k1");
    }

    #[tokio::test]
    async fn search_orders_by_reuse_count_desc() {
        let repo = setup_repo().await;
        repo.insert(krec("low", "Rust A", "...", "lesson", "published", None, 1))
            .await
            .unwrap();
        repo.insert(krec("high", "Rust B", "...", "lesson", "published", None, 50))
            .await
            .unwrap();
        repo.insert(krec("mid", "Rust C", "...", "lesson", "published", None, 10))
            .await
            .unwrap();

        let hits = repo.search("Rust", None, 10).await.unwrap();
        let ids: Vec<_> = hits.iter().map(|h| h.id.as_str()).collect();
        assert_eq!(ids, vec!["high", "mid", "low"]);
    }

    #[tokio::test]
    async fn search_truncates_to_limit() {
        let repo = setup_repo().await;
        for i in 0..5 {
            repo.insert(krec(&format!("k{i}"), &format!("Rust-{i}"), "...", "lesson", "published", None, i))
                .await
                .unwrap();
        }
        let hits = repo.search("Rust", None, 2).await.unwrap();
        assert_eq!(hits.len(), 2);
        // reuse_count 最高的两条(k4=4, k3=3)
        assert_eq!(hits[0].id, "k4");
        assert_eq!(hits[1].id, "k3");
    }

    #[tokio::test]
    async fn list_by_status_orders_by_confidence_semantics() {
        // confidence 字典序 high>low>medium 非预期,实现用 CASE 强制 high>medium>low
        let repo = setup_repo().await;
        repo.insert(krec("low", "t", "c", "lesson", "pending_review", Some("low"), 0))
            .await
            .unwrap();
        repo.insert(krec("high", "t", "c", "lesson", "pending_review", Some("high"), 0))
            .await
            .unwrap();
        repo.insert(krec("medium", "t", "c", "lesson", "pending_review", Some("medium"), 0))
            .await
            .unwrap();
        // created_at 相同,纯靠 confidence 排序

        let list = repo.list_by_status("pending_review").await.unwrap();
        let confidences: Vec<_> = list.iter().map(|r| r.confidence.as_deref().unwrap_or("")).collect();
        assert_eq!(confidences, vec!["high", "medium", "low"]);
    }

    #[tokio::test]
    async fn list_by_status_filters_other_status() {
        let repo = setup_repo().await;
        repo.insert(krec("a", "t", "c", "lesson", "pending_review", Some("high"), 0))
            .await
            .unwrap();
        repo.insert(krec("b", "t", "c", "lesson", "published", Some("high"), 0))
            .await
            .unwrap();

        let list = repo.list_by_status("pending_review").await.unwrap();
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].id, "a");
    }

    #[tokio::test]
    async fn increment_reuse_count_is_atomic_plus_one() {
        let repo = setup_repo().await;
        repo.insert(krec("k1", "Rust", "...", "lesson", "published", None, 0))
            .await
            .unwrap();

        // 连续 +1 两次
        assert!(repo.increment_reuse_count("k1").await.unwrap());
        assert!(repo.increment_reuse_count("k1").await.unwrap());

        let rec = repo.get_by_id("k1").await.unwrap().expect("记录存在");
        assert_eq!(rec.reuse_count, 2);

        // 不存在的 id → false
        assert!(!repo.increment_reuse_count("nope").await.unwrap());
    }

    #[tokio::test]
    async fn search_vector_orders_by_cosine_and_truncates() {
        let repo = setup_repo().await;
        // 三条 published 记录,embedding 维度均为 2
        repo.insert(krec("exact", "e", "c", "lesson", "published", None, 0))
            .await
            .unwrap();
        repo.insert(krec("orth", "e", "c", "lesson", "published", None, 0))
            .await
            .unwrap();
        repo.insert(krec("neg", "e", "c", "lesson", "published", None, 0))
            .await
            .unwrap();
        // 一条非 published(应被过滤)
        repo.insert(krec("hidden", "e", "c", "lesson", "candidate", None, 0))
            .await
            .unwrap();

        repo.set_embedding("exact", &[1.0, 0.0]).await.unwrap(); // 与 query 完全同向
        repo.set_embedding("orth", &[0.0, 1.0]).await.unwrap(); // 正交 ≈ 0
        repo.set_embedding("neg", &[-1.0, 0.0]).await.unwrap(); // 反向 ≈ -1
        repo.set_embedding("hidden", &[1.0, 0.0]).await.unwrap(); // 非 published 过滤掉

        let results = repo.search_vector(&[1.0, 0.0], 10).await.unwrap();
        // hidden 被 status 过滤,剩 3 条
        assert_eq!(results.len(), 3);
        // 按相似度降序: exact(≈1) > orth(≈0) > neg(≈-1)
        assert_eq!(results[0].0.id, "exact");
        assert_eq!(results[1].0.id, "orth");
        assert_eq!(results[2].0.id, "neg");
        assert!((results[0].1 - 1.0).abs() < 1e-5);
        assert!(results[1].1.abs() < 1e-5);
        assert!((results[2].1 + 1.0).abs() < 1e-5);
    }

    #[tokio::test]
    async fn search_vector_skips_dimension_mismatch() {
        let repo = setup_repo().await;
        repo.insert(krec("dim2", "e", "c", "lesson", "published", None, 0))
            .await
            .unwrap();
        repo.insert(krec("dim3", "e", "c", "lesson", "published", None, 0))
            .await
            .unwrap();

        repo.set_embedding("dim2", &[1.0, 0.0]).await.unwrap();
        repo.set_embedding("dim3", &[1.0, 0.0, 0.0]).await.unwrap(); // 维度不匹配

        // query 是 2 维,dim3 被跳过
        let results = repo.search_vector(&[1.0, 0.0], 10).await.unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].0.id, "dim2");
    }

    #[tokio::test]
    async fn search_vector_truncates_to_limit() {
        let repo = setup_repo().await;
        for i in 0..4 {
            repo.insert(krec(&format!("k{i}"), "e", "c", "lesson", "published", None, 0))
                .await
                .unwrap();
            // 全部与 query 同向,相似度相同 → 仅验证 truncate 生效
            repo.set_embedding(&format!("k{i}"), &[1.0, 0.0]).await.unwrap();
        }
        let results = repo.search_vector(&[1.0, 0.0], 2).await.unwrap();
        assert_eq!(results.len(), 2);
    }

    #[tokio::test]
    async fn search_vector_ignores_records_without_embedding() {
        let repo = setup_repo().await;
        // published 但未写 embedding
        repo.insert(krec("noemb", "e", "c", "lesson", "published", None, 0))
            .await
            .unwrap();
        let results = repo.search_vector(&[1.0, 0.0], 10).await.unwrap();
        assert!(results.is_empty());
    }
}