feat(health+workflow): 行动分发→工作流启动集成 — 事件驱动 BPMN 实例化

- create_pending_action 新增 workflow.ai_action.start_requested 事件发布
- 根据 action_type 映射到对应 BPMN 流程定义 key
- erp-workflow 消费启动请求，自动创建审批流程实例
- 流程变量包含 risk_level/patient_id/action_type/params

crates/erp-health/src/event.rs

@@ -445,6 +445,83 @@ pub fn register_handlers_with_state(state: crate::state::HealthState) {
         }
     });
 
+    // ai.analysis.completed → AI→行动闭环消费者（行动分发）
+    let (mut ai_action_rx, _ai_action_handle) = state.event_bus.subscribe_filtered("ai.".to_string());
+    let action_db = state.db.clone();
+    let action_event_bus = state.event_bus.clone();
+    tokio::spawn(async move {
+        loop {
+            match ai_action_rx.recv().await {
+                Some(event) if event.event_type == "ai.analysis.completed" => {
+                    if erp_core::events::is_event_processed(&action_db, event.id, "ai_action_dispatcher").await.unwrap_or(false) {
+                        continue;
+                    }
+
+                    let tenant_id = event.tenant_id;
+                    let analysis_id = event.payload.get("analysis_id")
+                        .and_then(|v| v.as_str())
+                        .and_then(|s| Uuid::parse_str(s).ok());
+                    let patient_id = event.payload.get("patient_id")
+                        .and_then(|v| v.as_str())
+                        .and_then(|s| Uuid::parse_str(s).ok());
+                    let doctor_id = event.payload.get("doctor_id")
+                        .and_then(|v| v.as_str())
+                        .and_then(|s| Uuid::parse_str(s).ok());
+                    let risk_level = event.payload.get("risk_level")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("medium");
+                    let suggestion_count = event.payload.get("suggestion_count")
+                        .and_then(|v| v.as_u64())
+                        .unwrap_or(0);
+
+                    if suggestion_count > 0 {
+                        if let (Some(aid), Some(pid)) = (analysis_id, patient_id) {
+                            let loader_result: Result<Vec<serde_json::Value>, sea_orm::DbErr> =
+                                crate::service::ai_suggestion_loader::load_by_analysis(
+                                    &action_db, tenant_id, aid,
+                                ).await;
+                            match loader_result {
+                                Ok(suggestions) if !suggestions.is_empty() => {
+                                    crate::service::ai_action_dispatcher::handle_ai_suggestions(
+                                        &action_db,
+                                        &action_event_bus,
+                                        tenant_id,
+                                        aid,
+                                        pid,
+                                        doctor_id,
+                                        &suggestions,
+                                        risk_level,
+                                    ).await;
+                                    tracing::info!(
+                                        analysis_id = %aid,
+                                        patient_id = %pid,
+                                        suggestion_count = suggestions.len(),
+                                        risk_level = %risk_level,
+                                        "AI 行动分发完成"
+                                    );
+                                }
+                                Ok(_) => {
+                                    tracing::info!(analysis_id = %aid, "建议列表为空，跳过行动分发");
+                                }
+                                Err(e) => {
+                                    tracing::warn!(
+                                        analysis_id = %aid,
+                                        error = %e,
+                                        "加载建议列表失败"
+                                    );
+                                }
+                            }
+                        }
+                    }
+
+                    let _ = erp_core::events::mark_event_processed(&action_db, event.id, "ai_action_dispatcher").await;
+                }
+                Some(_) => {}
+                None => break,
+            }
+        }
+    });
+
     // consent.granted/revoked → 通知关联医生
     let (mut consent_rx, _consent_handle) = state.event_bus.subscribe_filtered("consent.".to_string());
     let consent_db = state.db.clone();

crates/erp-health/src/service/ai_action_dispatcher.rs

