//! AI 编排节点 — 直连 agentic 会话执行工作流任务
//! AI 编排节点 — 直连 agentic 会话执行工作流任务
//!
//! 与 AiNode(单次 LLM 补全)的区别:AiAgentNode 在节点 execute 时以编程方式启动一个
//! 独立 agentic 会话(新 conv_id,与用户活跃会话隔离,防 epoch 互踩),由 run_agentic_loop
//! 完整驱动(多轮迭代 + 工具调用),会话收敛后取最终 assistant 报告作为节点输出。
//!
//! 阻塞语义与 HumanNode 同构:tokio::select! 三路等待
//!   - 会话完成通知(AppState.agent_completion_bridge 的 oneshot,由 agentic loop 收敛
//!     统一出口 emit_ai_completed_once 投递,epoch 守卫保证仅 owner loop 触发)
//!   - deadline 超时(config.timeout_secs,默认 1800s)→ Err(工作流 failed → 推进链退回)
//!   - 500ms tick 轮询 is_cancelled
//!
//! spawn 通道经 [`SpawnFn`] 注入(src-tauri 节点工厂提供,内部取 AppState 起会话),
//! 本 crate 不依赖 src-tauri 类型,依赖方向与 AiNode 持 Arc<Database> 同款。

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use df_storage::crud::TaskRepo;
use df_storage::db::Database;
use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema};

// 任务文本兜底 prompt(与 AiNode 同款,经 ai_node_helpers 复用)
use crate::ai_node_helpers::{default_fallback_prompt, resolve_fallback_prompt};

/// 会话启动结果:(conv_id, 完成通知接收端)。接收端值 = 最终 assistant 报告文本。
pub type SpawnedSession = (String, tokio::sync::oneshot::Receiver<String>);

/// 会话启动 future(Err = 启动失败,如 provider 缺失/应用句柄未就绪)。
pub type SpawnFuture = Pin<Box<dyn Future<Output = Result<SpawnedSession, String>> + Send>>;

/// 会话启动通道:入参任务文本,由 src-tauri 节点工厂注入实现(见 chat.rs
/// orchestration_session_spawner)。
pub type SpawnFn = Arc<dyn Fn(String) -> SpawnFuture + Send + Sync>;

/// 默认编排超时(秒)。1800s = 30min,与 HumanNode 默认审批窗口同量级;
/// agentic 会话含多轮工具调用,给宽值,到点 fail 退回任务状态。
const DEFAULT_TIMEOUT_SECS: u64 = 1800;

/// AI 编排节点
///
/// 持 Arc<Database> 落 task.output_json(有 task_id 时,与 AiNode 同模式),
/// 持 SpawnFn 启动独立编排会话。
pub struct AiAgentNode {
    db: Arc<Database>,
    spawn: SpawnFn,
}

impl AiAgentNode {
    /// 构造节点(注册表工厂调用,注入数据库句柄与会话启动通道)
    pub fn new(db: Arc<Database>, spawn: SpawnFn) -> Self {
        Self { db, spawn }
    }
}

