MoviePilot/app/chain/workflow.py

"""
工作流的加载、调度、执行与结果通知编排。
"""

import ast
import base64
import binascii
import copy
import inspect
import pickletools
import threading
from collections import defaultdict, deque
from concurrent.futures import Future
from contextvars import Context, copy_context
from dataclasses import is_dataclass
from datetime import date, datetime
from functools import partial
from io import BytesIO, StringIO
from pickle import HIGHEST_PROTOCOL
from time import monotonic, sleep
from typing import Any, Callable, List, Optional, Tuple

from pydantic import BaseModel

from app.application.workflow import (
    WorkflowExecutionPort,
    WorkflowSnapshot,
    get_configured_workflow_execution,
    get_configured_workflow_query,
    get_workflow_manager,
    validate_workflow_graph,
)
from app.chain.base import ChainBase
from app.runtime.events import Event, eventmanager
from app.runtime.execution import OwnedThreadPoolExecutor
from app.runtime.log import logger
from app.runtime.stop import runtime_stop_state
from app.schemas.types import EventType
from app.schemas.workflow import Action, ActionContext, ActionExecution, ActionFlow, ActionResult

ARTIFACT_FIELDS = {"torrents", "medias", "fileitems", "downloads", "sites", "subscribes"}
DEFAULT_WORKFLOW_MAX_WORKERS = 4
WORKFLOW_EXECUTOR_STOP_TIMEOUT_SECONDS = 10.0
CIRCULAR_REFERENCE_PLACEHOLDER = "[Circular]"


def _serialize_workflow_key(key: Any) -> Any:
    """
    将映射键转换为 JSON 安全值。
    """
    if key is None or isinstance(key, (str, int, float, bool)):
        return key
    return str(key)


def _serialize_workflow_value(value: Any, stack: Optional[set[int]] = None) -> Any:
    """
    把工作流上下文和值转换为可持久化的 JSON 结构。
    """
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, (date, datetime)):
        return value.isoformat()

    stack = stack or set()
    object_id = id(value)
    if object_id in stack:
        return CIRCULAR_REFERENCE_PLACEHOLDER

    if isinstance(value, BaseModel):
        stack.add(object_id)
        try:
            return {
                field_name: _serialize_workflow_value(getattr(value, field_name, None), stack)
                for field_name in value.__class__.model_fields
            }
        finally:
            stack.remove(object_id)

    to_dict = getattr(value, "to_dict", None) if is_dataclass(value) and not isinstance(value, type) else None
    if callable(to_dict):
        # RSS 等动作按设计把领域 dataclass（如 app.domain.context.Context）放进上下文，其 to_dict
        # 与工作流 Schema 字段对齐，恢复时可经 ActionContext 校验还原；落到 str() 会存成无法恢复的文本。
        stack.add(object_id)
        try:
            return _serialize_workflow_value(to_dict(), stack)
        finally:
            stack.remove(object_id)

    if isinstance(value, dict):
        stack.add(object_id)
        try:
            return {
                _serialize_workflow_key(key): _serialize_workflow_value(item, stack)
                for key, item in value.items()
            }
        finally:
            stack.remove(object_id)

    if isinstance(value, (list, tuple, set)):
        stack.add(object_id)
        try:
            return [_serialize_workflow_value(item, stack) for item in value]
        finally:
            stack.remove(object_id)

    return str(value)


def _serialize_workflow_context(context: ActionContext) -> dict[str, Any]:
    """
    构建可写入数据库的工作流上下文字典。
    """
    serialized = _serialize_workflow_value(context)
    return serialized if isinstance(serialized, dict) else {}


class WorkflowCancelToken:
    """
    工作流取消令牌。
    """

    def __init__(
            self,
            workflow_id: int,
            stop_event: Optional[threading.Event] = None,
    ) -> None:
        """
        初始化取消令牌。

        :param workflow_id: 工作流ID
        :param stop_event: 单次执行 任务持有者 的本地停止信号
        """
        self.workflow_id = workflow_id
        self.stop_event = stop_event

    def is_cancelled(self) -> bool:
        """
        判断工作流是否已被取消。
        """
        return bool(
            (self.stop_event and self.stop_event.is_set())
            or runtime_stop_state.is_workflow_stopped(self.workflow_id)
        )


