""" 管理工作流实例及动作执行生命周期。 """ import threading from time import monotonic, sleep from typing import Any, Dict, List, Optional, Tuple, cast from pydantic import BaseModel from app.application.workflow import ( WorkflowExecutionOwner, WorkflowSnapshot, get_configured_workflow_query, ) from app.foundation.reflection import ModuleHelper from app.foundation.singleton import Singleton from app.runtime.events import Event, eventmanager 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, ActionResult _WORKFLOW_STOP_TIMEOUT_SECONDS = 10.0 class _ActionDeadlineToken: """ 把单次动作截止时间与工作流取消信号组合为协作停止令牌。 """ def __init__(self, workflow_id: int, deadline: float, cancel_token: Optional[Any]) -> None: """ 保存单调时钟截止时间和上层工作流令牌。 """ self.workflow_id = workflow_id self.deadline = deadline self.cancel_token = cancel_token def is_cancelled(self) -> bool: """ 到达截止时间或收到上层停止请求时通知支持取消的动作退出。 """ return bool( monotonic() >= self.deadline or (self.cancel_token and self.cancel_token.is_cancelled()) or runtime_stop_state.is_workflow_stopped(self.workflow_id) ) class WorkflowManager(metaclass=Singleton): """ 工作流管理器 """ def __init__(self) -> None: """ 创建动作、事件触发器和活动执行 owner 注册表。 """ # 所有动作定义 self._lock = threading.RLock() self._actions: Dict[str, Any] = {} self._event_workflows: Dict[str, List[int]] = {} self._accepting_executions = True self._executions: Dict[int, WorkflowExecutionOwner] = {} self.init() def init(self) -> None: """ 扫描并登记工作流动作与事件触发器。 """ def filter_func(obj: object) -> bool: """ 只接纳在工作流动作包中定义的具体动作类。 """ if not isinstance(obj, type): return False if not hasattr(obj, 'execute') or not hasattr(obj, "name"): return False if obj.__name__ == "BaseAction": return False return obj.__module__.startswith("app.workflow.actions") # 加载所有动作 self._actions = {} actions = ModuleHelper.load( "app.workflow.actions", filter_func=lambda _, obj: filter_func(obj) ) for action in actions: logger.debug(f"加载动作: {action.__name__}") try: self._actions[action.__name__] = action except Exception as err: logger.error(f"加载动作失败: {action.__name__} - {err}") # 加载工作流事件触发器 self.load_workflow_events() def register_execution(self, owner: WorkflowExecutionOwner) -> bool: """ 在同一锁内登记执行，拒绝停机封口和同一工作流的重复准入。 """ with self._lock: if not self._accepting_executions: return False if any(execution.workflow_id == owner.workflow_id for execution in self._executions.values()): return False self._executions[id(owner)] = owner return True def unregister_execution(self, owner: WorkflowExecutionOwner) -> None: """ 仅在执行及其节点线程池真实终止后释放 owner。 """ with self._lock: self._executions.pop(id(owner), None) def stop( self, timeout: float = _WORKFLOW_STOP_TIMEOUT_SECONDS, ) -> bool: """ 封口工作流入口并有限等待全部活动执行。 :param timeout: 等待活动执行真实终止的最长秒数 :return: 全部执行终止并安全释放动作注册表时返回 True """ with self._lock: self._accepting_executions = False event_type_values = tuple(self._event_workflows) for event_type_str in event_type_values: self.remove_workflow_event(event_type_str=event_type_str) with self._lock: self._event_workflows = {} executions = tuple(self._executions.values()) converged = True for execution in executions: try: execution.request_stop() except Exception as err: converged = False logger.error("请求停止工作流执行失败：%s", err) deadline = monotonic() + max(0.0, timeout) for execution in executions: try: if not execution.wait_stopped( timeout=max(0.0, deadline - monotonic()), ): converged = False except Exception as err: converged = False logger.error("等待工作流执行停止失败：%s", err) with self._lock: converged = converged and not self._executions if converged: self._actions = {} return converged def execute(self, workflow_id: int, action: Action, context: Optional[ActionContext] = None, inputs: Optional[dict[str, Any]] = None, runtime: Optional[dict[str, Any]] = None, cancel_token: Optional[Any] = None) -> ActionResult: """ 执行工作流动作 """ if not context: context = ActionContext() action_type = action.type if not action_type or action_type not in self._actions: logger.error(f"未找到动作: {action.type} - {action.name}") return ActionResult(success=False, message=" ", context=context) retry_config = self._get_retry_config(action) max_attempts = int(retry_config["max_attempts"]) interval = retry_config["interval"] backoff = retry_config["backoff"] action_result = ActionResult(success=False, message="", context=context) for attempt in range(1, max_attempts + 1): if self._is_cancelled(workflow_id, cancel_token): return ActionResult(success=False, message="工作流已取消", context=context) runtime_data = { **(runtime or {}), "attempt": attempt, "max_attempts": max_attempts, "cancel_token": cancel_token, } action_result = self._execute_action_once( workflow_id=workflow_id, action=action, context=context, inputs=inputs or {}, runtime=runtime_data, cancel_token=cancel_token ) action_result.attempts = attempt context = action_result.context or context if action_result.success: logger.info(f"{action.name} 执行成功") return action_result if attempt < max_attempts and not self._is_cancelled(workflow_id, cancel_token): wait_seconds = interval * (backoff ** (attempt - 1)) logger.info(f"{action.name} 执行失败，{wait_seconds} 秒后重试（{attempt}/{max_attempts}）...") self._sleep_with_cancel(workflow_id, wait_seconds, cancel_token) logger.error(f"{action.name} 执行失败！") return action_result def excute(self, workflow_id: int, action: Action, context: Optional[ActionContext] = None) -> Tuple[bool, str, ActionContext]: """ 执行工作流动作，兼容历史拼写错误的方法名。 """ action_result = self.execute(workflow_id=workflow_id, action=action, context=context) return bool(action_result.success), action_result.message or "", action_result.context or context or ActionContext() @staticmethod def _normalize_action_result(result: Any, action_obj: Any, fallback_context: ActionContext) -> ActionResult: """ 将旧版动作上下文与新版结构化结果统一为动作执行结果。 """ if isinstance(result, ActionResult): result.context = result.context or fallback_context if result.message is None: result.message = action_obj.message return result return ActionResult( success=action_obj.success, message=action_obj.message, context=result or fallback_context ) def _execute_action_once(self, workflow_id: int, action: Action, context: ActionContext, inputs: dict[str, Any], runtime: dict[str, Any], cancel_token: Optional[Any]) -> ActionResult: """ 运行单次尝试，保留同步动作真实返回后再判断失败和超时的语义。 """ action_type = action.type if not action_type or action_type not in self._actions: return ActionResult(success=False, message="未找到动作", context=context) action_obj = self._actions[action_type](action.id) logger.info(f"执行动作: {action.id} - {action.name}") try: action_result = self._run_action_with_loop( workflow_id=workflow_id, action=action, action_obj=action_obj, context=context, inputs=inputs, runtime=runtime, cancel_token=cancel_token ) except Exception as err: logger.error(f"{action.name} 执行失败: {err}") return ActionResult(success=False, message=f"{err}", context=context) return action_result def _run_action_with_loop(self, workflow_id: int, action: Action, action_obj: Any, context: ActionContext, inputs: dict[str, Any], runtime: dict[str, Any], cancel_token: Optional[Any]) -> ActionResult: """ 循环前后均检查截止时间，并向动作传递协作令牌；不分离阻塞节点线程。 """ timeout = self._get_action_timeout(action) deadline = monotonic() + timeout if timeout else None action_token = _ActionDeadlineToken(workflow_id, deadline, cancel_token) if deadline is not None else cancel_token runtime = {**runtime, "cancel_token": action_token, "deadline": deadline} loop = self._get_action_data_value(action, "loop") loop_interval = self._get_action_data_value(action, "loop_interval") action_result = ActionResult(success=False, context=context) while True: if self._is_cancelled(workflow_id, cancel_token): return ActionResult(success=False, message="工作流已取消", context=action_result.context or context) if deadline is not None and monotonic() >= deadline: return ActionResult(success=False, message=f"动作执行超时（{timeout}秒）", context=action_result.context or context) action_result = self._call_action( workflow_id=workflow_id, action=action, action_obj=action_obj, context=action_result.context or context, inputs=inputs, runtime=runtime ) if self._is_cancelled(workflow_id, cancel_token): return ActionResult(success=False, message="工作流已取消", context=action_result.context or context) if deadline is not None and monotonic() >= deadline: return ActionResult(success=False, message=f"动作执行超时（{timeout}秒）", context=action_result.context or context) if not (loop and loop_interval and not action_obj.done): return action_result wait_seconds = float(loop_interval) if deadline is not None: wait_seconds = min(wait_seconds, max(0.0, deadline - monotonic())) logger.info(f"{action.name} 等待 {wait_seconds} 秒后继续执行 ...") self._sleep_with_cancel(workflow_id, wait_seconds, action_token) def _call_action(self, workflow_id: int, action: Action, action_obj: Any, context: ActionContext, inputs: dict[str, Any], runtime: dict[str, Any]) -> ActionResult: """ 兼容新旧动作入口并把真实同步返回值转换为统一结果。 """ if hasattr(action_obj, "execute_with_inputs"): result = action_obj.execute_with_inputs(workflow_id, action.data, inputs, runtime, context) else: result = action_obj.execute(workflow_id, action.data, context) return self._normalize_action_result(result, action_obj, context) @staticmethod def _get_action_data_value(action: Action, key: str) -> Any: """ 读取兼容旧版存放在 data 中的动作参数。 """ data = action.data or {} return data.get(key) if isinstance(data, dict) else None def _get_action_timeout(self, action: Action) -> Optional[int]: """ 读取新旧动作配置中的单次尝试超时秒数。 """ timeout = action.timeout or self._get_action_data_value(action, "timeout") return int(timeout) if timeout else None def _get_retry_config(self, action: Action) -> dict[str, float | int]: """ 归一化动作重试次数、间隔和退避配置。 """ retry_config = action.retry or self._get_action_data_value(action, "retry") or {} if isinstance(retry_config, BaseModel): retry_config = retry_config.model_dump(exclude_none=True) if not isinstance(retry_config, dict): retry_config = {} return { "max_attempts": max(int(retry_config.get("max_attempts") or 1), 1), "interval": max(float(retry_config.get("interval") or 0), 0), "backoff": max(float(retry_config.get("backoff") or 1), 1), } @staticmethod def _is_cancelled(workflow_id: int, cancel_token: Optional[Any]) -> bool: """ 组合调用方取消令牌和全局工作流停止状态。 """ if cancel_token and cancel_token.is_cancelled(): return True return runtime_stop_state.is_workflow_stopped(workflow_id) def _sleep_with_cancel(self, workflow_id: int, seconds: float, cancel_token: Optional[Any]) -> None: """ 分段等待，并在取消或截止时立即停止继续等待。 """ deadline = monotonic() + seconds while monotonic() < deadline: if self._is_cancelled(workflow_id, cancel_token): return sleep(max(0.0, min(0.1, deadline - monotonic()))) def list_actions(self) -> List[dict[str, Any]]: """ 获取所有动作 """ return [ { "type": key, "name": action.name, "description": action.description, "contract": action.get_contract(), "data": { "label": action.name, **action.data } } for key, action in self._actions.items() ] def get_action_contract(self, action_type: str) -> dict[str, Any]: """ 获取动作输入输出契约。 """ action = self._actions.get(action_type) if not action or not hasattr(action, "get_contract"): return {} return cast(dict[str, Any], action.get_contract()) def update_workflow_event(self, workflow: WorkflowSnapshot) -> None: """ 根据工作流当前配置更新事件触发器。 :param workflow: 待更新的工作流快照 """ # 工作流可能切换触发事件，先按工作流ID从所有事件映射中移除。 self.remove_workflow_event(workflow_id=workflow.id) # 如果工作流是事件触发类型且未被禁用 if workflow.trigger_type == "event" and workflow.state != 'P': # 注册事件触发器 self.register_workflow_event(workflow.id, workflow.event_type) def load_workflow_events(self, workflow_id: Optional[int] = None) -> None: """ 从持久化快照加载工作流触发事件。 :param workflow_id: 指定工作流标识；None 加载全部已启用的事件工作流 """ workflows = [] if workflow_id: workflow = get_configured_workflow_query().get_sync(workflow_id) if workflow: workflows = [workflow] else: workflows = get_configured_workflow_query().list_event_enabled() try: for workflow in workflows: self.update_workflow_event(workflow) except Exception as e: logger.error(f"加载事件触发工作流失败: {e}") def register_workflow_event( self, workflow_id: int, event_type_str: Optional[str] ) -> None: """ 为有效事件类型注册工作流触发器。 :param workflow_id: 待触发的工作流标识 :param event_type_str: 事件类型值；为空时不登记 """ if not event_type_str: return try: event_type = EventType(event_type_str) except ValueError: logger.error(f"无效的事件类型: {event_type_str}") return if event_type in EventType: with self._lock: if event_type.value not in self._event_workflows: self._event_workflows[event_type.value] = [] eventmanager.add_event_listener(event_type, self._handle_event) # 记录工作流事件触发器 if workflow_id not in self._event_workflows[event_type.value]: self._event_workflows[event_type.value].append(workflow_id) logger.info(f"已注册工作流 {workflow_id} 事件触发器: {event_type.value}") def remove_workflow_event( self, workflow_id: Optional[int] = None, event_type_str: Optional[str] = None, ) -> None: """ 移除指定工作流或事件类型对应的触发器。 :param workflow_id: 指定工作流标识；None 清空匹配事件的工作流 :param event_type_str: 指定事件类型；None 遍历已登记的全部事件 """ event_type_values = [event_type_str] if event_type_str else list(self._event_workflows.keys()) for event_type_value in event_type_values: try: event_type = EventType(event_type_value) except ValueError: logger.error(f"无效的事件类型: {event_type_value}") continue with self._lock: workflow_ids = self._event_workflows.get(event_type.value) if not workflow_ids: continue if workflow_id is None: workflow_ids.clear() elif workflow_id in workflow_ids: workflow_ids.remove(workflow_id) if not workflow_ids: self._event_workflows.pop(event_type.value, None) eventmanager.remove_event_listener(event_type, self._handle_event) logger.info(f"已移除工作流 {workflow_id or ''} 事件触发器") def _handle_event(self, event: Event) -> None: """ 收到事件后触发已登记且匹配的工作流。 :param event: 当前事件及其载荷 """ try: event_type = event.event_type event_type_str = str(getattr(event_type, "value", event_type)) with self._lock: if event_type_str not in self._event_workflows: return workflow_ids = self._event_workflows[event_type_str].copy() for workflow_id in workflow_ids: self._trigger_workflow(workflow_id, event) except Exception as e: logger.error(f"处理工作流事件失败: {e}") def _trigger_workflow(self, workflow_id: int, event: Event) -> None: """ 确认工作流状态与事件条件后发出执行事件。 :param workflow_id: 候选工作流标识 :param event: 触发执行的原始事件 """ try: # 检查工作流是否存在且启用 workflow = get_configured_workflow_query().get_sync(workflow_id) if not workflow or workflow.state == 'P': return # 检查事件条件 if not self._check_event_conditions(workflow, event): logger.debug(f"工作流 {workflow.name} 事件条件不匹配，跳过执行") return # 检查工作流是否正在运行 if workflow.state == 'R': logger.warning(f"工作流 {workflow.name} 正在运行中，跳过重复触发") return event_type = event.event_type logger.info( f"事件 {getattr(event_type, 'value', event_type)} 触发工作流: {workflow.name}" ) # 发送工作流执行事件以启动工作流 eventmanager.send_event(EventType.WorkflowExecute, { "workflow_id": workflow_id, }) except Exception as e: logger.error(f"触发工作流 {workflow_id} 失败: {e}") def _check_event_conditions(self, workflow: Any, event: Event) -> bool: """ 检查事件是否满足工作流的触发条件 """ if not workflow.event_conditions: return True conditions = workflow.event_conditions event_data: Any = event.event_data if isinstance(event_data, BaseModel): event_data = event_data.model_dump() if not isinstance(event_data, dict): return False # 检查字段匹配条件 for field, expected_value in conditions.items(): if field not in event_data: return False actual_value = event_data[field] # 支持多种条件匹配方式 if isinstance(expected_value, dict): # 复杂条件匹配 if not self._check_complex_condition(actual_value, expected_value): return False else: # 简单值匹配 if actual_value != expected_value: return False return True @staticmethod def _check_complex_condition( actual_value: Any, condition: dict[str, Any] ) -> bool: """ 检查复杂条件匹配 支持的操作符：equals, not_equals, contains, not_contains, in, not_in, regex """ for operator, expected_value in condition.items(): if operator == "equals": if actual_value != expected_value: return False elif operator == "not_equals": if actual_value == expected_value: return False elif operator == "contains": if expected_value not in str(actual_value): return False elif operator == "not_contains": if expected_value in str(actual_value): return False elif operator == "in": if actual_value not in expected_value: return False elif operator == "not_in": if actual_value in expected_value: return False elif operator == "regex": import re if not re.search(expected_value, str(actual_value)): return False return True def get_event_workflows(self) -> dict[str, list[int]]: """ 获取所有事件触发的工作流 """ with self._lock: return self._event_workflows.copy()