feat: 接入本地执行内核与运行面板
This commit is contained in:
yu committed
1 parent
32ca0a1af8
commit
6a2cc24898
23 files changed
+2937
-41
No files matched your search
@@ -0,0 +1,467 @@
|
||||
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<void> {
|
||||
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<T>(promise: Promise<T>, timeoutMs: number, signal: AbortSignal, label: string): Promise<T> {
|
||||
return new Promise<T>((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<string, unknown>): Record<string, unknown> {
|
||||
const args: Record<string, unknown> = {};
|
||||
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<string, unknown>);
|
||||
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<string, unknown>;
|
||||
}
|
||||
): Promise<RunResult> {
|
||||
const startedAt = new Date().toISOString();
|
||||
const startMs = Date.now();
|
||||
const vars: Record<string, unknown> = { ...(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<string, NodeState>();
|
||||
let status: RunResult["status"] = "succeeded";
|
||||
let failure: RunError | undefined;
|
||||
|
||||
options.emit({ type: "run.started", startedAt });
|
||||
|
||||
const order = topLevelOrder(flow);
|
||||
try {
|
||||
for (const node of order) {
|
||||
if (options.signal.aborted) throw new RuntimeError("cancelled", "运行已取消");
|
||||
await executeNode(node, 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
|
||||
};
|
||||
}
|
||||
|
||||
function topLevelOrder(flow: FlowDefinition): FlowNode[] {
|
||||
const top = flow.nodes.filter((node) => !node.parentId);
|
||||
if (top.length <= 1) return top;
|
||||
const topIds = new Set(top.map((node) => node.id));
|
||||
const incoming = new Map<string, number>();
|
||||
for (const edge of flow.edges) {
|
||||
if (topIds.has(edge.source) && topIds.has(edge.target)) {
|
||||
incoming.set(edge.target, (incoming.get(edge.target) || 0) + 1);
|
||||
}
|
||||
}
|
||||
const start =
|
||||
top.find((node) => !incoming.get(node.id) && node.componentId.startsWith("Trigger.")) ||
|
||||
top.find((node) => !incoming.get(node.id)) ||
|
||||
top[0];
|
||||
const ordered: FlowNode[] = [];
|
||||
const visited = new Set<string>();
|
||||
let current: FlowNode | undefined = start;
|
||||
while (current && !visited.has(current.id)) {
|
||||
ordered.push(current);
|
||||
visited.add(current.id);
|
||||
const edge = flow.edges.find((item) => item.source === current!.id && topIds.has(item.target));
|
||||
current = edge ? top.find((node) => node.id === edge.target) : undefined;
|
||||
}
|
||||
for (const node of [...top].sort((a, b) => a.position.y - b.position.y)) {
|
||||
if (!visited.has(node.id)) ordered.push(node);
|
||||
}
|
||||
return ordered;
|
||||
}
|
||||
|
||||
async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string, NodeState>): Promise<unknown> {
|
||||
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<string, NodeState>
|
||||
): Promise<unknown> {
|
||||
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 end = node.values?.end_index === "" || node.values?.end_index == null ? list.length : toNumber(node.values?.end_index, list.length);
|
||||
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 };
|
||||
Reference in new issue
Block a user