@@ -0,0 +1,208 @@
+use std::time::Duration;
+
+use erp_core::events::EventBus;
+use sea_orm::DatabaseConnection;
+use uuid::Uuid;
+
+/// 执行模式
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ExecutionMode {
+    AutoExecute,
+    DoctorReview,
+    UrgentConfirm,
+}
+
+/// 分发决策
+#[derive(Debug, Clone)]
+pub struct DispatchDecision {
+    pub execution_mode: ExecutionMode,
+    pub response_timeout: Option<Duration>,
+}
+
+/// 根据风险等级和建议类型生成执行决策
+pub fn dispatch_decision(risk_level: &str, _suggestion_type: &str) -> DispatchDecision {
+    match risk_level {
+        "low" => DispatchDecision {
+            execution_mode: ExecutionMode::AutoExecute,
+            response_timeout: None,
+        },
+        "medium" => DispatchDecision {
+            execution_mode: ExecutionMode::DoctorReview,
+            response_timeout: Some(Duration::from_secs(86400)),
+        },
+        "high" => DispatchDecision {
+            execution_mode: ExecutionMode::UrgentConfirm,
+            response_timeout: Some(Duration::from_secs(14400)),
+        },
+        _ => DispatchDecision {
+            execution_mode: ExecutionMode::DoctorReview,
+            response_timeout: Some(Duration::from_secs(86400)),
+        },
+    }
+}
+
+/// 处理 AI 建议事件：根据风险等级分发到不同执行路径
+pub async fn handle_ai_suggestions(
+    db: &DatabaseConnection,
+    event_bus: &EventBus,
+    tenant_id: Uuid,
+    _analysis_id: Uuid,
+    patient_id: Uuid,
+    doctor_id: Option<Uuid>,
+    suggestions: &[serde_json::Value],
+    risk_level: &str,
+) {
+    for suggestion in suggestions {
+        let suggestion_type = suggestion["type"].as_str().unwrap_or("alert");
+        let decision = dispatch_decision(risk_level, suggestion_type);
+
+        match decision.execution_mode {
+            ExecutionMode::AutoExecute => {
+                execute_action(
+                    db,
+                    event_bus,
+                    tenant_id,
+                    patient_id,
+                    suggestion_type,
+                    suggestion,
+                )
+                .await;
+            }
+            ExecutionMode::DoctorReview | ExecutionMode::UrgentConfirm => {
+                create_pending_action(
+                    db,
+                    event_bus,
+                    tenant_id,
+                    patient_id,
+                    doctor_id,
+                    suggestion_type,
+                    suggestion,
+                    risk_level,
+                    &decision,
+                )
+                .await;
+            }
+        }
+    }
+}
+
+async fn execute_action(
+    db: &DatabaseConnection,
+    event_bus: &EventBus,
+    tenant_id: Uuid,
+    patient_id: Uuid,
+    action_type: &str,
+    params: &serde_json::Value,
+) {
+    match action_type {
+        "alert" => {
+            let event = erp_core::events::DomainEvent::new(
+                "health.ai_alert.sent",
+                tenant_id,
+                erp_core::events::build_event_payload(serde_json::json!({
+                    "patient_id": patient_id,
+                    "alert_type": "ai_risk_warning",
+                    "severity": params.get("severity").and_then(|v| v.as_str()).unwrap_or("warning"),
+                    "message": params.get("message").and_then(|v| v.as_str()).unwrap_or(""),
+                    "source": "ai_analysis",
+                })),
+            );
+            event_bus.publish(event, db).await;
+        }
+        "followup" | "appointment" => {
+            let event = erp_core::events::DomainEvent::new(
+                "health.ai_action.auto_executed",
+                tenant_id,
+                erp_core::events::build_event_payload(serde_json::json!({
+                    "patient_id": patient_id,
+                    "action_type": action_type,
+                    "params": params,
+                })),
+            );
+            event_bus.publish(event, db).await;
+        }
+        _ => {}
+    }
+}
+
+async fn create_pending_action(
+    db: &DatabaseConnection,
+    event_bus: &EventBus,
+    tenant_id: Uuid,
+    patient_id: Uuid,
+    doctor_id: Option<Uuid>,
+    action_type: &str,
+    params: &serde_json::Value,
+    risk_level: &str,
+    decision: &DispatchDecision,
+) {
+    // 发布待审批事件（通知/日志用）
+    let event = erp_core::events::DomainEvent::new(
+        "health.ai_action.pending_approval",
+        tenant_id,
+        erp_core::events::build_event_payload(serde_json::json!({
+            "patient_id": patient_id,
+            "doctor_id": doctor_id,
+            "action_type": action_type,
+            "risk_level": risk_level,
+            "timeout_seconds": decision.response_timeout.map(|d| d.as_secs()),
+            "params": params,
+        })),
+    );
+    event_bus.publish(event, db).await;
+
+    // 发布工作流启动请求事件（触发 BPMN 审批流程）
+    let workflow_key = match action_type {
+        "followup" => "ai_followup_workflow",
+        "appointment" => "ai_appointment_workflow",
+        "alert" => "ai_alert_workflow",
+        _ => return,
+    };
+
+    let workflow_event = erp_core::events::DomainEvent::new(
+        "workflow.ai_action.start_requested",
+        tenant_id,
+        erp_core::events::build_event_payload(serde_json::json!({
+            "workflow_key": workflow_key,
+            "patient_id": patient_id,
+            "doctor_id": doctor_id,
+            "risk_level": risk_level,
+            "action_type": action_type,
+            "params": params,
+        })),
+    );
+    event_bus.publish(workflow_event, db).await;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn route_low_risk_to_auto_execute() {
+        let decision = dispatch_decision("low", "alert");
+        assert_eq!(decision.execution_mode, ExecutionMode::AutoExecute);
+        assert_eq!(decision.response_timeout, None);
+    }
+
+    #[test]
+    fn route_medium_risk_to_doctor_review() {
+        let decision = dispatch_decision("medium", "followup");
+        assert_eq!(decision.execution_mode, ExecutionMode::DoctorReview);
+        assert_eq!(decision.response_timeout, Some(Duration::from_secs(86400)));
+    }
+
+    #[test]
+    fn route_high_risk_to_urgent_confirm() {
+        let decision = dispatch_decision("high", "alert");
+        assert_eq!(decision.execution_mode, ExecutionMode::UrgentConfirm);
+        assert_eq!(decision.response_timeout, Some(Duration::from_secs(14400)));
+    }
+
+    #[test]
+    fn route_unknown_defaults_to_doctor_review() {
+        let decision = dispatch_decision("unknown", "followup");
+        assert_eq!(decision.execution_mode, ExecutionMode::DoctorReview);
+        assert_eq!(decision.response_timeout, Some(Duration::from_secs(86400)));
+    }
+}