class WorkflowExecutor:
    """
    工作流执行器
    """

    def __init__(
            self,
            workflow: WorkflowSnapshot,
            step_callback: Optional[Callable[..., object]] = None,
            *,
            from_begin: bool = False,
            completion_callback: Optional[Callable[["WorkflowExecutor"], None]] = None,
    ):
        """
        初始化工作流执行器

        :param workflow: 工作流对象
        :param step_callback: 步骤回调函数
        :param from_begin: 忽略历史节点和上下文，从头构建本次执行
        :param completion_callback: 节点真实终止后、释放同 ID 所有权前提交终态
        """
        validate_workflow_graph(workflow.actions, workflow.flows)
        self.workflow = workflow
        self._from_begin = from_begin
        self._completion_callback = completion_callback
        self.step_callback = step_callback
        self.step_callback_extended = (
            len(inspect.signature(step_callback).parameters) > 2
            if step_callback
            else False
        )
        self.actions: dict[str, Action] = {}
        for action_data in workflow.actions or ():
            action = Action.model_validate(dict(action_data))
            if not action.id:
                raise ValueError("工作流动作缺少 ID")
            self.actions[action.id] = action
        self.flows = [ActionFlow.model_validate(dict(flow)) for flow in workflow.flows or ()]
        execution_config = getattr(workflow, "execution_config", None) or {}
        execution_state = {} if from_begin else (getattr(workflow, "execution_state", None) or {})
        self.execution_config = (
            execution_config.model_dump(exclude_none=True)
            if isinstance(execution_config, BaseModel)
            else execution_config
        )
        self.restored_execution_state = (
            execution_state.model_dump(exclude_none=True)
            if isinstance(execution_state, BaseModel)
            else execution_state
        )
        self.total_actions = len(self.actions)
        self.success = True
        self.has_failure = False
        self.stopped = False
        self.errmsg = ""
        self.errors = self.get_restored_errors()
        self.node_metadata = self.get_restored_node_metadata()
        self.node_attempts = self.get_restored_attempts()
        self.node_states = self.get_restored_node_states()
        self.completed_actions = {
            action_id for action_id, state in self.node_states.items()
            if state == "success"
        }
        self.finished_actions = len([
            state for state in self.node_states.values()
            if state in ("success", "failed", "skipped")
        ])
        self.flow_finished: set[str] = set()
        self.flow_satisfied: set[str] = set()
        self.flow_failed: set[str] = set()

        self.workflowmanager = get_workflow_manager()
        # 具体管理器登记活动执行；旧自定义 能力提供方 不实现 任务持有者 接口时保持原调用兼容。
        self._execution_lock = threading.RLock()
        self._admission_state = "pending"
        self._registered_execution = False
        self._stop_event = threading.Event()
        self._execute_returned = threading.Event()
        self._stopped_event = threading.Event()
        self.queue: deque[str] = deque()
        self.queued_actions: set[str] = set()
        self.active_concurrency_keys: set[str] = set()
        self.lock = threading.Lock()
        self.executor = OwnedThreadPoolExecutor(max_workers=self.get_workflow_max_workers())
        self.cancel_token = WorkflowCancelToken(self.workflow.id, self._stop_event)
        self.running_tasks = 0

        # 构建出边与入边表，用于条件流转和多上游汇合。
        self.outgoing_flows = defaultdict(list)
        self.incoming_flows = defaultdict(list)
        for flow in self.flows:
            if not flow.source or not flow.target:
                continue
            self.outgoing_flows[flow.source].append(flow)
            self.incoming_flows[flow.target].append(flow)

        self.context = self.restore_context()
        self.ensure_context_partitions()

        # 恢复时重新释放已终态节点的出边，使后继节点能继续执行或保持跳过传播。
        for action_id, state in self.node_states.items():
            if state == "success":
                self.release_successors(action_id, source_success=True)
            elif state in ("failed", "skipped"):
                self.release_successors(action_id, source_success=False)
        # 初始化队列，添加没有入边的起始节点。
        for action_id in self.actions:
            if self.node_states.get(action_id) == "pending" and not self.incoming_flows.get(action_id):
                self.enqueue_node(action_id)

    def admit(self) -> bool:
        """
        向 concrete manager 登记本次执行，并保持旧 能力提供方 可直接运行。
        """
        with self._execution_lock:
            if self._admission_state == "admitted":
                return True
            if self._admission_state == "rejected":
                return False
            register = getattr(self.workflowmanager, "register_execution", None)
            if callable(register) and not register(self):
                self._admission_state = "rejected"
                self.success = False
                self.stopped = True
                self.errmsg = "工作流正在执行或服务正在停止"
                self.executor.shutdown_bounded(timeout=0.0)
                self._execute_returned.set()
                self._stopped_event.set()
                return False
            self._registered_execution = callable(register)
            self._admission_state = "admitted"
        # 只有获得执行准入后才能清除历史单工作流停止标记。
        runtime_stop_state.resume_workflow(self.workflow.id)
        return True

    @property
    def workflow_id(self) -> int:
        """
        返回执行 任务持有者 互斥准入所使用的工作流 ID。
        """
        return self.workflow.id

    def request_stop(self) -> None:
        """
        停止调度新节点，并通过本地令牌通知支持取消的活动动作。
        """
        self._stop_event.set()

    def abort_before_execute(self) -> None:
        """
        执行状态启动失败时释放尚未使用的节点池和 manager 任务持有者。
        """
        self.request_stop()
        # reset/start 失败尚未真正执行，不提交误导性的成功或失败终态。
        self._completion_callback = None
        converged = self.executor.shutdown_bounded(timeout=0.0)
        self._execute_returned.set()
        if converged:
            self._release_execution()

    def wait_stopped(self, timeout: float) -> bool:
        """
        有限等待 execute 返回，并在需要时重试节点线程池收敛。
        """
        deadline = monotonic() + max(0.0, timeout)
        if self._stopped_event.is_set():
            return True
        if not self._execute_returned.wait(
                timeout=max(0.0, deadline - monotonic()),
        ):
            return False
        if self._stopped_event.is_set():
            return True
        if not self.executor.shutdown_bounded(
                timeout=max(0.0, deadline - monotonic()),
        ):
            return False
        self._release_execution()
        return self._stopped_event.is_set()

    def _release_execution(self) -> None:
        """
        先完成终态事务，再释放已真实终止的执行 任务持有者；事务失败允许重试。
        """
        with self._execution_lock:
            if self._stopped_event.is_set():
                return
            if self._completion_callback is not None:
                self._completion_callback(self)
                self._completion_callback = None
            if self._registered_execution:
                unregister = getattr(self.workflowmanager, "unregister_execution", None)
                if callable(unregister):
                    unregister(self)
                self._registered_execution = False
            self._stopped_event.set()

    def _stop_requested(self) -> bool:
        """
        判断本次执行或全局工作流是否已收到停止请求。
        """
        return bool(
            self._stop_event.is_set()
            or runtime_stop_state.is_workflow_stopped(self.workflow.id)
        )

    def get_workflow_max_workers(self) -> int:
        """
        获取工作流最大并发数。
        """
        max_workers = self.execution_config.get("max_workers") if isinstance(self.execution_config, dict) else None
        try:
            return max(int(max_workers or DEFAULT_WORKFLOW_MAX_WORKERS), 1)
        except (TypeError, ValueError):
            return DEFAULT_WORKFLOW_MAX_WORKERS

    def get_restored_node_metadata(self) -> dict[str, Any]:
        """
        获取已持久化的节点状态元数据。
        """
        nodes = self.restored_execution_state.get("nodes") if isinstance(self.restored_execution_state, dict) else {}
        return nodes if isinstance(nodes, dict) else {}

    def get_restored_errors(self) -> dict[str, Any]:
        """
        获取已持久化的错误状态。
        """
        errors = self.restored_execution_state.get("errors") if isinstance(self.restored_execution_state, dict) else {}
        return errors if isinstance(errors, dict) else {}

    def get_restored_attempts(self) -> dict[str, Any]:
        """
        获取已持久化的节点尝试次数。
        """
        attempts = {}
        for action_id, metadata in self.get_restored_node_metadata().items():
            if isinstance(metadata, dict) and metadata.get("attempt"):
                attempts[action_id] = int(metadata.get("attempt") or 0)
        return attempts

    def get_restored_node_states(self) -> dict[str, Any]:
        """
        获取结构化节点状态，兼容旧版 current_action 字符串。
        """
        legacy_actions = {
            action_id for action_id in (self.workflow.current_action or "").split(",")
            if not self._from_begin and action_id in self.actions
        }
        states = {}
        for action_id in self.actions:
            metadata = self.node_metadata.get(action_id) or {}
            state = metadata.get("state") if isinstance(metadata, dict) else None
            if state == "completed":
                state = "success"
            if state in ("running", "queued", "failed", "skipped"):
                state = "pending"
            if not state and action_id in legacy_actions:
                state = "success"
            states[action_id] = state or "pending"
        return states

    def restore_context(self) -> ActionContext:
        """
        恢复工作流上下文。

        只接受结构化 JSON 上下文；旧版 Base64 Pickle 格式（``content`` 字段）不再反序列化，
        按空上下文从头执行，避免执行数据库中来源不可信的序列化载荷。
        """
        context = ActionContext()
        if self._from_begin:
            return context
        if self.workflow.context:
            if self.workflow.current_action:
                logger.info(f"工作流已执行动作：{self.workflow.current_action}")
            try:
                if self.is_legacy_context(self.workflow.context):
                    logger.warning(f"工作流 {self.workflow.name} 的上下文为已废弃的旧版格式，将从空上下文开始执行")
                elif isinstance(self.workflow.context, dict):
                    context = ActionContext.model_validate(self.workflow.context)
            except Exception:
                context = ActionContext()
        outputs = self.restored_execution_state.get("outputs") if isinstance(self.restored_execution_state, dict) else {}
        if outputs and not getattr(context, "node_outputs", None):
            context.node_outputs = outputs
        return context

    @staticmethod
    def is_legacy_context(context: Any) -> bool:
        """
        只识别单字段的旧 Base64 Pickle 包装，绝不反序列化载荷。

        当前文本 content 即使是唯一字段也属于合法结构，不能据其非空丢弃上下文。

        :param context: 持久化的工作流上下文字典
        :return: 仅旧包装内含支持协议的完整指令流、栈合法且无尾部数据时为 True
        """
        if not isinstance(context, dict) or set(context) != {"content"}:
            return False
        content = context.get("content")
        if not isinstance(content, str):
            return False
        try:
            payload = base64.b64decode(content, validate=True)
            if len(payload) < 2 or payload[0] != 0x80 or not 2 <= payload[1] <= HIGHEST_PROTOCOL:
                return False
            stream = BytesIO(payload)
            # 静态解析校验指令、memo 与栈，不导入载荷引用，也不执行其 callable。
            pickletools.dis(stream, out=StringIO())
            return stream.tell() == len(payload)
        except (ValueError, binascii.Error):
            return False

    def ensure_context_partitions(self) -> None:
        """
        确保上下文具备新版分区结构，并把旧字段映射到 artifacts。
        """
        self.context.workflow_context = self.context.workflow_context or {}
        self.context.node_outputs = self.context.node_outputs or {}
        self.context.runtime_state = self.context.runtime_state or {}
        self.context.artifacts = self.context.artifacts or {}
        for key in ARTIFACT_FIELDS:
            value = getattr(self.context, key, None)
            if value not in (None, "", [], {}) and key not in self.context.artifacts:
                self.context.artifacts[key] = value
        self.update_runtime_state()

    def update_runtime_state(self) -> None:
        """
        更新上下文中的运行期状态分区。
        """
        if self.context.runtime_state is None:
            self.context.runtime_state = {}
        self.context.runtime_state.update({
            "progress": self.context.progress,
            "finished_actions": self.finished_actions,
            "running_tasks": self.running_tasks,
            "errors": self.errors,
            "node_states": self.node_states,
            "attempts": self.node_attempts,
        })

    def set_node_state(self, action_id: str, state: str, message: Optional[str] = None) -> None:
        """
        更新节点结构化状态。
        """
        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        metadata = self.node_metadata.setdefault(action_id, {})
        metadata["state"] = state
        if state == "success":
            self.errors.pop(action_id, None)
        metadata["attempt"] = self.node_attempts.get(action_id, metadata.get("attempt") or 0)
        if state == "running":
            metadata["started_at"] = now
        if state in ("success", "failed", "skipped"):
            metadata["finished_at"] = now
        if message is not None:
            metadata["message"] = message
        self.node_states[action_id] = state
        self.update_runtime_state()

    def build_execution_state(self) -> dict[str, Any]:
        """
        构建可持久化的结构化执行状态。
        """
        self.update_runtime_state()
        execution_state = _serialize_workflow_value({
            "version": 1,
            "nodes": self.node_metadata,
            "outputs": self.context.node_outputs,
            "errors": self.errors,
            "runtime": self.context.runtime_state,
        })
        return execution_state if isinstance(execution_state, dict) else {}

    def execute(self) -> None:
        """
        执行工作流
        """
        if not self.admit():
            return
        try:
            while True:
                should_sleep = False
                node_id = None
                with self.lock:
                    if self._stop_requested():
                        self.success = False
                        self.stopped = True
                        self.errmsg = "工作流已停止"
                        if self.running_tasks == 0:
                            break
                        should_sleep = True
                    # 退出条件：队列为空且无运行任务
                    elif not self.queue and self.running_tasks == 0:
                        break
                    # 出错后不再调度新节点，但等待已提交节点完成，避免后台线程继续写状态。
                    if not self.success:
                        if self.running_tasks == 0:
                            break
                        should_sleep = True
                    elif not self.queue:
                        should_sleep = True
                    else:
                        node_id = self.pop_dispatchable_node()
                        if not node_id:
                            should_sleep = True

                if should_sleep:
                    sleep(0.1)
                    continue

                if not node_id:
                    continue

                # 节点分别复制业务上下文和调用上下文，避免共享可变状态或丢失触发链路。
                context = copy_context()
                future = Context().run(
                    self.executor.submit,
                    context.run,
                    self.execute_node,
                    self.workflow.id,
                    node_id,
                    copy.deepcopy(self.context),
                )
                future.add_done_callback(partial(self._complete_in_context, context))
        except Exception as err:
            self.success = False
            self.has_failure = True
            self.errmsg = str(err)
            raise
        finally:
            converged = self.executor.shutdown_bounded(
                timeout=WORKFLOW_EXECUTOR_STOP_TIMEOUT_SECONDS,
            )
            try:
                if converged:
                    self._release_execution()
            finally:
                # 终态事务仍在进行时，停机等待不能把节点线程结束等同于 任务持有者 已释放。
                self._execute_returned.set()

    def pop_dispatchable_node(self) -> Optional[str]:
        """
        从队列中取出当前可调度节点。
        """
        for _ in range(len(self.queue)):
            node_id = self.queue.popleft()
            self.queued_actions.discard(node_id)
            if self.node_states.get(node_id) != "queued":
                continue
            concurrency_key = self.get_action_concurrency_key(self.actions[node_id])
            if concurrency_key and concurrency_key in self.active_concurrency_keys:
                self.queue.append(node_id)
                self.queued_actions.add(node_id)
                continue
            if concurrency_key:
                self.active_concurrency_keys.add(concurrency_key)
            self.running_tasks += 1
            self.set_node_state(node_id, "running")
            return node_id
        return None

    def execute_node(self, workflow_id: int, node_id: str,
                     context: ActionContext) -> Tuple[Action, ActionResult]:
        """
        执行单个节点操作，返回修改后的上下文和节点ID
        """
        action = self.actions[node_id]
        action_result = self.workflowmanager.execute(
            workflow_id,
            action,
            context=context,
            inputs=self.build_action_inputs(action),
            runtime=self.build_action_runtime(action),
            cancel_token=self.cancel_token
        )
        return action, action_result

    def _complete_in_context(
            self,
            context: Context,
            future: Future[Tuple[Action, ActionResult]],
    ) -> None:
        """
        在节点提交时的上下文中消费完成结果。
        """
        context.run(self.on_node_complete, future)

    def on_node_complete(self, future: Future[Tuple[Action, ActionResult]]) -> None:
        """
        节点完成回调：更新上下文、处理后继节点
        """
        action = None
        try:
            action, action_result = future.result()
            if not action.id:
                raise ValueError("工作流动作缺少 ID")
            with self.lock:
                if self._stop_requested():
                    self.success = False
                    self.stopped = True
                    self.errmsg = "工作流已停止"
                    return
                state = bool(action_result.success)
                message = action_result.message or ""
                result_ctx = action_result.context or ActionContext()
                self.node_attempts[action.id] = action_result.attempts or self.node_attempts.get(action.id, 1)

                self.finished_actions += 1
                self.update_progress()
                if self.context.execute_history is None:
                    self.context.execute_history = []
                self.context.execute_history.append(
                    ActionExecution(
                        action=action.name,
                        result=state,
                        message=message
                    )
                )

                # 节点执行失败时默认停止；显式配置 continue/ignore 时继续释放后续 all_done 汇合。
                if not state:
                    self.errors[action.id] = message or f"{action.name} 失败"
                    self.set_node_state(action.id, "failed", message=message)
                    fail_policy = self.get_action_fail_policy(action)
                    if fail_policy != "ignore":
                        self.has_failure = True
                        self.errmsg = f"{action.name} 失败"
                    if fail_policy == "stop":
                        self.success = False
                        self.call_step_callback(action, completed=False)
                        return
                    if fail_policy not in ("continue", "ignore"):
                        self.success = False
                        self.errmsg = f"{action.name} 失败：无效失败策略 {fail_policy}"
                        self.call_step_callback(action, completed=False)
                        return
                    self.release_successors(action.id, source_success=False)
                    self.call_step_callback(action, completed=False)
                    return

                self.ensure_result_context_partitions(result_ctx)
                outputs = self.normalize_action_outputs(action, action_result, result_ctx)
                self.merge_context_partitions(result_ctx)
                self.merge_action_outputs(action, outputs)
                self.record_node_outputs(action.id, outputs)
                self.completed_actions.add(action.id)
                self.set_node_state(action.id, "success", message=message)
                self.release_successors(action.id, source_success=True)
                self.call_step_callback(action, completed=True)
        except Exception as err:
            logger.error(f"工作流节点执行回调失败: {str(err)}")
            with self.lock:
                self.success = False
                self.errmsg = str(err)
        finally:
            with self.lock:
                if action:
                    concurrency_key = self.get_action_concurrency_key(action)
                    if concurrency_key:
                        self.active_concurrency_keys.discard(concurrency_key)
                self.running_tasks -= 1
                self.update_runtime_state()

    def enqueue_node(self, node_id: str) -> None:
        """
        将满足条件的节点加入待执行队列。
        """
        if node_id not in self.actions:
            return
        if self.node_states.get(node_id) != "pending" or node_id in self.queued_actions:
            return
        self.queue.append(node_id)
        self.queued_actions.add(node_id)
        self.set_node_state(node_id, "queued")

    def skip_node(self, node_id: str, message: str) -> None:
        """
        将不可达节点标记为跳过，并把跳过状态继续传递给后继节点。
        """
        if node_id not in self.actions:
            return
        if self.node_states.get(node_id) not in ("pending", "queued"):
            return
        self.queued_actions.discard(node_id)
        self.finished_actions += 1
        self.update_progress()
        self.set_node_state(node_id, "skipped", message=message)
        if self.context.execute_history is None:
            self.context.execute_history = []
        self.context.execute_history.append(
            ActionExecution(
                action=self.actions[node_id].name,
                result=True,
                message=message
            )
        )
        self.call_step_callback(self.actions[node_id], completed=False)
        self.release_successors(node_id, source_success=False)

    def release_successors(self, source_id: str, source_success: bool) -> None:
        """
        根据源节点状态释放出边，并重新判断目标节点是否可运行。
        """
        flows = self.outgoing_flows.get(source_id, [])
        branch_policy = self.get_action_branch_policy(self.actions.get(source_id), flows)
        matched_exclusive_flow = None
        for flow in flows:
            flow_key = self.get_flow_key(flow)
            if flow_key in self.flow_finished:
                continue
            condition_matched = False
            if source_success:
                try:
                    condition_matched = self.evaluate_condition(self.get_flow_condition(flow))
                except ValueError as err:
                    self.success = False
                    self.errmsg = f"流程条件判断失败：{err}"
                    return
                if branch_policy == "exclusive" and condition_matched and matched_exclusive_flow:
                    condition_matched = False
                elif branch_policy == "exclusive" and condition_matched:
                    matched_exclusive_flow = flow_key
            self.flow_finished.add(flow_key)
            if source_success and condition_matched:
                self.flow_satisfied.add(flow_key)
            if not source_success and self.node_states.get(source_id) == "failed":
                self.flow_failed.add(flow_key)
            if flow.target:
                self.evaluate_target_state(flow.target)

    def evaluate_target_state(self, target_id: str) -> None:
        """
        按目标节点汇合策略判断节点是否入队或跳过。
        """
        if not target_id or target_id not in self.actions:
            return
        if self.node_states.get(target_id) != "pending":
            return
        incoming_flows = self.incoming_flows.get(target_id, [])
        if not incoming_flows:
            self.enqueue_node(target_id)
            return

        total_count = len(incoming_flows)
        finished_count = sum(1 for flow in incoming_flows if self.get_flow_key(flow) in self.flow_finished)
        satisfied_count = sum(1 for flow in incoming_flows if self.get_flow_key(flow) in self.flow_satisfied)
        failed_count = sum(1 for flow in incoming_flows if self.get_flow_key(flow) in self.flow_failed)
        join_policy = self.get_action_join_policy(self.actions[target_id], incoming_flows)

        if join_policy == "fail_fast":
            if failed_count > 0:
                self.skip_node(target_id, "上游失败触发 fail_fast，已取消后续节点")
            elif finished_count == total_count and satisfied_count == total_count:
                self.enqueue_node(target_id)
            elif finished_count == total_count:
                self.skip_node(target_id, "上游条件未全部满足，已跳过")
            return

        if join_policy == "any_success":
            if satisfied_count > 0:
                self.enqueue_node(target_id)
            elif finished_count == total_count:
                self.skip_node(target_id, "所有上游条件均未满足，已跳过")
            return

        if join_policy == "all_done":
            if finished_count == total_count:
                self.enqueue_node(target_id)
            return

        if join_policy != "all_success":
            self.success = False
            self.errmsg = f"{self.actions[target_id].name} 汇合策略无效：{join_policy}"
            return

        if finished_count != total_count:
            return
        if satisfied_count == total_count:
            self.enqueue_node(target_id)
        else:
            self.skip_node(target_id, "上游条件未全部满足，已跳过")

    def update_progress(self) -> None:
        """
        根据已完成和已跳过节点数量更新整体进度。
        """
        self.context.progress = round(self.finished_actions / self.total_actions * 100) if self.total_actions else 100
        self.update_runtime_state()

    def build_action_inputs(self, action: Action) -> dict[str, Any]:
        """
        根据动作输入声明读取上游节点输出。
        """
        inputs = {}
        input_paths = action.inputs or self.get_action_data_value(action, "inputs")
        if not input_paths:
            input_paths = [
                field["name"] for field in self.get_action_contract(action).get("inputs") or []
            ]
        if isinstance(input_paths, str):
            input_paths = [item.strip() for item in input_paths.splitlines() if item.strip()]
        for input_path in input_paths:
            inputs[input_path] = self.resolve_action_input(input_path)
        return inputs

    def resolve_action_input(self, input_path: str) -> Any:
        """
        解析动作输入声明。
        """
        if input_path in ActionContext.model_fields:
            value = getattr(self.context, input_path, None)
            if value not in (None, "", [], {}):
                return value
            return self.context.artifacts.get(input_path) if self.context.artifacts else value
        return self.resolve_context_path(input_path)

    def build_action_runtime(self, action: Action) -> dict[str, Any]:
        """
        构建传递给动作的新运行期数据。
        """
        return {
            "workflow_id": self.workflow.id,
            "action_id": action.id,
            "execution_config": self.execution_config,
            "runtime_state": self.context.runtime_state,
        }

    @staticmethod
    def ensure_result_context_partitions(context: ActionContext) -> None:
        """
        确保动作返回上下文具备新版分区字段。
        """
        context.workflow_context = context.workflow_context or {}
        context.node_outputs = context.node_outputs or {}
        context.runtime_state = context.runtime_state or {}
        context.artifacts = context.artifacts or {}

    def normalize_action_outputs(self, action: Action, action_result: ActionResult,
                                 result_context: ActionContext) -> dict[str, Any]:
        """
        根据动作输出声明整理当前节点输出。
        """
        outputs = action_result.outputs or self.extract_context_outputs(result_context)
        declared_outputs = self.get_action_output_declarations(action)
        if isinstance(declared_outputs, list):
            normalized_outputs = {}
            missing = object()
            for item in declared_outputs:
                key = item.get("name") if isinstance(item, dict) else item
                if not key:
                    continue
                value = outputs.get(key, missing)
                if value is missing and key in result_context.__class__.model_fields:
                    value = getattr(result_context, key, missing)
                if value is not missing and self.should_keep_output_value(action, key, value):
                    normalized_outputs[key] = value
            return normalized_outputs or outputs
        if isinstance(declared_outputs, dict):
            normalized_outputs = {}
            missing = object()
            for key in declared_outputs:
                value = outputs.get(key, missing)
                if value is missing and key in result_context.__class__.model_fields:
                    value = getattr(result_context, key, missing)
                if value is not missing and self.should_keep_output_value(action, key, value):
                    normalized_outputs[key] = value
            return normalized_outputs or outputs
        return outputs

    def should_keep_output_value(self, action: Action, key: str, value: Any) -> bool:
        """
        判断输出值是否应参与后续合并。
        """
        if value not in (None, "", [], {}):
            return True
        output_config = self.get_action_output_config(action, key)
        target_key = output_config.get("target") or key
        merge_policy = output_config.get("merge") or self.get_default_merge_policy(action, target_key, value)
        return merge_policy == "replace"

    def record_node_outputs(self, action_id: str, outputs: dict[str, Any]) -> None:
        """
        记录当前节点输出，供后续条件表达式读取。
        """
        if outputs:
            if self.context.node_outputs is None:
                self.context.node_outputs = {}
            if self.context.runtime_state is None:
                self.context.runtime_state = {}
            self.context.node_outputs[action_id] = outputs
            self.context.runtime_state["last_outputs"] = outputs

    def merge_context_partitions(self, context: ActionContext) -> None:
        """
        合并动作返回的新分区上下文。
        """
        for key in ("workflow_context", "runtime_state", "artifacts"):
            value = getattr(context, key, None)
            if not value:
                continue
            current_value = getattr(self.context, key, None) or {}
            current_value.update(value)
            setattr(self.context, key, current_value)

    def merge_action_outputs(self, action: Action, outputs: dict[str, Any]) -> None:
        """
        按声明式合并策略写入全局上下文和 artifacts 分区。
        """
        for key, value in outputs.items():
            output_config = self.get_action_output_config(action, key)
            target_key = output_config.get("target") or key
            merge_policy = output_config.get("merge") or self.get_default_merge_policy(action, target_key, value)
            if value in (None, "", [], {}) and merge_policy != "replace":
                continue
            identity = output_config.get("identity")
            self.merge_output_value(target_key, value, merge_policy, identity)

    def merge_output_value(self, key: str, value: Any, merge_policy: str, identity: Optional[str] = None) -> None:
        """
        按指定策略合并单个输出值。
        """
        current_value = getattr(self.context, key, None) if key in ActionContext.model_fields else None
        merged_value = self.apply_merge_policy(current_value, value, merge_policy, identity)
        if key in ActionContext.model_fields:
            setattr(self.context, key, merged_value)
        if key in ARTIFACT_FIELDS:
            if self.context.artifacts is None:
                self.context.artifacts = {}
            current_artifact = self.context.artifacts.get(key)
            self.context.artifacts[key] = self.apply_merge_policy(current_artifact, value, merge_policy, identity)

    def get_action_output_config(self, action: Action, output_key: str) -> dict[str, Any]:
        """
        获取动作输出声明配置。
        """
        outputs_config = self.get_action_output_declarations(action)
        if isinstance(outputs_config, dict):
            value = outputs_config.get(output_key) or {}
            return value if isinstance(value, dict) else {}
        if isinstance(outputs_config, list):
            for item in outputs_config:
                if isinstance(item, dict) and item.get("name") == output_key:
                    return {
                        key: value for key, value in item.items()
                        if key not in ("name", "label", "kind") and value not in (None, "", [], {})
                    }
        return {}

    def get_action_output_declarations(self, action: Action) -> Any:
        """
        获取动作输出声明，优先使用节点显式配置，其次使用动作固定契约。
        """
        outputs_config = action.outputs or self.get_action_data_value(action, "outputs")
        if outputs_config:
            return outputs_config
        return self.get_action_contract(action).get("outputs") or {}

    @staticmethod
    def get_default_merge_policy(action: Action, key: str, value: Any) -> str:
        """
        获取输出默认合并策略。
        """
        if isinstance(value, list):
            action_type = action.type or ""
            action_name = action.name or ""
            if key in ARTIFACT_FIELDS and (
                action_type.startswith("Filter")
                or action_name.startswith("过滤")
            ):
                return "replace"
            return "append_unique"
        if isinstance(value, dict):
            return "merge_dict"
        return "first_non_empty"

    def apply_merge_policy(self, current_value: Any, value: Any, merge_policy: str,
                           identity: Optional[str] = None) -> Any:
        """
        应用声明式合并策略。
        """
        if merge_policy == "replace":
            return value
        if merge_policy == "merge_dict":
            merged = current_value.copy() if isinstance(current_value, dict) else {}
            if isinstance(value, dict):
                merged.update(value)
                return merged
            return current_value or value
        if merge_policy == "append_unique":
            return self.append_unique_values(current_value, value, identity)
        if merge_policy == "first_non_empty":
            return current_value or value
        return current_value or value

    def append_unique_values(self, current_value: Any, value: Any, identity: Optional[str] = None) -> list[Any]:
        """
        追加列表并按身份字段去重。
        """
        current_list = list(current_value or [])
        incoming_list = value if isinstance(value, list) else [value]
        seen = {self.get_identity_value(item, identity) for item in current_list}
        for item in incoming_list:
            identity_value = self.get_identity_value(item, identity)
            if identity_value in seen:
                continue
            current_list.append(item)
            seen.add(identity_value)
        return current_list

    def get_identity_value(self, item: Any, identity: Optional[str] = None) -> Any:
        """
        获取列表元素去重身份。
        """
        if not identity:
            return self.make_hashable_identity(item)
        value = item
        for part in identity.split("."):
            value = self.read_value(value, int(part) if part.isdigit() else part)
        return self.make_hashable_identity(value)

    @staticmethod
    def make_hashable_identity(value: Any) -> Any:
        """
        将身份值转换为可哈希对象。
        """
        try:
            hash(value)
            return value
        except TypeError:
            return repr(value)

    def call_step_callback(self, action: Action, completed: bool) -> None:
        """
        持久化当前步骤上下文和结构化执行状态。
        """
        if not self.step_callback:
            return
        if self.step_callback_extended:
            self.step_callback(action, self.context, self.build_execution_state(), completed)
            return
        self.step_callback(action, self.context)

    @staticmethod
    def extract_context_outputs(context: ActionContext) -> dict[str, Any]:
        """
        从动作上下文中提取非空业务字段作为节点默认输出。
        """
        if not context:
            return {}
        outputs = {}
        for key in context.__class__.model_fields:
            if key in ("execute_history", "progress", "node_outputs", "runtime_state"):
                continue
            value = getattr(context, key, None)
            if value in (None, "", [], {}):
                continue
            outputs[key] = value
        return outputs

    @staticmethod
    def get_flow_key(flow: ActionFlow) -> str:
        """
        生成流程边的运行期唯一标识。
        """
        return flow.id or f"{flow.source}->{flow.target}:{id(flow)}"

    def get_action_join_policy(self, action: Action, incoming_flows: List[ActionFlow]) -> str:
        """
        获取动作汇合策略，优先使用动作配置，其次兼容流程边配置。
        """
        join_policy = action.join_policy or self.get_action_data_value(action, "join_policy")
        if join_policy:
            return join_policy
        for flow in incoming_flows:
            join_policy = flow.join_policy or self.get_flow_data_value(flow, "join_policy")
            if join_policy:
                return join_policy
        return "all_success"

    def get_action_branch_policy(self, action: Optional[Action], outgoing_flows: List[ActionFlow]) -> str:
        """
        获取动作出边分支策略。
        """
        if action:
            branch_policy = action.branch_policy or self.get_action_data_value(action, "branch_policy")
            if branch_policy:
                return branch_policy
        for flow in outgoing_flows:
            branch_policy = flow.branch_policy or self.get_flow_data_value(flow, "branch_policy")
            if branch_policy:
                return branch_policy
        return "parallel"

    def get_action_fail_policy(self, action: Action) -> str:
        """
        获取动作失败策略。
        """
        return action.fail_policy or self.get_action_data_value(action, "fail_policy") or "stop"

    def get_action_concurrency_key(self, action: Action) -> Optional[str]:
        """
        获取动作并发互斥键。
        """
        return (
            action.concurrency_key
            or self.get_action_data_value(action, "concurrency_key")
            or self.get_action_contract(action).get("concurrency_key")
        )

    def get_action_contract(self, action: Action) -> dict[str, Any]:
        """
        获取动作固定输入输出契约。
        """
        get_contract = getattr(self.workflowmanager, "get_action_contract", None)
        if not get_contract:
            return {}
        return get_contract(action.type) or {}

    def get_flow_condition(self, flow: ActionFlow) -> Optional[str]:
        """
        获取流程边条件表达式。
        """
        return flow.condition or self.get_flow_data_value(flow, "condition")

    @staticmethod
    def get_action_data_value(action: Optional[Action], key: str) -> Any:
        """
        从动作 data 中读取扩展配置。
        """
        if not action:
            return None
        data = action.data or {}
        return data.get(key) if isinstance(data, dict) else None

    @staticmethod
    def get_flow_data_value(flow: ActionFlow, key: str) -> Any:
        """
        从流程边 data 中读取扩展配置。
        """
        data = flow.data or {}
        return data.get(key) if isinstance(data, dict) else None

    def evaluate_condition(self, condition: Optional[str]) -> bool:
        """
        安全计算流程边条件表达式。
        """
        if not condition:
            return True
        expression = condition.strip()
        if not expression:
            return True
        expression = expression.replace("&&", " and ").replace("||", " or ")
        try:
            tree = ast.parse(expression, mode="eval")
        except SyntaxError as err:
            raise ValueError(f"{condition} 语法错误") from err
        return bool(self.evaluate_condition_node(tree.body))

    def evaluate_condition_node(self, node: ast.AST) -> Any:
        """
        递归计算受限 AST 节点，避免执行任意代码。
        """
        if isinstance(node, ast.BoolOp):
            values = [bool(self.evaluate_condition_node(value)) for value in node.values]
            if isinstance(node.op, ast.And):
                return all(values)
            if isinstance(node.op, ast.Or):
                return any(values)
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not):
            return not bool(self.evaluate_condition_node(node.operand))
        if isinstance(node, ast.Compare):
            return self.evaluate_compare_node(node)
        if isinstance(node, ast.Name):
            return self.resolve_condition_name(node.id)
        if isinstance(node, ast.Attribute):
            return self.read_value(self.evaluate_condition_node(node.value), node.attr)
        if isinstance(node, ast.Subscript):
            return self.read_subscript_node(node)
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.List):
            return [self.evaluate_condition_node(item) for item in node.elts]
        if isinstance(node, ast.Tuple):
            return tuple(self.evaluate_condition_node(item) for item in node.elts)
        if isinstance(node, ast.Set):
            return {self.evaluate_condition_node(item) for item in node.elts}
        if isinstance(node, ast.Dict):
            if any(key is None for key in node.keys):
                raise ValueError("条件表达式不支持字典展开")
            return {
                self.evaluate_condition_node(key): self.evaluate_condition_node(value)
                for key, value in zip(node.keys, node.values)
                if key is not None
            }
        raise ValueError(f"不支持的条件表达式：{ast.dump(node)}")

    def evaluate_compare_node(self, node: ast.Compare) -> bool:
        """
        计算比较表达式，支持链式比较和成员判断。
        """
        left = self.evaluate_condition_node(node.left)
        for operator, comparator in zip(node.ops, node.comparators):
            right = self.evaluate_condition_node(comparator)
            if not self.compare_values(left, operator, right):
                return False
            left = right
        return True

    def read_subscript_node(self, node: ast.Subscript) -> Any:
        """
        读取下标访问表达式。
        """
        if isinstance(node.slice, ast.Slice):
            raise ValueError("条件表达式不支持切片访问")
        container = self.evaluate_condition_node(node.value)
        key = self.evaluate_condition_node(node.slice)
        return self.read_value(container, key)

    def resolve_condition_name(self, name: str) -> Any:
        """
        将条件表达式中的根名称映射到当前工作流上下文。
        """
        if name in ("true", "True"):
            return True
        if name in ("false", "False"):
            return False
        if name in ("none", "None", "null"):
            return None
        if name == "context":
            return self.context
        if name == "workflow_context":
            return self.context.workflow_context or {}
        if name == "runtime_state":
            return self.context.runtime_state or {}
        if name == "artifacts":
            return self.context.artifacts or {}
        if name in ("outputs", "node_outputs"):
            return self.context.node_outputs or {}
        if name == "last":
            return self.context.runtime_state.get("last_outputs") if self.context.runtime_state else {}
        node_outputs = self.context.node_outputs or {}
        if name in node_outputs:
            return node_outputs[name]
        if name in ActionContext.model_fields:
            return getattr(self.context, name, None)
        raise ValueError(f"未知上下文变量 {name}")

    def resolve_context_path(self, path: str) -> Any:
        """
        按点分路径读取工作流上下文数据。
        """
        if not path:
            return None
        value = None
        for index, part in enumerate(path.split(".")):
            if index == 0:
                value = self.resolve_condition_name(part)
                continue
            key = int(part) if part.isdigit() else part
            value = self.read_value(value, key)
        return value

    @staticmethod
    def read_value(value: Any, key: Any) -> Any:
        """
        从 dict、对象或序列中读取属性值。
        """
        if value is None:
            return None
        if isinstance(key, str) and key in ("count", "length") and hasattr(value, "__len__"):
            return len(value)
        if isinstance(value, dict):
            return value.get(key)
        if isinstance(value, (list, tuple)):
            if isinstance(key, int) and 0 <= key < len(value):
                return value[key]
            return None
        if isinstance(key, str) and hasattr(value, key):
            return getattr(value, key)
        return None

    @staticmethod
    def compare_values(left: Any, operator: ast.cmpop, right: Any) -> bool:
        """
        比较两个条件表达式值。
        """
        try:
            if isinstance(operator, ast.Eq):
                return bool(left == right)
            if isinstance(operator, ast.NotEq):
                return bool(left != right)
            if isinstance(operator, ast.Gt):
                return bool(left > right)
            if isinstance(operator, ast.GtE):
                return bool(left >= right)
            if isinstance(operator, ast.Lt):
                return bool(left < right)
            if isinstance(operator, ast.LtE):
                return bool(left <= right)
            if isinstance(operator, ast.In):
                return left in right
            if isinstance(operator, ast.NotIn):
                return left not in right
        except TypeError:
            return False
        raise ValueError(f"不支持的比较操作符：{operator.__class__.__name__}")

    def merge_context(self, context: ActionContext) -> None:
        """
        合并上下文
        """
        if not context:
            return
        for key in context.__class__.model_fields:
            value = getattr(context, key, None)
            if key in ("execute_history", "progress") or value in (None, "", [], {}):
                continue
            current_value = getattr(self.context, key, None)
            if isinstance(value, list):
                if current_value is None:
                    setattr(self.context, key, value)
                    continue
                for item in value:
                    if item not in current_value:
                        current_value.append(item)
            elif isinstance(value, dict):
                if not current_value:
                    setattr(self.context, key, value)
                else:
                    current_value.update(value)
            elif not current_value:
                setattr(self.context, key, value)