#[async_trait]
impl Node for AiAgentNode {
    async fn execute(&self, ctx: NodeContext) -> NodeResult {
        tracing::info!("AiAgentNode 执行: node_id={}", ctx.node_id);

        // 任务文本取值优先级与 AiNode 一致:上游 prompt 输出 > config.prompt >
        // task_id 兜底(读 TaskRecord 生成) > 通用默认。推进链模板 config 无 prompt,
        // 兜底链保证单节点 DAG 也能拿到任务书。
        let task = match ctx
            .inputs
            .get("prompt")
            .and_then(|o| o.data.as_str())
            .map(String::from)
            .or_else(|| {
                ctx.config
                    .get("prompt")
                    .and_then(|v| v.as_str())
                    .map(String::from)
            }) {
            Some(t) if !t.trim().is_empty() => t,
            _ => {
                resolve_fallback_prompt(&self.db, &ctx.config)
                    .await
                    .unwrap_or_else(default_fallback_prompt)
            }
        };

        let timeout_secs = ctx
            .config
            .get("timeout_secs")
            .and_then(|v| v.as_u64())
            .unwrap_or(DEFAULT_TIMEOUT_SECS);
        let task_id = ctx
            .config
            .get("task_id")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();

        // 启动独立编排会话(内部生成新 conv_id 并注册完成通知桥,注册先于 spawn)
        let (conv_id, mut done_rx) = (self.spawn)(task)
            .await
            .map_err(|e| anyhow::anyhow!("AI 编排会话启动失败: {}", e))?;

        // select! 骨架与 HumanNode 同构:完成通知 / deadline 超时 / 500ms 取消轮询。
        // oneshot 接收端在 loop 里重 poll,须借 &mut(移动进 select! 会破坏循环重入)。
        let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
        let mut cancel_tick = tokio::time::interval(Duration::from_millis(500));
        cancel_tick.tick().await; // 丢弃首个立即触发的 tick

        let report = loop {
            tokio::select! {
                done = &mut done_rx => match done {
                    Ok(text) => break text,
                    Err(_) => {
                        // 发送端被 drop 且未投递:会话条目被覆盖注册或异常清除,无报告可取
                        return Err(anyhow::anyhow!(
                            "AI 编排会话完成信号丢失(conv_id={})",
                            conv_id
                        ));
                    }
                },
                _ = tokio::time::sleep_until(deadline) => {
                    // 超时 fail:工作流 failed → 推进链按 regression_target 退回任务状态
                    return Err(anyhow::anyhow!(
                        "AI 编排会话超时({}s), conv_id={}",
                        timeout_secs,
                        conv_id
                    ));
                }
                _ = cancel_tick.tick() => {
                    if ctx.node_status.is_cancelled(&ctx.node_id) {
                        return Err(anyhow::anyhow!("AI 编排会话被取消"));
                    }
                }
            }
        };

        // 空报告判失败:stop 在任何输出前 / 会话被清理的场景,不作为有效产出推进任务
        if report.trim().is_empty() {
            return Err(anyhow::anyhow!(
                "AI 编排会话无输出(会话被停止或未产出报告), conv_id={}",
                conv_id
            ));
        }

        let output = serde_json::json!({
            "text": report,
            "conv_id": conv_id,
            "task_id": task_id,
        });

        // 有 task_id 落 task.output_json(与 AiNode 同模式:通用 update_field 白名单字段,
        // 落库失败不阻断节点返回,产出已在内存工作流可继续)。
        if !task_id.is_empty() {
            if let Ok(json_str) = serde_json::to_string(&output) {
                let repo = TaskRepo::new(&self.db);
                if let Err(e) = repo.update_field(&task_id, "output_json", &json_str).await {
                    tracing::warn!(
                        task_id = task_id,
                        error = %e,
                        "AiAgentNode 落 output_json 失败(产出已在内存,节点继续)"
                    );
                }
            }
        }

        Ok(NodeOutput::from_value(output))
    }

    fn schema(&self) -> NodeSchema {
        NodeSchema {
            params: serde_json::json!({
                "type": "object",
                "properties": {
                    "prompt": { "type": "string", "description": "任务书(缺省依次取上游 prompt 输出 / 按 task_id 读任务生成 / 通用默认)" },
                    "timeout_secs": { "type": "integer", "description": "编排超时秒数(默认 1800,到点节点 fail 触发工作流退回)" },
                    "task_id": { "type": "string", "description": "关联任务 ID(run_workflow 注入;有值时产出落 task.output_json)" }
                },
                "required": []
            }),
            output: serde_json::json!({
                "type": "object",
                "properties": {
                    "text": { "type": "string", "description": "最终 assistant 报告文本" },
                    "conv_id": { "type": "string", "description": "编排会话 ID(独立于用户活跃会话)" },
                    "task_id": { "type": "string" }
                }
            }),
        }
    }

    fn is_blocking(&self) -> bool {
        true
    }