crates/erp-health/src/service/ai_suggestion_loader.rs

@@ -0,0 +1,40 @@
+use sea_orm::{DatabaseConnection, FromQueryResult, Statement};
+use uuid::Uuid;
+
+#[derive(Debug, FromQueryResult)]
+struct SuggestionRow {
+    params: Option<serde_json::Value>,
+    suggestion_type: Option<String>,
+    risk_level: Option<String>,
+}
+
+/// 跨 crate 读取 ai_suggestion 表（通过 raw SQL）
+pub async fn load_by_analysis(
+    db: &DatabaseConnection,
+    tenant_id: Uuid,
+    analysis_id: Uuid,
+) -> Result<Vec<serde_json::Value>, sea_orm::DbErr> {
+    let rows: Vec<SuggestionRow> = SuggestionRow::find_by_statement(Statement::from_sql_and_values(
+        sea_orm::DatabaseBackend::Postgres,
+        r#"
+            SELECT params, suggestion_type, risk_level
+            FROM ai_suggestion
+            WHERE tenant_id = $1 AND analysis_id = $2 AND deleted_at IS NULL
+            ORDER BY created_at ASC
+        "#,
+        [tenant_id.into(), analysis_id.into()],
+    ))
+    .all(db)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|r| {
+            serde_json::json!({
+                "params": r.params.unwrap_or(serde_json::Value::Null),
+                "suggestion_type": r.suggestion_type.unwrap_or_default(),
+                "risk_level": r.risk_level.unwrap_or_default(),
+            })
+        })
+        .collect())
+}

