PRA-only-demo/packages/rpa-core/test/engine/runtime.cancel.test.ts

import { describe, expect, it } from "vitest";
import type { CatalogComponent, FlowDefinition, FlowNode } from "../../src/types";
import { validateFlowGraph, hasBlockingIssues } from "../../src/engine/graph-validator";
import { register } from "../../src/engine/registry";
import { RuntimeError } from "../../src/engine/types";
import { runFlow } from "../../src/engine/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 || [],
    meta: { createdAt: "2026-01-01T00:00:00Z", updatedAt: "2026-01-01T00:00:00Z", appVersion: "test" }
  };
}

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: { label: "", key: "", suggestedName: "", valueTypes: [], fields: [] },
    slots: [],
    functionCalls: []
  }) as unknown as CatalogComponent;

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);
  });
});