starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-08 16:03:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(frontend): add flow log view (#6330)

Browse Source 
* componentize detail module

* Add logs tab

* add flow log viewer

* fetch logs

* handle subflows

* add forloop iteration picker

* polish

* expand steps log by default

* move logic handling to wrapper component

* improve iteration picker

* clean code

* polishing

* Add flow start and flow end rows

* fix infinite loop

* nit

* use list instead of table

* use custom id for collapsing subflow

* remove debug logs

* Use status dot instead of text

* fetch log from moduleState

* wip

* only fetch subflow jobs from cache if job is completed

* Add job polling for expanded steps

* handle subflows

* Init logs for steps

* update localModuleState logs

* use selected iteration from local module state

* handle branchone

* Add branch one and branch all label

* remove redondant innerModule prop

* Improve UX

* Add expand/collapse

* Add filter to hide result and inputs

* Steps are now flow children

* improve UX

* Open flow and steps sction when executing

* Handle empty subflows

* remove unnecessary sequence viewer component

* nit

* use iteration picker in log view

* Replace dot with step type icon

* indicate subflows

* add step number and progression

* Incorporate inputs and results in the list of steps

* Add error indicator when subflow has error

* improve topbar

* improve log polling

* Improve log polling

* Add root flow log fetching and polling

* Add debounce for loading subflow jobs

* write a function to build the tree view from the graph

* remove unnecessary log polling

* fix flow result display

* flag errors

* preprocessor

* remove all flow logs drawer

* grenerate graph from component

* wip

* Check module change before building graph

* nit

* fix log overflow

* fix log viewer borders

* mini jobs run preview fix

* elegent job logs loading

* nit

* nit

* nit

* all

* all

* all

* all

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
Guilhem
2025-08-14 00:39:21 +00:00
committed by GitHub
co-authored by Ruben Fiszel
parent 3066bccad4
commit 235354fe13
24 changed files with 1175 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/openapi.yaml
+21
View File
@@ -7777,6 +7777,10 @@ paths:
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/JobId"
- name: remove_ansi_warnings
in: query
schema:
type: boolean
responses:
"200":
description: job details
@@ -7785,6 +7789,23 @@ paths:
schema:
type: string
/w/{workspace}/jobs_u/get_completed_logs_tail/{id}:
get:
summary: get completed job logs tail
operationId: getCompletedJobLogsTail
tags:
- job
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/JobId"
responses:
"200":
description: completed job logs tail
content:
text/plain:
schema:
type: string
/w/{workspace}/jobs_u/get_args/{id}:
get:
summary: get job args
backend/windmill-api/src/jobs.rs
+52 -3
View File
@@ -272,6 +272,7 @@ pub fn workspace_unauthed_service() -> Router {
.route("/get_root_job_id/:id", get(get_root_job))
.route("/get/:id", get(get_job))
.route("/get_logs/:id", get(get_job_logs))
.route("/get_completed_logs_tail/:id", get(get_completed_job_logs_tail))
.route("/get_args/:id", get(get_args))
.route("/get_flow_debug_info/:id", get(get_flow_job_debug_info))
.route("/completed/get/:id", get(get_completed_job))
@@ -1361,11 +1362,55 @@ async fn get_logs_from_disk(
return None;
}
async fn get_completed_job_logs_tail(
OptAuthed(opt_authed): OptAuthed,
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::JsonResult<String> {
let tags = opt_authed
.as_ref()
.map(|authed| get_scope_tags(authed).map(|v| v.iter().map(|s| s.to_string()).collect_vec()))
.flatten();
let record = sqlx::query!(
"SELECT created_by AS \"created_by!\", coalesce(job_logs.logs, '') as logs
FROM v2_job
LEFT JOIN job_logs ON job_logs.job_id = v2_job.id
WHERE v2_job.id = $1 AND v2_job.workspace_id = $2 AND ($3::text[] IS NULL OR v2_job.tag = ANY($3))
ORDER BY job_logs.log_offset DESC
LIMIT 100",
id,
w_id,
tags.as_ref().map(|v| v.as_slice())
)
.fetch_optional(&db)
.await?;
if let Some(record) = record {
if opt_authed.is_none() && record.created_by != "anonymous" {
return Err(Error::BadRequest(
"As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
));
}
let logs = record.logs.unwrap_or_default();
Ok(Json(logs))
} else {
Err(Error::NotFound("Job not found".to_string()).into())
}
}
#[derive(Debug, Deserialize)]
struct QueryJobLogs {
remove_ansi_warnings: Option<bool>,
}
async fn get_job_logs(
OptAuthed(opt_authed): OptAuthed,
opt_tokened: OptTokened,
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
Query(query_job_logs): Query<QueryJobLogs>,
) -> error::Result<Response> {
// let audit_author: AuditAuthor = match opt_authed {
// Some(authed) => (&authed).into(),
@@ -1419,10 +1464,14 @@ async fn get_job_logs(
{
return r.map(content_plain);
}
let logs = format!(
"to remove ansi colors, use: | sed 's/\\x1B\\[[0-9;]\\{{1,\\}}[A-Za-z]//g'\n{}",
let logs = if query_job_logs.remove_ansi_warnings.unwrap_or(false) {
logs
);
} else {
format!(
"to remove ansi colors, use: | sed 's/\\x1B\\[[0-9;]\\{{1,\\}}[A-Za-z]//g'\n{}",
logs
)
};
Ok(content_plain(Body::from(logs)))
} else {
let text = sqlx::query!(
frontend/src/lib/components/AllFlowLogs.svelte
-39
View File
@@ -1,39 +0,0 @@
<script lang="ts">
import { Loader2 } from 'lucide-svelte'
import JobLogs from './JobLogs.svelte'
import { workspaceStore } from '$lib/stores'
import { base } from '$lib/base'
import type { DurationStatus } from './graph'
import type { Writable } from 'svelte/store'
interface Props {
states: Writable<Record<string, DurationStatus>> | undefined
}
let { states }: Props = $props()
</script>
<div class="flex flex-col">
{#if states != undefined}
{#each Object.entries($states ?? {}) as [id, status] (id)}
<div class="pb-12"
><h1 class="mb-2">Step {id}</h1>
{#each Object.entries(status.byJob ?? {}) as jobS}
{@const job = jobS[0]}
<div>
<a
class="text-xs"
rel="noreferrer"
target="_blank"
href="{base}/run/{job}?workspace={$workspaceStore}"
>
{job}
</a><JobLogs jobId={job} /></div
>
{/each}
</div>
{/each}
{:else}
<Loader2 size={14} class="animate-spin " />
{/if}
</div>
frontend/src/lib/components/Dev.svelte
+2 -5
View File
@@ -50,7 +50,7 @@
import { Triggers } from './triggers/triggers.svelte'
import { TestSteps } from './flows/testSteps.svelte'
import { ModulesTestStates } from './modulesTest.svelte'
import type { DurationStatus, GraphModuleState } from './graph'
import type { GraphModuleState } from './graph'
import { updateDerivedModuleStatesFromTestJobs } from './flows/utils'
let flowCopilotContext: FlowCopilotContext = {
@@ -589,9 +589,7 @@
const localModuleStates: Writable<Record<string, GraphModuleState>> = $derived(
flowPreviewContent?.getLocalModuleStates() ?? writable({})
)
const localDurationStatuses: Writable<Record<string, DurationStatus>> = $derived(
flowPreviewContent?.getLocalDurationStatuses() ?? writable({})
)
const suspendStatus: Writable<Record<string, { job: Job; nb: number }>> = $derived(
flowPreviewContent?.getSuspendStatus() ?? writable({})
)
@@ -839,7 +837,6 @@
onTestFlow={flowPreviewButtons?.runPreview}
{job}
isOwner={flowPreviewContent?.getIsOwner()}
{localDurationStatuses}
{suspendStatus}
onOpenDetails={flowPreviewButtons?.openPreview}
/>
frontend/src/lib/components/FlowBuilder.svelte
+1 -5
View File
@@ -78,7 +78,7 @@
import { Triggers } from './triggers/triggers.svelte'
import { TestSteps } from './flows/testSteps.svelte'
import { aiChatManager } from './copilot/chat/AIChatManager.svelte'
import type { DurationStatus, GraphModuleState } from './graph'
import type { GraphModuleState } from './graph'
import {
setStepHistoryLoaderContext,
StepHistoryLoader,
@@ -936,9 +936,6 @@
const localModuleStates: Writable<Record<string, GraphModuleState>> = $derived(
flowPreviewContent?.getLocalModuleStates() ?? writable({})
)
const localDurationStatuses: Writable<Record<string, DurationStatus>> = $derived(
flowPreviewContent?.getLocalDurationStatuses() ?? writable({})
)
const suspendStatus: Writable<Record<string, { job: Job; nb: number }>> = $derived(
flowPreviewContent?.getSuspendStatus() ?? writable({})
)
@@ -1239,7 +1236,6 @@
onHideJobStatus={resetModulesStates}
{individualStepTests}
{job}
{localDurationStatuses}
{suspendStatus}
{showJobStatus}
onDelete={(id) => {
frontend/src/lib/components/FlowJobResult.svelte
+1 -66
View File
@@ -2,14 +2,6 @@
import { Loader2 } from 'lucide-svelte'
import DisplayResult from './DisplayResult.svelte'
import LogViewer from './LogViewer.svelte'
import { JobService } from '$lib/gen'
import { workspaceStore } from '$lib/stores'
import DrawerContent from './common/drawer/DrawerContent.svelte'
import { Drawer } from './common'
import AllFlowLogs from './AllFlowLogs.svelte'
import type { DurationStatus } from './graph'
import type { Writable } from 'svelte/store'
import { untrack } from 'svelte'
interface Props {
waitingForExecutor?: boolean
@@ -24,7 +16,6 @@
tag?: string | undefined
workspaceId?: string | undefined
refreshLog?: boolean
durationStates: Writable<Record<string, DurationStatus>> | undefined
downloadLogs?: boolean
tagLabel?: string | undefined
}
@@ -33,7 +24,7 @@
waitingForExecutor = false,
result,
result_stream,
logs = $bindable(),
logs,
col = false,
noBorder = false,
loading,
@@ -41,64 +32,11 @@
jobId = undefined,
tag = undefined,
workspaceId = undefined,
refreshLog = false,
durationStates,
downloadLogs = true,
tagLabel = undefined
}: Props = $props()
let lastJobId: string | undefined = $state(undefined)
let drawer: Drawer | undefined = $state(undefined)
let iteration = 0
let logOffset = 0
async function diffJobId() {
if (jobId != lastJobId) {
lastJobId = jobId
logs = undefined
logOffset = 0
iteration = 0
getLogs()
}
}
async function getLogs() {
iteration += 1
if (jobId) {
const getUpdate = await JobService.getJobUpdates({
workspace: workspaceId ?? $workspaceStore!,
id: jobId,
running: loading ?? false,
logOffset: logOffset == 0 ? (logs?.length ? logs?.length + 1 : 0) : logOffset
})
logs = (logs ?? '').concat(getUpdate.new_logs ?? '')
logOffset = getUpdate.log_offset ?? 0
}
if (refreshLog) {
setTimeout(
() => {
if (refreshLog) {
getLogs()
}
},
iteration < 10 ? 1000 : iteration < 20 ? 2000 : 5000
)
}
}
$effect(() => {
jobId
untrack(() => {
jobId != lastJobId && diffJobId()
})
})
</script>
<Drawer bind:this={drawer}>
<DrawerContent title="Explore all steps' logs" on:close={drawer.closeDrawer}
><AllFlowLogs states={durationStates} /></DrawerContent
>
</Drawer>
<div
class:border={!noBorder}
class="grid {!col
@@ -116,9 +54,6 @@
{/if}
</div>
<div class="overflow-auto {col ? '' : 'max-h-80'} relative">
<div class="absolute z-40 text-xs top-0 left-1"
><button class="" onclick={drawer.openDrawer}>explore all steps' logs</button></div
>
<LogViewer
{tagLabel}
download={downloadLogs}
frontend/src/lib/components/FlowLogUtils.ts
+11
View File
@@ -0,0 +1,11 @@
import type { FlowModuleValue } from '$lib/gen'
export interface FlowLogEntry {
id: string
stepId: string
stepNumber?: number
summary?: string
stepType?: FlowModuleValue['type']
subflows?: FlowLogEntry[][]
subflowsSummary?: string[]
}
frontend/src/lib/components/FlowLogViewer.svelte
+676
View File
@@ -0,0 +1,676 @@
<script lang="ts">
import {
ChevronDown,
ChevronRight,
GitBranch,
Repeat,
Code,
ArrowDownToLine,
ArrowDownFromLine,
FoldVertical,
UnfoldVertical
} from 'lucide-svelte'
import { base } from '$lib/base'
import { workspaceStore } from '$lib/stores'
import { truncateRev } from '$lib/utils'
import ObjectViewer from './propertyPicker/ObjectViewer.svelte'
import LogViewer from './LogViewer.svelte'
import FlowLogViewer from './FlowLogViewer.svelte'
import type { FlowModuleValue, FlowStatusModule, Job } from '$lib/gen'
import { twMerge } from 'tailwind-merge'
import FlowJobsMenu from './flows/map/FlowJobsMenu.svelte'
import BarsStaggered from './icons/BarsStaggered.svelte'
import type { GraphModuleState } from './graph/model'
import type { Writable } from 'svelte/store'
import type { FlowLogEntry } from './FlowLogUtils'
type RootJobData = Partial<Job>
interface Props {
logEntries: FlowLogEntry[]
localModuleStates: Writable<Record<string, GraphModuleState>>
rootJob: RootJobData
expandedRows: Record<string, boolean>
allExpanded?: boolean
showResultsInputs?: boolean
toggleExpanded: (id: string) => void
toggleExpandAll?: () => void
workspaceId: string | undefined
render: boolean
level?: number
flowId: string
onSelectedIteration: (
detail:
| { id: string; index: number; manuallySet: true; moduleId: string }
| { manuallySet: false; moduleId: string }
) => Promise<void>
getSelectedIteration: (stepId: string) => number
flowSummary?: string
}
let {
logEntries,
localModuleStates,
rootJob,
expandedRows,
allExpanded,
showResultsInputs,
toggleExpanded,
toggleExpandAll,
workspaceId,
render,
level = 0,
flowId = 'root',
onSelectedIteration,
getSelectedIteration,
flowSummary
}: Props = $props()
function getJobLink(jobId: string | undefined): string {
if (!jobId) return ''
return `${base}/run/${jobId}?workspace=${workspaceId ?? $workspaceStore}`
}
function getStatusColor(status: FlowStatusModule['type'] | undefined): string {
const statusColors = {
Success: 'text-green-500',
Failure: 'text-red-500',
InProgress: 'text-yellow-500',
WaitingForPriorSteps: 'text-gray-400',
WaitingForEvents: 'text-purple-400',
WaitingForExecutor: 'text-gray-400'
}
return status ? statusColors[status] : 'text-gray-400'
}
function getFlowStatus(job: RootJobData): FlowStatusModule['type'] | undefined {
if (job.type === 'CompletedJob') {
return job.success ? 'Success' : 'Failure'
} else if (job.type === 'QueuedJob') {
return 'InProgress'
} else {
return undefined
}
}
function getStepProgress(job: RootJobData, totalSteps: number): string {
if (totalSteps === 0) return ''
// If flow is completed, show total steps
if (job.type === 'CompletedJob') {
return ` (${totalSteps} step${totalSteps === 1 ? '' : 's'})`
}
// If flow is running, use flow_status.step if available (like JobStatus.svelte)
if (job.type === 'QueuedJob') {
if (job.flow_status?.step !== undefined) {
const currentStep = (job.flow_status.step ?? 0) + 1
return ` (step ${currentStep} of ${totalSteps})`
}
return ''
}
return ''
}
function isExpanded(id: string, isRunning: boolean = false): boolean {
// If explicitly set in expandedRows, use that value
// Otherwise, fall back to allExpanded
return expandedRows[id] ?? (allExpanded || isRunning)
}
function hasEmptySubflow(stepId: string, stepType: FlowModuleValue['type'] | undefined): boolean {
const state = $localModuleStates[stepId]
if (!state || !stepType) return false
return (
['forloopflow', 'whileloopflow'].includes(stepType) &&
(!state.flow_jobs || state.flow_jobs.length === 0)
)
}
// Find all parents of error steps
function findParentsOfErrors(entries: FlowLogEntry[]): Set<string> {
const parentsWithErrors = new Set<string>()
function traverseEntries(entryList: FlowLogEntry[], parentId?: string) {
let hasChildError = false
for (const entry of entryList) {
let currentEntryHasError = false
// Check if this entry has subflows with errors
if (entry.subflows && entry.subflows.length > 0) {
for (const subflow of entry.subflows) {
const subflowHasError = traverseEntries(subflow, entry.stepId)
if (subflowHasError) {
currentEntryHasError = true
parentsWithErrors.add(entry.stepId)
}
}
}
// Check if this entry itself has an error (but don't flag it - only its parents)
const stepStatus = $localModuleStates[entry.stepId]?.type
if (stepStatus === 'Failure') {
currentEntryHasError = true
// Don't add the entry itself to parentsWithErrors
}
// If this entry has an error, mark its parent
if (currentEntryHasError && parentId) {
parentsWithErrors.add(parentId)
hasChildError = true
}
}
return hasChildError
}
traverseEntries(entries, flowId)
return parentsWithErrors
}
// Get flow info for display
const flowInfo = $derived.by(() => {
const parentsWithErrors = findParentsOfErrors(logEntries)
return {
jobId: rootJob.id,
inputs: rootJob.args || {},
result: rootJob.type === 'CompletedJob' ? rootJob.result : undefined,
logs: rootJob.logs || '',
status: rootJob.type,
label: flowSummary,
hasErrors: parentsWithErrors.has(flowId),
parentsWithErrors
}
})
</script>
{#if render}
{#if level === 0 && toggleExpandAll}
<div class="flex justify-end gap-4 items-center p-2 bg-surface-secondary border-b">
<div class="flex items-center gap-2 whitespace-nowrap">
<label
for="showResultsInputs"
class="text-xs text-tertiary hover:text-primary transition-colors"
>Show inputs/results</label
>
<div class="flex-shrink-0">
<input
type="checkbox"
name="showResultsInputs"
id="showResultsInputs"
bind:checked={showResultsInputs}
class="w-3 h-4 accent-primary -my-1"
/>
</div>
</div>
<button
onclick={toggleExpandAll}
class="text-xs text-tertiary hover:text-primary transition-colors flex items-center gap-2 min-w-24 justify-end"
>
{allExpanded ? 'Collapse All' : 'Expand All'}
{#if allExpanded}
<FoldVertical size={16} />
{:else}
<UnfoldVertical size={16} />
{/if}
</button>
</div>
{/if}
<ul class="w-full font-mono text-xs bg-surface-secondary list-none">
<!-- Flow entry -->
<li class="border-b flex flex-row">
<div class="py-2 leading-tight align-top">
{#if level > 0}
<button
class="w-4 flex items-center justify-center text-xs text-tertiary hover:text-primary transition-colors"
onclick={() => toggleExpanded(`flow-${flowId}`)}
>
{#if isExpanded(`flow-${flowId}`, rootJob.type === 'QueuedJob')}
<ChevronDown size={8} />
{:else}
<ChevronRight size={8} />
{/if}
</button>
{:else}
<!-- Root flow - no collapse button, just spacing -->
<div class="w-4"></div>
{/if}
</div>
<div class="grow min-w-0 leading-tigh">
<!-- svelte-ignore a11y_click_events_have_key_events -->
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class={twMerge(
'py-1 flex items-center justify-between pr-2',
level > 0 ? 'cursor-pointer' : '',
rootJob.type === undefined ? 'opacity-50' : ''
)}
onclick={level > 0 ? () => toggleExpanded(`flow-${flowId}`) : undefined}
>
<div class="flex items-center gap-2 grow min-w-0">
<!-- Flow icon -->
{@render flowIcon(getFlowStatus(rootJob), flowInfo.hasErrors)}
<div class="flex items-center gap-2">
<span class="text-xs font-mono">
{flowId === 'root' ? 'Flow' : 'Subflow'}
{#if flowInfo.label}
: {flowInfo.label}
{/if}
<span class="text-tertiary">{getStepProgress(rootJob, logEntries.length)}</span>
</span>
</div>
</div>
{#if flowInfo.jobId}
<a
href={getJobLink(flowInfo.jobId)}
class="text-xs text-primary hover:underline font-mono"
target="_blank"
rel="noopener noreferrer"
onclick={(e) => e.stopPropagation()}
>
{truncateRev(flowInfo.jobId, 6)}
</a>
{/if}
</div>
{#if level === 0 || isExpanded(`flow-${flowId}`, rootJob.type === 'QueuedJob')}
<div class="mb-2 transition-all duration-200 ease-in-out w-full">
<!-- Flow logs -->
{#if flowInfo.logs}
<LogViewer
content={flowInfo.logs}
jobId={flowInfo.jobId}
isLoading={false}
small={true}
download={false}
noAutoScroll={true}
tag={undefined}
noPadding
wrapperClass="w-full mb-2 pr-2"
/>
{/if}
<!-- Flow steps - nested as children -->
<ul class="w-full font-mono text-xs bg-surface-secondary list-none border-l">
<!-- Flow inputs as first row entry -->
{#if showResultsInputs && flowInfo.inputs && Object.keys(flowInfo.inputs).length > 0}
<li class="border-b flex flex-row w-full">
<div class="py-2 leading-tight align-top">
<button
class="w-4 flex items-center justify-center text-xs text-tertiary hover:text-primary transition-colors"
onclick={() => toggleExpanded(`flow-${flowId}-input`)}
>
{#if isExpanded(`flow-${flowId}-input`)}
<ChevronDown size={8} />
{:else}
<ChevronRight size={8} />
{/if}
</button>
</div>
<div class="grow min-w-0 leading-tight">
<!-- svelte-ignore a11y_no_static_element_interactions -->
<!-- svelte-ignore a11y_click_events_have_key_events -->
<div
class="py-1 flex items-center justify-between pr-2 cursor-pointer"
onclick={() => toggleExpanded(`flow-${flowId}-input`)}
>
<div class="flex items-center gap-2 grow min-w-0">
<ArrowDownToLine size={10} />
<span class="text-xs font-mono">Inputs</span>
</div>
</div>
{#if isExpanded(`flow-${flowId}-input`)}
<div class="my-1 transition-all duration-200 ease-in-out">
<div class="pl-4">
<ObjectViewer json={flowInfo.inputs} pureViewer={true} />
</div>
</div>
{/if}
</div>
</li>
{/if}
{#if logEntries.length > 0}
{#each logEntries as entry (entry.id)}
{@const isLeafStep =
entry.stepType !== 'branchall' &&
entry.stepType !== 'branchone' &&
entry.stepType !== 'forloopflow' &&
entry.stepType !== 'whileloopflow'}
{@const status = $localModuleStates[entry.stepId]?.type}
{@const isRunning = status === 'InProgress' || status === 'WaitingForExecutor'}
{@const hasEmptySubflowValue = hasEmptySubflow(entry.stepId, entry.stepType)}
{@const isCollapsible = !hasEmptySubflowValue}
<li class="border-b flex flex-row">
<div class="py-2 leading-tight align-top">
{#if isCollapsible}
<button
class="w-4 flex items-center justify-center text-xs text-tertiary hover:text-primary transition-colors"
onclick={() => toggleExpanded(entry.id)}
>
{#if isExpanded(entry.id, isRunning)}
<ChevronDown size={8} />
{:else}
<ChevronRight size={8} />
{/if}
</button>
{:else}
<!-- Empty subflow - no collapse button, just spacing -->
<div class="w-4"></div>
{/if}
</div>
<div class="w-full leading-tight grow min-w-0">
<!-- svelte-ignore a11y_click_events_have_key_events -->
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class={twMerge(
'py-1 flex items-center justify-between pr-2',
isCollapsible ? 'cursor-pointer' : '',
status === 'WaitingForPriorSteps' ||
status === 'WaitingForEvents' ||
status === 'WaitingForExecutor' ||
status === undefined
? 'opacity-50'
: ''
)}
onclick={isCollapsible ? () => toggleExpanded(entry.id) : undefined}
>
<div class="flex items-center gap-2 grow min-w-0">
<!-- Step icon -->
{@render stepIcon(
entry.stepType,
status as FlowStatusModule['type'],
flowInfo.parentsWithErrors.has(entry.stepId)
)}
<div class="flex items-center gap-2">
<span class="text-xs font-mono">
<b>
{entry.stepId}
</b>
{#if entry.stepType === 'forloopflow'}
For loop
{:else if entry.stepType === 'whileloopflow'}
While loop
{:else if entry.stepType === 'branchall'}
Branch to all
{:else if entry.stepType === 'branchone'}
Branch to one
{:else if entry.stepType === 'flow'}
Subflow
{:else}
Step
{/if}
{#if entry.summary}
: {entry.summary}
{/if}
{#if hasEmptySubflowValue}
<span class="text-tertiary">
{#if entry.stepType === 'forloopflow' || entry.stepType === 'whileloopflow'}
(empty loop)
{:else if entry.stepType === 'branchall' || entry.stepType === 'branchone'}
(no branch)
{/if}
</span>
{/if}
</span>
{#if !hasEmptySubflowValue && $localModuleStates[entry.stepId]?.flow_jobs && (entry.stepType === 'forloopflow' || entry.stepType === 'whileloopflow')}
<span
class="text-xs font-mono font-medium inline-flex items-center grow min-w-0 -my-2"
>
<span onclick={(e) => e.stopPropagation()}>
<FlowJobsMenu
moduleId={entry.stepId}
id={entry.stepId}
{onSelectedIteration}
flowJobsSuccess={$localModuleStates[entry.stepId]
?.flow_jobs_success}
flowJobs={$localModuleStates[entry.stepId]?.flow_jobs}
selected={$localModuleStates[entry.stepId]
?.selectedForloopIndex ?? 0}
selectedManually={$localModuleStates[entry.stepId]
?.selectedForLoopSetManually ?? false}
showIcon={false}
/>
</span>
{#if entry.stepType === 'forloopflow'}
{`/${$localModuleStates[entry.stepId]?.iteration_total ?? 0}`}
{/if}
</span>
{/if}
</div>
</div>
{#if isLeafStep}
{@const jobId = $localModuleStates[entry.stepId]?.job_id}
<a
href={getJobLink(jobId ?? '')}
class="text-xs text-primary hover:underline font-mono"
target="_blank"
rel="noopener noreferrer"
>
{truncateRev(jobId ?? '', 6)}
</a>
{/if}
</div>
{#if isCollapsible && isExpanded(entry.id, isRunning)}
{@const args = $localModuleStates[entry.stepId]?.args}
{@const logs = $localModuleStates[entry.stepId]?.logs}
{@const result = $localModuleStates[entry.stepId]?.result}
{@const jobId = $localModuleStates[entry.stepId]?.job_id}
<div class="my-1 transition-all duration-200 ease-in-out">
<!-- Show child steps if they exist -->
{#if entry.subflows && entry.subflows.length > 0}
{#each entry.subflows as subflow, index}
{@const subflowLabel = entry.subflowsSummary?.[index]}
{@const subflowJob = {
id: jobId,
type:
$localModuleStates[entry.stepId]?.type === 'Failure' ||
$localModuleStates[entry.stepId]?.type === 'Success'
? 'CompletedJob'
: ('QueuedJob' as Job['type']),
logs,
result,
args,
success: $localModuleStates[entry.stepId]?.type === 'Success'
}}
<div class="border-l mb-2">
<!-- Recursively render child steps using FlowLogViewer -->
<FlowLogViewer
logEntries={subflow}
{localModuleStates}
rootJob={subflowJob}
{expandedRows}
{allExpanded}
{showResultsInputs}
{toggleExpanded}
toggleExpandAll={undefined}
{workspaceId}
{render}
level={level + 1}
flowId={`${entry.stepId}-subflow-${index}`}
flowSummary={subflowLabel}
{onSelectedIteration}
{getSelectedIteration}
/>
</div>
{/each}
<!-- Show input arguments -->
{:else}
{#if showResultsInputs && isLeafStep && args && Object.keys(args).length > 0}
<div class="mb-2">
<!-- svelte-ignore a11y_click_events_have_key_events -->
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class="flex items-center gap-1 cursor-pointer hover:text-primary text-xs font-mono font-medium mb-1"
onclick={() => toggleExpanded(`${entry.id}-input`)}
>
{#if isExpanded(`${entry.id}-input`)}
<ChevronDown size={8} />
{:else}
<ChevronRight size={8} />
{/if}
Input
</div>
{#if isExpanded(`${entry.id}-input`)}
<div class="pl-4">
<ObjectViewer json={args} pureViewer={true} />
</div>
{/if}
</div>
{/if}
<!-- Show logs if they exist -->
{#if logs}
<LogViewer
content={logs}
jobId={jobId ?? ''}
isLoading={false}
small={true}
download={false}
noAutoScroll={true}
tag={undefined}
noPadding
wrapperClass="w-full mb-2 pr-2"
/>
{:else if jobId && !entry.subflows?.[0]?.length}
<div class="mb-2">
<div class="text-xs text-tertiary font-mono">
No logs available
</div>
</div>
{/if}
<!-- Show result if completed -->
{#if showResultsInputs && isLeafStep && result !== undefined && (status === 'Success' || status === 'Failure')}
<div class="mb-2 mt-2">
<!-- svelte-ignore a11y_click_events_have_key_events -->
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class="flex items-center gap-1 cursor-pointer hover:text-primary text-xs font-mono font-medium mb-1"
onclick={() => toggleExpanded(`${entry.id}-result`)}
>
{#if isExpanded(`${entry.id}-result`)}
<ChevronDown size={8} />
{:else}
<ChevronRight size={8} />
{/if}
Result
</div>
{#if isExpanded(`${entry.id}-result`)}
<div class="pl-4">
<ObjectViewer json={result} pureViewer={true} />
</div>
{/if}
</div>
{/if}
{/if}
</div>
{/if}
</div>
</li>
{/each}
{/if}
<!-- Flow result as last row entry -->
{#if showResultsInputs && flowInfo.result !== undefined && rootJob.type === 'CompletedJob'}
<li class="border-b flex">
<div class="py-2 leading-tight align-top">
<button
class="w-4 flex items-center justify-center text-xs text-tertiary hover:text-primary transition-colors"
onclick={() => toggleExpanded(`flow-${flowId}-result`)}
>
{#if isExpanded(`flow-${flowId}-result`)}
<ChevronDown size={8} />
{:else}
<ChevronRight size={8} />
{/if}
</button>
</div>
<div class="w-full leading-tight">
<!-- svelte-ignore a11y_no_static_element_interactions -->
<!-- svelte-ignore a11y_click_events_have_key_events -->
<div
class="py-1 flex items-center justify-between pr-2 cursor-pointer"
onclick={() => toggleExpanded(`flow-${flowId}-result`)}
>
<div class="flex items-center gap-2 grow min-w-0">
<ArrowDownFromLine size={10} />
<span class="text-xs font-mono">Results</span>
</div>
</div>
{#if isExpanded(`flow-${flowId}-result`)}
<div class="my-1 transition-all duration-200 ease-in-out">
<div class="pl-4">
<ObjectViewer json={flowInfo.result} pureViewer={true} />
</div>
</div>
{/if}
</div>
</li>
{/if}
</ul>
</div>
{/if}
</div>
</li>
</ul>
{/if}
{#snippet flowIcon(status: FlowStatusModule['type'] | undefined, hasErrors: boolean)}
{@const colorClass = getStatusColor(status)}
<div class="relative flex items-center">
<BarsStaggered
size={10}
class={twMerge(colorClass, status === 'InProgress' ? 'animate-pulse' : '', 'flex-shrink-0')}
/>
{#if hasErrors && status !== 'Failure'}
<span
class="text-red-500 -ml-0.5 -mr-1.5"
title="A subflow or a step has failed but failure was skipped">!</span
>
{/if}
</div>
{/snippet}
{#snippet stepIcon(
stepType: string | undefined,
status: FlowStatusModule['type'] | undefined,
hasErrors: boolean
)}
{@const colorClass = getStatusColor(status)}
{@const animationClass = status === 'InProgress' ? 'animate-pulse' : ''}
{@const classes = `${colorClass} ${animationClass} flex-shrink-0`}
<div class="relative flex items-center">
{#if stepType === 'flow'}
<BarsStaggered size={10} class={classes} />
{:else if stepType === 'forloopflow' || stepType === 'whileloopflow'}
<Repeat size={10} class={classes} />
{:else if stepType === 'branchall' || stepType === 'branchone'}
<GitBranch size={10} class={classes} />
{:else}
<Code strokeWidth={2.5} size={10} class={classes} />
{/if}
{#if hasErrors && status !== 'Failure'}
<span class="text-red-500 -ml-0.5 -mr-1.5" title="A subflow or a step has failed">!</span>
{/if}
</div>
{/snippet}
<style>
.transition-all {
transition: all 0.2s ease-in-out;
}
</style>
frontend/src/lib/components/FlowLogViewerWrapper.svelte
+216
View File
@@ -0,0 +1,216 @@
<script lang="ts">
import type { Job } from '$lib/gen'
import { writable, type Writable } from 'svelte/store'
import type { GraphModuleState } from './graph'
import FlowLogViewer from './FlowLogViewer.svelte'
import { graphBuilder, type NodeLayout } from './graph/graphBuilder.svelte'
import type { FlowLogEntry } from './FlowLogUtils'
import { untrack } from 'svelte'
import { ChangeTracker } from '$lib/svelte5Utils.svelte'
import { readFieldsRecursively } from '$lib/utils'
interface Props {
job: Job
localModuleStates: Writable<Record<string, GraphModuleState>>
workspaceId: string | undefined
render: boolean
onSelectedIteration: (
detail:
| { id: string; index: number; manuallySet: true; moduleId: string }
| { manuallySet: false; moduleId: string }
) => Promise<void>
}
let { job, localModuleStates, workspaceId, render, onSelectedIteration }: Props = $props()
// State for tracking expanded rows - using Record to allow explicit control
let expandedRows: Record<string, boolean> = $state({})
let allExpanded = $state(false)
let showResultsInputs = $state(true)
let emptyEventHandler = {
deleteBranch: () => {},
insert: () => {},
select: () => {},
changeId: () => {},
delete: () => {},
newBranch: () => {},
move: () => {},
selectedIteration: () => {},
simplifyFlow: () => {},
expandSubflow: () => {},
minimizeSubflow: () => {},
updateMock: () => {},
testUpTo: () => {},
editInput: () => {},
testFlow: () => {},
cancelTestFlow: () => {},
openPreview: () => {},
hideJobStatus: () => {}
}
let modules = $derived(job.raw_flow?.modules ?? [])
let moduleTracker = new ChangeTracker($state.snapshot(job.raw_flow?.modules ?? []))
$effect(() => {
readFieldsRecursively(modules)
untrack(() => moduleTracker.track($state.snapshot(modules)))
})
let nodes: NodeLayout[] | undefined = $derived.by(() => {
moduleTracker.counter
const graph = graphBuilder(
untrack(() => modules),
{
disableAi: false,
insertable: false,
flowModuleStates: undefined,
selectedId: undefined,
path: undefined,
newFlow: false,
cache: false,
earlyStop: false,
editMode: false,
isOwner: false,
isRunning: false,
individualStepTests: false,
flowJob: undefined,
showJobStatus: false,
suspendStatus: writable({}),
flowHasChanged: false
},
untrack(() => job.raw_flow?.failure_module),
untrack(() => job.raw_flow?.preprocessor_module),
emptyEventHandler, // eventHandler - empty for logs view
undefined,
false, // useDataflow
undefined, // selectedId
undefined, // moving
undefined, // simplifiableFlow
undefined, // triggerNode path
{} // expandedSubflows
)
return graph.error ? undefined : graph.nodes
})
function toggleExpanded(id: string) {
// If not in record, use opposite of allExpanded as new state
// If in record, toggle the current state
const currentState = expandedRows[id] ?? allExpanded
expandedRows[id] = !currentState
}
function getSelectedIteration(stepId: string): number {
return $localModuleStates[stepId]?.selectedForloopIndex ?? 0
}
function toggleExpandAll() {
allExpanded = !allExpanded
expandedRows = {}
}
// Build tree structure from modules using bottom-up traversal from result node
function buildFlowTree(nodes: NodeLayout[]): FlowLogEntry[] {
// Index nodes for quick access
const nodeById: Record<string, NodeLayout> = {}
for (const n of nodes) nodeById[n.id] = n
function traverseFromId(id: string): FlowLogEntry[] {
const entries: FlowLogEntry[] = []
const currentNode = nodeById[id]
if (currentNode.type === 'module') {
entries.push({
id: currentNode.id,
stepId: currentNode.id,
stepNumber: 0,
summary: currentNode.data.module.summary ?? '',
stepType: currentNode.data.module.value.type
})
}
if (
currentNode.type === 'whileLoopStart' ||
currentNode.type === 'branchOneStart' ||
currentNode.type === 'branchAllStart'
) {
// Reaching the end of a subflow
return entries
}
let nextParentId = currentNode.parentIds?.[0]
if (!nextParentId) {
// Reached the root of the flow
return entries
}
if (
currentNode.type === 'forLoopEnd' ||
currentNode.type === 'whileLoopEnd' ||
currentNode.type === 'branchOneEnd'
) {
const subflow = traverseFromId(nextParentId)
const subflowId = currentNode.id.slice(0, -4) // Remove '-end' suffix
const subflowNode = nodeById[subflowId]
const subflowSummary = currentNode.type === 'branchOneEnd' ? 'branch' : 'iteration'
if (subflowNode.type === 'module') {
entries.push({
id: subflowId,
stepId: subflowId,
subflows: [subflow],
stepType:
currentNode.type === 'forLoopEnd'
? 'forloopflow'
: currentNode.type === 'whileLoopEnd'
? 'whileloopflow'
: 'branchone',
summary: subflowNode.data.module.summary ?? '',
subflowsSummary: [subflowSummary]
})
nextParentId = subflowNode.parentIds?.[0] ?? ''
}
} else if (currentNode.type === 'branchAllEnd') {
const subflowId = currentNode.id.slice(0, -4) // Remove '-end' suffix
const subflowNode = nodeById[subflowId]
if (subflowNode.type === 'module' && subflowNode.data.module.value.type === 'branchall') {
const subflows = currentNode.parentIds?.map((id) => traverseFromId(id)) ?? []
const subflowsSummary = subflowNode.data.module.value.branches.map((b) => b.summary ?? '')
entries.push({
id: subflowId,
stepId: subflowId,
subflows: subflows,
stepType: 'branchall',
subflowsSummary
})
nextParentId = subflowNode.parentIds?.[0] ?? '' // a module can only have one parent
}
}
// Get entries from parent nodes
const parentEntries = traverseFromId(nextParentId)
return [...parentEntries, ...entries]
}
// Start from the result node and traverse backwards
return traverseFromId('result')
}
let logEntries = $derived(nodes ? buildFlowTree(nodes) : [])
</script>
<div class="w-full rounded-md overflow-hidden border">
<FlowLogViewer
{logEntries}
{localModuleStates}
rootJob={job}
{expandedRows}
{allExpanded}
{showResultsInputs}
{toggleExpanded}
{toggleExpandAll}
{onSelectedIteration}
{workspaceId}
{render}
{getSelectedIteration}
flowId="root"
/>
</div>
frontend/src/lib/components/FlowPreviewResult.svelte
-4
View File
@@ -5,7 +5,6 @@
import FlowStatusWaitingForEvents from './FlowStatusWaitingForEvents.svelte'
import type { FlowStatusModule, Job } from '$lib/gen'
import { emptyString } from '$lib/utils'
import type { DurationStatus } from './graph'
import type { Writable } from 'svelte/store'
import Badge from './common/badge/Badge.svelte'
@@ -15,7 +14,6 @@
isOwner: boolean
hideFlowResult: boolean
hideDownloadLogs: boolean
localDurationStatuses: Writable<Record<string, DurationStatus>>
innerModules: FlowStatusModule[]
suspendStatus: Writable<Record<string, { job: Job; nb: number }>>
hideJobId?: boolean
@@ -29,7 +27,6 @@
isOwner,
hideFlowResult,
hideDownloadLogs,
localDurationStatuses,
innerModules,
suspendStatus,
hideJobId,
@@ -54,7 +51,6 @@
loading={job['running'] == true}
result={job.result}
logs={job.logs}
durationStates={localDurationStatuses}
downloadLogs={!hideDownloadLogs}
/>
</div>
frontend/src/lib/components/FlowStatusViewer.svelte
+2
View File
@@ -120,4 +120,6 @@
bind:rightColumnSelect
{render}
{customUi}
graphTabOpen={true}
isNodeSelected={true}
/>
frontend/src/lib/components/FlowStatusViewerInner.svelte
+83 -33
View File
@@ -34,6 +34,7 @@
import { buildPrefix } from './graph/graphBuilder.svelte'
import { parseInputArgsAssets } from './assets/lib'
import FlowPreviewResult from './FlowPreviewResult.svelte'
import FlowLogViewerWrapper from './FlowLogViewerWrapper.svelte'
import type { FlowGraphAssetContext } from './flows/types'
import { createState } from '$lib/svelte5Utils.svelte'
import JobLoader from './JobLoader.svelte'
@@ -97,6 +98,9 @@
customUi?: {
tagLabel?: string | undefined
}
graphTabOpen: boolean
isNodeSelected: boolean
loadExtraLogs?: (logs: string) => void
}
let {
@@ -127,7 +131,10 @@
localModuleStates = writable({}),
localDurationStatuses = writable({}),
customUi,
onResultStreamUpdate = undefined
onResultStreamUpdate = undefined,
graphTabOpen,
isNodeSelected,
loadExtraLogs = undefined
}: Props = $props()
let resultStreams: Record<string, string | undefined> = $state({})
@@ -542,6 +549,14 @@
},
resultStreamUpdate({ id, result_stream }: { id: string; result_stream?: string }) {
onResultStreamUpdate?.({ jobId: id, result_stream })
},
loadExtraLogs({ id, logs }: { id: string; logs: string }) {
if (id == jobId && job) {
job.logs = logs
}
if (loadExtraLogs) {
loadExtraLogs(logs)
}
}
})
}
@@ -926,6 +941,38 @@
let subflowsSize = $state(500)
async function onSelectedIteration(
detail:
| { id: string; index: number; manuallySet: true; moduleId: string }
| { manuallySet: false; moduleId: string }
) {
if (detail.manuallySet) {
let rootJobId = detail.id
await tick()
let previousId = $localModuleStates[detail.moduleId]?.selectedForloop
if (previousId) {
await globalRefreshes?.[detail.moduleId]?.(true, previousId)
}
$localModuleStates[detail.moduleId] = {
...$localModuleStates[detail.moduleId],
selectedForloop: detail.id,
selectedForloopIndex: detail.index,
selectedForLoopSetManually: true
}
await tick()
await globalRefreshes?.[detail.moduleId]?.(false, rootJobId)
} else {
$localModuleStates[detail.moduleId] = {
...$localModuleStates[detail.moduleId],
selectedForLoopSetManually: false
}
}
}
$effect(() => {
flowJobIds?.moduleId && untrack(() => onFlowModuleId())
})
@@ -939,10 +986,14 @@
$effect(() => {
flowJobIds?.moduleId && untrack(() => onModuleIdChange())
})
let selected = $derived(isListJob ? 'sequence' : 'graph')
let selected = $derived(isListJob ? 'sequence' : 'graph') as 'sequence' | 'graph' | 'logs'
let animateLogsTab = $state(false)
let noLogs = $derived(graphTabOpen && !isNodeSelected)
</script>
<JobLoader workspaceOverride={workspaceId} noCode noLogs bind:this={jobLoader} />
<JobLoader workspaceOverride={workspaceId} {noLogs} noCode bind:this={jobLoader} />
{#if notAnonynmous}
<Alert type="error" title="Required Auth">
As a non logged in user, you can only see jobs ran by anonymous users like you
@@ -999,7 +1050,6 @@
{isOwner}
{hideFlowResult}
{hideDownloadLogs}
{localDurationStatuses}
{innerModules}
{suspendStatus}
{hideJobId}
@@ -1011,6 +1061,12 @@
{#if innerModules.length > 0 && !isListJob}
<Tabs class="mx-auto {wideResults ? '' : 'max-w-7xl'}" bind:selected>
<Tab value="graph"><span class="font-semibold text-md">Graph</span></Tab>
<Tab
value="logs"
class={animateLogsTab
? 'animate-pulse animate-duration-1000 bg-surface-inverse text-primary-inverse'
: ''}><span class="font-semibold">Logs</span></Tab
>
<Tab value="sequence"><span class="font-semibold">Details</span></Tab>
</Tabs>
{:else}
@@ -1096,6 +1152,8 @@
innerJobLoaded(job, j, false, force)
}}
{onResultStreamUpdate}
graphTabOpen={selected == 'graph' && graphTabOpen}
isNodeSelected={forloop_selected == loopJobId}
/>
</div>
{/if}
@@ -1172,6 +1230,8 @@
{workspaceId}
jobId={failedRetry}
{onResultStreamUpdate}
graphTabOpen={selected == 'graph' && graphTabOpen}
isNodeSelected={retry_selected == failedRetry}
/>
</div>
{/each}
@@ -1205,6 +1265,8 @@
onJobsLoaded(mod, job, force)
}}
{onResultStreamUpdate}
graphTabOpen={selected == 'graph' && graphTabOpen}
isNodeSelected={false}
/>
{:else if mod.flow_jobs?.length == 0 && mod.job == '00000000-0000-0000-0000-000000000000'}
<div class="text-secondary">no subflow (empty loop?)</div>
@@ -1236,7 +1298,14 @@
let { job, force } = e.detail
onJobsLoaded(mod, job, force)
}}
loadExtraLogs={(logs) => {
setModuleState(mod.id ?? '', {
logs
})
}}
{onResultStreamUpdate}
graphTabOpen={selected == 'graph' && graphTabOpen}
isNodeSelected={$localModuleStates?.[selectedNode ?? '']?.job_id == mod.job}
/>
{/if}
{:else}
@@ -1252,6 +1321,15 @@
<div class="p-2 text-tertiary text-sm italic">Empty flow</div>
{/if}
</div>
<div class="{selected != 'logs' ? 'hidden' : ''} mx-auto h-[800px]">
<FlowLogViewerWrapper
{job}
{localModuleStates}
{workspaceId}
{render}
{onSelectedIteration}
/>
</div>
</div>
{#if render}
{#if job.raw_flow && !isListJob}
@@ -1305,33 +1383,7 @@
selectedNode = e.id
}
}}
onSelectedIteration={async (detail) => {
if (detail.manuallySet) {
let rootJobId = detail.id
await tick()
let previousId = $localModuleStates[detail.moduleId]?.selectedForloop
if (previousId) {
await globalRefreshes?.[detail.moduleId]?.(true, previousId)
}
$localModuleStates[detail.moduleId] = {
...$localModuleStates[detail.moduleId],
selectedForloop: detail.id,
selectedForloopIndex: detail.index,
selectedForLoopSetManually: true
}
await tick()
await globalRefreshes?.[detail.moduleId]?.(false, rootJobId)
} else {
$localModuleStates[detail.moduleId] = {
...$localModuleStates[detail.moduleId],
selectedForLoopSetManually: false
}
}
}}
{onSelectedIteration}
earlyStop={job.raw_flow?.skip_expr !== undefined}
cache={job.raw_flow?.cache_ttl !== undefined}
modules={job.raw_flow?.modules ?? []}
@@ -1386,7 +1438,6 @@
col
result={job['result']}
logs={job.logs ?? ''}
durationStates={localDurationStatuses}
downloadLogs={!hideDownloadLogs}
/>
{:else if selectedNode == 'start'}
@@ -1464,7 +1515,6 @@
result={node.result}
tag={node.tag}
logs={node.logs}
durationStates={localDurationStatuses}
downloadLogs={!hideDownloadLogs}
/>
{:else}
frontend/src/lib/components/JobLoader.svelte
+48 -6
View File
@@ -1,3 +1,9 @@
<script context="module" lang="ts">
import pLimit from 'p-limit'
const plimit = pLimit(5)
</script>
<script lang="ts">
import {
type Job,
@@ -24,6 +30,7 @@
started?: ({ id }: { id: string }) => void
running?: ({ id }: { id: string }) => void
resultStreamUpdate?: ({ id, result_stream }: { id: string; result_stream?: string }) => void
loadExtraLogs?: ({ id, logs }: { id: string; logs: string }) => void
}
interface Props {
@@ -83,6 +90,8 @@
let lastStartedAt: number = Date.now()
let currentId: string | undefined = $state(undefined)
let noPingTimeout: NodeJS.Timeout | undefined = undefined
let lastNoLogs = $state(noLogs)
let lastCompletedJobId = $state<string | undefined>(undefined)
$effect(() => {
let newIsLoading = currentId !== undefined
@@ -93,6 +102,32 @@
})
})
const noLogsChangeRestartEvent = 'SSE restart after no logs change'
$effect(() => {
if (noLogs != lastNoLogs) {
lastNoLogs = noLogs
if (!noLogs) {
currentEventSource?.onerror?.(new Event(noLogsChangeRestartEvent))
const lastJobId = lastCompletedJobId
if (lastJobId && (job || lastCallbacks?.loadExtraLogs)) {
plimit(() =>
JobService.getCompletedJobLogsTail({
workspace: $workspaceStore!,
id: lastJobId
})
).then((res) => {
if (res && job) {
job.logs = res
}
if (res && lastCallbacks?.loadExtraLogs) {
lastCallbacks.loadExtraLogs({ id: lastJobId, logs: res })
}
})
}
}
}
})
function clearCurrentId() {
if (currentId) {
if (allowConcurentRequests) {
@@ -106,6 +141,7 @@
export async function abstractRun(fn: () => Promise<string>, callbacks?: Callbacks) {
try {
isLoading = true
lastCompletedJobId = undefined
clearCurrentJob()
lastCallbacks = callbacks
noPingTimeout = undefined
@@ -251,7 +287,6 @@
const id = currentId
if (id) {
lastCallbacks?.cancel?.({ id })
lastCallbacks = undefined
clearCurrentId()
// Clean up SSE connection
currentEventSource?.close()
@@ -272,7 +307,6 @@
if (currentId && !allowConcurentRequests) {
job = undefined
lastCallbacks?.cancel?.({ id: currentId })
lastCallbacks = undefined
await cancelJob()
}
}
@@ -298,7 +332,7 @@
// Clean up any existing SSE connection
currentEventSource?.close()
currentEventSource = undefined
lastCallbacks = callbacks
// Try SSE first, fall back to polling if needed
if (supportsSSE()) {
await loadTestJobWithSSE(testId, 0, callbacks)
@@ -497,6 +531,7 @@
callbacks?.change?.(job)
clearCurrentId()
lastCompletedJobId = id
}
}
@@ -673,9 +708,16 @@
console.warn('SSE error:', error)
currentEventSource?.close()
currentEventSource = undefined
if (attempt < 3) {
console.log(`SSE error (1), retrying ... attempt: ${attempt + 1}/3`)
setTimeout(() => loadTestJobWithSSE(id, attempt + 1, callbacks), 1000)
let delay = 1000
let isNoLogsChange = error.type == noLogsChangeRestartEvent
if (isNoLogsChange) {
delay = 0
}
if (attempt < 3 || isNoLogsChange) {
if (!isNoLogsChange) {
console.log(`SSE error (1), retrying ... attempt: ${attempt + 1}/3`)
}
setTimeout(() => loadTestJobWithSSE(id, attempt + 1, callbacks), delay)
} else {
// Fall back to polling on error
setTimeout(() => syncer(id, callbacks), 1000)
frontend/src/lib/components/LogViewer.svelte
+10 -2
View File
@@ -16,6 +16,7 @@
import NoWorkerWithTagWarning from './runs/NoWorkerWithTagWarning.svelte'
import { JobService } from '$lib/gen'
import Tooltip from './Tooltip.svelte'
import { twMerge } from 'tailwind-merge'
interface Props {
content: string | undefined
@@ -32,6 +33,7 @@
download?: boolean
customEmptyMessage?: string
tagLabel?: string
noPadding?: boolean
}
let {
@@ -48,7 +50,8 @@
noAutoScroll = false,
download = true,
customEmptyMessage = 'No logs are available yet',
tagLabel = undefined
tagLabel = undefined,
noPadding = false
}: Props = $props()
// @ts-ignore
@@ -271,7 +274,12 @@
: 'top-2'} left-36">mem peak: {(mem / 1024).toPrecision(4)}MB</span
>
{/if}
<pre class="whitespace-pre break-words {small ? '!text-2xs' : '!text-xs'} w-full p-2"
<pre
class={twMerge(
'whitespace-pre break-words w-full',
small ? '!text-2xs' : '!text-xs',
noPadding ? '' : 'p-2'
)}
>{#if content}{@const len =
(content?.length ?? 0) +
(loadedFromObjectStore?.length ?? 0)}{#if downloadStartUrl}<button onclick={getStoreLogs}
frontend/src/lib/components/flows/FlowEditor.svelte
+1 -4
View File
@@ -14,7 +14,7 @@
import type { Trigger } from '$lib/components/triggers/utils'
import FlowAIChat from '../copilot/chat/flow/FlowAIChat.svelte'
import { aiChatManager, AIMode } from '../copilot/chat/AIChatManager.svelte'
import type { DurationStatus, GraphModuleState } from '../graph'
import type { GraphModuleState } from '../graph'
import { triggerableByAI } from '$lib/actions/triggerableByAI.svelte'
const { flowStore } = getContext<FlowEditorContext>('FlowEditorContext')
@@ -49,7 +49,6 @@
onHideJobStatus?: () => void
individualStepTests?: boolean
job?: Job
localDurationStatuses?: Writable<Record<string, DurationStatus>>
suspendStatus?: Writable<Record<string, { job: Job; nb: number }>>
showJobStatus?: boolean
onDelete?: (id: string) => void
@@ -83,7 +82,6 @@
onHideJobStatus,
individualStepTests = false,
job,
localDurationStatuses,
suspendStatus,
showJobStatus,
onDelete,
@@ -186,7 +184,6 @@
{onTestFlow}
{job}
{isOwner}
{localDurationStatuses}
{suspendStatus}
onOpenDetails={onOpenPreview}
/>
frontend/src/lib/components/flows/content/FlowEditorPanel.svelte
+1 -4
View File
@@ -15,7 +15,6 @@
import { computeMissingInputWarnings } from '../missingInputWarnings'
import FlowResult from './FlowResult.svelte'
import type { Writable } from 'svelte/store'
import type { DurationStatus } from '$lib/components/graph'
interface Props {
noEditor?: boolean
@@ -33,7 +32,6 @@
onTestFlow?: () => void
job?: Job
isOwner?: boolean
localDurationStatuses?: Writable<Record<string, DurationStatus>>
suspendStatus?: Writable<Record<string, { job: Job; nb: number }>>
onOpenDetails?: () => void
}
@@ -50,7 +48,6 @@
onTestFlow,
job,
isOwner,
localDurationStatuses,
suspendStatus,
onOpenDetails
}: Props = $props()
@@ -100,7 +97,7 @@
{onTestFlow}
/>
{:else if $selectedId === 'Result'}
<FlowResult {noEditor} {job} {isOwner} {localDurationStatuses} {suspendStatus} {onOpenDetails} />
<FlowResult {noEditor} {job} {isOwner} {suspendStatus} {onOpenDetails} />
{:else if $selectedId === 'constants'}
<FlowConstants {noEditor} />
{:else if $selectedId === 'failure'}
frontend/src/lib/components/flows/content/FlowResult.svelte
+2 -6
View File
@@ -1,6 +1,5 @@
<script lang="ts">
import FlowPreviewResult from '$lib/components/FlowPreviewResult.svelte'
import type { DurationStatus } from '$lib/components/graph'
import type { Job } from '$lib/gen'
import { workspaceStore } from '$lib/stores'
import type { Writable } from 'svelte/store'
@@ -10,18 +9,16 @@
interface Props {
job?: Job
isOwner?: boolean
localDurationStatuses?: Writable<Record<string, DurationStatus>>
suspendStatus?: Writable<Record<string, { job: Job; nb: number }>>
noEditor: boolean
onOpenDetails?: () => void
}
let { job, isOwner, localDurationStatuses, suspendStatus, noEditor, onOpenDetails }: Props =
$props()
let { job, isOwner, suspendStatus, noEditor, onOpenDetails }: Props = $props()
</script>
<FlowCard {noEditor} title="Flow result">
{#if job && isOwner !== undefined && localDurationStatuses && suspendStatus}
{#if job && isOwner !== undefined && suspendStatus}
<div class="px-4 py-2">
<FlowPreviewResult
{job}
@@ -29,7 +26,6 @@
{isOwner}
hideFlowResult={false}
hideDownloadLogs={false}
{localDurationStatuses}
innerModules={[]}
{suspendStatus}
{extra}
frontend/src/lib/components/flows/map/FlowJobsMenu.svelte
+6 -2
View File
@@ -15,6 +15,7 @@
selected: number
selectedManually: boolean | undefined
onSelectedIteration: onSelectedIteration
showIcon?: boolean
}
let {
@@ -24,7 +25,8 @@
selected,
selectedManually,
onSelectedIteration,
moduleId
moduleId,
showIcon = true
}: Props = $props()
let filter: number | undefined = $state(undefined)
@@ -127,7 +129,9 @@
meltElement={trigger}
>
#{selected == -1 ? '?' : selected + 1}
<ListFilter size={15} />
{#if showIcon}
<ListFilter size={15} />
{/if}
</MeltButton>
{/snippet}
frontend/src/lib/components/graph/FlowGraphV2.svelte
+4 -11
View File
@@ -48,11 +48,11 @@
import type { TriggerContext } from '../triggers'
import { workspaceStore } from '$lib/stores'
import SubflowBound from './renderers/nodes/SubflowBound.svelte'
import { deepEqual } from 'fast-equals'
import ViewportResizer from './ViewportResizer.svelte'
import AssetNode, { computeAssetNodes } from './renderers/nodes/AssetNode.svelte'
import AssetsOverflowedNode from './renderers/nodes/AssetsOverflowedNode.svelte'
import type { FlowGraphAssetContext } from '../flows/types'
import { ChangeTracker } from '$lib/svelte5Utils.svelte'
let useDataflow: Writable<boolean | undefined> = writable<boolean | undefined>(false)
@@ -348,14 +348,7 @@
}
}
let lastModules = $state.snapshot(modules)
let moduleCounter = $state(0)
function onModulesChange2(modules) {
if (!deepEqual(modules, lastModules)) {
lastModules = $state.snapshot(modules)
moduleCounter++
}
}
let moduleTracker = new ChangeTracker($state.snapshot(modules))
let nodes = $state.raw<Node[]>([])
let edges = $state.raw<Edge[]>([])
@@ -426,10 +419,10 @@
})
$effect(() => {
readFieldsRecursively(modules)
untrack(() => onModulesChange2(modules))
untrack(() => moduleTracker.track($state.snapshot(modules)))
})
let graph = $derived.by(() => {
moduleCounter
moduleTracker.counter
return graphBuilder(
untrack(() => modules),
{
frontend/src/lib/components/preview/FlowPreviewStatus.svelte
+3 -1
View File
@@ -26,7 +26,9 @@
{#if job && !hideJobId}
<div>
<div class="text-primary whitespace-nowrap truncate text-sm">
<span class="font-semibold mr-1">Flow:</span>
{#if ['flow', 'flowpreview', 'flownode'].includes(job.job_kind)}
<span class="font-semibold mr-1">Flow:</span>
{/if}
<a
rel="noreferrer"
target="_blank"
frontend/src/lib/components/runs/JobPreview.svelte → frontend/src/lib/components/runs/JobRunsPreview.svelte
+2 -4
View File
@@ -72,9 +72,7 @@
})
)
})
$effect(() => {
job?.logs == undefined && job && viewTab == 'logs' && untrack(() => jobLoader?.getLogs())
})
$effect(() => {
job?.id && lastJobId !== job.id && untrack(() => job && getConcurrencyKey(job))
})
@@ -82,7 +80,7 @@
let jobLoader: JobLoader | undefined = $state(undefined)
</script>
<JobLoader noLogs workspaceOverride={workspace} bind:job={currentJob} bind:this={jobLoader} />
<JobLoader workspaceOverride={workspace} bind:job={currentJob} bind:this={jobLoader} />
<div class="p-4 flex flex-col gap-2 items-start h-full">
{#if job}
frontend/src/lib/components/search/RunsSearch.svelte
+1 -1
View File
@@ -9,7 +9,7 @@
import QuickMenuItem from './QuickMenuItem.svelte'
import { goto } from '$app/navigation'
import { displayDateOnly } from '$lib/utils'
import JobPreview from '../runs/JobPreview.svelte'
import JobPreview from '../runs/JobRunsPreview.svelte'
let debounceTimeout: any = undefined
const debouncePeriod: number = 1000
frontend/src/lib/svelte5Utils.svelte.ts
+29
View File
@@ -1,6 +1,7 @@
// https://github.com/sveltejs/svelte/issues/14600
import { untrack } from 'svelte'
import { deepEqual } from 'fast-equals'
import type { StateStore } from './utils'
export function withProps<Component, Props>(component: Component, props: Props) {
@@ -69,3 +70,31 @@ export function usePromise<T>(
return ret
}
/**
* Generic change tracker class that monitors changes in state using deep equality comparison
* and provides a counter to trigger Svelte 5 reactivity. Similar to the pattern used in
* FlowGraphV2.svelte's onModulesChange2 function.
*/
export class ChangeTracker<T> {
counter = $state(0)
#lastState: T | undefined
constructor(initialValue?: T) {
this.#lastState = initialValue ? initialValue : undefined
}
/**
* Check if the value has changed and update the counter to trigger reactivity
* @param value - The current value to check for changes
* @returns true if the value changed, false otherwise
*/
track(value: T): boolean {
if (!deepEqual(value, this.#lastState)) {
this.#lastState = value
this.counter++
return true
}
return false
}
}
frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte
+3 -3
View File
@@ -16,7 +16,7 @@
import { Button, Drawer, DrawerContent, Skeleton } from '$lib/components/common'
import RunChart from '$lib/components/RunChart.svelte'
import JobPreview from '$lib/components/runs/JobPreview.svelte'
import JobRunsPreview from '$lib/components/runs/JobRunsPreview.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import CalendarPicker from '$lib/components/common/calendarPicker/CalendarPicker.svelte'
@@ -834,7 +834,7 @@
{#if selectedIds[0] === '-'}
<div class="p-4">There is no information available for this job</div>
{:else}
<JobPreview blankLink id={selectedIds[0]} workspace={selectedWorkspace} />
<JobRunsPreview blankLink id={selectedIds[0]} workspace={selectedWorkspace} />
{/if}
{/if}
</DrawerContent>
@@ -1179,7 +1179,7 @@
{#if selectedIds[0] === '-'}
<div class="p-4">There is no information available for this job</div>
{:else}
<JobPreview
<JobRunsPreview
on:filterByConcurrencyKey={filterByConcurrencyKey}
on:filterByWorker={filterByWorker}
id={selectedIds[0]}