import type { FlowDefinition, FlowNode, RunError } from "../../shared/types"; import { evaluate, evaluateCondition, interpolate, toNumber } from "./expression"; import { getHandler } from "./registry"; import { asRuntimeError, BreakSignal, ContinueSignal, RuntimeError, StopSignal, type NodeState, type RunContext, type RunResult } from "./types"; const MAX_LOOP = 10000; function sleep(ms: number, signal?: AbortSignal): Promise { if (ms <= 0) return Promise.resolve(); return new Promise((resolve, reject) => { const timer = setTimeout(() => { signal?.removeEventListener("abort", onAbort); resolve(); }, ms); const onAbort = () => { clearTimeout(timer); reject(new RuntimeError("cancelled", "运行已取消")); }; if (signal?.aborted) return onAbort(); signal?.addEventListener("abort", onAbort, { once: true }); }); } function withTimeout(promise: Promise, timeoutMs: number, signal: AbortSignal, label: string): Promise { return new Promise((resolve, reject) => { let settled = false; const finish = (fn: () => void) => { if (settled) return; settled = true; clearTimeout(timer); signal.removeEventListener("abort", onAbort); fn(); }; const timer = setTimeout(() => { finish(() => reject(new RuntimeError("timeout", `${label} 执行超时(${timeoutMs}ms)`, { recommendation: "检查目标是否可达,或调大该节点的超时时间" }))); }, timeoutMs > 0 ? timeoutMs : 30000); const onAbort = () => finish(() => reject(new RuntimeError("cancelled", "运行已取消"))); if (signal.aborted) return onAbort(); signal.addEventListener("abort", onAbort, { once: true }); promise.then( (value) => finish(() => resolve(value)), (error) => finish(() => reject(error)) ); }); } function isRetryable(code: string): boolean { return !["validation_error", "capability_not_implemented", "blocked_url", "cancelled", "expression_error"].includes(code); } function resolveArgs(node: FlowNode, component: { fields: Array<{ key: string; defaultValue: string }> }, vars: Record): Record { const args: Record = {}; for (const field of component.fields) { const raw = Object.prototype.hasOwnProperty.call(node.values, field.key) ? node.values[field.key] : field.defaultValue; try { args[field.key] = interpolate(raw, vars); } catch (error) { throw new RuntimeError("expression_error", `参数 ${field.key} 求值失败:${(error as Error).message}`, { arg: field.key, recommendation: "检查 ${...} 中的变量名与表达式语法" }); } } for (const [key, value] of Object.entries(node.values || {})) { if (key in args) continue; try { args[key] = interpolate(value, vars); } catch (error) { throw new RuntimeError("expression_error", `参数 ${key} 求值失败:${(error as Error).message}`, { arg: key }); } } return args; } function childChain(flow: FlowDefinition, parentId: string, slot: string): FlowNode[] { return flow.nodes.filter((node) => node.parentId === parentId && node.slot === slot); } function asArray(value: unknown): unknown[] { if (Array.isArray(value)) return value; if (value == null || value === "") return []; if (typeof value === "string") { try { const parsed = JSON.parse(value); return Array.isArray(parsed) ? parsed : [parsed]; } catch { return value.split(/[,,\n]/).map((item) => item.trim()).filter(Boolean); } } if (typeof value === "object") return Object.values(value as Record); return [value]; } export async function runFlow( flow: FlowDefinition, options: { runId: string; evidenceDir: string; signal: AbortSignal; emit: RunContext["emit"]; log: RunContext["log"]; componentOf: RunContext["componentOf"]; browser?: RunContext["browser"]; variables?: Record; } ): Promise { const startedAt = new Date().toISOString(); const startMs = Date.now(); const vars: Record = { ...(options.variables || {}) }; const ctx: RunContext = { runId: options.runId, flowId: flow.id, flow, vars, secrets: {}, signal: options.signal, evidenceDir: options.evidenceDir, browser: options.browser, emit: options.emit, log: options.log, componentOf: options.componentOf }; const states = new Map(); let status: RunResult["status"] = "succeeded"; let failure: RunError | undefined; options.emit({ type: "run.started", startedAt }); try { await executeTopLevelGraph(flow, ctx, states); } catch (error) { if (error instanceof StopSignal) { options.log("流程被「退出流程」节点终止", "warn"); status = "succeeded"; } else if (error instanceof BreakSignal || error instanceof ContinueSignal) { failure = { code: "validation_error", reason: "跳出/继续语句不在循环体内" }; status = "failed"; } else { const runtimeError = asRuntimeError(error); failure = runtimeError; status = runtimeError.code === "cancelled" ? "cancelled" : "failed"; } } const finishedAt = new Date().toISOString(); const durationMs = Date.now() - startMs; options.emit({ type: "run.finished", status, durationMs, error: failure, startedAt }); return { runId: options.runId, status, startedAt, finishedAt, durationMs, nodes: [...states.values()], variables: vars, error: failure }; } async function executeTopLevelGraph(flow: FlowDefinition, ctx: RunContext, states: Map): Promise { const top = flow.nodes.filter((node) => !node.parentId); if (!top.length) return; if (!flow.edges.length) { for (const node of [...top].sort((a, b) => a.position.y - b.position.y || a.position.x - b.position.x)) { if (ctx.signal.aborted) throw new RuntimeError("cancelled", "运行已取消"); await executeNode(node, ctx, states); } return; } const topIds = new Set(top.map((node) => node.id)); const topById = new Map(top.map((node) => [node.id, node])); const edges = flow.edges.filter((edge) => topIds.has(edge.source) && topIds.has(edge.target)); const incoming = new Set(edges.map((edge) => edge.target)); const start = top.find((node) => !incoming.has(node.id) && node.componentId.startsWith("Trigger.")) || top.find((node) => !incoming.has(node.id)) || top[0]; const visited = new Set(); const walk = async (node: FlowNode): Promise => { if (visited.has(node.id)) return; if (ctx.signal.aborted) throw new RuntimeError("cancelled", "运行已取消"); visited.add(node.id); try { await executeNode(node, ctx, states); } catch (error) { const errorEdge = edges.find((edge) => edge.source === node.id && edge.sourceHandle === "error"); const target = errorEdge ? topById.get(errorEdge.target) : undefined; if (!target) throw error; ctx.vars.error = asRuntimeError(error, node, node.componentId); ctx.log(`节点失败,转入异常分支:${target.label || target.componentId}`, "warn", node.id); await walk(target); return; } const normalEdge = edges.find((edge) => edge.source === node.id && edge.sourceHandle !== "error"); const target = normalEdge ? topById.get(normalEdge.target) : undefined; if (target) await walk(target); }; await walk(start); } async function executeNode(node: FlowNode, ctx: RunContext, states: Map): Promise { const component = ctx.componentOf(node); const state: NodeState = { nodeId: node.id, componentId: node.componentId, status: "running", startedAt: new Date().toISOString(), evidence: [], attempt: 1 }; states.set(node.id, state); ctx.emit({ type: "node.started", nodeId: node.id, attempt: 1 }); if (!component) { const error = new RuntimeError("validation_error", `组件 ${node.componentId} 不在目录中`, { nodeId: node.id, componentId: node.componentId, recommendation: "重新添加该节点" }); state.status = "failed"; state.error = error.toRunError(); state.finishedAt = new Date().toISOString(); ctx.emit({ type: "node.finished", nodeId: node.id, status: "failed", error: state.error, durationMs: 0, evidence: [] }); throw error; } if (node.enabled === false) { state.status = "skipped"; state.finishedAt = new Date().toISOString(); ctx.emit({ type: "node.finished", nodeId: node.id, status: "skipped", durationMs: 0, evidence: [] }); return undefined; } const startMs = Date.now(); try { let result: unknown; if (component.slots.length > 0) { result = await executeContainer(node, component.id, ctx, states); } else { const handler = getHandler(node.componentId) || getHandler(component.functionName); if (!handler) { throw new RuntimeError( "capability_not_implemented", `组件「${component.name}」当前版本未接入执行器`, { nodeId: node.id, componentId: node.componentId, recommendation: "请检查组件配置或重新添加节点" } ); } const args = resolveArgs(node, component, ctx.vars); const delayBefore = Math.max(0, Number(node.options?.delayBeforeMs || 0)); const delayAfter = Math.max(0, Number(node.options?.delayAfterMs || 0)); const timeoutMs = Math.max(0, Number(node.options?.timeoutMs || 30000)); const retries = Math.max(0, Math.min(100, Number(node.options?.retryCount || 0))); const interval = Math.max(1000, Number(node.options?.retryIntervalMs || 1000)); await sleep(delayBefore, ctx.signal); for (let attempt = 1; attempt <= retries + 1; attempt += 1) { state.attempt = attempt; try { result = await withTimeout( Promise.resolve(handler(ctx, node, args, component)), timeoutMs, ctx.signal, component.name ); break; } catch (error) { if (error instanceof BreakSignal || error instanceof ContinueSignal || error instanceof StopSignal) throw error; const runtimeError = asRuntimeError(error, node, component.id); if (attempt > retries || !isRetryable(runtimeError.code)) { throw new RuntimeError(runtimeError.code, runtimeError.reason, { recommendation: runtimeError.recommendation, nodeId: node.id, componentId: component.id, detail: runtimeError.detail }); } ctx.log( `第 ${attempt} 次执行失败:${runtimeError.reason},${Math.round(interval / 1000)} 秒后重试`, "warn", node.id ); await sleep(interval, ctx.signal); } } if (node.options?.outputVar) ctx.vars[node.options.outputVar] = result; await sleep(delayAfter, ctx.signal); } state.status = "succeeded"; state.output = result; state.finishedAt = new Date().toISOString(); state.durationMs = Date.now() - startMs; ctx.emit({ type: "node.finished", nodeId: node.id, status: "succeeded", output: result, durationMs: state.durationMs, evidence: state.evidence }); return result; } catch (error) { if (error instanceof BreakSignal || error instanceof ContinueSignal || error instanceof StopSignal) { state.status = "skipped"; state.finishedAt = new Date().toISOString(); state.durationMs = Date.now() - startMs; ctx.emit({ type: "node.finished", nodeId: node.id, status: "skipped", durationMs: state.durationMs, evidence: [] }); throw error; } const runtimeError = asRuntimeError(error, node, component.id); state.error = runtimeError; state.finishedAt = new Date().toISOString(); state.durationMs = Date.now() - startMs; const skip = Boolean(node.options?.skipOnError); state.status = skip ? "skipped" : "failed"; ctx.log(`${component.name} ${skip ? "失败但已跳过" : "失败"}:${runtimeError.reason}`, "error", node.id); if (node.options?.logException && runtimeError.detail) ctx.log(runtimeError.detail, "error", node.id); ctx.emit({ type: "node.finished", nodeId: node.id, status: state.status, error: runtimeError, durationMs: state.durationMs, evidence: [] }); if (skip) return undefined; throw error; } } async function executeContainer( node: FlowNode, componentId: string, ctx: RunContext, states: Map ): Promise { const runSlot = async (slot: string) => { for (const child of childChain(ctx.flow, node.id, slot)) { if (ctx.signal.aborted) throw new RuntimeError("cancelled", "运行已取消"); await executeNode(child, ctx, states); } }; switch (componentId) { case "logic_control_if_else": { const condition = evaluateCondition(node.values?.condition, ctx.vars); ctx.log(`条件 ${condition ? "成立" : "不成立"}:${String(node.values?.condition ?? "")}`, "info", node.id); await runSlot(condition ? "statement_if" : "statement_else"); return condition; } case "logic_control_count_for": { const start = toNumber(interpolate(node.values?.count_start_value, ctx.vars), 1); const end = toNumber(interpolate(node.values?.count_end_value, ctx.vars), 1); const step = toNumber(interpolate(node.values?.count_step, ctx.vars), 1) || 1; const indexName = String(interpolate(node.values?.count_index, ctx.vars) || "index"); let iterations = 0; for (let value = start; step > 0 ? value <= end : value >= end; value += step) { if (iterations++ > MAX_LOOP) throw new RuntimeError("validation_error", "循环次数超过上限 10000"); ctx.vars[indexName] = value; try { await runSlot("count_for_fn"); } catch (error) { if (error instanceof BreakSignal) break; if (error instanceof ContinueSignal) continue; throw error; } } return iterations; } case "logic_control_array_for_v3": case "logic_control_dictionary_for_v1": { const raw = node.values?.array_name; let value = interpolate(raw, ctx.vars); if (typeof value === "string") { const name = value.trim(); if (name in ctx.vars) value = ctx.vars[name]; else { try { value = JSON.parse(name); } catch { value = asArray(name); } } } const list = asArray(value); const reverse = String(node.values?.reverse ?? "") === "true" || String(node.values?.reverse ?? "") === "1"; const start = Math.max(0, toNumber(node.values?.start_index, 0)); const requestedEnd = node.values?.end_index === "" || node.values?.end_index == null ? list.length : toNumber(node.values?.end_index, list.length); const end = requestedEnd < 0 ? Math.max(start, list.length + requestedEnd + 1) : requestedEnd; const slice = (reverse ? [...list].reverse() : list).slice(start, end); let iterations = 0; for (let index = 0; index < slice.length; index += 1) { if (iterations++ > MAX_LOOP) throw new RuntimeError("validation_error", "循环次数超过上限 10000"); ctx.vars.item = slice[index]; ctx.vars.index = index; ctx.vars.value = slice[index]; try { await runSlot("array_for_fn"); } catch (error) { if (error instanceof BreakSignal) break; if (error instanceof ContinueSignal) continue; throw error; } } return iterations; } case "logic_control_while": { let iterations = 0; while (evaluateCondition(node.values?.if_for_condition, ctx.vars)) { if (iterations++ > MAX_LOOP) throw new RuntimeError("validation_error", "循环次数超过上限 10000"); try { await runSlot("if_for_fn"); } catch (error) { if (error instanceof BreakSignal) break; if (error instanceof ContinueSignal) continue; throw error; } } return iterations; } case "infinite_loop": { for (let iterations = 0; ; iterations += 1) { if (iterations > MAX_LOOP) throw new RuntimeError("validation_error", "循环次数超过上限 10000"); try { await runSlot("loop_body"); } catch (error) { if (error instanceof BreakSignal) break; if (error instanceof ContinueSignal) continue; throw error; } } return undefined; } case "logic_control_try_catch": { try { await runSlot("try_fn"); await runSlot("else_fn"); return "else"; } catch (error) { if (error instanceof BreakSignal || error instanceof ContinueSignal || error instanceof StopSignal) throw error; ctx.vars.error = asRuntimeError(error, node, componentId); await runSlot("except_fn"); return "catch"; } } default: { const slot = componentId === "Element_exception_monitor_v2" ? "exception_monitor_params" : "body"; await runSlot(slot); return undefined; } } } export { evaluate, evaluateCondition, interpolate };