feat: 优化条件不满足时的报错信息展示

1. 新增格式化reason的工具函数统一处理报错信息
2. 支持从条件函数中提取自定义的失败原因
3. 完善NOT和OR条件的失败原因传递逻辑
4. 移除任务跳过的冗余打印输出

src/pyflowx/conditions.py

@@ -211,7 +211,13 @@ class BuiltinConditions:
         """对条件取反."""
 
         def _cond(ctx: Context) -> bool:
-            return not condition(ctx)
+            result = condition(ctx)
+            if result:
+                # inner 为 True 时 NOT 会失败，记录 inner 的具体原因
+                inner_reason = getattr(condition, "_reason", None)
+                if inner_reason is not None:
+                    _cond._reason = inner_reason  # type: ignore[attr-defined]
+            return not result
 
         _cond.__name__ = f"NOT({getattr(condition, '__name__', repr(condition))})"
         return _cond
@@ -232,7 +238,16 @@ class BuiltinConditions:
         """多个条件的逻辑或."""
 
         def _cond(ctx: Context) -> bool:
-            return any(c(ctx) for c in conditions)
+            matched: list[str] = []
+            for c in conditions:
+                if c(ctx):
+                    matched.append(
+                        getattr(c, "_reason", None) or getattr(c, "__name__", repr(c)),
+                    )
+            if matched:
+                _cond._reason = matched  # type: ignore[attr-defined]
+                return True
+            return False
 
         names = [getattr(c, "__name__", repr(c)) for c in conditions]
         _cond.__name__ = f"OR({', '.join(names)})"

src/pyflowx/executors.py

@@ -116,6 +116,13 @@ def _check_upstream_skipped(
     return False, None  # pragma: no cover
 
 
+def _format_reason(reason: Any) -> str:
+    """将 _reason 格式化为可读字符串."""
+    if isinstance(reason, list):
+        return ", ".join(str(r) for r in reason)
+    return str(reason)
+
+
 def _evaluate_conditions(spec: TaskSpec[Any], context: Mapping[str, Any]) -> str | None:
     """求值所有条件，返回跳过原因或 ``None``。
 
@@ -132,7 +139,11 @@ def _evaluate_conditions(spec: TaskSpec[Any], context: Mapping[str, Any]) -> str
             continue
 
         if not ok:
-            failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
+            reason = getattr(condition, "_reason", None)
+            if reason is not None:
+                failed_conditions.append(_format_reason(reason))
+            else:
+                failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
 
     if failed_conditions:
         if len(failed_conditions) <= 2:
@@ -159,8 +170,6 @@ def _make_skipped_result(
         reason=reason,
     )
     _emit(on_event, result)
-    if spec.verbose:
-        print(f"[skip] 任务 '{spec.name}' 跳过: {reason}", flush=True)
     logger.info("task %r skipped (%s)", spec.name, reason)
     return result
 

src/pyflowx/task.py

@@ -327,7 +327,13 @@ class TaskSpec(Generic[T]):
                 failed_conditions.append(name)
                 continue
             if not ok:
-                failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
+                reason = getattr(condition, "_reason", None)
+                if reason is not None:
+                    failed_conditions.append(
+                        ", ".join(str(r) for r in reason) if isinstance(reason, list) else str(reason),
+                    )
+                else:
+                    failed_conditions.append(getattr(condition, "__name__", None) or "匿名条件")
 
         if failed_conditions:
             return False, f"条件不满足: {', '.join(failed_conditions)}"