mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
fix(console): match child-workflow inner blocks by instanceId when reconciling dropped SSE events (#4575)
* fix(console): match child-workflow inner blocks by instanceId when reconciling dropped SSE events * fix(console): drop noisy warn when reconcile finds no matching entry
This commit is contained in:
+368
@@ -0,0 +1,368 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*
|
||||
* Integration tests that exercise `reconcileFinalBlockLogs` against the real
|
||||
* `useTerminalConsoleStore` to validate end-to-end matching behavior. The
|
||||
* sibling unit-test file mocks the store and only verifies call args, which
|
||||
* cannot catch identity-mismatch regressions of the kind that produced the
|
||||
* 34.57s wall-clock symptom.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
vi.unmock('@/stores/terminal')
|
||||
vi.unmock('@/stores/terminal/console/store')
|
||||
|
||||
import { reconcileFinalBlockLogs } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
|
||||
import { useExecutionStore } from '@/stores/execution'
|
||||
import { useTerminalConsoleStore } from '@/stores/terminal/console/store'
|
||||
|
||||
describe('reconcileFinalBlockLogs (real store)', () => {
|
||||
beforeEach(() => {
|
||||
useTerminalConsoleStore.setState({
|
||||
workflowEntries: {},
|
||||
entryIdsByBlockExecution: {},
|
||||
entryLocationById: {},
|
||||
isOpen: false,
|
||||
_hasHydrated: true,
|
||||
})
|
||||
vi.mocked(useExecutionStore.getState).mockReturnValue({
|
||||
getCurrentExecutionId: vi.fn(() => 'exec-1'),
|
||||
} as any)
|
||||
})
|
||||
|
||||
it('actually flips a child-workflow inner block from running to success', () => {
|
||||
const store = useTerminalConsoleStore.getState()
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
isRunning: false,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'child-inst-1',
|
||||
})
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'set-projects',
|
||||
blockName: 'setProjects',
|
||||
blockType: 'variables',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 5,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
childWorkflowName: 'Workflow 1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 27).toISOString()
|
||||
|
||||
reconcileFinalBlockLogs(store.updateConsole, 'wf-1', 'exec-1', [
|
||||
{
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
startedAt,
|
||||
endedAt,
|
||||
durationMs: 100,
|
||||
success: true,
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'set-projects-span',
|
||||
name: 'setProjects',
|
||||
type: 'variables',
|
||||
blockId: 'set-projects',
|
||||
executionOrder: 5,
|
||||
status: 'success',
|
||||
duration: 27,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { value: [{ id: 'p1' }] },
|
||||
},
|
||||
],
|
||||
} as any,
|
||||
])
|
||||
|
||||
const innerEntry = useTerminalConsoleStore
|
||||
.getState()
|
||||
.getWorkflowEntries('wf-1')
|
||||
.find((e) => e.blockId === 'set-projects')
|
||||
|
||||
expect(innerEntry).toBeDefined()
|
||||
expect(innerEntry?.isRunning).toBe(false)
|
||||
expect(innerEntry?.success).toBe(true)
|
||||
expect(innerEntry?.durationMs).toBe(27)
|
||||
expect(innerEntry?.output).toEqual({ value: [{ id: 'p1' }] })
|
||||
})
|
||||
|
||||
it('targets the correct invocation when the same child nodeId runs twice', () => {
|
||||
const store = useTerminalConsoleStore.getState()
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
isRunning: false,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-A',
|
||||
})
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 2,
|
||||
isRunning: false,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-B',
|
||||
})
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-inner',
|
||||
blockName: 'Inner',
|
||||
blockType: 'function',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 3,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'inst-A',
|
||||
})
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-inner',
|
||||
blockName: 'Inner',
|
||||
blockType: 'function',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 4,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'inst-B',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 5).toISOString()
|
||||
const baseLog = {
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
startedAt,
|
||||
endedAt,
|
||||
durationMs: 50,
|
||||
success: true,
|
||||
}
|
||||
|
||||
reconcileFinalBlockLogs(store.updateConsole, 'wf-1', 'exec-1', [
|
||||
{
|
||||
...baseLog,
|
||||
blockId: 'workflow-1',
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'a',
|
||||
name: 'Inner',
|
||||
type: 'function',
|
||||
blockId: 'fn-inner',
|
||||
executionOrder: 3,
|
||||
status: 'success',
|
||||
duration: 5,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { result: 'A' },
|
||||
},
|
||||
],
|
||||
} as any,
|
||||
{
|
||||
...baseLog,
|
||||
blockId: 'workflow-1',
|
||||
executionOrder: 2,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'b',
|
||||
name: 'Inner',
|
||||
type: 'function',
|
||||
blockId: 'fn-inner',
|
||||
executionOrder: 4,
|
||||
status: 'success',
|
||||
duration: 5,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { result: 'B' },
|
||||
},
|
||||
],
|
||||
} as any,
|
||||
])
|
||||
|
||||
const entries = useTerminalConsoleStore.getState().getWorkflowEntries('wf-1')
|
||||
const a = entries.find((e) => e.blockId === 'fn-inner' && e.childWorkflowBlockId === 'inst-A')
|
||||
const b = entries.find((e) => e.blockId === 'fn-inner' && e.childWorkflowBlockId === 'inst-B')
|
||||
|
||||
expect(a?.isRunning).toBe(false)
|
||||
expect(a?.output).toEqual({ result: 'A' })
|
||||
expect(b?.isRunning).toBe(false)
|
||||
expect(b?.output).toEqual({ result: 'B' })
|
||||
})
|
||||
|
||||
it('propagates error state for spans with error status', () => {
|
||||
const store = useTerminalConsoleStore.getState()
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
isRunning: false,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-1',
|
||||
})
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'http-1',
|
||||
blockName: 'API',
|
||||
blockType: 'api',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 2,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 30).toISOString()
|
||||
|
||||
reconcileFinalBlockLogs(store.updateConsole, 'wf-1', 'exec-1', [
|
||||
{
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
startedAt,
|
||||
endedAt,
|
||||
durationMs: 100,
|
||||
success: true,
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'http-span',
|
||||
name: 'API',
|
||||
type: 'api',
|
||||
blockId: 'http-1',
|
||||
executionOrder: 2,
|
||||
status: 'error',
|
||||
duration: 30,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { error: 'Connection refused' },
|
||||
},
|
||||
],
|
||||
} as any,
|
||||
])
|
||||
|
||||
const entry = useTerminalConsoleStore
|
||||
.getState()
|
||||
.getWorkflowEntries('wf-1')
|
||||
.find((e) => e.blockId === 'http-1')
|
||||
|
||||
expect(entry?.isRunning).toBe(false)
|
||||
expect(entry?.success).toBe(false)
|
||||
expect(entry?.error).toBe('Connection refused')
|
||||
})
|
||||
|
||||
it('matches the correct iteration row inside a child workflow loop', () => {
|
||||
const store = useTerminalConsoleStore.getState()
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
isRunning: false,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-1',
|
||||
})
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-leaf',
|
||||
blockName: 'Leaf',
|
||||
blockType: 'function',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 2,
|
||||
isRunning: false,
|
||||
success: true,
|
||||
iterationCurrent: 0,
|
||||
iterationType: 'loop',
|
||||
iterationContainerId: 'loop-1',
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
output: { i: 0 },
|
||||
})
|
||||
store.addConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-leaf',
|
||||
blockName: 'Leaf',
|
||||
blockType: 'function',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 3,
|
||||
isRunning: true,
|
||||
iterationCurrent: 1,
|
||||
iterationType: 'loop',
|
||||
iterationContainerId: 'loop-1',
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 12).toISOString()
|
||||
|
||||
reconcileFinalBlockLogs(store.updateConsole, 'wf-1', 'exec-1', [
|
||||
{
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
startedAt,
|
||||
endedAt,
|
||||
durationMs: 100,
|
||||
success: true,
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'leaf-0',
|
||||
name: 'Leaf',
|
||||
type: 'function',
|
||||
blockId: 'fn-leaf',
|
||||
executionOrder: 2,
|
||||
loopId: 'loop-1',
|
||||
iterationIndex: 0,
|
||||
status: 'success',
|
||||
duration: 5,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { i: 0 },
|
||||
},
|
||||
{
|
||||
id: 'leaf-1',
|
||||
name: 'Leaf',
|
||||
type: 'function',
|
||||
blockId: 'fn-leaf',
|
||||
executionOrder: 3,
|
||||
loopId: 'loop-1',
|
||||
iterationIndex: 1,
|
||||
status: 'success',
|
||||
duration: 12,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { i: 1 },
|
||||
},
|
||||
],
|
||||
} as any,
|
||||
])
|
||||
|
||||
const entries = useTerminalConsoleStore.getState().getWorkflowEntries('wf-1')
|
||||
const iter0 = entries.find((e) => e.blockId === 'fn-leaf' && e.iterationCurrent === 0)
|
||||
const iter1 = entries.find((e) => e.blockId === 'fn-leaf' && e.iterationCurrent === 1)
|
||||
|
||||
expect(iter0?.isRunning).toBe(false)
|
||||
expect(iter0?.output).toEqual({ i: 0 })
|
||||
expect(iter1?.isRunning).toBe(false)
|
||||
expect(iter1?.output).toEqual({ i: 1 })
|
||||
})
|
||||
})
|
||||
+389
-8
@@ -427,7 +427,7 @@ describe('workflow-execution-utils', () => {
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 3,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'workflow-1',
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
childWorkflowName: 'Workflow 1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
@@ -489,7 +489,7 @@ describe('workflow-execution-utils', () => {
|
||||
success: true,
|
||||
isRunning: false,
|
||||
isCanceled: false,
|
||||
childWorkflowBlockId: 'workflow-1',
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
}),
|
||||
'exec-1',
|
||||
])
|
||||
@@ -501,7 +501,7 @@ describe('workflow-execution-utils', () => {
|
||||
error: 'Request failed',
|
||||
isRunning: false,
|
||||
isCanceled: false,
|
||||
childWorkflowBlockId: 'workflow-1',
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
}),
|
||||
'exec-1',
|
||||
])
|
||||
@@ -529,7 +529,7 @@ describe('workflow-execution-utils', () => {
|
||||
iterationCurrent: 0,
|
||||
iterationType: 'loop',
|
||||
iterationContainerId: 'loop-1',
|
||||
childWorkflowBlockId: 'workflow-1',
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
@@ -542,7 +542,7 @@ describe('workflow-execution-utils', () => {
|
||||
iterationCurrent: 1,
|
||||
iterationType: 'loop',
|
||||
iterationContainerId: 'loop-1',
|
||||
childWorkflowBlockId: 'workflow-1',
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
@@ -632,7 +632,7 @@ describe('workflow-execution-utils', () => {
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 3,
|
||||
isRunning: false,
|
||||
childWorkflowBlockId: 'workflow-1',
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
childWorkflowInstanceId: 'nested-inst-1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
@@ -643,7 +643,7 @@ describe('workflow-execution-utils', () => {
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'nested-workflow',
|
||||
childWorkflowBlockId: 'nested-inst-1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
@@ -688,7 +688,7 @@ describe('workflow-execution-utils', () => {
|
||||
expect(updateConsole.mock.calls[1]).toEqual([
|
||||
'nested-api',
|
||||
expect.objectContaining({
|
||||
childWorkflowBlockId: 'nested-workflow',
|
||||
childWorkflowBlockId: 'nested-inst-1',
|
||||
success: true,
|
||||
isRunning: false,
|
||||
isCanceled: false,
|
||||
@@ -697,6 +697,387 @@ describe('workflow-execution-utils', () => {
|
||||
])
|
||||
})
|
||||
|
||||
it('rescues a child-workflow block whose block:completed SSE event was dropped', () => {
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
success: true,
|
||||
isRunning: false,
|
||||
childWorkflowInstanceId: 'child-inst-1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'set-projects',
|
||||
blockName: 'setProjects',
|
||||
blockType: 'variables',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 5,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
childWorkflowName: 'Workflow 1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 27).toISOString()
|
||||
const updateConsole = vi.fn()
|
||||
reconcileFinalBlockLogs(updateConsole, 'wf-1', 'exec-1', [
|
||||
makeLog({
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'set-projects-span',
|
||||
name: 'setProjects',
|
||||
type: 'variables',
|
||||
blockId: 'set-projects',
|
||||
executionOrder: 5,
|
||||
status: 'success',
|
||||
duration: 27,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { value: [{ id: 'p1' }, { id: 'p2' }] },
|
||||
},
|
||||
],
|
||||
}),
|
||||
])
|
||||
|
||||
expect(updateConsole).toHaveBeenCalledTimes(1)
|
||||
expect(updateConsole.mock.calls[0]).toEqual([
|
||||
'set-projects',
|
||||
expect.objectContaining({
|
||||
executionOrder: 5,
|
||||
childWorkflowBlockId: 'child-inst-1',
|
||||
replaceOutput: { value: [{ id: 'p1' }, { id: 'p2' }] },
|
||||
success: true,
|
||||
isRunning: false,
|
||||
isCanceled: false,
|
||||
durationMs: 27,
|
||||
startedAt,
|
||||
endedAt,
|
||||
}),
|
||||
'exec-1',
|
||||
])
|
||||
})
|
||||
|
||||
it('matches per-invocation when the same child workflow nodeId runs twice', () => {
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-A',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockName: 'Workflow 1',
|
||||
blockType: 'workflow',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 2,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-B',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-inner',
|
||||
blockName: 'Inner',
|
||||
blockType: 'function',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 3,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'inst-A',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-inner',
|
||||
blockName: 'Inner',
|
||||
blockType: 'function',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 4,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'inst-B',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 10).toISOString()
|
||||
const updateConsole = vi.fn()
|
||||
reconcileFinalBlockLogs(updateConsole, 'wf-1', 'exec-1', [
|
||||
makeLog({
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'a',
|
||||
name: 'Inner',
|
||||
type: 'function',
|
||||
blockId: 'fn-inner',
|
||||
executionOrder: 3,
|
||||
status: 'success',
|
||||
duration: 5,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { result: 'A' },
|
||||
},
|
||||
],
|
||||
}),
|
||||
makeLog({
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
executionOrder: 2,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'b',
|
||||
name: 'Inner',
|
||||
type: 'function',
|
||||
blockId: 'fn-inner',
|
||||
executionOrder: 4,
|
||||
status: 'success',
|
||||
duration: 5,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { result: 'B' },
|
||||
},
|
||||
],
|
||||
}),
|
||||
])
|
||||
|
||||
expect(updateConsole).toHaveBeenCalledTimes(2)
|
||||
expect(updateConsole.mock.calls[0][1]).toMatchObject({
|
||||
executionOrder: 3,
|
||||
childWorkflowBlockId: 'inst-A',
|
||||
replaceOutput: { result: 'A' },
|
||||
})
|
||||
expect(updateConsole.mock.calls[1][1]).toMatchObject({
|
||||
executionOrder: 4,
|
||||
childWorkflowBlockId: 'inst-B',
|
||||
replaceOutput: { result: 'B' },
|
||||
})
|
||||
})
|
||||
|
||||
it('reconciles parallel-iteration spans inside a child workflow', () => {
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
blockName: 'Workflow 1',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-leaf',
|
||||
blockType: 'function',
|
||||
blockName: 'Leaf',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 2,
|
||||
isRunning: true,
|
||||
iterationCurrent: 0,
|
||||
iterationType: 'parallel',
|
||||
iterationContainerId: 'par-1',
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 8).toISOString()
|
||||
const updateConsole = vi.fn()
|
||||
reconcileFinalBlockLogs(updateConsole, 'wf-1', 'exec-1', [
|
||||
makeLog({
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'leaf-span',
|
||||
name: 'Leaf',
|
||||
type: 'function',
|
||||
blockId: 'fn-leaf',
|
||||
executionOrder: 2,
|
||||
parallelId: 'par-1',
|
||||
iterationIndex: 0,
|
||||
status: 'success',
|
||||
duration: 8,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { ok: true },
|
||||
},
|
||||
],
|
||||
}),
|
||||
])
|
||||
|
||||
expect(updateConsole).toHaveBeenCalledTimes(1)
|
||||
expect(updateConsole.mock.calls[0][1]).toMatchObject({
|
||||
executionOrder: 2,
|
||||
iterationCurrent: 0,
|
||||
iterationType: 'parallel',
|
||||
iterationContainerId: 'par-1',
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
success: true,
|
||||
})
|
||||
})
|
||||
|
||||
it('rescues only the iteration whose terminal SSE event was dropped', () => {
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
blockName: 'Workflow 1',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-leaf',
|
||||
blockType: 'function',
|
||||
blockName: 'Leaf',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 2,
|
||||
isRunning: false,
|
||||
success: true,
|
||||
iterationCurrent: 0,
|
||||
iterationType: 'loop',
|
||||
iterationContainerId: 'loop-1',
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'fn-leaf',
|
||||
blockType: 'function',
|
||||
blockName: 'Leaf',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 3,
|
||||
isRunning: true,
|
||||
iterationCurrent: 1,
|
||||
iterationType: 'loop',
|
||||
iterationContainerId: 'loop-1',
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 12).toISOString()
|
||||
const updateConsole = vi.fn()
|
||||
reconcileFinalBlockLogs(updateConsole, 'wf-1', 'exec-1', [
|
||||
makeLog({
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'leaf-0',
|
||||
name: 'Leaf',
|
||||
type: 'function',
|
||||
blockId: 'fn-leaf',
|
||||
executionOrder: 2,
|
||||
loopId: 'loop-1',
|
||||
iterationIndex: 0,
|
||||
status: 'success',
|
||||
duration: 5,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { i: 0 },
|
||||
},
|
||||
{
|
||||
id: 'leaf-1',
|
||||
name: 'Leaf',
|
||||
type: 'function',
|
||||
blockId: 'fn-leaf',
|
||||
executionOrder: 3,
|
||||
loopId: 'loop-1',
|
||||
iterationIndex: 1,
|
||||
status: 'success',
|
||||
duration: 12,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { i: 1 },
|
||||
},
|
||||
],
|
||||
}),
|
||||
])
|
||||
|
||||
// updateConsole is called for both spans (idempotent re-application), but
|
||||
// production matchesEntryForUpdate filters by the identity so only the
|
||||
// still-running iteration is actually mutated. We assert the args carry
|
||||
// distinct iteration identities so the store can target the right row.
|
||||
expect(updateConsole.mock.calls[0][1]).toMatchObject({
|
||||
executionOrder: 2,
|
||||
iterationCurrent: 0,
|
||||
})
|
||||
expect(updateConsole.mock.calls[1][1]).toMatchObject({
|
||||
executionOrder: 3,
|
||||
iterationCurrent: 1,
|
||||
replaceOutput: { i: 1 },
|
||||
})
|
||||
})
|
||||
|
||||
it('propagates span error state when the block:error SSE was lost', () => {
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
blockName: 'Workflow 1',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 1,
|
||||
success: true,
|
||||
childWorkflowInstanceId: 'inst-1',
|
||||
})
|
||||
terminalConsoleMockFns.mockAddConsole({
|
||||
workflowId: 'wf-1',
|
||||
blockId: 'http-1',
|
||||
blockType: 'api',
|
||||
blockName: 'API',
|
||||
executionId: 'exec-1',
|
||||
executionOrder: 2,
|
||||
isRunning: true,
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
})
|
||||
|
||||
const startedAt = new Date().toISOString()
|
||||
const endedAt = new Date(Date.now() + 30).toISOString()
|
||||
const updateConsole = vi.fn()
|
||||
reconcileFinalBlockLogs(updateConsole, 'wf-1', 'exec-1', [
|
||||
makeLog({
|
||||
blockId: 'workflow-1',
|
||||
blockType: 'workflow',
|
||||
executionOrder: 1,
|
||||
childTraceSpans: [
|
||||
{
|
||||
id: 'http-span',
|
||||
name: 'API',
|
||||
type: 'api',
|
||||
blockId: 'http-1',
|
||||
executionOrder: 2,
|
||||
status: 'error',
|
||||
duration: 30,
|
||||
startTime: startedAt,
|
||||
endTime: endedAt,
|
||||
output: { error: 'Connection refused' },
|
||||
},
|
||||
],
|
||||
}),
|
||||
])
|
||||
|
||||
expect(updateConsole).toHaveBeenCalledTimes(1)
|
||||
expect(updateConsole.mock.calls[0][1]).toMatchObject({
|
||||
success: false,
|
||||
error: 'Connection refused',
|
||||
childWorkflowBlockId: 'inst-1',
|
||||
isRunning: false,
|
||||
isCanceled: false,
|
||||
})
|
||||
})
|
||||
|
||||
it('is a no-op when finalBlockLogs is empty or executionId is missing', () => {
|
||||
const updateConsole = vi.fn()
|
||||
reconcileFinalBlockLogs(updateConsole, 'wf-1', 'exec-1', [])
|
||||
|
||||
+17
-9
@@ -512,7 +512,6 @@ export function reconcileFinalBlockLogs(
|
||||
reconcileChildTraceSpans(
|
||||
updateConsole,
|
||||
workflowId,
|
||||
log.blockId,
|
||||
childWorkflowInstanceId,
|
||||
executionId,
|
||||
log.childTraceSpans
|
||||
@@ -521,24 +520,34 @@ export function reconcileFinalBlockLogs(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Reconciles trace spans for blocks inside a child workflow.
|
||||
*
|
||||
* Inner-block console entries are created from SSE `block:started` events whose
|
||||
* `childWorkflowBlockId` field carries the parent's per-invocation instanceId
|
||||
* (see `execute/route.ts` where the server emits `childWorkflowContext.parentBlockId`).
|
||||
* The matcher must therefore key on that instanceId — using the parent workflow
|
||||
* block's static nodeId would never match and the rescue silently no-ops, leaving
|
||||
* inner blocks stuck `isRunning: true` until `finishRunningEntries` sweeps them
|
||||
* with a wall-clock duration.
|
||||
*/
|
||||
function reconcileChildTraceSpans(
|
||||
updateConsole: UpdateConsoleFn,
|
||||
workflowId: string,
|
||||
childWorkflowBlockId: string,
|
||||
childWorkflowInstanceId: string,
|
||||
executionId: string,
|
||||
spans: TraceSpan[]
|
||||
): void {
|
||||
for (const span of spans) {
|
||||
const matchingEntry = span.blockId
|
||||
? findConsoleEntryForSpan(workflowId, executionId, childWorkflowBlockId, span)
|
||||
? findConsoleEntryForSpan(workflowId, executionId, childWorkflowInstanceId, span)
|
||||
: undefined
|
||||
if (span.blockId) {
|
||||
const errorMessage = normalizeSpanError(span.output?.error)
|
||||
updateConsole(
|
||||
span.blockId,
|
||||
{
|
||||
...spanConsoleIdentity(span, childWorkflowBlockId),
|
||||
...spanConsoleIdentity(span, childWorkflowInstanceId),
|
||||
replaceOutput: (span.output ?? {}) as Record<string, unknown>,
|
||||
success: span.status !== 'error',
|
||||
...(errorMessage !== undefined ? { error: errorMessage } : {}),
|
||||
@@ -555,7 +564,6 @@ function reconcileChildTraceSpans(
|
||||
reconcileChildTraceSpans(
|
||||
updateConsole,
|
||||
workflowId,
|
||||
matchingEntry?.blockId ?? childWorkflowBlockId,
|
||||
matchingEntry?.childWorkflowInstanceId ?? childWorkflowInstanceId,
|
||||
executionId,
|
||||
span.children
|
||||
@@ -564,7 +572,7 @@ function reconcileChildTraceSpans(
|
||||
}
|
||||
}
|
||||
|
||||
function spanConsoleIdentity(span: TraceSpan, childWorkflowBlockId: string): ConsoleUpdate {
|
||||
function spanConsoleIdentity(span: TraceSpan, childWorkflowInstanceId: string): ConsoleUpdate {
|
||||
const iterationContainerId = span.loopId ?? span.parallelId
|
||||
const iterationType = span.loopId ? 'loop' : span.parallelId ? 'parallel' : undefined
|
||||
return {
|
||||
@@ -573,18 +581,18 @@ function spanConsoleIdentity(span: TraceSpan, childWorkflowBlockId: string): Con
|
||||
...(iterationType !== undefined && { iterationType }),
|
||||
...(iterationContainerId !== undefined && { iterationContainerId }),
|
||||
...(span.parentIterations !== undefined && { parentIterations: span.parentIterations }),
|
||||
childWorkflowBlockId,
|
||||
childWorkflowBlockId: childWorkflowInstanceId,
|
||||
}
|
||||
}
|
||||
|
||||
function findConsoleEntryForSpan(
|
||||
workflowId: string,
|
||||
executionId: string,
|
||||
childWorkflowBlockId: string,
|
||||
childWorkflowInstanceId: string,
|
||||
span: TraceSpan
|
||||
): ConsoleEntry | undefined {
|
||||
if (!span.blockId) return undefined
|
||||
const identity = spanConsoleIdentity(span, childWorkflowBlockId)
|
||||
const identity = spanConsoleIdentity(span, childWorkflowInstanceId)
|
||||
return useTerminalConsoleStore
|
||||
.getState()
|
||||
.getWorkflowEntries(workflowId)
|
||||
|
||||
Reference in New Issue
Block a user