crates/erp-health/src/service/mod.rs

@@ -1,3 +1,5 @@
+pub mod ai_action_dispatcher;
+pub mod ai_suggestion_loader;
 pub mod alert_engine;
 pub mod alert_rule_service;
 pub mod alert_service;

crates/erp-server/src/main.rs

@@ -289,6 +289,12 @@ async fn main() -> anyhow::Result<()> {
                 .map_err(|e| anyhow::anyhow!("Failed to seed auth data: {}", e))?;
 
                 tracing::info!(tenant_id = %new_tenant_id, "Default tenant ready with auth seed data");
+
+                // Seed AI workflow definitions
+                if let Err(e) = erp_workflow::service::ai_workflow_seed::ensure_ai_workflows(&db, new_tenant_id).await {
+                    tracing::warn!(error = %e, "Failed to seed AI workflow definitions");
+                }
+
                 new_tenant_id
             }
         }

crates/erp-workflow/src/module.rs

@@ -134,6 +134,120 @@ impl WorkflowModule {
     }
 }
 
+/// 处理 AI 行动工作流启动请求
+async fn handle_ai_action_start(
+    db: &sea_orm::DatabaseConnection,
+    event_bus: &EventBus,
+    event: &erp_core::events::DomainEvent,
+) {
+    let workflow_key = match event.payload.get("workflow_key").and_then(|v| v.as_str()) {
+        Some(k) => k,
+        None => {
+            tracing::warn!("AI 行动工作流事件缺少 workflow_key，跳过");
+            return;
+        }
+    };
+
+    let tenant_id = event.tenant_id;
+
+    // 查找对应的流程定义
+    use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
+    let def = crate::entity::process_definition::Entity::find()
+        .filter(crate::entity::process_definition::Column::TenantId.eq(tenant_id))
+        .filter(crate::entity::process_definition::Column::Key.eq(workflow_key))
+        .filter(crate::entity::process_definition::Column::DeletedAt.is_null())
+        .filter(crate::entity::process_definition::Column::Status.eq("published"))
+        .one(db)
+        .await;
+
+    let def = match def {
+        Ok(Some(d)) => d,
+        Ok(None) => {
+            tracing::warn!(
+                key = %workflow_key,
+                tenant_id = %tenant_id,
+                "AI 行动工作流定义未找到或未发布，跳过"
+            );
+            return;
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "查询工作流定义失败");
+            return;
+        }
+    };
+
+    // 构造启动变量
+    let risk_level = event.payload.get("risk_level")
+        .and_then(|v| v.as_str())
+        .unwrap_or("medium")
+        .to_string();
+
+    let variables = vec![
+        crate::dto::SetVariableReq {
+            name: "risk_level".into(),
+            var_type: Some("string".into()),
+            value: serde_json::Value::String(risk_level.clone()),
+        },
+        crate::dto::SetVariableReq {
+            name: "patient_id".into(),
+            var_type: Some("string".into()),
+            value: event.payload.get("patient_id")
+                .cloned()
+                .unwrap_or(serde_json::Value::Null),
+        },
+        crate::dto::SetVariableReq {
+            name: "action_type".into(),
+            var_type: Some("string".into()),
+            value: event.payload.get("action_type")
+                .and_then(|v| v.as_str())
+                .map(|s| serde_json::Value::String(s.to_string()))
+                .unwrap_or(serde_json::Value::Null),
+        },
+        crate::dto::SetVariableReq {
+            name: "params".into(),
+            var_type: Some("string".into()),
+            value: event.payload.get("params")
+                .cloned()
+                .unwrap_or(serde_json::Value::Null),
+        },
+    ];
+
+    let req = crate::dto::StartInstanceReq {
+        definition_id: def.id,
+        business_key: Some(format!("ai_action_{}", chrono::Utc::now().timestamp_millis())),
+        variables: Some(variables),
+    };
+
+    let system_id = Uuid::parse_str("00000000-0000-0000-0000-000000000000").unwrap();
+
+    match crate::service::instance_service::InstanceService::start(
+        tenant_id,
+        system_id,
+        &req,
+        db,
+        event_bus,
+    )
+    .await
+    {
+        Ok(instance) => {
+            tracing::info!(
+                key = %workflow_key,
+                instance_id = %instance.id,
+                tenant_id = %tenant_id,
+                risk_level = %risk_level,
+                "AI 行动工作流实例已启动"
+            );
+        }
+        Err(e) => {
+            tracing::warn!(
+                key = %workflow_key,
+                error = %e,
+                "AI 行动工作流实例启动失败"
+            );
+        }
+    }
+}
+
 impl Default for WorkflowModule {
     fn default() -> Self {
         Self::new()
@@ -286,6 +400,27 @@ impl ErpModule for WorkflowModule {
         });
 
         tracing::info!(module = "workflow", "Workflow 事件处理器已注册（监听 user.deleted）");
+
+        // 订阅 AI 行动工作流启动请求
+        let (mut ai_rx, _ai_handle) = bus.subscribe_filtered("workflow.ai_action.".to_string());
+        let ai_db = ctx.db.clone();
+        let ai_bus = bus.clone();
+
+        tokio::spawn(async move {
+            loop {
+                match ai_rx.recv().await {
+                    Some(event) if event.event_type == "workflow.ai_action.start_requested" => {
+                        handle_ai_action_start(&ai_db, &ai_bus, &event).await;
+                    }
+                    Some(_) => {}
+                    None => {
+                        tracing::info!("AI 行动工作流事件订阅通道已关闭");
+                        break;
+                    }
+                }
+            }
+        });
+
         Ok(())
     }
 

