修复: unwrap 吞错 + lock 中毒 panic 高危项(数据吞 warn 跳过 / lock 降级返 Err)

crates/df-workflow/src/state.rs

@@ -41,13 +41,21 @@ impl StateMachine {
     }
 
     /// 获取节点状态
+    ///
+    /// 锁中毒时降级返回 `NodeStatus::Pending`（保守默认：视为未启动，
+    /// 执行器不会误判为已完成/失败），并 `tracing::error!` 记录，不 panic。
     pub fn get(&self, node_id: &NodeId) -> NodeStatus {
-        self.states
-            .lock()
-            .expect("状态机锁中毒")
-            .get(node_id)
-            .cloned()
-            .unwrap_or(NodeStatus::Pending)
+        match self.states.lock() {
+            Ok(states) => states.get(node_id).cloned().unwrap_or(NodeStatus::Pending),
+            Err(poisoned) => {
+                tracing::error!(
+                    "状态机锁中毒，get({}) 降级返回 Pending：{}",
+                    node_id,
+                    poisoned
+                );
+                NodeStatus::Pending
+            }
+        }
     }
 
     /// 判断状态转换是否合法
@@ -61,8 +69,23 @@ impl StateMachine {
     }
 
     /// 状态转换 — 校验合法性后更新，非法转换返回错误
+    ///
+    /// 锁中毒时降级返回 `Err`（携带「状态机锁中毒」上下文）并 `tracing::error!` 记录，
+    /// 由调用方决定如何处理（通常是 `set_running`/`set_completed`/`set_failed`
+    /// 把 Err 上抛 → 节点执行流捕获后置 Failed），不 panic 拖垮 runtime。
     pub fn transition(&self, node_id: NodeId, target: NodeStatus) -> anyhow::Result<()> {
-        let mut states = self.states.lock().expect("状态机锁中毒");
+        let mut states = match self.states.lock() {
+            Ok(guard) => guard,
+            Err(poisoned) => {
+                tracing::error!(
+                    "状态机锁中毒，transition({}, {}) 降级返 Err：{}",
+                    node_id,
+                    target.as_str(),
+                    poisoned
+                );
+                anyhow::bail!("状态机锁中毒，节点 {} 状态转换失败", node_id);
+            }
+        };
         let current = states
             .get(&node_id)
             .cloned()
@@ -104,15 +127,35 @@ impl StateMachine {
     ///
     /// 注：原 set_waiting/set_skipped 同为旁路置位但全仓零调用，已删除。
     pub fn set_cancelled(&self, node_id: NodeId) {
-        self.states
-            .lock()
-            .expect("状态机锁中毒")
-            .insert(node_id, NodeStatus::Cancelled);
+        match self.states.lock() {
+            Ok(mut states) => {
+                states.insert(node_id, NodeStatus::Cancelled);
+            }
+            Err(poisoned) => {
+                tracing::error!(
+                    "状态机锁中毒，set_cancelled({}) 降级丢弃取消信号：{}",
+                    node_id,
+                    poisoned
+                );
+            }
+        }
     }
 
     /// 获取所有状态快照（clone 返回，调用方持独立副本）
+    ///
+    /// 锁中毒时降级返回空 HashMap（调用方遍历视为无已完成节点，保守安全），
+    /// 并 `tracing::error!` 记录，不 panic。
     pub fn snapshot(&self) -> HashMap<NodeId, NodeStatus> {
-        self.states.lock().expect("状态机锁中毒").clone()
+        match self.states.lock() {
+            Ok(states) => states.clone(),
+            Err(poisoned) => {
+                tracing::error!(
+                    "状态机锁中毒，snapshot() 降级返回空 HashMap：{}",
+                    poisoned
+                );
+                HashMap::new()
+            }
+        }
     }
 
     /// 检查节点是否被取消