lxy
f736f435bc
持久化(P1-c):新建 usePersistedRef composable,Tasks/AuditLog/ProjectDetail 等接入 localStorage
AuditLog(P1-d):后端 list_tool_executions 加 WHERE 筛选+返 {items,total,has_more},前端对接+长列折叠+筛选持久化
数据源解耦(P1-g):ProjectDetail projectTasks 按 project_id 独立加载 + ChatInput @项目联想独立加载(不读 store.tasks 当前页)
GitChanges(12a):后端 get_module_commits 加 git rev-list --count 返 total,前端显真实总数
原12大改:Dashboard 统计卡压底行(1)/Projects 列表卡片视图(2)/project_event 埋点排序(3)/TaskDetail 重设计(4)/IdeaDetail 重设计(5)/KnowledgeDetail 重设计(6)/界面持久化+侧栏Ctrl+B+审批数字键(7)/ProjectDetail 三栏改两栏(10)
P2打磨:文件浏览器(FileTree去重/FilePreview行号.md Diff/selectedFilePath归位)/settings反馈(假保存/端口校验)/AI会话(try-catch/scrollIntoView)/后端计数(move_queue事件/timeline total/workflow分页/import_batch分块)/杂项(TopBar/ConfirmDialog键盘/CIStatus i18n/ToolResultBody/ModuleNode/ApprovalDialog全选)
580 lines
26 KiB
Rust
580 lines
26 KiB
Rust
//! 工作流相关命令 — 触发执行、查询执行记录
|
||
//!
|
||
//! 事件转发：run_workflow 在执行前订阅事件总线，
|
||
//! 将 WorkflowEvent 包装为 `{ execution_id, event }` 通过 `workflow-event` 事件发给前端。
|
||
|
||
use serde::Serialize;
|
||
use tauri::{AppHandle, Emitter, State};
|
||
use tokio::sync::broadcast::error::RecvError;
|
||
|
||
use df_types::events::{WorkflowEvent, HumanApprovalResponse, SelectType};
|
||
use df_types::types::{new_id, ExecutionId, NodeStatus};
|
||
use df_nodes::task_advance_node::advance_task_atomic;
|
||
// F-260616-06 ②-4: 工作流失败时按 target_status 推算任务退回态。
|
||
// regression_target 已收敛到 df-nodes::task_state_machine(状态机业务契约,单一可信源),
|
||
// 不再在 app crate 维护私有副本(防 DRY 漂移)。
|
||
use df_nodes::task_state_machine::regression_target;
|
||
// F-260616-06 ①-1: 空 dag + target_status 时按目标态自动选推进链模板
|
||
use df_nodes::task_workflow_templates::template_for;
|
||
use df_storage::crud::{TaskRepo, WorkflowRepo};
|
||
use df_storage::models::WorkflowRecord;
|
||
use df_workflow::dag_def::DagDef;
|
||
use df_workflow::executor::DagExecutor;
|
||
|
||
use crate::state::AppState;
|
||
|
||
use super::{err_str, now_millis};
|
||
|
||
/// 转发到前端的事件载荷
|
||
#[derive(Debug, Clone, Serialize)]
|
||
struct WorkflowEventPayload {
|
||
/// 工作流执行记录 ID
|
||
execution_id: ExecutionId,
|
||
/// 原始工作流事件（serde tag = "type"）
|
||
event: WorkflowEvent,
|
||
}
|
||
|
||
/// Lagged 兜底：查 DB 终态，命中返回 `(要补发的事件, status)`；未命中返回 None。
|
||
///
|
||
/// broadcast `Lagged` 不暴露被丢事件类型，关键终态事件（WorkflowCompleted/Failed）可能已丢，
|
||
/// forward 循环会永久等不到。达阈值时查 DB 终态补发。三种"未命中"（仍 running / 无记录 /
|
||
/// 查询失败）统一返回 None 并各自 warn，主循环据此重置 `lagged_total` 继续等。
|
||
async fn probe_lagged_terminal(
|
||
db: &df_storage::db::Database,
|
||
exec_id: &ExecutionId,
|
||
) -> Option<(WorkflowEvent, String)> {
|
||
let record = match WorkflowRepo::new(db).get_by_id(exec_id).await {
|
||
Ok(Some(r)) => r,
|
||
Ok(None) => {
|
||
tracing::warn!(
|
||
execution_id = %exec_id,
|
||
"Lagged 兜底查询未找到执行记录，继续等待事件"
|
||
);
|
||
return None;
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
execution_id = %exec_id,
|
||
error = %e,
|
||
"Lagged 兜底查询 DB 失败，继续等待事件"
|
||
);
|
||
return None;
|
||
}
|
||
};
|
||
let event = match record.status.as_str() {
|
||
"completed" => WorkflowEvent::WorkflowCompleted {
|
||
execution_id: exec_id.clone(),
|
||
total_duration_ms: 0,
|
||
},
|
||
"failed" => WorkflowEvent::WorkflowFailed {
|
||
execution_id: exec_id.clone(),
|
||
error: "工作流执行失败（Lagged 兜底补发，详情见 DB）".to_string(),
|
||
failed_node: String::new(),
|
||
},
|
||
"cancelled" => WorkflowEvent::WorkflowFailed {
|
||
execution_id: exec_id.clone(),
|
||
error: "工作流被取消（Lagged 兜底补发）".to_string(),
|
||
failed_node: String::new(),
|
||
},
|
||
_ => return None,
|
||
};
|
||
Some((event, record.status))
|
||
}
|
||
|
||
/// 触发工作流执行（核心命令）
|
||
///
|
||
/// 流程：build_dag 校验 → 写入执行记录(status=running) → 后台异步执行 →
|
||
/// 事件经 EventBus 转发到前端 → 完成后更新执行记录状态。
|
||
/// 立即返回执行记录 ID，前端凭此关联后续事件。
|
||
///
|
||
/// F-260616-06 阶段2 工作流联动(②-2):
|
||
/// - `task_id` / `target_status` 同时 Some 时,完成后按 target_status 推进任务(②-3),
|
||
/// 失败时按 regression_target 退回(②-4)。
|
||
/// - Option 向后兼容:旧调用方不传即 None,不触发任何任务回调,零行为破坏。
|
||
#[tauri::command]
|
||
pub async fn run_workflow(
|
||
app: AppHandle,
|
||
state: State<'_, AppState>,
|
||
name: String,
|
||
dag: DagDef,
|
||
config: serde_json::Value,
|
||
// F-260616-06 ②-2: 工作流联动任务的关联 ID 与目标态(同时 Some 才联动)
|
||
task_id: Option<String>,
|
||
target_status: Option<String>,
|
||
) -> Result<String, String> {
|
||
// 瘦转发到 run_workflow_inner(B-260617-01:抽取共用核心供 ai_approve 经
|
||
// execute_run_workflow_for_tool 直接调,无需构造 tauri::State)。
|
||
// CR-52: 命令层(前端 invoke run_workflow)是人工手动触发 → triggered_by="manual"。
|
||
run_workflow_inner(&app, &state, name, dag, config, task_id, target_status, "manual").await
|
||
}
|
||
|
||
/// 工作流执行核心逻辑(B-260617-01 抽取,命令层瘦转发)。
|
||
///
|
||
/// 原 `run_workflow` 命令体拆出,使 ai_approve 经 execute_run_workflow_for_tool 直接调用
|
||
/// (持 &AppState 而非 tauri::State,绕开 tauri::State 在非命令上下文不可构造的限制)。
|
||
/// 语义与原命令完全一致:模板选 DAG → 落执行记录 → spawn 事件转发 + DAG 执行 + 任务联动推进。
|
||
///
|
||
/// `triggered_by` 落入 WorkflowRecord.triggered_by,区分触发来源:
|
||
/// - 命令层 run_workflow(前端 invoke,人工手动)传 "manual"
|
||
/// - execute_run_workflow_for_tool(AI 工具经 ai_approve 审批后执行)传 "ai"
|
||
/// CR-52:原硬编码 "manual" 致 AI 路径误标,改由调用方透传。
|
||
pub async fn run_workflow_inner(
|
||
app: &AppHandle,
|
||
state: &AppState,
|
||
name: String,
|
||
dag: DagDef,
|
||
config: serde_json::Value,
|
||
task_id: Option<String>,
|
||
target_status: Option<String>,
|
||
triggered_by: &str,
|
||
) -> Result<String, String> {
|
||
// 0. F-260616-06 ①-1: DagDef 来源选模板(方案A·最小)。
|
||
// 规则:
|
||
// - dag.nodes 非空 → 用传入 dag(向后兼容,旧调用方/编辑器自定义工作流不变)
|
||
// - dag.nodes 为空 + target_status Some → 调 template_for 自动选推进链模板;
|
||
// 模板不存在(target 非三合法态) → Err「无对应工作流模板」
|
||
// - dag.nodes 为空 + target_status None → 原行为(空 dag 让 build_dag 自行报错,零新分支)
|
||
// 前端选 target_status 后传空 dag,零模板知识零新 IPC,后端收敛选模板逻辑。
|
||
let dag = if dag.nodes.is_empty() {
|
||
match target_status.as_deref() {
|
||
Some(target) => template_for(target)
|
||
.ok_or_else(|| format!("无对应工作流模板: target_status={}", target))?,
|
||
None => dag, // 空dag+无target: 走原路径由 build_dag 校验
|
||
}
|
||
} else {
|
||
dag
|
||
};
|
||
|
||
// 1. 先构建运行时 DAG，校验失败直接返回，不落库
|
||
let runtime_dag = state.registry.build_dag(&dag).map_err(err_str)?;
|
||
|
||
// 2. 写入执行记录（status=running）
|
||
let execution_id: ExecutionId = new_id().into();
|
||
let dag_json = serde_json::to_string(&dag).map_err(err_str)?;
|
||
let record = WorkflowRecord {
|
||
id: execution_id.to_string(),
|
||
name,
|
||
dag_json,
|
||
status: "running".to_string(),
|
||
// CR-52: triggered_by 由调用方透传(命令层 run_workflow="manual",
|
||
// execute_run_workflow_for_tool="ai"),不再硬编码 "manual" 致 AI 路径误标。
|
||
triggered_by: Some(triggered_by.to_string()),
|
||
project_id: None,
|
||
task_id: task_id.clone(),
|
||
created_at: now_millis(),
|
||
completed_at: None,
|
||
};
|
||
state
|
||
.workflows
|
||
.insert(record)
|
||
.await
|
||
.map_err(err_str)?;
|
||
|
||
// 3. 订阅事件总线，把事件转发给前端（收到完成/失败事件后退出）
|
||
let mut rx = state.event_bus.subscribe();
|
||
let forward_app = app.clone();
|
||
let forward_exec_id = execution_id.clone();
|
||
// B-260615-35: forward 任务内需查 DB 终态兜底 Lagged 丢终态事件，故 move 进 db 句柄
|
||
let forward_db = state.db.clone();
|
||
tauri::async_runtime::spawn(async move {
|
||
// Lagged 累计次数（用于"累计达阈值"触发兜底查询）
|
||
let mut lagged_total: u64 = 0;
|
||
// Lagged 单次/累计触发 DB 终态查询兜底的阈值
|
||
const LAGGED_PROBE_THRESHOLD: u64 = 8;
|
||
loop {
|
||
match rx.recv().await {
|
||
Ok(event) => {
|
||
// B-03b-R10 ③(波17 治本): 完成判定加 execution_id 匹配。
|
||
// AppState.event_bus 是全局单例,并发工作流各自订阅同一总线;
|
||
// 此前 matches! 只看变体不看 exec_id,任一工作流的 WorkflowCompleted/Failed
|
||
// 都会触发本 forward 循环 break,过早断他人链(并发串扰根因)。
|
||
// 现在按变体+exec_id 双重匹配:只处理属于本 forward_exec_id 的终态事件,
|
||
// 他人终态事件忽略(emit 给前端后继续循环),不再误 break。
|
||
// execution_id 字段 #[serde(default)] 向后兼容老事件(空串):
|
||
// 自家终态事件本就带正确 exec_id;空串匹配只可能漏(自家老事件),不会误 break。
|
||
let finished = matches!(
|
||
&event,
|
||
WorkflowEvent::WorkflowCompleted { execution_id, .. }
|
||
| WorkflowEvent::WorkflowFailed { execution_id, .. }
|
||
// execution_id: WorkflowEvent 中反序列化的 ExecutionId(#[serde(default)]),
|
||
// forward_exec_id: 本 forward 循环所属的 ExecutionId,同类型直接比较
|
||
if execution_id == &forward_exec_id
|
||
);
|
||
let payload = WorkflowEventPayload {
|
||
execution_id: forward_exec_id.clone(),
|
||
event,
|
||
};
|
||
if let Err(e) = forward_app.emit("workflow-event", &payload) {
|
||
tracing::warn!("工作流事件转发失败: {}", e);
|
||
}
|
||
if finished {
|
||
break;
|
||
}
|
||
}
|
||
// R-PD-13: Lagged 表示消费过慢，broadcast 滑动窗口丢 n 条最旧事件。
|
||
// broadcast 不暴露被丢事件的具体类型，无法在此精确判断是否为关键终态事件。
|
||
// 风险：若关键终态事件（WorkflowCompleted/WorkflowFailed）被丢，finished 判定不触发，
|
||
// 循环不会 break，前端永久收不到工作流结束信号。
|
||
// B-260615-35 兜底：收到 Lagged 时，累计次数或单次 n 达阈值则查 DB 终态，
|
||
// 若工作流已终态则补发对应 workflow-event 并 break 退出 forward。
|
||
// 避免依赖实时性差的 DB 轮询兜底导致前端审批/完成弹窗卡死。
|
||
Err(RecvError::Lagged(n)) => {
|
||
lagged_total = lagged_total.saturating_add(n);
|
||
tracing::warn!(
|
||
execution_id = %forward_exec_id,
|
||
lagged = n,
|
||
lagged_total = lagged_total,
|
||
"工作流事件消费滞后，丢失 {} 条事件（broadcast 不暴露被丢事件类型，关键终态事件可能已丢失）",
|
||
n
|
||
);
|
||
if lagged_total < LAGGED_PROBE_THRESHOLD {
|
||
continue;
|
||
}
|
||
// 累计/单次达阈值：查 DB 终态兜底（probe_lagged_terminal 收口三种未命中情况）
|
||
match probe_lagged_terminal(&forward_db, &forward_exec_id).await {
|
||
Some((synth_event, status)) => {
|
||
let payload = WorkflowEventPayload {
|
||
execution_id: forward_exec_id.clone(),
|
||
event: synth_event.clone(),
|
||
};
|
||
if let Err(e) = forward_app.emit("workflow-event", &payload) {
|
||
tracing::warn!("Lagged 终态兜底事件转发失败: {}", e);
|
||
}
|
||
tracing::info!(
|
||
execution_id = %forward_exec_id,
|
||
status = %status,
|
||
"Lagged 兜底命中终态，补发 {} 并退出 forward",
|
||
match synth_event {
|
||
WorkflowEvent::WorkflowCompleted { .. } => "WorkflowCompleted",
|
||
_ => "WorkflowFailed",
|
||
}
|
||
);
|
||
break;
|
||
}
|
||
None => {
|
||
// 未命中(仍 running / 无记录 / 查询失败):重置累计计数继续等待
|
||
lagged_total = 0;
|
||
}
|
||
}
|
||
}
|
||
Err(RecvError::Closed) => break,
|
||
}
|
||
}
|
||
});
|
||
|
||
// 4. 后台异步执行 DAG，完成后更新执行记录状态
|
||
let event_bus = state.event_bus.clone();
|
||
let db = state.db.clone();
|
||
let exec_id = execution_id.clone();
|
||
let state_registry = state.workflow_state_registry.clone();
|
||
// F-260616-06 ②-2: move task_id / target_status 进闭包供完成/失败回调使用
|
||
let cb_task_id = task_id.clone();
|
||
let cb_target_status = target_status.clone();
|
||
// panic-guard:执行 spawn 闭包最外层包 catch_unwind(AssertUnwindSafe(..))。
|
||
// panic 时走已存在 failed 分支(update status=failed + emit WorkflowFailed + state_registry.remove),
|
||
// 防 panic 后终态不发/registry 不清永久卡。主逻辑/调度零改动,仅在 panic 时补发终态。
|
||
// 注:依赖 panic=unwind(tokio 默认),若构建设 panic=abort 则 catch_unwind 不生效(尽力兜底)。
|
||
tauri::async_runtime::spawn({
|
||
// panic 兜底需要这些引用:move 进 catch_unwind 闭包前各取一份 clone 供 panic 分支使用。
|
||
let panic_exec_id = exec_id.clone();
|
||
let panic_db = db.clone();
|
||
let panic_event_bus = event_bus.clone();
|
||
let panic_registry = state_registry.clone();
|
||
async move {
|
||
use std::panic::AssertUnwindSafe;
|
||
use futures::FutureExt;
|
||
// 原闭包主体包进 catch_unwind:panic 以 Err(Box<dyn Any>) 返回,Ok 走正常路径。
|
||
let outcome = AssertUnwindSafe(async move {
|
||
let mut executor = DagExecutor::new(event_bus.clone(), exec_id.clone());
|
||
// 注册执行器状态机：StateMachine 内部 Arc<Mutex>，clone 共享底层 HashMap，
|
||
// cancel_workflow_node IPC 经 execution_id 取此引用 set_cancelled，直达运行中 HumanNode
|
||
state_registry
|
||
.lock()
|
||
.await
|
||
.insert(exec_id.clone(), executor.state_machine());
|
||
let result = executor.run(&runtime_dag, config).await;
|
||
|
||
let (status, error) = match &result {
|
||
Ok(_) => ("completed", None),
|
||
Err(e) => ("failed", Some(format!("{:#}", e))),
|
||
};
|
||
|
||
// 更新执行记录（Repo 内部仅持有 Arc 连接，重建开销可忽略）
|
||
let workflows = WorkflowRepo::new(&db);
|
||
if let Err(e) = workflows.update_field(&exec_id, "status", status).await {
|
||
tracing::error!("更新工作流状态失败: {}", e);
|
||
}
|
||
if let Err(e) = workflows
|
||
.update_field(&exec_id, "completed_at", &now_millis())
|
||
.await
|
||
{
|
||
tracing::error!("更新工作流完成时间失败: {}", e);
|
||
}
|
||
|
||
// F-260616-06 ②-3 / ②-4: 工作流联动任务推进回调
|
||
// - 成功(completed): task_id 与 target_status 都 Some 时推进到 target_status
|
||
// - 失败(failed): 按 regression_target 推算退回态推进;无映射则跳过
|
||
// task_id / target_status 任一 None → 跳过(向后兼容,旧工作流不联动任务)
|
||
// 回调失败仅 tracing::warn!,不回滚(工作流成功语义与任务推进解耦 — 回调失败不撤销
|
||
// 已完成工作流,前端可后续手动处理)
|
||
match (cb_task_id.as_ref(), cb_target_status.as_deref()) {
|
||
(Some(tid), Some(target)) => {
|
||
let advance_target = match status {
|
||
"completed" => Some(target),
|
||
"failed" => regression_target(target),
|
||
_ => None,
|
||
};
|
||
if let Some(to) = advance_target {
|
||
let repo = TaskRepo::new(&db);
|
||
if let Err(e) = advance_task_atomic(&repo, tid, to).await {
|
||
tracing::warn!(
|
||
execution_id = %exec_id,
|
||
task_id = %tid,
|
||
target_status = %to,
|
||
workflow_status = %status,
|
||
"工作流联动推进任务失败(不回滚): {}",
|
||
e
|
||
);
|
||
} else {
|
||
tracing::info!(
|
||
execution_id = %exec_id,
|
||
task_id = %tid,
|
||
target_status = %to,
|
||
workflow_status = %status,
|
||
"工作流联动推进任务成功"
|
||
);
|
||
}
|
||
} else {
|
||
// failed 且无退回映射:warn 跳过(常见:target 是 todo 起点无可退态)
|
||
tracing::warn!(
|
||
execution_id = %exec_id,
|
||
task_id = %tid,
|
||
target_status = %target,
|
||
"工作流失败但 target_status 无退回映射,跳过任务回调"
|
||
);
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
|
||
// 执行结束（成功/失败）清理状态注册表，防内存泄漏
|
||
state_registry.lock().await.remove(&exec_id);
|
||
|
||
// 执行失败时补发 WorkflowFailed（执行器内部只发 NodeFailed）
|
||
// failed_node 从状态机取首个失败/取消节点(R9⑪: 原 String::new() 空值)
|
||
if let Some(error) = error {
|
||
let failed_node = executor
|
||
.state_machine()
|
||
.snapshot()
|
||
.into_iter()
|
||
.find(|(_, s)| matches!(s, NodeStatus::Failed | NodeStatus::Cancelled))
|
||
.map(|(id, _)| id)
|
||
.unwrap_or_default();
|
||
event_bus
|
||
.send(WorkflowEvent::WorkflowFailed {
|
||
execution_id: exec_id.clone(),
|
||
error,
|
||
failed_node,
|
||
})
|
||
.await;
|
||
}
|
||
}) // 闭 catch_unwind 内 async move
|
||
.catch_unwind()
|
||
.await;
|
||
|
||
// panic 分支:catch_unwind 返回 Err(Box<dyn Any + Send>),走已存在 failed 清理路径。
|
||
match outcome {
|
||
Ok(_) => {}
|
||
Err(panic_payload) => {
|
||
// 提取 panic 消息(String/&'static str 常见,其他类型用兜底文案)
|
||
let msg = panic_payload
|
||
.downcast_ref::<String>()
|
||
.map(|s| s.clone())
|
||
.or_else(|| panic_payload.downcast_ref::<&'static str>().map(|s| s.to_string()))
|
||
.unwrap_or_else(|| "工作流执行 panic".to_string());
|
||
tracing::error!(
|
||
execution_id = %panic_exec_id,
|
||
"工作流执行 panic,走 failed 分支兜底: {}",
|
||
msg
|
||
);
|
||
let workflows = WorkflowRepo::new(&panic_db);
|
||
if let Err(e) = workflows.update_field(&panic_exec_id, "status", "failed").await {
|
||
tracing::error!("更新工作流状态失败(panic 兜底): {}", e);
|
||
}
|
||
if let Err(e) = workflows
|
||
.update_field(&panic_exec_id, "completed_at", &now_millis())
|
||
.await
|
||
{
|
||
tracing::error!("更新工作流完成时间失败(panic 兜底): {}", e);
|
||
}
|
||
panic_registry.lock().await.remove(&panic_exec_id);
|
||
panic_event_bus
|
||
.send(WorkflowEvent::WorkflowFailed {
|
||
execution_id: panic_exec_id.clone(),
|
||
error: msg,
|
||
failed_node: String::new(),
|
||
})
|
||
.await;
|
||
}
|
||
}
|
||
}
|
||
});
|
||
|
||
Ok(execution_id.to_string())
|
||
}
|
||
|
||
/// 列出工作流执行记录(最近 N 条,对标设计 §10.5 单用户桌面应用分页)。
|
||
///
|
||
/// M19:原实现调 `list_all()` 无 limit/分页,工作流执行记录随使用累积(workflow_executions 表
|
||
/// 只插不删,每次 run_workflow 落一条),长期使用后全量返回会撑爆前端列表 + IPC 传输 + 内存。
|
||
/// 加可选 `limit` 参数(默认 100,钳制 ≤ 500),list_all 已按 created_at DESC 排序,截断取最近 N 条。
|
||
/// 前端旧调用方不传 limit 走默认 100,零破坏(工作流历史本就按时间倒序展示,截断尾部老记录无感)。
|
||
#[tauri::command]
|
||
pub async fn list_workflow_executions(
|
||
state: State<'_, AppState>,
|
||
limit: Option<u32>,
|
||
) -> Result<Vec<WorkflowRecord>, String> {
|
||
let safe_limit = limit.unwrap_or(100).min(500) as usize;
|
||
let mut records = state.workflows.list_all().await.map_err(err_str)?;
|
||
if records.len() > safe_limit {
|
||
records.truncate(safe_limit);
|
||
}
|
||
Ok(records)
|
||
}
|
||
|
||
/// 按 ID 查询工作流执行记录
|
||
#[tauri::command]
|
||
pub async fn get_workflow_execution(
|
||
state: State<'_, AppState>,
|
||
id: String,
|
||
) -> Result<Option<WorkflowRecord>, String> {
|
||
state
|
||
.workflows
|
||
.get_by_id(&id)
|
||
.await
|
||
.map_err(err_str)
|
||
}
|
||
|
||
/// 发送人工审批响应
|
||
///
|
||
/// F-260615-01: 支持 single/multiple。
|
||
/// - single: decision 单值,decisions 为空(向后兼容);
|
||
/// - multiple: decisions 数组(≥1 项),decision 留空。
|
||
/// IPC 早校验规则与 HumanNode 下游兜底一致,避免「IPC 成功+工作流失败」割裂。
|
||
#[tauri::command]
|
||
pub async fn approve_human_approval(
|
||
app: AppHandle,
|
||
state: State<'_, AppState>,
|
||
execution_id: ExecutionId,
|
||
node_id: String,
|
||
decision: String,
|
||
// F-260615-01: 多选结果数组,缺省空数组(单选调用方不传)
|
||
decisions: Option<Vec<String>>,
|
||
comment: Option<String>,
|
||
// R-PD-5: options 由前端从收到的 HumanApprovalRequest 事件透传（IPC 无法访问节点 config）。
|
||
options: Vec<String>,
|
||
// F-260615-01: 选择类型,缺省 single
|
||
select_type: Option<String>,
|
||
) -> Result<(), String> {
|
||
let _ = &app; // 原签名含 app 参数(未使用),保留避免 invoke_handler 注册签名变化
|
||
let select_type = match select_type.as_deref() {
|
||
Some("multiple") => SelectType::Multiple,
|
||
_ => SelectType::Single,
|
||
};
|
||
let decisions = decisions.unwrap_or_default();
|
||
|
||
// 归一化决策集合(与 HumanNode 下游一致)
|
||
let mut picked: Vec<String> = decisions;
|
||
if picked.is_empty() && !decision.is_empty() {
|
||
picked.push(decision.clone());
|
||
}
|
||
|
||
// 数量校验
|
||
match select_type {
|
||
SelectType::Single if picked.len() != 1 => {
|
||
return Err("单选审批只能提交一个决策".to_string());
|
||
}
|
||
SelectType::Multiple if picked.is_empty() => {
|
||
return Err("多选审批至少提交一个决策".to_string());
|
||
}
|
||
_ => {}
|
||
}
|
||
|
||
// 每项非空 + ∈options(options 非空时)
|
||
for d in &picked {
|
||
if d.trim().is_empty() {
|
||
return Err("审批决策不能为空".to_string());
|
||
}
|
||
if !options.is_empty() && !options.contains(d) {
|
||
return Err(format!("审批决策非法: {}", d));
|
||
}
|
||
}
|
||
|
||
// 发送审批响应到事件总线(decision 取首项保留,decisions 透传全量)
|
||
let primary = picked.first().cloned().unwrap_or_default();
|
||
let response = HumanApprovalResponse {
|
||
execution_id: execution_id.clone(),
|
||
node_id: node_id.clone(),
|
||
decision: primary,
|
||
decisions: picked,
|
||
comment,
|
||
};
|
||
|
||
let event = WorkflowEvent::HumanApprovalResponse {
|
||
execution_id: response.execution_id,
|
||
node_id: response.node_id,
|
||
decision: response.decision,
|
||
decisions: response.decisions,
|
||
comment: response.comment,
|
||
};
|
||
|
||
// 直接发送到全局事件总线
|
||
state.event_bus.send(event).await;
|
||
Ok(())
|
||
}
|
||
|
||
/// 取消工作流节点（人工审批取消）
|
||
///
|
||
/// 从执行器状态机注册表取出共享引用，调 `set_cancelled` 置目标节点为 Cancelled。
|
||
/// StateMachine 内部 Arc<Mutex> 共享底层 HashMap，IPC 写入直达运行中阻塞节点（HumanNode）
|
||
/// 的 select! 轮询分支，其 is_cancelled 检测到后返回 Err "人工审批被取消"。
|
||
///
|
||
/// 终态前置守卫：仅 Pending（排队中）/ Running（执行中，含阻塞等审批）允许取消；
|
||
/// 终态节点（Completed/Failed/Skipped/Cancelled）返 Err，避免静默覆盖终态。
|
||
/// 守卫放 IPC 层（非 set_cancelled 内）：保留 set_cancelled 作为 executor 内部受控旁路语义，
|
||
/// 让状态变更的合法入口收敛到 IPC 这一道。
|
||
#[tauri::command]
|
||
pub async fn cancel_workflow_node(
|
||
state: State<'_, AppState>,
|
||
execution_id: ExecutionId,
|
||
node_id: String,
|
||
) -> Result<(), String> {
|
||
// 经 execution_id 取执行器状态机（与 NodeContext.node_status 共享同一 HashMap）
|
||
let registry = state.workflow_state_registry.lock().await;
|
||
let sm = match registry.get(&execution_id) {
|
||
Some(sm) => sm,
|
||
None => return Err(format!("工作流 {} 不存在或已结束", execution_id)),
|
||
};
|
||
// 终态守卫：get() 返回缺失条目默认 Pending（尚未执行）→ 允许取消
|
||
let current = sm.get(&node_id);
|
||
match current {
|
||
NodeStatus::Pending | NodeStatus::Running | NodeStatus::Waiting => {
|
||
sm.set_cancelled(node_id.clone());
|
||
tracing::info!(
|
||
"工作流节点取消：execution_id={}, node_id={}",
|
||
execution_id,
|
||
node_id
|
||
);
|
||
Ok(())
|
||
}
|
||
// 终态：不允许取消（避免静默覆盖终态）
|
||
NodeStatus::Completed | NodeStatus::Failed | NodeStatus::Skipped | NodeStatus::Cancelled => {
|
||
Err(format!(
|
||
"节点 {} 已终态({})，无法取消",
|
||
node_id,
|
||
current.as_str()
|
||
))
|
||
}
|
||
}
|
||
}
|