class WorkflowChain(ChainBase):
    """
    工作流链
    """

    @eventmanager.register(EventType.WorkflowExecute)
    def event_process(self, event: Event) -> None:
        """
        事件触发工作流执行
        """
        if not isinstance(event.event_data, dict):
            return
        workflow_id = event.event_data.get('workflow_id')
        if not workflow_id:
            return
        self.process(workflow_id, from_begin=False)

    @staticmethod
    def process(
            workflow_id: int,
            from_begin: Optional[bool] = True,
            progress_callback: Optional[Callable[..., None]] = None,
    ) -> Tuple[bool, str]:
        """
        处理工作流

        :param workflow_id: 工作流ID
        :param from_begin: 是否从头开始，默认为True
        :param progress_callback: 定时服务进度更新回调
        """
        workflow_execution = get_configured_workflow_execution()

        workflow = get_configured_workflow_query().get_sync(workflow_id)
        if not workflow:
            logger.warn(f"工作流 {workflow_id} 不存在")
            return False, "工作流不存在"
        try:
            validate_workflow_graph(workflow.actions, workflow.flows)
        except ValueError as err:
            return False, str(err)
        if not workflow.actions:
            logger.warn(f"工作流 {workflow.name} 无动作")
            return False, "工作流无动作"

        def save_step(
                action: Action,
                context: ActionContext,
                execution_state: dict[str, Any],
                completed: bool,
        ) -> None:
            """
            保存动作上下文和结构化执行状态。
            """
            persisted_action_id = (action.id or "") if completed else ""
            workflow_execution.step(
                workflow_id,
                action_id=persisted_action_id,
                context=_serialize_workflow_context(context),
                execution_state=_serialize_workflow_value(execution_state)
            )
            if progress_callback:
                runtime = execution_state.get("runtime") or {}
                finished_actions = int(runtime.get("finished_actions") or 0)
                total_actions = len(workflow.actions)
                progress_callback(
                    value=finished_actions / total_actions * 100,
                    text=(
                        f"工作流动作（{finished_actions}/{total_actions}）"
                        f"{action.name or action.type or action.id} "
                        f"{'执行完成' if completed else '执行中'}"
                    ),
                    data={
                        "total": total_actions,
                        "finished": finished_actions,
                        "current": action.id,
                    },
                )

        def commit_terminal(execution: WorkflowExecutor) -> None:
            """
            同 ID 所有权持有期间提交最后一次结果，避免覆盖后续执行的状态。
            """
            if execution.stopped:
                return
            if not execution.success or execution.has_failure:
                committed = workflow_execution.fail(workflow_id, result=execution.errmsg)
            else:
                committed = workflow_execution.success(workflow_id)
            if not committed:
                raise RuntimeError("工作流终态写入失败")

        executor = WorkflowExecutor(
            workflow, step_callback=save_step, from_begin=bool(from_begin), completion_callback=commit_terminal,
        )
        if not executor.admit():
            logger.warning("%s，拒绝执行 %s", executor.errmsg, workflow.name)
            return False, executor.errmsg
        started, message = WorkflowChain._start_execution(
            executor, workflow_execution, bool(from_begin), progress_callback,
        )
        if not started:
            return False, message
        executor.execute()

        if executor.stopped:
            logger.info(f"工作流 {workflow.name} 已停止")
            return False, executor.errmsg

        if not executor.success or executor.has_failure:
            logger.info(f"工作流 {workflow.name} 执行失败：{executor.errmsg}")
            return False, executor.errmsg
        logger.info(f"工作流 {workflow.name} 执行完成")
        if progress_callback:
            progress_callback(value=100, text=f"工作流 {workflow.name} 执行完成")
        return True, ""

    @staticmethod
    def _start_execution(
        executor: WorkflowExecutor,
        execution: WorkflowExecutionPort,
        from_begin: bool,
        progress_callback: Optional[Callable[..., None]],
    ) -> Tuple[bool, str]:
        """
        仅在获得同 ID 准入后重置并启动；事务拒绝或异常均释放未执行 任务持有者。
        """
        workflow = executor.workflow
        try:
            if from_begin and not execution.reset(workflow.id):
                executor.abort_before_execute()
                return False, "工作流重置失败"
            if not execution.start(workflow.id):
                executor.abort_before_execute()
                return False, "工作流启动失败"
            logger.info(f"开始执行工作流 {workflow.name}，共 {len(workflow.actions)} 个动作 ...")
            if progress_callback:
                progress_callback(
                    value=0,
                    text=f"开始执行工作流 {workflow.name} ...",
                    data={"total": len(workflow.actions), "finished": 0},
                )
        except Exception:
            executor.abort_before_execute()
            raise
        return True, ""

    @staticmethod
    def get_workflows() -> List[WorkflowSnapshot]:
        """
        获取工作流列表
        """
        return get_configured_workflow_query().list_enabled()

    @staticmethod
    def get_timer_workflows() -> List[WorkflowSnapshot]:
        """
        获取定时触发的工作流列表
        """
        return get_configured_workflow_query().list_timer_enabled()

    @staticmethod
    def get_event_workflows() -> List[WorkflowSnapshot]:
        """
        获取事件触发的工作流列表
        """
        return get_configured_workflow_query().list_event_enabled()