mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-09 08:03:50 +00:00
refactor
This commit is contained in:
Diego Imbert
@@ -49,6 +49,7 @@
|
||||
import JobAssetsViewer from './assets/JobAssetsViewer.svelte'
|
||||
import McpToolCallDetails from './McpToolCallDetails.svelte'
|
||||
import { SelectionManager } from './graph/selectionUtils.svelte'
|
||||
import { useThrottle } from 'runed'
|
||||
|
||||
let {
|
||||
flowState: flowStateStore,
|
||||
@@ -503,211 +504,187 @@
|
||||
}
|
||||
|
||||
let jobMissingStartedAt: Record<string, number | 'P'> = {}
|
||||
let lastSelectedLoopSwitch: number | undefined
|
||||
let selectedLoopSwitchTimeout: number | undefined = undefined
|
||||
|
||||
let setSelectedLoopSwitch = useThrottle((lastStarted: string, mod: FlowStatusModule) => {
|
||||
let position = mod.flow_jobs?.indexOf(lastStarted)
|
||||
if (!position) return
|
||||
// we wait a bit for suspended steps, to make sure the current iteration is updated
|
||||
// when resumed before going forward. This avoid the duplicate resume forms bug.
|
||||
setIteration(position, lastStarted, false, mod.id ?? '', true)
|
||||
}, 2000)
|
||||
|
||||
function updateInnerModules() {
|
||||
if (localModuleStates) {
|
||||
innerModules?.forEach((mod, i) => {
|
||||
if (mod.type === 'WaitingForEvents' && innerModules?.[i - 1]?.type === 'Success') {
|
||||
setModuleState(mod.id ?? '', {
|
||||
type: mod.type,
|
||||
args: job?.args,
|
||||
tag: job?.tag,
|
||||
script_hash: job?.script_hash
|
||||
if (!localModuleStates || !innerModules) return
|
||||
for (let i = 0; i < innerModules.length; i++) {
|
||||
let mod = innerModules[i]
|
||||
let prev = innerModules[i - 1]
|
||||
if (mod.type === 'WaitingForEvents' && prev?.type === 'Success') {
|
||||
setModuleState(mod.id ?? '', {
|
||||
type: mod.type,
|
||||
args: job?.args,
|
||||
tag: job?.tag,
|
||||
script_hash: job?.script_hash
|
||||
})
|
||||
} else if (
|
||||
mod.type === 'WaitingForExecutor' &&
|
||||
localModuleStates[mod.id ?? '']?.scheduled_for == undefined
|
||||
) {
|
||||
JobService.getJob({
|
||||
workspace: workspaceId ?? $workspaceStore ?? '',
|
||||
id: mod.job ?? '',
|
||||
noLogs: true,
|
||||
noCode: true
|
||||
})
|
||||
.then((job) => {
|
||||
const newState = {
|
||||
type: mod.type,
|
||||
scheduled_for: job?.['scheduled_for'],
|
||||
job_id: job?.id,
|
||||
parent_module: mod['parent_module'],
|
||||
args: job?.args,
|
||||
tag: job?.tag,
|
||||
script_hash: job?.script_hash
|
||||
}
|
||||
setModuleState(mod.id ?? '', newState)
|
||||
})
|
||||
} else if (
|
||||
mod.type === 'WaitingForExecutor' &&
|
||||
localModuleStates[mod.id ?? '']?.scheduled_for == undefined
|
||||
) {
|
||||
JobService.getJob({
|
||||
workspace: workspaceId ?? $workspaceStore ?? '',
|
||||
id: mod.job ?? '',
|
||||
noLogs: true,
|
||||
noCode: true
|
||||
.catch((e) => {
|
||||
console.error(`Could not load inner module for job ${mod.job}`, e)
|
||||
})
|
||||
.then((job) => {
|
||||
const newState = {
|
||||
type: mod.type,
|
||||
scheduled_for: job?.['scheduled_for'],
|
||||
job_id: job?.id,
|
||||
parent_module: mod['parent_module'],
|
||||
args: job?.args,
|
||||
tag: job?.tag,
|
||||
script_hash: job?.script_hash
|
||||
}
|
||||
} else if (
|
||||
(mod.flow_jobs || mod.branch_chosen) &&
|
||||
(mod.type == 'Success' || mod.type == 'Failure') &&
|
||||
!['Success', 'Failure'].includes(localModuleStates?.[mod.id ?? '']?.type)
|
||||
) {
|
||||
let branchChosen = mod.branch_chosen
|
||||
? {
|
||||
branchChosen:
|
||||
mod.branch_chosen.type == 'default' ? 0 : (mod.branch_chosen.branch ?? 0) + 1
|
||||
}
|
||||
: {}
|
||||
setModuleState(mod.id ?? '', { type: mod.type, ...branchChosen }, true)
|
||||
} else if (isForloopSelected) {
|
||||
setModuleState(mod.id ?? '', {}, true)
|
||||
}
|
||||
|
||||
setModuleState(mod.id ?? '', newState)
|
||||
if (mod.flow_jobs_success || mod.flow_jobs_duration || mod.flow_jobs) {
|
||||
setModuleState(mod.id ?? '', {
|
||||
flow_jobs_success: mod.flow_jobs_success,
|
||||
flow_jobs_duration: mod.flow_jobs_duration,
|
||||
flow_jobs: mod.flow_jobs,
|
||||
iteration_total: mod.iterator?.itered_len ?? mod.flow_jobs?.length
|
||||
})
|
||||
}
|
||||
|
||||
if (mod.flow_jobs_duration && mod.flow_jobs) {
|
||||
let key = buildSubflowKey(mod.id ?? '', prefix)
|
||||
let durationStatuses = Object.fromEntries(
|
||||
mod.flow_jobs.map((flowJobId, idx) => {
|
||||
let started_at_str = mod.flow_jobs_duration?.started_at?.[idx]
|
||||
let started_at = started_at_str ? new Date(started_at_str).getTime() : undefined
|
||||
let duration_ms = mod.flow_jobs_duration?.duration_ms?.[idx]
|
||||
if (started_at == undefined) {
|
||||
let missingStartedAt = jobMissingStartedAt[flowJobId]
|
||||
if (missingStartedAt != 'P') {
|
||||
started_at = missingStartedAt
|
||||
}
|
||||
} else {
|
||||
delete jobMissingStartedAt[flowJobId]
|
||||
}
|
||||
return [
|
||||
flowJobId,
|
||||
{
|
||||
created_at: started_at,
|
||||
started_at: started_at,
|
||||
duration_ms: duration_ms
|
||||
}
|
||||
]
|
||||
})
|
||||
)
|
||||
let missingStartedAtIds = Object.keys(durationStatuses)
|
||||
.filter(
|
||||
(id) => durationStatuses[id].created_at == undefined && jobMissingStartedAt[id] != 'P'
|
||||
)
|
||||
.slice(0, 100)
|
||||
|
||||
updateDurationStatuses(key, durationStatuses)
|
||||
|
||||
if (missingStartedAtIds.length > 0) {
|
||||
missingStartedAtIds.forEach((id) => {
|
||||
jobMissingStartedAt[id] = 'P'
|
||||
})
|
||||
JobService.getStartedAtByIds({
|
||||
workspace: workspaceId ?? $workspaceStore ?? '',
|
||||
requestBody: missingStartedAtIds
|
||||
})
|
||||
.then((jobs) => {
|
||||
let lastStarted: string | undefined = undefined
|
||||
let anySet = false
|
||||
let nDurationStatuses = localDurationStatuses[key]?.byJob
|
||||
missingStartedAtIds.forEach((id, idx) => {
|
||||
const startedAt = jobs[idx]
|
||||
const time = startedAt ? new Date(startedAt).getTime() : undefined
|
||||
if (time) {
|
||||
jobMissingStartedAt[id] = time
|
||||
} else {
|
||||
delete jobMissingStartedAt[id]
|
||||
}
|
||||
if (nDurationStatuses && time) {
|
||||
if (!nDurationStatuses[id]?.duration_ms) {
|
||||
anySet = true
|
||||
lastStarted = id
|
||||
nDurationStatuses[id] = {
|
||||
created_at: time,
|
||||
started_at: time
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
if (anySet) {
|
||||
updateDurationStatuses(key, nDurationStatuses)
|
||||
lastStarted && setSelectedLoopSwitch(lastStarted, mod)
|
||||
}
|
||||
})
|
||||
.catch((e) => {
|
||||
console.error(`Could not load inner module for job ${mod.job}`, e)
|
||||
console.error(`Could not load inner module duration status for job ${mod.job}`, e)
|
||||
})
|
||||
} else if (
|
||||
(mod.flow_jobs || mod.branch_chosen) &&
|
||||
(mod.type == 'Success' || mod.type == 'Failure') &&
|
||||
!['Success', 'Failure'].includes(localModuleStates?.[mod.id ?? '']?.type)
|
||||
) {
|
||||
let branchChosen = mod.branch_chosen
|
||||
? {
|
||||
branchChosen:
|
||||
mod.branch_chosen.type == 'default' ? 0 : (mod.branch_chosen.branch ?? 0) + 1
|
||||
}
|
||||
: {}
|
||||
// console.debug('updateInnerModules', mod.id, mod.type, branchChosen)
|
||||
setModuleState(
|
||||
mod.id ?? '',
|
||||
{
|
||||
type: mod.type,
|
||||
...branchChosen
|
||||
},
|
||||
true
|
||||
)
|
||||
} else if (isForloopSelected) {
|
||||
setModuleState(mod.id ?? '', {}, true)
|
||||
} else {
|
||||
setIteration(0, mod.flow_jobs?.[0] ?? '', false, mod.id ?? '', true)
|
||||
}
|
||||
}
|
||||
|
||||
if (mod.flow_jobs_success || mod.flow_jobs_duration || mod.flow_jobs) {
|
||||
setModuleState(mod.id ?? '', {
|
||||
flow_jobs_success: mod.flow_jobs_success,
|
||||
flow_jobs_duration: mod.flow_jobs_duration,
|
||||
flow_jobs: mod.flow_jobs,
|
||||
iteration_total: mod.iterator?.itered_len ?? mod.flow_jobs?.length
|
||||
})
|
||||
}
|
||||
|
||||
if (mod.flow_jobs_duration && mod.flow_jobs) {
|
||||
let key = buildSubflowKey(mod.id ?? '', prefix)
|
||||
let durationStatuses = Object.fromEntries(
|
||||
mod.flow_jobs.map((flowJobId, idx) => {
|
||||
let started_at_str = mod.flow_jobs_duration?.started_at?.[idx]
|
||||
let started_at = started_at_str ? new Date(started_at_str).getTime() : undefined
|
||||
let duration_ms = mod.flow_jobs_duration?.duration_ms?.[idx]
|
||||
if (started_at == undefined) {
|
||||
let missingStartedAt = jobMissingStartedAt[flowJobId]
|
||||
if (missingStartedAt != 'P') {
|
||||
started_at = missingStartedAt
|
||||
}
|
||||
} else {
|
||||
delete jobMissingStartedAt[flowJobId]
|
||||
}
|
||||
return [
|
||||
flowJobId,
|
||||
{
|
||||
created_at: started_at,
|
||||
started_at: started_at,
|
||||
duration_ms: duration_ms
|
||||
}
|
||||
]
|
||||
})
|
||||
)
|
||||
let missingStartedAtIds = Object.keys(durationStatuses)
|
||||
.filter(
|
||||
(id) => durationStatuses[id].created_at == undefined && jobMissingStartedAt[id] != 'P'
|
||||
)
|
||||
.slice(0, 100)
|
||||
|
||||
updateDurationStatuses(key, durationStatuses)
|
||||
|
||||
if (missingStartedAtIds.length > 0) {
|
||||
missingStartedAtIds.forEach((id) => {
|
||||
jobMissingStartedAt[id] = 'P'
|
||||
})
|
||||
JobService.getStartedAtByIds({
|
||||
workspace: workspaceId ?? $workspaceStore ?? '',
|
||||
requestBody: missingStartedAtIds
|
||||
})
|
||||
.then((jobs) => {
|
||||
let lastStarted: string | undefined = undefined
|
||||
let anySet = false
|
||||
let nDurationStatuses = localDurationStatuses[key]?.byJob
|
||||
missingStartedAtIds.forEach((id, idx) => {
|
||||
const startedAt = jobs[idx]
|
||||
const time = startedAt ? new Date(startedAt).getTime() : undefined
|
||||
if (time) {
|
||||
jobMissingStartedAt[id] = time
|
||||
} else {
|
||||
delete jobMissingStartedAt[id]
|
||||
}
|
||||
if (nDurationStatuses && time) {
|
||||
if (!nDurationStatuses[id]?.duration_ms) {
|
||||
anySet = true
|
||||
lastStarted = id
|
||||
nDurationStatuses[id] = {
|
||||
created_at: time,
|
||||
started_at: time
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
if (anySet) {
|
||||
updateDurationStatuses(key, nDurationStatuses)
|
||||
selectedLoopSwitchTimeout && clearTimeout(selectedLoopSwitchTimeout)
|
||||
function setSelectedLoopSwitch() {
|
||||
if (lastStarted) {
|
||||
let position = mod.flow_jobs?.indexOf(lastStarted)
|
||||
if (position != undefined) {
|
||||
lastSelectedLoopSwitch = new Date().getTime()
|
||||
console.log('setSelectedLoopSwitch', position, lastStarted)
|
||||
setIteration(position, lastStarted, false, mod.id ?? '', true)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
lastSelectedLoopSwitch &&
|
||||
new Date().getTime() - lastSelectedLoopSwitch < 3000
|
||||
) {
|
||||
selectedLoopSwitchTimeout = setTimeout(() => {
|
||||
setSelectedLoopSwitch()
|
||||
}, 2000)
|
||||
} else {
|
||||
console.log('setSelectedLoopSwitch')
|
||||
setSelectedLoopSwitch()
|
||||
}
|
||||
}
|
||||
if (mod.agent_actions && mod.id) {
|
||||
setModuleState(mod.id, {
|
||||
agent_actions: mod.agent_actions
|
||||
})
|
||||
mod.agent_actions.forEach((action, idx) => {
|
||||
if (mod.id) {
|
||||
if (action.type == 'tool_call') {
|
||||
const toolCallId = getToolCallId(idx, mod.id, action.module_id)
|
||||
const success = mod.agent_actions_success?.[idx]
|
||||
setModuleState(toolCallId, {
|
||||
job_id: action.job_id,
|
||||
type: success != undefined ? (success ? 'Success' : 'Failure') : 'InProgress'
|
||||
})
|
||||
.catch((e) => {
|
||||
console.error(`Could not load inner module duration status for job ${mod.job}`, e)
|
||||
} else if (action.type == 'mcp_tool_call') {
|
||||
const mcpToolCallId = AI_MCP_TOOL_CALL_PREFIX + '-' + mod.id + '-' + idx
|
||||
const success = mod.agent_actions_success?.[idx]
|
||||
setModuleState(mcpToolCallId, {
|
||||
type: success != undefined ? (success ? 'Success' : 'Failure') : 'InProgress'
|
||||
})
|
||||
} else if (action.type == 'web_search') {
|
||||
const websearchId = AI_WEBSEARCH_PREFIX + '-' + mod.id + '-' + idx
|
||||
setModuleState(websearchId, {
|
||||
type: 'Success'
|
||||
})
|
||||
} else if (action.type == 'message') {
|
||||
const toolCallId = getToolCallId(idx, mod.id)
|
||||
setModuleState(toolCallId, {
|
||||
type: 'Success'
|
||||
})
|
||||
} else {
|
||||
setIteration(0, mod.flow_jobs?.[0] ?? '', false, mod.id ?? '', true)
|
||||
}
|
||||
}
|
||||
|
||||
if (mod.agent_actions && mod.id) {
|
||||
setModuleState(mod.id, {
|
||||
agent_actions: mod.agent_actions
|
||||
})
|
||||
mod.agent_actions.forEach((action, idx) => {
|
||||
if (mod.id) {
|
||||
if (action.type == 'tool_call') {
|
||||
const toolCallId = getToolCallId(idx, mod.id, action.module_id)
|
||||
const success = mod.agent_actions_success?.[idx]
|
||||
setModuleState(toolCallId, {
|
||||
job_id: action.job_id,
|
||||
type: success != undefined ? (success ? 'Success' : 'Failure') : 'InProgress'
|
||||
})
|
||||
} else if (action.type == 'mcp_tool_call') {
|
||||
const mcpToolCallId = AI_MCP_TOOL_CALL_PREFIX + '-' + mod.id + '-' + idx
|
||||
const success = mod.agent_actions_success?.[idx]
|
||||
setModuleState(mcpToolCallId, {
|
||||
type: success != undefined ? (success ? 'Success' : 'Failure') : 'InProgress'
|
||||
})
|
||||
} else if (action.type == 'web_search') {
|
||||
const websearchId = AI_WEBSEARCH_PREFIX + '-' + mod.id + '-' + idx
|
||||
setModuleState(websearchId, {
|
||||
type: 'Success'
|
||||
})
|
||||
} else if (action.type == 'message') {
|
||||
const toolCallId = getToolCallId(idx, mod.id)
|
||||
setModuleState(toolCallId, {
|
||||
type: 'Success'
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -746,7 +723,7 @@
|
||||
} else {
|
||||
job = newJob
|
||||
}
|
||||
job?.flow_status && updateStatus(job?.flow_status)
|
||||
if (job?.flow_status) updateStatus(job?.flow_status)
|
||||
onJobsLoaded?.({ job, force: false })
|
||||
notAnonynmous = false
|
||||
if (job?.type == 'CompletedJob' && !destroyed) {
|
||||
|
||||
Reference in New Issue
Block a user