feat(engine): 取消贯通/超时真中止/重试门控/图校验（迁移阶段1-A3）
- 节点尝试级 AbortController：run 取消与节点超时汇聚到同一信号， handler 收到的 ctx.signal 即尝试信号，其内部 sleep/HTTP/子进程可被真正中止 - 超时中止底层后 handler 先抛 cancelled 的竞态归一为 timeout - executeNode 取消映射 NodeState=cancelled：不适用 skipOnError、不进异常分支 - 重试门控：CapabilityMeta.idempotent===false 时无视 retryCount（防重复副作用） - 新增 validateFlowGraph：重复ID/悬空边/自环/顶层环/孤儿节点/不可达节点， 阻断性错误在 runFlow 内统一收口为 validation_error - 新增 11 项测试；既有 error-route/integration 回归通过
This commit is contained in:
yu
parent
d972a18bda
commit
9b76dc6987
117
packages/rpa-core/src/engine/graph-validator.ts
Normal file
117
packages/rpa-core/src/engine/graph-validator.ts
Normal file
@ -0,0 +1,117 @@
|
||||
import type { FlowDefinition } from "../types";
|
||||
|
||||
/**
|
||||
* 流程图静态校验（任务：审查报告 P0-03）。
|
||||
* 在执行前拦截结构性错误，避免运行期以诡异方式失败。
|
||||
* 只做纯结构检查，不读目录/注册表（组件是否存在由运行时校验并给出友好错误）。
|
||||
*/
|
||||
|
||||
export interface FlowIssue {
|
||||
severity: "error" | "warning";
|
||||
code: string;
|
||||
nodeId?: string;
|
||||
edgeId?: string;
|
||||
message: string;
|
||||
}
|
||||
|
||||
const CONTROL_FLOW_COMPONENTS = new Set([
|
||||
"logic_control_if_else",
|
||||
"logic_control_count_for",
|
||||
"logic_control_array_for_v3",
|
||||
"logic_control_dictionary_for_v1",
|
||||
"logic_control_while",
|
||||
"infinite_loop",
|
||||
"logic_control_try_catch"
|
||||
]);
|
||||
|
||||
export function validateFlowGraph(flow: FlowDefinition): FlowIssue[] {
|
||||
const issues: FlowIssue[] = [];
|
||||
const nodeIds = new Set<string>();
|
||||
|
||||
for (const node of flow.nodes) {
|
||||
if (nodeIds.has(node.id)) {
|
||||
issues.push({ severity: "error", code: "duplicate_node_id", nodeId: node.id, message: `节点 ID 重复：${node.id}` });
|
||||
}
|
||||
nodeIds.add(node.id);
|
||||
if (node.parentId && !flow.nodes.some((candidate) => candidate.id === node.parentId)) {
|
||||
issues.push({
|
||||
severity: "error",
|
||||
code: "orphan_child",
|
||||
nodeId: node.id,
|
||||
message: `节点「${node.label || node.id}」挂在不存在的父节点 ${node.parentId} 下`
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
for (const edge of flow.edges) {
|
||||
if (!nodeIds.has(edge.source)) {
|
||||
issues.push({ severity: "error", code: "dangling_edge", edgeId: edge.id, message: `连线 ${edge.id} 的起点 ${edge.source} 不存在于画布` });
|
||||
}
|
||||
if (!nodeIds.has(edge.target)) {
|
||||
issues.push({ severity: "error", code: "dangling_edge", edgeId: edge.id, message: `连线 ${edge.id} 的终点 ${edge.target} 不存在于画布` });
|
||||
}
|
||||
if (edge.source === edge.target && nodeIds.has(edge.source)) {
|
||||
issues.push({
|
||||
severity: "error",
|
||||
code: "self_loop",
|
||||
edgeId: edge.id,
|
||||
nodeId: edge.source,
|
||||
message: `节点「${edge.source}」连接到自己，执行时会立即跳过`
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// 顶层图环检测：执行引擎从起点沿边游走且不重访，环意味着存在永远不会执行的分支或意图不明的回边
|
||||
const top = flow.nodes.filter((node) => !node.parentId);
|
||||
const topIds = new Set(top.map((node) => node.id));
|
||||
const edges = flow.edges.filter((edge) => topIds.has(edge.source) && topIds.has(edge.target));
|
||||
const adjacency = new Map<string, string[]>();
|
||||
for (const edge of edges) {
|
||||
if (edge.source === edge.target) continue;
|
||||
adjacency.set(edge.source, [...(adjacency.get(edge.source) || []), 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];
|
||||
|
||||
if (start) {
|
||||
const state = new Map<string, "visiting" | "done">();
|
||||
const reachable = new Set<string>();
|
||||
const visit = (nodeId: string, path: string[]): void => {
|
||||
if (state.get(nodeId) === "done") return;
|
||||
if (state.get(nodeId) === "visiting") {
|
||||
const cycleStart = path.indexOf(nodeId);
|
||||
const cycle = [...path.slice(cycleStart), nodeId].join(" → ");
|
||||
issues.push({ severity: "error", code: "cycle", nodeId, message: `顶层流程存在环，环上的节点执行顺序不明确：${cycle}` });
|
||||
return;
|
||||
}
|
||||
state.set(nodeId, "visiting");
|
||||
reachable.add(nodeId);
|
||||
for (const next of adjacency.get(nodeId) || []) visit(next, [...path, nodeId]);
|
||||
state.set(nodeId, "done");
|
||||
};
|
||||
visit(start.id, [start.id]);
|
||||
|
||||
for (const node of top) {
|
||||
if (!reachable.has(node.id)) {
|
||||
const knownContainer = CONTROL_FLOW_COMPONENTS.has(node.componentId);
|
||||
if (!knownContainer && !flow.edges.some((edge) => edge.target === node.id)) {
|
||||
issues.push({
|
||||
severity: "warning",
|
||||
code: "unreachable_node",
|
||||
nodeId: node.id,
|
||||
message: `节点「${node.label || node.id}」没有从起点可达的连线，执行时不会运行`
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return issues;
|
||||
}
|
||||
|
||||
export function hasBlockingIssues(issues: FlowIssue[]): boolean {
|
||||
return issues.some((issue) => issue.severity === "error");
|
||||
}
|
||||
@ -29,6 +29,12 @@ export interface CapabilityMeta {
|
||||
errorCodes?: string[];
|
||||
/** 执行中是否需要回传截图（Agent 侧视觉判断用） */
|
||||
needsScreenshot?: boolean;
|
||||
/**
|
||||
* 能力是否可安全重试（无副作用或副作用可重复）。
|
||||
* false 时运行时强制忽略节点的 retryCount（重试门控，防重复副作用）；
|
||||
* 缺省表示未知，遵循流程显式配置。
|
||||
*/
|
||||
idempotent?: boolean;
|
||||
/** 平台限制；省略表示跨平台 */
|
||||
platforms?: CapabilityPlatform[];
|
||||
}
|
||||
|
||||
202
packages/rpa-core/src/engine/runtime.cancel.test.ts
Normal file
202
packages/rpa-core/src/engine/runtime.cancel.test.ts
Normal file
@ -0,0 +1,202 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { CatalogComponent, FlowDefinition, FlowNode } from "../../types";
|
||||
import { validateFlowGraph, hasBlockingIssues } from "./graph-validator";
|
||||
import { register } from "./registry";
|
||||
import { RuntimeError } from "./types";
|
||||
import { runFlow } from "./runtime";
|
||||
|
||||
type RunFlowOptions = Parameters<typeof runFlow>[1];
|
||||
|
||||
function makeFlow(partial: Partial<FlowDefinition> & { nodes: FlowNode[] }): FlowDefinition {
|
||||
return {
|
||||
schemaVersion: 1,
|
||||
id: "flow_test",
|
||||
name: "测试流程",
|
||||
variables: [],
|
||||
nodes: partial.nodes,
|
||||
edges: partial.edges || []
|
||||
} as FlowDefinition;
|
||||
}
|
||||
|
||||
function node(id: string, componentId: string, overrides: Partial<FlowNode> = {}): FlowNode {
|
||||
return {
|
||||
id,
|
||||
componentId,
|
||||
position: { x: 0, y: 0 },
|
||||
values: {},
|
||||
options: {},
|
||||
...overrides
|
||||
};
|
||||
}
|
||||
|
||||
const fakeComponent = (id: string): CatalogComponent => ({
|
||||
id,
|
||||
name: id,
|
||||
version: "1.0.0",
|
||||
tags: [],
|
||||
platformTag: null,
|
||||
icon: "",
|
||||
risk: "low",
|
||||
fields: [],
|
||||
output: { type: "any", description: "" },
|
||||
slots: [],
|
||||
functionCalls: []
|
||||
});
|
||||
|
||||
const baseOptions = (signal: AbortSignal): RunFlowOptions => ({
|
||||
runId: "run_test",
|
||||
evidenceDir: ".",
|
||||
signal,
|
||||
emit: () => {},
|
||||
log: () => {},
|
||||
componentOf: (n: FlowNode) => fakeComponent(n.componentId),
|
||||
variables: {}
|
||||
});
|
||||
|
||||
const hangUntilAborted = (ctx: { signal: AbortSignal }): Promise<never> =>
|
||||
new Promise((_, reject) => {
|
||||
const onAbort = () => reject(new RuntimeError("cancelled", "运行已取消"));
|
||||
if (ctx.signal.aborted) return onAbort();
|
||||
ctx.signal.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
|
||||
register("Test_HangHandler", (ctx) => hangUntilAborted(ctx));
|
||||
|
||||
register("Test_SlowHandler", async (ctx) => {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const timer = setTimeout(resolve, 5000);
|
||||
const onAbort = () => {
|
||||
clearTimeout(timer);
|
||||
reject(new RuntimeError("cancelled", "运行已取消"));
|
||||
};
|
||||
if (ctx.signal.aborted) return onAbort();
|
||||
ctx.signal.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
return "done";
|
||||
});
|
||||
|
||||
let nonIdempotentCalls = 0;
|
||||
register("Test_NonIdempotent", async () => {
|
||||
nonIdempotentCalls += 1;
|
||||
throw new Error("boom");
|
||||
}, { idempotent: false });
|
||||
|
||||
let idempotentCalls = 0;
|
||||
register("Test_Idempotent", async () => {
|
||||
idempotentCalls += 1;
|
||||
if (idempotentCalls < 3) throw new Error("transient");
|
||||
return "ok";
|
||||
}, { idempotent: true });
|
||||
|
||||
describe("validateFlowGraph", () => {
|
||||
it("正常图无问题", () => {
|
||||
const flow = makeFlow({
|
||||
nodes: [node("a", "Trigger.Manual"), node("b", "File_Read")],
|
||||
edges: [{ id: "e1", source: "a", target: "b" }]
|
||||
});
|
||||
expect(hasBlockingIssues(validateFlowGraph(flow))).toBe(false);
|
||||
});
|
||||
|
||||
it("重复节点 ID 记为错误", () => {
|
||||
const flow = makeFlow({ nodes: [node("a", "Trigger.Manual"), node("a", "File_Read")] });
|
||||
const issues = validateFlowGraph(flow);
|
||||
expect(issues.some((issue) => issue.code === "duplicate_node_id" && issue.severity === "error")).toBe(true);
|
||||
});
|
||||
|
||||
it("悬空边记为错误", () => {
|
||||
const flow = makeFlow({
|
||||
nodes: [node("a", "Trigger.Manual")],
|
||||
edges: [{ id: "e1", source: "a", target: "ghost" }]
|
||||
});
|
||||
expect(validateFlowGraph(flow).some((issue) => issue.code === "dangling_edge")).toBe(true);
|
||||
});
|
||||
|
||||
it("自环记为错误", () => {
|
||||
const flow = makeFlow({
|
||||
nodes: [node("a", "Trigger.Manual")],
|
||||
edges: [{ id: "e1", source: "a", target: "a" }]
|
||||
});
|
||||
expect(validateFlowGraph(flow).some((issue) => issue.code === "self_loop")).toBe(true);
|
||||
});
|
||||
|
||||
it("顶层环记为错误", () => {
|
||||
const flow = makeFlow({
|
||||
nodes: [node("a", "Trigger.Manual"), node("b", "File_Read"), node("c", "File_Write")],
|
||||
edges: [
|
||||
{ id: "e1", source: "a", target: "b" },
|
||||
{ id: "e2", source: "b", target: "c" },
|
||||
{ id: "e3", source: "c", target: "b" }
|
||||
]
|
||||
});
|
||||
expect(validateFlowGraph(flow).some((issue) => issue.code === "cycle" && issue.severity === "error")).toBe(true);
|
||||
});
|
||||
|
||||
it("孤儿子节点与不可达节点", () => {
|
||||
const flow = makeFlow({
|
||||
nodes: [node("a", "Trigger.Manual"), { ...node("x", "File_Read"), parentId: "ghost" }, node("y", "Http_Get")]
|
||||
});
|
||||
const issues = validateFlowGraph(flow);
|
||||
expect(issues.some((issue) => issue.code === "orphan_child")).toBe(true);
|
||||
expect(issues.some((issue) => issue.code === "unreachable_node")).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("runtime 取消/超时/重试门控", () => {
|
||||
it("超时会中止监听尝试信号的 handler", async () => {
|
||||
const flow = makeFlow({
|
||||
nodes: [{ ...node("slow", "Test_SlowHandler"), options: { timeoutMs: 120 } }]
|
||||
});
|
||||
const controller = new AbortController();
|
||||
const result = await runFlow(flow, baseOptions(controller.signal));
|
||||
expect(result.status).toBe("failed");
|
||||
expect(result.error?.code).toBe("timeout");
|
||||
expect(result.error?.reason).toContain("执行超时");
|
||||
});
|
||||
|
||||
it("运行取消时节点状态为 cancelled 且不走 error route", async () => {
|
||||
const flow = makeFlow({
|
||||
nodes: [node("n1", "Test_HangHandler"), node("err", "File_Write")],
|
||||
edges: [{ id: "e1", source: "n1", target: "err", sourceHandle: "error" }]
|
||||
});
|
||||
const controller = new AbortController();
|
||||
setTimeout(() => controller.abort(), 50);
|
||||
const result = await runFlow(flow, baseOptions(controller.signal));
|
||||
expect(result.status).toBe("cancelled");
|
||||
const nodeState = result.nodes.find((item) => item.nodeId === "n1");
|
||||
expect(nodeState?.status).toBe("cancelled");
|
||||
// error 边存在，但取消不应路由到 err 节点
|
||||
expect(result.nodes.some((item) => item.nodeId === "err")).toBe(false);
|
||||
});
|
||||
|
||||
it("非幂等能力忽略 retryCount", async () => {
|
||||
nonIdempotentCalls = 0;
|
||||
const flow = makeFlow({
|
||||
nodes: [{ ...node("r", "Test_NonIdempotent"), options: { retryCount: 3, retryIntervalMs: 1000 } }]
|
||||
});
|
||||
const controller = new AbortController();
|
||||
const result = await runFlow(flow, baseOptions(controller.signal));
|
||||
expect(result.status).toBe("failed");
|
||||
expect(nonIdempotentCalls).toBe(1);
|
||||
});
|
||||
|
||||
it("幂等能力遵循 retryCount 重试后成功", async () => {
|
||||
idempotentCalls = 0;
|
||||
const flow = makeFlow({
|
||||
nodes: [{ ...node("r", "Test_Idempotent"), options: { retryCount: 3, retryIntervalMs: 1000, timeoutMs: 5000 } }]
|
||||
});
|
||||
const controller = new AbortController();
|
||||
const result = await runFlow(flow, baseOptions(controller.signal));
|
||||
expect(result.status).toBe("succeeded");
|
||||
expect(idempotentCalls).toBe(3);
|
||||
});
|
||||
|
||||
it("流程图校验失败返回 validation_error 且不执行任何节点", async () => {
|
||||
const flow = makeFlow({ nodes: [node("a", "Trigger.Manual"), node("a", "File_Read")] });
|
||||
const controller = new AbortController();
|
||||
const result = await runFlow(flow, baseOptions(controller.signal));
|
||||
expect(result.status).toBe("failed");
|
||||
expect(result.error?.code).toBe("validation_error");
|
||||
expect(result.error?.reason).toContain("流程图校验未通过");
|
||||
expect(result.nodes).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
@ -1,6 +1,7 @@
|
||||
import type { FlowDefinition, FlowNode, RunError } from "../types";
|
||||
import { evaluate, evaluateCondition, interpolate, toNumber } from "./expression";
|
||||
import { getHandler } from "./registry";
|
||||
import { hasBlockingIssues, validateFlowGraph } from "./graph-validator";
|
||||
import { getCapabilityMeta, getHandler } from "./registry";
|
||||
import {
|
||||
asRuntimeError,
|
||||
BreakSignal,
|
||||
@ -138,6 +139,17 @@ export async function runFlow(
|
||||
options.emit({ type: "run.started", startedAt });
|
||||
|
||||
try {
|
||||
const issues = validateFlowGraph(flow);
|
||||
for (const issue of issues.filter((item) => item.severity === "warning")) {
|
||||
options.log(issue.message, "warn");
|
||||
}
|
||||
if (hasBlockingIssues(issues)) {
|
||||
const summary = issues.filter((issue) => issue.severity === "error").map((item) => item.message).join("；");
|
||||
throw new RuntimeError("validation_error", `流程图校验未通过：${summary}`, {
|
||||
recommendation: "修复流程图结构后再运行",
|
||||
issues: JSON.stringify(issues.filter((issue) => issue.severity === "error"))
|
||||
});
|
||||
}
|
||||
await executeTopLevelGraph(flow, ctx, states);
|
||||
} catch (error) {
|
||||
if (error instanceof StopSignal) {
|
||||
@ -202,6 +214,9 @@ async function executeTopLevelGraph(flow: FlowDefinition, ctx: RunContext, state
|
||||
try {
|
||||
await executeNode(node, ctx, states);
|
||||
} catch (error) {
|
||||
if (error instanceof BreakSignal || error instanceof ContinueSignal || error instanceof StopSignal) throw error;
|
||||
// 取消不进异常分支：中止运行时直接上抛，由 runFlow 统一收口为 cancelled
|
||||
if (asRuntimeError(error).code === "cancelled") throw 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;
|
||||
@ -273,23 +288,48 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
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 effectiveTimeout = timeoutMs > 0 ? timeoutMs : 30000;
|
||||
const meta = getCapabilityMeta(node.componentId) || getCapabilityMeta(component.functionName);
|
||||
let retries = Math.max(0, Math.min(100, Number(node.options?.retryCount || 0)));
|
||||
// 重试门控：实现方声明非幂等的能力，无视流程里的重试配置（防重复副作用）
|
||||
if (retries > 0 && meta?.idempotent === false) {
|
||||
retries = 0;
|
||||
ctx.log(`能力「${component.name}」为非幂等操作，已忽略重试设置（retryCount）`, "warn", node.id);
|
||||
}
|
||||
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;
|
||||
// 尝试级取消源：run 取消与节点超时都汇聚到它，真正中止底层操作
|
||||
const attemptCtl = new AbortController();
|
||||
const onRunAbort = () => attemptCtl.abort();
|
||||
if (ctx.signal.aborted) attemptCtl.abort();
|
||||
else ctx.signal.addEventListener("abort", onRunAbort, { once: true });
|
||||
let timeoutError: RuntimeError | undefined;
|
||||
const timeoutTimer = setTimeout(() => {
|
||||
timeoutError = new RuntimeError("timeout", `${component.name} 执行超时（${effectiveTimeout}ms）`, {
|
||||
recommendation: "检查目标是否可达，或调大该节点的超时时间",
|
||||
nodeId: node.id,
|
||||
componentId: component.id
|
||||
});
|
||||
attemptCtl.abort(timeoutError);
|
||||
}, effectiveTimeout);
|
||||
// handler 收到的 ctx.signal 是尝试级信号：它内部的 sleep/HTTP/子进程监听它即可被中止
|
||||
const attemptCtx: RunContext = { ...ctx, signal: attemptCtl.signal };
|
||||
try {
|
||||
result = await withTimeout(
|
||||
Promise.resolve(handler(ctx, node, args, component)),
|
||||
timeoutMs,
|
||||
ctx.signal,
|
||||
Promise.resolve(handler(attemptCtx, node, args, component)),
|
||||
effectiveTimeout,
|
||||
attemptCtl.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);
|
||||
// 超时中止底层后，handler 可能先抛出 cancelled；若中止原因是超时则归一为 timeout
|
||||
if (attemptCtl.signal.aborted && timeoutError && runtimeError.code === "cancelled") throw timeoutError;
|
||||
if (attempt > retries || !isRetryable(runtimeError.code)) {
|
||||
throw new RuntimeError(runtimeError.code, runtimeError.reason, {
|
||||
recommendation: runtimeError.recommendation,
|
||||
@ -304,6 +344,9 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
node.id
|
||||
);
|
||||
await sleep(interval, ctx.signal);
|
||||
} finally {
|
||||
clearTimeout(timeoutTimer);
|
||||
ctx.signal.removeEventListener("abort", onRunAbort);
|
||||
}
|
||||
}
|
||||
if (node.options?.outputVar) ctx.vars[node.options.outputVar] = result;
|
||||
@ -332,6 +375,22 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
throw error;
|
||||
}
|
||||
const runtimeError = asRuntimeError(error, node, component.id);
|
||||
// 取消语义：用户中止运行时节点状态为 cancelled，不适用 skipOnError，也不进异常分支
|
||||
if (runtimeError.code === "cancelled" || ctx.signal.aborted) {
|
||||
state.error = runtimeError;
|
||||
state.status = "cancelled";
|
||||
state.finishedAt = new Date().toISOString();
|
||||
state.durationMs = Date.now() - startMs;
|
||||
ctx.emit({
|
||||
type: "node.finished",
|
||||
nodeId: node.id,
|
||||
status: "cancelled",
|
||||
error: runtimeError,
|
||||
durationMs: state.durationMs,
|
||||
evidence: []
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
state.error = runtimeError;
|
||||
state.finishedAt = new Date().toISOString();
|
||||
state.durationMs = Date.now() - startMs;
|
||||
|
||||
Loading…
Reference in New Issue
Block a user