diff --git a/packages/rpa-core/src/engine/graph-validator.ts b/packages/rpa-core/src/engine/graph-validator.ts new file mode 100644 index 0000000..cd57128 --- /dev/null +++ b/packages/rpa-core/src/engine/graph-validator.ts @@ -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(); + + 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(); + 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(); + const reachable = new Set(); + 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"); +} diff --git a/packages/rpa-core/src/engine/registry.ts b/packages/rpa-core/src/engine/registry.ts index 539a299..8b24144 100644 --- a/packages/rpa-core/src/engine/registry.ts +++ b/packages/rpa-core/src/engine/registry.ts @@ -29,6 +29,12 @@ export interface CapabilityMeta { errorCodes?: string[]; /** 执行中是否需要回传截图（Agent 侧视觉判断用） */ needsScreenshot?: boolean; + /** + * 能力是否可安全重试（无副作用或副作用可重复）。 + * false 时运行时强制忽略节点的 retryCount（重试门控，防重复副作用）； + * 缺省表示未知，遵循流程显式配置。 + */ + idempotent?: boolean; /** 平台限制；省略表示跨平台 */ platforms?: CapabilityPlatform[]; } diff --git a/packages/rpa-core/src/engine/runtime.cancel.test.ts b/packages/rpa-core/src/engine/runtime.cancel.test.ts new file mode 100644 index 0000000..7e41aaf --- /dev/null +++ b/packages/rpa-core/src/engine/runtime.cancel.test.ts @@ -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[1]; + +function makeFlow(partial: Partial & { 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 { + 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 => + 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((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); + }); +}); diff --git a/packages/rpa-core/src/engine/runtime.ts b/packages/rpa-core/src/engine/runtime.ts index 9e2e519..c42af09 100644 --- a/packages/rpa-core/src/engine/runtime.ts +++ b/packages/rpa-core/src/engine/runtime.ts @@ -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 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