crates/erp-workflow/src/service/ai_workflow_seed.rs

@@ -0,0 +1,199 @@
+//! AI 行动闭环 BPMN 流程定义种子数据
+//!
+//! 三条流程：
+//! - ai_followup_workflow — AI 随访建议审批
+//! - ai_appointment_workflow — AI 预约建议审批
+//! - ai_alert_workflow — AI 预警确认
+
+use chrono::Utc;
+use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
+use uuid::Uuid;
+
+use crate::entity::process_definition;
+
+/// AI 随访审批流程
+///
+/// ```text
+/// Start → ExclusiveGateway(风险分级)
+///   → [low]    → End (自动执行，由分发器直接处理)
+///   → [medium] → UserTask(医生审批) → ExclusiveGateway → [approved] → End
+///                                                         → [rejected] → End
+///   → [high]   → UserTask(紧急确认) → ExclusiveGateway → [approved] → End
+///                                                         → [rejected] → End
+/// ```
+fn followup_nodes() -> Vec<serde_json::Value> {
+    serde_json::from_value(serde_json::json!([
+        {"id": "start", "type": "StartEvent", "name": "AI 随访建议"},
+        {"id": "gw_risk", "type": "ExclusiveGateway", "name": "风险分级"},
+        {"id": "end_auto", "type": "EndEvent", "name": "自动完成"},
+        {"id": "doctor_review", "type": "UserTask", "name": "医生审批随访建议",
+         "candidate_groups": ["doctor"]},
+        {"id": "gw_outcome", "type": "ExclusiveGateway", "name": "审批结果"},
+        {"id": "end_approved", "type": "EndEvent", "name": "已批准"},
+        {"id": "end_rejected", "type": "EndEvent", "name": "已拒绝"}
+    ])).unwrap()
+}
+
+fn followup_edges() -> Vec<serde_json::Value> {
+    serde_json::from_value(serde_json::json!([
+        {"id": "e1", "source": "start", "target": "gw_risk"},
+        {"id": "e2", "source": "gw_risk", "target": "end_auto",
+         "condition": "risk_level == \"low\"", "label": "低风险"},
+        {"id": "e3", "source": "gw_risk", "target": "doctor_review",
+         "label": "中/高风险"},
+        {"id": "e4", "source": "doctor_review", "target": "gw_outcome"},
+        {"id": "e5", "source": "gw_outcome", "target": "end_approved",
+         "condition": "outcome == \"approved\"", "label": "批准"},
+        {"id": "e6", "source": "gw_outcome", "target": "end_rejected",
+         "condition": "outcome == \"rejected\"", "label": "拒绝"}
+    ])).unwrap()
+}
+
+/// AI 预约审批流程
+fn appointment_nodes() -> Vec<serde_json::Value> {
+    serde_json::from_value(serde_json::json!([
+        {"id": "start", "type": "StartEvent", "name": "AI 预约建议"},
+        {"id": "gw_risk", "type": "ExclusiveGateway", "name": "风险分级"},
+        {"id": "end_auto", "type": "EndEvent", "name": "自动完成"},
+        {"id": "doctor_confirm", "type": "UserTask", "name": "医生确认预约建议",
+         "candidate_groups": ["doctor"]},
+        {"id": "gw_outcome", "type": "ExclusiveGateway", "name": "确认结果"},
+        {"id": "end_approved", "type": "EndEvent", "name": "已确认"},
+        {"id": "end_rejected", "type": "EndEvent", "name": "已拒绝"}
+    ])).unwrap()
+}
+
+fn appointment_edges() -> Vec<serde_json::Value> {
+    serde_json::from_value(serde_json::json!([
+        {"id": "e1", "source": "start", "target": "gw_risk"},
+        {"id": "e2", "source": "gw_risk", "target": "end_auto",
+         "condition": "risk_level == \"low\"", "label": "低风险"},
+        {"id": "e3", "source": "gw_risk", "target": "doctor_confirm",
+         "label": "中/高风险"},
+        {"id": "e4", "source": "doctor_confirm", "target": "gw_outcome"},
+        {"id": "e5", "source": "gw_outcome", "target": "end_approved",
+         "condition": "outcome == \"approved\"", "label": "确认"},
+        {"id": "e6", "source": "gw_outcome", "target": "end_rejected",
+         "condition": "outcome == \"rejected\"", "label": "拒绝"}
+    ])).unwrap()
+}
+
+/// AI 预警确认流程
+fn alert_nodes() -> Vec<serde_json::Value> {
+    serde_json::from_value(serde_json::json!([
+        {"id": "start", "type": "StartEvent", "name": "AI 预警"},
+        {"id": "gw_risk", "type": "ExclusiveGateway", "name": "风险分级"},
+        {"id": "end_auto", "type": "EndEvent", "name": "已发送"},
+        {"id": "doctor_ack", "type": "UserTask", "name": "医生确认预警",
+         "candidate_groups": ["doctor"]},
+        {"id": "gw_outcome", "type": "ExclusiveGateway", "name": "确认结果"},
+        {"id": "end_acknowledged", "type": "EndEvent", "name": "已确认"},
+        {"id": "end_escalated", "type": "EndEvent", "name": "已升级"}
+    ])).unwrap()
+}
+
+fn alert_edges() -> Vec<serde_json::Value> {
+    serde_json::from_value(serde_json::json!([
+        {"id": "e1", "source": "start", "target": "gw_risk"},
+        {"id": "e2", "source": "gw_risk", "target": "end_auto",
+         "condition": "risk_level == \"low\"", "label": "低风险"},
+        {"id": "e3", "source": "gw_risk", "target": "doctor_ack",
+         "label": "中/高风险"},
+        {"id": "e4", "source": "doctor_ack", "target": "gw_outcome"},
+        {"id": "e5", "source": "gw_outcome", "target": "end_acknowledged",
+         "condition": "outcome == \"approved\"", "label": "确认"},
+        {"id": "e6", "source": "gw_outcome", "target": "end_escalated",
+         "condition": "outcome == \"rejected\"", "label": "升级"}
+    ])).unwrap()
+}
+
+struct WorkflowTemplate {
+    key: &'static str,
+    name: &'static str,
+    category: &'static str,
+    description: &'static str,
+    nodes: Vec<serde_json::Value>,
+    edges: Vec<serde_json::Value>,
+}
+
+fn all_templates() -> Vec<WorkflowTemplate> {
+    vec![
+        WorkflowTemplate {
+            key: "ai_followup_workflow",
+            name: "AI 随访建议审批",
+            category: "ai_action",
+            description: "AI 分析生成的随访建议，按风险等级自动执行或提交医生审批",
+            nodes: followup_nodes(),
+            edges: followup_edges(),
+        },
+        WorkflowTemplate {
+            key: "ai_appointment_workflow",
+            name: "AI 预约建议审批",
+            category: "ai_action",
+            description: "AI 分析生成的预约建议，按风险等级自动执行或提交医生确认",
+            nodes: appointment_nodes(),
+            edges: appointment_edges(),
+        },
+        WorkflowTemplate {
+            key: "ai_alert_workflow",
+            name: "AI 预警确认",
+            category: "ai_action",
+            description: "AI 分析生成的预警通知，按风险等级自动发送或提交医生确认",
+            nodes: alert_nodes(),
+            edges: alert_edges(),
+        },
+    ]
+}
+
+/// 确保 AI 行动闭环的工作流定义存在（幂等）。
+///
+/// 对每个 tenant_id 检查 key 是否已存在，不存在则创建并发布。
+pub async fn ensure_ai_workflows(
+    db: &sea_orm::DatabaseConnection,
+    tenant_id: Uuid,
+) -> Result<(), sea_orm::DbErr> {
+    let system_id = Uuid::parse_str("00000000-0000-0000-0000-000000000000").unwrap();
+
+    for tmpl in all_templates() {
+        let exists = process_definition::Entity::find()
+            .filter(process_definition::Column::TenantId.eq(tenant_id))
+            .filter(process_definition::Column::Key.eq(tmpl.key))
+            .filter(process_definition::Column::DeletedAt.is_null())
+            .one(db)
+            .await?
+            .is_some();
+
+        if exists {
+            continue;
+        }
+
+        let now = Utc::now();
+        let id = Uuid::now_v7();
+
+        let active = process_definition::ActiveModel {
+            id: Set(id),
+            tenant_id: Set(tenant_id),
+            name: Set(tmpl.name.to_string()),
+            key: Set(tmpl.key.to_string()),
+            version: Set(1),
+            category: Set(Some(tmpl.category.to_string())),
+            description: Set(Some(tmpl.description.to_string())),
+            nodes: Set(serde_json::json!(tmpl.nodes)),
+            edges: Set(serde_json::json!(tmpl.edges)),
+            status: Set("published".to_string()),
+            created_at: Set(now),
+            updated_at: Set(now),
+            created_by: Set(system_id),
+            updated_by: Set(system_id),
+            deleted_at: Set(None),
+            version_field: Set(1),
+        };
+        active.insert(db).await?;
+        tracing::info!(
+            key = %tmpl.key,
+            tenant_id = %tenant_id,
+            "AI 工作流定义已创建"
+        );
+    }
+    Ok(())
+}

crates/erp-workflow/src/service/mod.rs

@@ -1,3 +1,4 @@
+pub mod ai_workflow_seed;
 pub mod definition_service;
 pub mod instance_service;
 pub mod task_service;