fix: 修复工作流试运行节点状态未随编辑更新

每轮执行刷新轮询节点并复用运行快照,隔离旧请求响应,缓存步骤排序。

补充新增与删除节点、配置更新、暂停恢复及轮询复用回归测试。
This commit is contained in:
2026-09-08 14:25:19 +08:00
parent c7c301d3b9
commit b5a355247b
4 changed files with 448 additions and 37 deletions

View File

@@ -94,6 +94,10 @@ const workflowInfo = ref<any>({});
const initializationError = ref(false);
const runParams = ref<any>(null);
const tinyFlowData = shallowRef<any>(null);
const runFlowData = shallowRef<any>(null);
const runNodes = computed(() =>
runFlowData.value ? sortNodes(runFlowData.value) : [],
);
const onlyRenderVisibleWorkflowElements = computed(
() =>
(tinyFlowData.value?.nodes?.length || 0) >=
@@ -592,6 +596,9 @@ function getRunningParams() {
.get(`/api/v1/workflow/getRunningParameters?id=${workflowId.value}`)
.then((res) => {
if (res.errorCode === 0) {
workflowForm.value?.reset();
runFlowData.value = tinyFlowData.value;
onSubmit();
runParams.value = res.data;
drawerVisible.value = true;
}
@@ -752,6 +759,7 @@ async function handlePublishAction() {
}
}
function onSubmit() {
chainInfo.value = null;
initState.value = !initState.value;
}
async function runIndependently(node: any) {
@@ -872,12 +880,12 @@ function onAsyncExecute(info: any) {
:workflow-params="runParams"
:on-submit="onSubmit"
:on-async-execute="onAsyncExecute"
:tiny-flow-data="tinyFlowData"
:tiny-flow-data="runFlowData"
/>
<div class="mb-2.5 font-semibold">{{ $t('aiWorkflow.steps') }}</div>
<WorkflowSteps
:workflow-id="workflowId"
:node-json="sortNodes(tinyFlowData)"
:node-json="runNodes"
:init-signal="initState"
:polling-data="chainInfo"
@resume="resumeChain"
@@ -887,7 +895,7 @@ function onAsyncExecute(info: any) {
</div>
<ExecResult
:workflow-id="workflowId"
:node-json="sortNodes(tinyFlowData)"
:node-json="runNodes"
:init-signal="initState"
:polling-data="chainInfo"
/>

View File

@@ -0,0 +1,194 @@
import { flushPromises, mount } from '@vue/test-utils';
import { afterEach, describe, expect, it, vi } from 'vitest';
import WorkflowForm from '../components/WorkflowForm.vue';
import WorkflowSteps from '../components/WorkflowSteps.vue';
import WorkflowDesign from '../WorkflowDesign.vue';
const { get, post, getData, sortNodes } = vi.hoisted(() => ({
get: vi.fn(),
post: vi.fn(),
getData: vi.fn(),
sortNodes: vi.fn((flow) =>
flow.nodes.map((node: any) => ({
key: node.id,
label: node.data.title,
original: node,
})),
),
}));
vi.mock('#/api/request', () => ({ api: { get, post } }));
vi.mock('#/router', () => ({ router: { replace: vi.fn() } }));
vi.mock('vue-router', () => ({
useRoute: () => ({ query: { id: 'workflow-1', navTitle: '测试' } }),
}));
vi.mock('@easyflow/preferences', () => ({
usePreferences: () => ({ isDark: false }),
}));
vi.mock('@easyflow/utils', () => ({ sortNodes }));
vi.mock('../customNode/index', () => ({ getCustomNode: async () => ({}) }));
vi.mock('#/views/ai/model/modelUtils/defaultIcon', () => ({
getIconByValue: () => '',
}));
vi.mock('#/components/commonSelectModal/CommonSelectDataModal.vue', () => ({
default: { template: '<div />' },
}));
vi.mock('../components/SingleRun.vue', () => ({
default: { template: '<div />' },
}));
vi.mock('../components/ExecResult.vue', () => ({
default: { props: ['nodeJson'], template: '<div />' },
}));
vi.mock('@tinyflow-ai/vue', async () => {
const { defineComponent, h } = await import('vue');
return {
Tinyflow: defineComponent({
props: {
data: { type: Object, required: true },
onRunTest: { type: Function, required: true },
},
setup(props, { expose }) {
expose({ getData });
return () =>
h(
'button',
{ 'data-test': 'try-run', onClick: props.onRunTest },
'试运行',
);
},
}),
};
});
afterEach(() => {
localStorage.clear();
vi.clearAllMocks();
});
describe('workflow designer run snapshot', () => {
it('saves new nodes and edited configuration, shares cached display nodes and clears old status', async () => {
const graph: any = {
nodes: [{ id: 'start', type: 'startNode', data: { title: '开始' } }],
edges: [],
};
let saved = structuredClone(graph);
getData.mockImplementation(() => structuredClone(graph));
get.mockImplementation(async (url) => {
let data = {};
if (url.includes('/detail')) {
data = {
id: 'workflow-1',
title: '测试',
content: JSON.stringify(saved),
};
} else if (url.includes('/getRunningParameters')) {
data = { parameters: [], startFormMeta: { submitText: '开始' } };
}
return { errorCode: 0, data };
});
post.mockImplementation(async (url, body) => {
if (url.endsWith('/update')) saved = structuredClone(body.content);
if (url.endsWith('/check'))
return { errorCode: 0, data: { passed: true } };
if (url.endsWith('/runAsync')) return { errorCode: 0, data: 'run-1' };
if (url.endsWith('/getChainStatus')) {
return {
errorCode: 0,
data: {
status: 5,
nodes: Object.fromEntries(
body.nodes.map((node: any) => [
node.nodeId,
{
status: node.nodeId === 'confirm' ? 5 : 20,
suspendForParameters:
node.nodeId === 'confirm'
? [
{
name: 'selection__confirm',
formType: 'radio',
required: true,
options: saved.nodes
.find((item: any) => item.id === 'confirm')
.data.options.map((value: string) => ({
label: value,
value,
})),
},
]
: [],
},
]),
),
},
};
}
return { errorCode: 0, data: {} };
});
const wrapper = mount(WorkflowDesign, {
global: {
directives: { loading: () => {} },
stubs: { ShowJson: true, WorkflowFormItem: true },
},
});
try {
await flushPromises();
const openRun = async () => {
await wrapper.get('[data-test="try-run"]').trigger('click');
await flushPromises();
};
const startRun = async () => {
await wrapper.getComponent(WorkflowForm).get('button').trigger('click');
await flushPromises();
};
await openRun();
await startRun();
const originalForm = wrapper.getComponent(WorkflowForm).vm.$.uid;
graph.nodes.push({
id: 'confirm',
type: 'confirmNode',
data: {
title: '用户确认',
message: '请选择旧模板',
multiple: false,
options: ['旧选项'],
outputDefs: [{ name: 'selection', dataType: 'String' }],
},
});
await openRun();
expect(wrapper.getComponent(WorkflowForm).vm.$.uid).toBe(originalForm);
expect(
wrapper.getComponent(WorkflowSteps).props('pollingData'),
).toBeNull();
const latestNodes = wrapper.getComponent(WorkflowSteps).props('nodeJson');
const sortCount = sortNodes.mock.calls.length;
await startRun();
expect(wrapper.getComponent(WorkflowSteps).text()).toContain(
'请选择旧模板',
);
expect(sortNodes).toHaveBeenCalledTimes(sortCount);
expect(wrapper.getComponent(WorkflowSteps).props('nodeJson')).toBe(
latestNodes,
);
graph.nodes[1].data.message = '请选择新模板';
graph.nodes[1].data.options = ['新选项'];
await openRun();
await startRun();
expect(saved.nodes[1].data.options).toEqual(['新选项']);
expect(wrapper.getComponent(WorkflowSteps).text()).toContain(
'请选择新模板',
);
expect(
wrapper.getComponent(WorkflowSteps).props('pollingData').nodes.confirm
.suspendForParameters[0].options,
).toEqual([{ label: '新选项', value: '新选项' }]);
expect(post.mock.calls.some(([url]) => url.includes('Publish'))).toBe(
false,
);
} finally {
wrapper.unmount();
}
});
});

View File

@@ -31,6 +31,7 @@ const props = withDefaults(defineProps<WorkflowFormProps>(), {
},
});
defineExpose({
reset,
resume,
});
const runForm = ref<FormInstance>();
@@ -88,53 +89,66 @@ watch(
);
const executeId = ref('');
async function resume(data: any) {
data.executeId = executeId.value;
if (submitLoading.value || !executeId.value) return false;
const generation = pollingGeneration;
submitLoading.value = true;
let accepted = false;
try {
const res = await api.post('/api/v1/workflow/resume', data);
const res = await api.post('/api/v1/workflow/resume', {
...data,
executeId: executeId.value,
});
if (generation !== pollingGeneration) return false;
if (res.errorCode === 0) {
accepted = true;
startPolling(executeId.value);
return true;
}
return accepted;
return false;
} finally {
if (!accepted) {
if (generation === pollingGeneration) {
submitLoading.value = false;
}
}
}
function submitV2() {
runForm.value?.validate((valid) => {
if (valid) {
const data = {
id: props.workflowId,
variables: {
...runParams.value,
},
};
props.onSubmit?.(runParams.value);
async function submitV2() {
if (submitLoading.value || !runForm.value) return;
stopPolling();
const generation = pollingGeneration;
submitLoading.value = true;
api.post('/api/v1/workflow/runAsync', data).then((res) => {
try {
const valid = await runForm.value.validate().catch(() => false);
if (!valid || generation !== pollingGeneration) return;
executeId.value = '';
// 每轮执行只生成一次轻量列表,后续轮询和暂停恢复复用同一份节点。
nodes = (props.tinyFlowData?.nodes || []).map((node: any) => ({
nodeId: node.id,
nodeName: node.data?.title || node.id,
}));
props.onSubmit?.(runParams.value);
const res = await api.post('/api/v1/workflow/runAsync', {
id: props.workflowId,
variables: { ...runParams.value },
});
if (generation !== pollingGeneration) return;
if (res.errorCode === 0 && res.data) {
// executeId
executeId.value = res.data;
startPolling(res.data);
}
});
} catch (error) {
if (generation === pollingGeneration) {
console.error('工作流启动失败', error);
}
} finally {
if (generation === pollingGeneration) {
submitLoading.value = false;
}
}
});
}
const POLLING_INTERVAL_MS = 1000;
const timer = ref<null | ReturnType<typeof setTimeout>>(null);
let pollingActive = false;
let pollingGeneration = 0;
const nodes = ref(
props.tinyFlowData.nodes.map((node: any) => ({
nodeId: node.id,
nodeName: node.data.title,
})),
);
let nodes: { nodeId: string; nodeName: string }[] = [];
// 轮询执行结果
function startPolling(executeId: any) {
if (pollingActive) return;
@@ -152,7 +166,7 @@ async function executePolling(executeId: any, generation: number) {
try {
const res = await api.post('/api/v1/workflow/getChainStatus', {
executeId,
nodes: nodes.value,
nodes,
});
if (!pollingActive || generation !== pollingGeneration) return;
@@ -180,9 +194,12 @@ function stopPolling() {
timer.value = null;
}
}
onUnmounted(() => {
function reset() {
stopPolling();
});
executeId.value = '';
nodes = [];
}
onUnmounted(reset);
</script>
<template>

View File

@@ -0,0 +1,192 @@
import { flushPromises, mount } from '@vue/test-utils';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import WorkflowForm from '../WorkflowForm.vue';
const { post } = vi.hoisted(() => ({ post: vi.fn() }));
vi.mock('#/api/request', () => ({ api: { post } }));
function flow(...ids: string[]) {
return {
nodes: ids.map((id) => ({ id, data: { title: id } })),
};
}
function deferred() {
let resolve!: (value: any) => void;
const promise = new Promise((done) => {
resolve = done;
});
return { promise, resolve };
}
function createForm(tinyFlowData = flow('start', 'end')) {
return mount(WorkflowForm, {
props: {
tinyFlowData,
workflowId: 'workflow-1',
workflowParams: { parameters: [], startFormMeta: { submitText: '开始' } },
onAsyncExecute: vi.fn(),
onSubmit: vi.fn(),
},
global: { stubs: { WorkflowFormItem: true } },
});
}
type FormWrapper = ReturnType<typeof createForm>;
const wrappers: FormWrapper[] = [];
function mountForm(tinyFlowData?: ReturnType<typeof flow>) {
const wrapper = createForm(tinyFlowData);
wrappers.push(wrapper);
return wrapper;
}
async function start(wrapper: FormWrapper) {
await wrapper.get('button').trigger('click');
await flushPromises();
}
function polls() {
return post.mock.calls
.filter(([url]) => url.endsWith('/getChainStatus'))
.map(([, body]) => body);
}
describe('workflowForm execution snapshot', () => {
beforeEach(() => {
vi.useFakeTimers();
post.mockReset();
post.mockImplementation(async (url, body) => {
if (url.endsWith('/runAsync')) return { errorCode: 0, data: 'run-1' };
if (url.endsWith('/resume')) return { errorCode: 0 };
return { errorCode: 0, data: { executeId: body.executeId, status: 5 } };
});
});
afterEach(() => {
for (const wrapper of wrappers.splice(0)) wrapper.unmount();
vi.useRealTimers();
vi.restoreAllMocks();
});
it('reuses the mounted form but polls newly added nodes on the next run', async () => {
const wrapper = mountForm();
await start(wrapper);
await wrapper.setProps({
tinyFlowData: flow('start', 'confirm-new', 'end'),
});
await start(wrapper);
expect(polls()[1].nodes.map((node: any) => node.nodeId)).toEqual([
'start',
'confirm-new',
'end',
]);
});
it('removes deleted nodes and uses the latest node name on a new run', async () => {
const wrapper = mountForm(flow('start', 'old-confirm', 'end'));
await start(wrapper);
const updated = flow('start', 'end');
updated.nodes = updated.nodes.map((node) => ({
...node,
data: { title: node.id === 'end' ? '新版结束节点' : node.id },
}));
await wrapper.setProps({ tinyFlowData: updated });
await start(wrapper);
expect(polls()[1].nodes).toEqual([
{ nodeId: 'start', nodeName: 'start' },
{ nodeId: 'end', nodeName: '新版结束节点' },
]);
});
it('reads a large node list once and reuses it for polling and resume', async () => {
const data = flow(...Array.from({ length: 1000 }, (_, i) => `node-${i}`));
const readNodes = vi.fn(() => data.nodes);
const wrapper = mountForm(
Object.defineProperty({}, 'nodes', { get: readNodes }) as typeof data,
);
readNodes.mockClear();
const states = [1, 1, 5, 20];
post.mockImplementation(async (url) => {
if (url.endsWith('/runAsync')) return { errorCode: 0, data: 'run-1' };
if (url.endsWith('/resume')) return { errorCode: 0 };
return { errorCode: 0, data: { status: states.shift() } };
});
await start(wrapper);
await vi.advanceTimersByTimeAsync(2000);
await wrapper.vm.resume({ confirmParams: { selection: '继续' } });
await flushPromises();
expect(readNodes).toHaveBeenCalledTimes(1);
expect(polls()).toHaveLength(4);
expect(polls().every((poll) => poll.nodes === polls()[0].nodes)).toBe(true);
expect(vi.getTimerCount()).toBe(0);
});
it('keeps the suspended execution nodes when props change before resume', async () => {
const wrapper = mountForm(flow('start', 'confirm', 'end'));
await start(wrapper);
await wrapper.setProps({ tinyFlowData: flow('different-node') });
await wrapper.vm.resume({ confirmParams: { selection__confirm: '继续' } });
await flushPromises();
expect(polls()[1].nodes).toBe(polls()[0].nodes);
expect(polls()[1].executeId).toBe('run-1');
});
it('ignores an old poll after reset and does not unlock a new request', async () => {
const oldPoll = deferred();
const newRun = deferred();
const wrapper = mountForm();
post.mockResolvedValueOnce({ errorCode: 0, data: 'old-run' });
post.mockReturnValueOnce(oldPoll.promise);
await start(wrapper);
wrapper.vm.reset();
await wrapper.vm.$nextTick();
post.mockReturnValueOnce(newRun.promise);
await start(wrapper);
oldPoll.resolve({ errorCode: 0, data: { status: 5 } });
await flushPromises();
expect(wrapper.props('onAsyncExecute')).not.toHaveBeenCalled();
expect(wrapper.get('button').classes()).toContain('is-loading');
newRun.resolve({ errorCode: 0, data: 'new-run' });
await flushPromises();
expect(polls().at(-1).executeId).toBe('new-run');
expect(wrapper.props('onAsyncExecute')).toHaveBeenCalledTimes(1);
});
it.each(['start', 'resume'])(
'ignores a late %s response after reset',
async (operation) => {
const wrapper = mountForm();
if (operation === 'resume') await start(wrapper);
const request = deferred();
post.mockReturnValueOnce(request.promise);
const pending =
operation === 'resume'
? wrapper.vm.resume({ confirmParams: {} })
: start(wrapper);
await flushPromises();
const previousPollCount = polls().length;
wrapper.vm.reset();
request.resolve({ errorCode: 0, data: 'old-run' });
await pending;
await flushPromises();
expect(polls()).toHaveLength(previousPollCount);
expect(wrapper.get('button').classes()).not.toContain('is-loading');
},
);
it('allows retry after a rejected start request and prevents duplicate starts', async () => {
vi.spyOn(console, 'error').mockImplementation(() => {});
const request = deferred();
const wrapper = mountForm();
post.mockReturnValueOnce(request.promise);
await start(wrapper);
await start(wrapper);
expect(post).toHaveBeenCalledTimes(1);
request.resolve({ errorCode: 1 });
await flushPromises();
expect(wrapper.get('button').classes()).not.toContain('is-loading');
post.mockRejectedValueOnce(new Error('请求失败'));
await start(wrapper);
expect(wrapper.get('button').classes()).not.toContain('is-loading');
await start(wrapper);
expect(polls()).toHaveLength(1);
});
});