    fn node_type(&self) -> &str {
        "ai_agent"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use df_storage::crud::ProjectRepo;
    use df_storage::models::{ProjectRecord, TaskRecord};
    use df_types::types::{ProjectStatus, TaskStatus};
    use df_workflow::eventbus::EventBus;
    use df_workflow::state::StateMachine;
    use serde_json::json;
    use std::collections::HashMap;

    /// 构造 NodeContext:自定义 config,共享指定状态机(取消轮询测试用)。
    fn make_ctx_with(config: serde_json::Value, sm: StateMachine) -> NodeContext {
        NodeContext {
            node_id: "node-a".to_string(),
            inputs: HashMap::new(),
            config,
            execution_id: "exec-1".into(),
            event_bus: EventBus::new(),
            node_status: sm,
        }
    }

    fn make_ctx(config: serde_json::Value) -> NodeContext {
        make_ctx_with(config, StateMachine::new())
    }

    /// 内存 DB(无表数据;task_id 为空时节点不触库)。
    async fn empty_db() -> Arc<Database> {
        Arc::new(Database::open_in_memory().await.expect("open_in_memory"))
    }

    /// 内存 DB helper:插父项目 + 指定 task,返回 (db, task_id)。
    async fn setup_task_db() -> (Arc<Database>, String) {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        ProjectRepo::new(&db)
            .insert(ProjectRecord {
                id: "p1".to_string(),
                name: "proj".to_string(),
                description: "".to_string(),
                status: ProjectStatus::Planning,
                idea_id: None,
                path: None,
                stack: None,
                archived: false,
                created_at: "0".to_string(),
                updated_at: "0".to_string(),
            })
            .await
            .expect("insert project");
        TaskRepo::new(&db)
            .insert(TaskRecord {
                id: "t1".to_string(),
                project_id: "p1".to_string(),
                title: "实现登录接口".to_string(),
                description: "含前端表单与后端校验".to_string(),
                status: TaskStatus::Todo,
                priority: 2,
                branch_name: None,
                assignee: None,
                workflow_def_id: None,
                base_branch: None,
                review_rounds: 0,
                output_json: None,
                idea_id: None,
                module_id: None,
                queue: "todo".to_string(),
                parent_id: None,
                content_json: None,
                archived: false,
                created_at: "0".to_string(),
                updated_at: "0".to_string(),
            })
            .await
            .expect("insert task");
        (Arc::new(db), "t1".to_string())
    }

    /// spawn 桩:启动后 50ms 投递指定报告文本。
    fn spawn_delivering(report: &'static str) -> SpawnFn {
        Arc::new(move |_task: String| {
            Box::pin(async move {
                let (tx, rx) = tokio::sync::oneshot::channel::<String>();
                tokio::spawn(async move {
                    tokio::time::sleep(Duration::from_millis(50)).await;
                    let _ = tx.send(report.to_string());
                });
                Ok(("conv-1".to_string(), rx))
            })
        })
    }

    /// spawn 桩:发送端保活不投递(模拟会话长时间未收敛)。
    fn spawn_silent() -> SpawnFn {
        Arc::new(|_task: String| {
            Box::pin(async move {
                let (tx, rx) = tokio::sync::oneshot::channel::<String>();
                tokio::spawn(async move {
                    tokio::time::sleep(Duration::from_secs(5)).await;
                    let _ = tx.send("迟到的报告".to_string());
                });
                Ok(("conv-silent".to_string(), rx))
            })
        })
    }

    /// 完成通知桥触发:spawn 后收到投递 → Ok 输出 {text, conv_id, task_id}。
    #[tokio::test]
    async fn completion_notify_returns_report() {
        let node = AiAgentNode::new(empty_db().await, spawn_delivering("任务完成报告"));
        let ctx = make_ctx(json!({ "task_id": "t1" }));
        let out = node.execute(ctx).await.unwrap();
        assert_eq!(out.data["text"], json!("任务完成报告"));
        assert_eq!(out.data["conv_id"], json!("conv-1"));
        assert_eq!(out.data["task_id"], json!("t1"));
    }

    /// deadline 超时:发送端保活不投递 + 短超时 → Err 超时(工作流 failed 退回路径)。
    #[tokio::test]
    async fn timeout_fails_node() {
        let node = AiAgentNode::new(empty_db().await, spawn_silent());
        let ctx = make_ctx(json!({ "timeout_secs": 1 }));
        let err = node.execute(ctx).await.unwrap_err().to_string();
        assert!(err.contains("超时"), "应报超时, 实际: {}", err);
        assert!(err.contains("conv-silent"), "超时信息应带 conv_id, 实际: {}", err);
    }

    /// 取消轮询:500ms tick 命中 is_cancelled → Err 取消(状态保持 Cancelled 由 executor 保证)。
    #[tokio::test]
    async fn cancelled_poll_fails_node() {
        let sm = StateMachine::new();
        let sm_external = sm.clone();
        let node = AiAgentNode::new(empty_db().await, spawn_silent());
        let handle = tokio::spawn(async move { node.execute(make_ctx_with(json!({}), sm)).await });
        // 等 execute 进入 select! 轮询后再取消(过早 set 会因 tick 未启动而漏检一轮,仍会被下轮命中)
        tokio::time::sleep(Duration::from_millis(50)).await;
        sm_external.set_cancelled("node-a".to_string());
        let err = handle.await.unwrap().unwrap_err().to_string();
        assert!(err.contains("取消"), "应报取消, 实际: {}", err);
    }

    /// 启动失败:spawn 返 Err(provider 缺失等)→ 节点 Err 带原因。
    #[tokio::test]
    async fn spawn_failure_fails_node() {
        let spawn: SpawnFn = Arc::new(|_task: String| {
            Box::pin(async { Err("未配置 AI Provider".to_string()) })
        });
        let node = AiAgentNode::new(empty_db().await, spawn);
        let err = node.execute(make_ctx(json!({}))).await.unwrap_err().to_string();
        assert!(err.contains("启动失败"), "应报启动失败, 实际: {}", err);
        assert!(err.contains("未配置 AI Provider"), "应携带原因, 实际: {}", err);
    }

    /// 信号丢失:发送端被 drop 且未投递 → Err 信号丢失(不误判为成功)。
    #[tokio::test]
    async fn dropped_sender_fails_node() {
        let spawn: SpawnFn = Arc::new(|_task: String| {
            Box::pin(async move {
                let (tx, rx) = tokio::sync::oneshot::channel::<String>();
                drop(tx);
                Ok(("conv-drop".to_string(), rx))
            })
        });
        let node = AiAgentNode::new(empty_db().await, spawn);
        let err = node.execute(make_ctx(json!({}))).await.unwrap_err().to_string();
        assert!(err.contains("信号丢失"), "应报信号丢失, 实际: {}", err);
    }

    /// 空报告:投递了纯空白文本 → Err 无输出(不作为有效产出推进任务)。
    #[tokio::test]
    async fn blank_report_fails_node() {
        let spawn: SpawnFn = Arc::new(|_task: String| {
            Box::pin(async move {
                let (tx, rx) = tokio::sync::oneshot::channel::<String>();
                let _ = tx.send("   ".to_string());
                Ok(("conv-blank".to_string(), rx))
            })
        });
        let node = AiAgentNode::new(empty_db().await, spawn);
        let err = node.execute(make_ctx(json!({}))).await.unwrap_err().to_string();
        assert!(err.contains("无输出"), "应报无输出, 实际: {}", err);
    }

    /// 有 task_id 落 task.output_json(与 AiNode 同一条 update_field 白名单路径),
    /// 且任务书兜底链生效:config 无 prompt 时按 task_id 读任务生成。
    #[tokio::test]
    async fn task_output_json_written_and_fallback_prompt_used() {
        let (db, task_id) = setup_task_db().await;
        let node = AiAgentNode::new(db.clone(), spawn_delivering("登录接口已完成"));
        // config 只有 task_id(推进链模板形态),prompt 走兜底生成
        let ctx = make_ctx(json!({ "task_id": task_id }));
        let out = node.execute(ctx).await.unwrap();
        assert_eq!(out.data["text"], json!("登录接口已完成"));

        let after = TaskRepo::new(&db)
            .get_by_id("t1")
            .await
            .unwrap()
            .expect("task 存在");
        let saved = after.output_json.expect("output_json 应被写入");
        let parsed: serde_json::Value = serde_json::from_str(&saved).unwrap();
        assert_eq!(parsed["text"], json!("登录接口已完成"));
        assert_eq!(parsed["task_id"], json!("t1"));
        assert!(parsed["conv_id"].as_str().is_some(), "conv_id 应落 output_json");
    }

    /// 默认编排超时 1800s(30min),与 HumanNode 默认审批窗口同量级。
    #[test]
    fn default_timeout_is_1800_secs() {
        assert_eq!(DEFAULT_TIMEOUT_SECS, 1800, "默认编排超时应为 1800s(30min)");
    }
}