PRA-only-demo/packages/rpa-core/test/engine/checkpoint-events.test.ts

import { describe, expect, it } from "vitest";
import type { CatalogComponent, FlowDefinition, FlowNode, RunEvent } from "../../src/types";
import { register } from "../../src/engine/registry";
import { NODE_STATUS_TO_STEP_STATUS, toStepStatus } from "../../src/engine/types";
import { runFlow } from "../../src/engine/runtime";

type EmittedEvent = Omit<RunEvent, "runId">;
type RunFlowOptions = Parameters<typeof runFlow>[1];

function makeFlow(partial: Partial<FlowDefinition> & { nodes: FlowNode[] }): FlowDefinition {
  return {
    schemaVersion: 1,
    id: "flow_checkpoint",
    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): FlowNode {
  return { id, componentId, position: { x: 0, y: 0 }, values: {}, options: {} };
}

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 = (emit: (event: EmittedEvent) => void): RunFlowOptions => ({
  runId: "run_cp",
  evidenceDir: ".",
  signal: new AbortController().signal,
  emit,
  log: () => {},
  componentOf: (n: FlowNode) => fakeComponent(n.componentId),
  variables: {}
});

register("Cp_Idempotent", async () => "ok", { idempotent: true });
register("Cp_NonIdempotent", async () => "ok", { idempotent: false });
register("Cp_DefaultMeta", async () => "ok");

describe("状态映射（NodeState ↔ 对面 StepStatus）", () => {
  it("8 个状态全分支映射，唯一非同名是 idle→pending", () => {
    expect(toStepStatus("idle")).toBe("pending");
    for (const status of ["running", "waiting_user", "succeeded", "failed", "cancelled", "unknown", "skipped"] as const) {
      expect(toStepStatus(status)).toBe(status);
    }
  });

  it("映射表为全量双射（无遗漏键）", () => {
    expect(Object.keys(NODE_STATUS_TO_STEP_STATUS).sort()).toEqual(
      ["cancelled", "failed", "idle", "running", "skipped", "succeeded", "unknown", "waiting_user"]
    );
  });
});

describe("A++ 检查点事件负载", () => {
  it("node.finished 携带 checkpointId/resumable/幂等键，非幂等不带键", async () => {
    const events: EmittedEvent[] = [];
    const flow = makeFlow({
      nodes: [node("first", "Cp_Idempotent"), node("second", "Cp_NonIdempotent"), node("third", "Cp_DefaultMeta")],
      edges: [
        { id: "e1", source: "first", target: "second" },
        { id: "e2", source: "second", target: "third" }
      ]
    });
    const result = await runFlow(flow, baseOptions((event) => events.push(event)));
    expect(result.status).toBe("succeeded");

    const finished = events.filter((e) => e.type === "node.finished" && e.status === "succeeded");
    expect(finished).toHaveLength(3);

    const first = finished.find((e) => e.nodeId === "first")!;
    expect(first.checkpointId).toBe("first");
    expect(first.resumable).toBe(true);
    expect(first.idempotencyKey).toBe("run_cp:first");
    // 双端透明调度（ADR-R6 补记）：本地执行驻地必须随事件上报，供审计区分
    expect(first.executor).toBe("local");
    // attempt 审计字段（无人值守审批对账：actionId=runId:stepId:attempt）
    expect(first.attempt).toBe(1);

    const second = finished.find((e) => e.nodeId === "second")!;
    expect(second.checkpointId).toBe("second");
    expect(second.resumable).toBe(false);
    expect(second.idempotencyKey).toBeUndefined();

    const third = finished.find((e) => e.nodeId === "third")!;
    expect(third.resumable).toBe(true);
    expect(third.idempotencyKey).toBe("run_cp:third");
  });

  it("幂等键由 runId+nodeId 派生，跨重入稳定（重入去重的基石）", async () => {
    const flow = makeFlow({ nodes: [node("only", "Cp_Idempotent")] });
    const runOnce = async (runId: string): Promise<string | undefined> => {
      const list: EmittedEvent[] = [];
      await runFlow(flow, { ...baseOptions((event) => list.push(event)), runId });
      return list.find((e) => e.type === "node.finished")?.idempotencyKey;
    };
    expect(await runOnce("run_a")).toBe("run_a:only");
    expect(await runOnce("run_a")).toBe("run_a:only");
    expect(await runOnce("run_b")).toBe("run_b:only");
  });

  it("attempt 字段全事件覆盖：失败/跳过/取消/成功都带尝试序号，重试节点终态 attempt 反映实际次数", async () => {
    let calls = 0;
    register("Cp_Flaky", async () => {
      calls += 1;
      if (calls < 3) throw new Error("transient");
      return "ok";
    }, { idempotent: true });
    const events: EmittedEvent[] = [];
    const flaky = { ...node("flaky", "Cp_Flaky"), options: { retryCount: 2, retryIntervalMs: 1 } };
    const flow = makeFlow({
      nodes: [flaky, { ...node("skip", "Cp_Idempotent"), enabled: false }],
      edges: [{ id: "e1", source: "flaky", target: "skip" }]
    });
    await runFlow(flow, { ...baseOptions((event) => events.push(event)) });

    const byNode = (id: string) => events.filter((e) => e.type === "node.finished" && e.nodeId === id);
    const flakyFinished = byNode("flaky");
    expect(flakyFinished).toHaveLength(1);
    expect(flakyFinished[0].status).toBe("succeeded");
    expect(flakyFinished[0].attempt).toBe(3); // 第 3 次尝试成功
    expect(byNode("skip")[0].attempt).toBe(1); // 跳过节点也是 1
    for (const event of events) {
      if (event.type === "node.finished") expect(event.attempt).toBeGreaterThanOrEqual(1);
    }
  }, 10_000);
});