feat(engine): node.finished 全事件补 attempt 审计字段(无人值守审批对账:对面 actionId=runId:stepId:attempt)
目标:审计侧可对账'第几次尝试成功/失败',为无人值守高危动作的审批绑定提供键 代码逻辑:六处 node.finished 发射点统一带 state.attempt(初值1,重试循环内递增);可选字段向后兼容,无事件对象精确断言受影响 验证:引擎域 24/24 全绿 + 新增契约测试(重试3次终态 attempt=3、跳过=1、全事件 ≥1);typecheck 归零 风险与回滚:低——纯新增可选字段,revert 即回
This commit is contained in:
parent
10d125ce9d
commit
43ae705629
@ -255,14 +255,14 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
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: [] });
|
||||
ctx.emit({ type: "node.finished", nodeId: node.id, status: "failed", error: state.error, durationMs: 0, evidence: [], attempt: state.attempt });
|
||||
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: [] });
|
||||
ctx.emit({ type: "node.finished", nodeId: node.id, status: "skipped", durationMs: 0, evidence: [], attempt: state.attempt });
|
||||
return undefined;
|
||||
}
|
||||
|
||||
@ -371,6 +371,7 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
checkpointId: node.id,
|
||||
resumable,
|
||||
executor: "local",
|
||||
attempt: state.attempt,
|
||||
...(resumable ? { idempotencyKey: `${ctx.runId}:${node.id}` } : {})
|
||||
});
|
||||
return result;
|
||||
@ -379,7 +380,7 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
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: [] });
|
||||
ctx.emit({ type: "node.finished", nodeId: node.id, status: "skipped", durationMs: state.durationMs, evidence: [], attempt: state.attempt });
|
||||
throw error;
|
||||
}
|
||||
const runtimeError = asRuntimeError(error, node, component.id);
|
||||
@ -395,7 +396,8 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
status: "cancelled",
|
||||
error: runtimeError,
|
||||
durationMs: state.durationMs,
|
||||
evidence: []
|
||||
evidence: [],
|
||||
attempt: state.attempt
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
@ -412,7 +414,8 @@ async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string,
|
||||
status: state.status,
|
||||
error: runtimeError,
|
||||
durationMs: state.durationMs,
|
||||
evidence: []
|
||||
evidence: [],
|
||||
attempt: state.attempt
|
||||
});
|
||||
if (skip) return undefined;
|
||||
throw error;
|
||||
|
||||
@ -89,6 +89,8 @@ describe("A++ 检查点事件负载", () => {
|
||||
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");
|
||||
@ -111,4 +113,30 @@ describe("A++ 检查点事件负载", () => {
|
||||
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);
|
||||
});
|
||||
|
||||
Loading…
Reference in New Issue
Block a user