修复失败策略运行逻辑

schedule/src/schedule/orchestrator.py

@@ -553,6 +553,7 @@ class DispatchOrchestrator:
                 latest[row.node_id] = row
 
         max_concurrency = max(1, int(snapshot.get("max_concurrency", 1)))
+        failure_policy = snapshot.get("failure_policy", "stop")
         while True:
             changed = False
             active_count = sum(
@@ -586,7 +587,7 @@ class DispatchOrchestrator:
                 for item in latest.values()
             )
             stop_all = (
-                snapshot.get("failure_policy", "stop") == "stop"
+                failure_policy == "stop"
                 and exhausted_failure
             )
             for node in nodes:
@@ -594,29 +595,15 @@ class DispatchOrchestrator:
                 if node_id in latest:
                     continue
                 parent_runs = [latest.get(parent) for parent in parents[node_id]]
-                # Decide whether this node should be skipped. A node
+                # 只有 ``stop`` 策略才会在失败后跳过尚未启动的节点。
-                # is only skipped when we know it can never run:
+                # ``continue`` 表示“前一个节点失败也继续往后执行”，因此
-                #   * ``stop_all`` — the whole run was aborted on the
+                # 下游只需等待所有上游结束，不要求它们全部成功。
-                #     first failure, so any not-yet-dispatched node is
-                #     dropped;
-                #   * all parents are terminal AND at least one
-                #     failed — there is no remaining success path.
-                # If even one parent is still ``queued`` or
-                # ``running`` we keep waiting: under ``failure_policy
-                # == 'continue'`` a sibling might still succeed and
-                # the failed parent does not block that.
                 parents_terminal = all(
                     item is not None
                     and item.node_status in TERMINAL_NODE_STATES
                     for item in parent_runs
                 )
-                any_parent_failed = any(
+                if stop_all:
-                    item is not None
-                    and item.node_status in FAILED_NODE_STATES
-                    for item in parent_runs
-                )
-                parents_blocked = parents_terminal and any_parent_failed
-                if stop_all or parents_blocked:
                     skipped = ScheduleNodeRuns(
                         node_run_id=new_ulid(),
                         run_id=run.run_id,
@@ -627,11 +614,7 @@ class DispatchOrchestrator:
                         state_version=1,
                         finished_at=utcnow(),
                         duration_ms=0,
-                        message=(
+                        message="调度失败策略为 stop，未再启动",
-                            "调度失败策略为 stop，未再启动"
-                            if stop_all
-                            else "上游节点全部终止且至少一个失败，已跳过"
-                        ),
                     )
                     session.add(skipped)
                     latest[node_id] = skipped
@@ -640,26 +623,9 @@ class DispatchOrchestrator:
                         "skipped node: run={} node={} reason={}",
                         run.run_id[-12:],
                         node_id[-12:],
-                        "stop_policy" if stop_all else "parents_blocked",
+                        "stop_policy",
                     )
-                elif (
+                elif parents_terminal and active_count < max_concurrency:
-                    # Dispatch only when every parent has actually
-                    # run to completion successfully. A None parent
-                    # means the parent has not even been dispatched
-                    # yet (e.g. upstream is still queued); the existing
-                    # parents_blocked branch above handles the case
-                    # where every parent is terminal but at least one
-                    # failed.
-                    # ``all([])`` is intentionally true: a root node has
-                    # no parents and must be eligible for the initial
-                    # dispatch that starts the DAG.
-                    all(
-                        item is not None
-                        and item.node_status == "succeeded"
-                        for item in parent_runs
-                    )
-                    and active_count < max_concurrency
-                ):
                     dispatched = await self._dispatch_node(
                         session,
                         run=run,