{
expect(node.classList.contains('active')).toBe(false)
}
})
+
+it('keeps a long detail inside its box, showing the tail and the whole on hover', () => {
+ const composed = {
+ steps: [
+ {
+ name: 'shot1_inventory',
+ workflow: { path: 'templates/minimax/composable-reference-shot.json' },
+ },
+ ],
+ }
+ const { container } = render(FlowView, { workflow: composed })
+ const node = nodeFor(container, 'shot1_inventory')
+ const detail = node.querySelector('.stepdetail')!
+ // A path is told apart by its end, so that is the part kept
+ expect(detail.textContent).toMatch(/^…/)
+ expect(detail.textContent).toMatch(/reference-shot\.json$/)
+ expect(detail.textContent!.length).toBeLessThan(
+ 'templates/minimax/composable-reference-shot.json'.length,
+ )
+ expect(node.querySelector('title')?.textContent).toBe(
+ 'templates/minimax/composable-reference-shot.json',
+ )
+ // The box clips whatever a wider glyph set still pushes past its edge
+ expect(node.getAttribute('clip-path')).toBe('url(#flow-nodebox)')
+})
+
+it('shortens a long step name from the end, leaving room for the entry tag', () => {
+ const long = {
+ steps: [
+ {
+ name: 'reference_to_video_audio_with_lipsync',
+ pipeline: { configuration: { component_type: 'Fake' } },
+ },
+ ],
+ }
+ const { container } = render(FlowView, { workflow: long })
+ const node = nodeFor(container, 'reference_to_video_audio_with_lipsync')
+ const name = node.querySelector('.stepname')!
+ expect(name.textContent).toMatch(/…$/)
+ expect(name.textContent!.length).toBeLessThan(
+ 'reference_to_video_audio_with_lipsync'.length,
+ )
+ // A name that fits gets no tooltip at all
+ expect(
+ nodeFor(render(FlowView, { workflow }).container, 'gen').querySelector(
+ 'title',
+ ),
+ ).toBeNull()
+ expect(node.querySelector('title')?.textContent).toBe(
+ 'reference_to_video_audio_with_lipsync',
+ )
+})
diff --git a/ui/src/lib/pages/JobPage.svelte b/ui/src/lib/pages/JobPage.svelte
index af543af6..ad91dc09 100644
--- a/ui/src/lib/pages/JobPage.svelte
+++ b/ui/src/lib/pages/JobPage.svelte
@@ -10,7 +10,12 @@
import { ApiError, api, outputUrl, streamJobEvents } from '../api'
import { confirmDialog } from '../confirm.svelte'
import { go } from '../router.svelte'
- import { groupResultFiles, sectionBySubfolder } from '../results'
+ import {
+ groupResultFiles,
+ sectionBySubfolder,
+ unsavedSteps,
+ } from '../results'
+ import { finishedNodes, flowNodeName } from '../runstate'
import { stepProgress } from '../progress'
import FlowView from '../editor/FlowView.svelte'
import CopyButton from '../CopyButton.svelte'
@@ -157,15 +162,27 @@
const steps = $derived(
(events.find((e) => e.event === 'workflow_start')?.steps as string[]) ?? [],
)
- const currentStep = $derived(
- [...events].reverse().find((e) => e.event === 'step_start')?.step as
- string | undefined,
+ // The step_start the run is on, as the engine named it: a for_each member
+ // keeps its '@', a sub-workflow's inner step its own name. The Progress
+ // list is written in these names, so it reads off this one.
+ const stepStart = $derived(
+ [...events].reverse().find((e) => e.event === 'step_start'),
)
- const finishedSteps = $derived(
- new Set(
- events.filter((e) => e.event === 'step_end').map((e) => e.step as string),
+ const currentStep = $derived(stepStart?.step as string | undefined)
+ // The same step in the name the definition gives it, which is what the
+ // flow graph's nodes are called - see runstate.ts for why the two differ
+ const activeNode = $derived(
+ flowNodeName(
+ stepStart?.step as string | undefined,
+ stepStart?.parent_step as string | undefined,
),
)
+ // A sub-workflow's inner step has no row of its own in the Progress list,
+ // so what lights up while one runs is the composed step that queued it
+ const listStep = $derived(
+ steps.includes(currentStep ?? '') ? currentStep : activeNode,
+ )
+ const finishedSteps = $derived(finishedNodes(events as JobEvent[], steps))
// Scoped to the step now running: its phase, and its own denoise counter
const progress = $derived(stepProgress(events as JobEvent[]))
const denoise = $derived(progress.denoise)
@@ -246,6 +263,13 @@
const sectioned = $derived(
sections.length > 1 || sections[0].subfolder !== '',
)
+ // Steps that ran and wrote no file, with the definition's reason for each.
+ // Without this a workflow that keeps most of its steps in memory shows a
+ // short Results list and no word about the rest, which reads as outputs
+ // that went missing rather than as a choice the workflow made
+ const unsaved = $derived(
+ unsavedSteps(job?.manifest, events as JobEvent[], definition),
+ )
const running = $derived(job !== null && !TERMINAL.includes(job.status))
// A cancel requested while loading a model or running a task step has no
// checkpoint to catch it until that phase finishes - without this the UI
@@ -352,13 +376,13 @@
-
{step}
- {#if step === currentStep && running}
+ {#if step === listStep && running}
{#if denoise}
Workflow
{/if}
- {#if fileGroups.length}
+ {#if fileGroups.length || unsaved.length}
Results
{#if allReused}
@@ -491,6 +515,27 @@
{/each}
{/each}
{/each}
+ {#if unsaved.length}
+
Steps that wrote nothing
+
+ {#each unsaved as entry (entry.node)}
+ -
+
{entry.node}
+ {#if entry.members.length}
+ × {entry.members.length}
+ {/if}
+
+ {#if entry.reason}{entry.reason.key}
+ {entry.reason.detail}{:else}no file written{/if}
+
+
+ {/each}
+
+ {/if}
{/if}
@@ -682,4 +727,25 @@
.subhead + .stephead {
margin-top: 0;
}
+ /* The steps whose files are deliberately absent: the step name is what
+ the engine resolves, the reason is written for a person */
+ .unsaved {
+ list-style: none;
+ margin: 0;
+ padding: 0;
+ font-size: var(--t-sm);
+ }
+ .unsaved li {
+ display: flex;
+ align-items: baseline;
+ flex-wrap: wrap;
+ gap: 0.5rem;
+ padding: 0.15rem 0;
+ }
+ .unsaved code {
+ font-size: 0.8rem;
+ }
+ .unsaved .why {
+ min-width: 0;
+ }
diff --git a/ui/src/lib/pages/JobPage.test.ts b/ui/src/lib/pages/JobPage.test.ts
index 84a20d16..bf4282b3 100644
--- a/ui/src/lib/pages/JobPage.test.ts
+++ b/ui/src/lib/pages/JobPage.test.ts
@@ -13,13 +13,17 @@ const stream = vi.hoisted(() => ({
const metadata = vi.hoisted(() => ({
byFile: {} as Record
>,
}))
+// The definition the job ran, for the flow view and the unsaved reasons
+const ran = vi.hoisted(() => ({
+ definition: null as Record | null,
+}))
vi.mock('../api', () => ({
ApiError: class ApiError extends Error {},
api: {
getJob: vi.fn(() => Promise.resolve(detail.job)),
getJobWorkflow: vi.fn(() =>
- Promise.resolve({ definition: null, seed_variable: null }),
+ Promise.resolve({ definition: ran.definition, seed_variable: null }),
),
galleryMetadata: vi.fn((name: string) =>
Promise.resolve({
@@ -62,9 +66,24 @@ afterEach(() => {
cleanup()
stream.onEvent = null
metadata.byFile = {}
+ ran.definition = null
vi.mocked(api.galleryMetadata).mockClear()
})
+/** The flow view's box for one step of the definition. */
+function nodeFor(container: HTMLElement, name: string) {
+ return [...container.querySelectorAll('g.node')].find((node) =>
+ node.getAttribute('aria-label')?.startsWith(`step ${name},`),
+ )!
+}
+
+/** The page's "wrote nothing" rows, whitespace flattened. */
+function unsavedRows(container: HTMLElement) {
+ return [...container.querySelectorAll('.unsaved li')].map((li) =>
+ li.textContent?.replace(/\s+/g, ' ').trim(),
+ )
+}
+
it('shows each image beside what made it, with a download link, and never probes a video', async () => {
metadata.byFile['a.png'] = {
model_name: 'org/model',
@@ -149,3 +168,120 @@ it('places a live step_end under its subfolder before the manifest arrives', asy
).toBeTruthy(),
)
})
+
+it('lights the for_each step in the flow chart while one of its members runs', async () => {
+ // The graph is drawn from the definition, where for_each is one step; the
+ // engine reports the members, so the two only meet at the group name
+ ran.definition = {
+ steps: [
+ {
+ name: 'shot',
+ pipeline: { configuration: { component_type: 'Fake' } },
+ for_each: 'variable:shots',
+ },
+ { name: 'episode', task: { command: 'mux' } },
+ ],
+ }
+ detail.job = { ...job([]), status: 'running', finished_at: null }
+ const { container } = render(JobPage, { jobId: 'j1' })
+ await waitFor(() => expect(stream.onEvent).not.toBeNull())
+ stream.onEvent!({
+ seq: 1,
+ event: 'workflow_start',
+ steps: ['shot@open', 'shot@reveal', 'episode'],
+ })
+ stream.onEvent!({ seq: 2, event: 'step_start', step: 'shot@open' })
+ await waitFor(() =>
+ expect(nodeFor(container, 'shot').classList.contains('active')).toBe(true),
+ )
+
+ // One member of two down: the group is not behind us yet
+ stream.onEvent!({ seq: 3, event: 'step_end', step: 'shot@open', files: [] })
+ await waitFor(() =>
+ expect(nodeFor(container, 'shot').classList.contains('active')).toBe(true),
+ )
+ expect(nodeFor(container, 'shot').classList.contains('done')).toBe(false)
+
+ stream.onEvent!({ seq: 4, event: 'step_end', step: 'shot@reveal', files: [] })
+ stream.onEvent!({ seq: 5, event: 'step_start', step: 'episode' })
+ await waitFor(() =>
+ expect(nodeFor(container, 'shot').classList.contains('done')).toBe(true),
+ )
+ expect(nodeFor(container, 'episode').classList.contains('active')).toBe(true)
+})
+
+it('keeps the Progress list on the composed step while its child runs', async () => {
+ ran.definition = {
+ steps: [
+ { name: 'shot1', workflow: { path: 'child.json' } },
+ { name: 'episode', task: { command: 'mux' } },
+ ],
+ }
+ detail.job = { ...job([]), status: 'running', finished_at: null }
+ const { container } = render(JobPage, { jobId: 'j1' })
+ await waitFor(() => expect(stream.onEvent).not.toBeNull())
+ stream.onEvent!({
+ seq: 1,
+ event: 'workflow_start',
+ steps: ['shot1', 'episode'],
+ })
+ // What a child emits: its own step name, with the parent's alongside
+ stream.onEvent!({
+ seq: 2,
+ event: 'step_start',
+ step: 'reference_to_video_audio',
+ parent_step: 'shot1',
+ })
+ await waitFor(() =>
+ expect(nodeFor(container, 'shot1').classList.contains('active')).toBe(true),
+ )
+ const dotFor = (name: string) =>
+ [...container.querySelectorAll('.step')]
+ .find(
+ (row) =>
+ row.querySelector('span:nth-child(2)')?.textContent?.trim() === name,
+ )!
+ .querySelector('.dot')!
+ expect(dotFor('shot1').classList.contains('active')).toBe(true)
+ expect(dotFor('episode').classList.contains('active')).toBe(false)
+})
+
+it('says why a step wrote nothing, so a deliberate non-output is not a missing one', async () => {
+ ran.definition = {
+ steps: [
+ { name: 'base', pipeline: {}, result: { save: false } },
+ { name: 'edit', task: { command: 'mux' } },
+ { name: 'film', pipeline: {}, result: { content_type: 'video/mp4' } },
+ ],
+ }
+ detail.job = job([
+ { step: 'base', files: [] },
+ { step: 'edit', files: [] },
+ { step: 'film', files: ['final/film.mp4'], subfolder: 'final' },
+ ])
+ const { container } = render(JobPage, { jobId: 'j1' })
+ await waitFor(() =>
+ expect(screen.getByText('Steps that wrote nothing')).toBeTruthy(),
+ )
+ expect(unsavedRows(container)).toEqual([
+ 'base result.save is false, so the step is kept in memory and never written',
+ 'edit result.content_type is not declared, so there is no file type to write',
+ ])
+ // The step that did write is in the results above, not in this list
+ expect(screen.getByRole('heading', { level: 3, name: 'final/' })).toBeTruthy()
+})
+
+it('explains a run that wrote nothing at all rather than showing no results', async () => {
+ ran.definition = { steps: [{ name: 'base', result: { save: false } }] }
+ detail.job = job([{ step: 'base', files: [] }])
+ const { container } = render(JobPage, { jobId: 'j1' })
+ await waitFor(() =>
+ expect(screen.getByText('Steps that wrote nothing')).toBeTruthy(),
+ )
+ expect(
+ screen.getByRole('heading', { level: 2, name: 'Results' }),
+ ).toBeTruthy()
+ expect(unsavedRows(container)).toEqual([
+ 'base result.save is false, so the step is kept in memory and never written',
+ ])
+})
diff --git a/ui/src/lib/results.test.ts b/ui/src/lib/results.test.ts
index 681eddd1..4681dbe4 100644
--- a/ui/src/lib/results.test.ts
+++ b/ui/src/lib/results.test.ts
@@ -1,5 +1,10 @@
import { describe, expect, it } from 'vitest'
-import { groupResultFiles, sectionBySubfolder } from './results'
+import {
+ groupResultFiles,
+ sectionBySubfolder,
+ unsavedReason,
+ unsavedSteps,
+} from './results'
import type { JobEvent } from './types'
const stepEnd = (
@@ -160,3 +165,116 @@ describe('sectionBySubfolder', () => {
expect(sections.map((s) => s.subfolder)).toEqual(['final', 'shots/act-1'])
})
})
+
+describe('unsavedReason', () => {
+ it('names result.save when the workflow asked for no file', () => {
+ expect(
+ unsavedReason({ result: { save: false, content_type: 'image/png' } }),
+ ).toEqual({
+ key: 'result.save',
+ detail: 'is false, so the step is kept in memory and never written',
+ })
+ })
+
+ it('names result.content_type when there is no type to write', () => {
+ expect(unsavedReason({ task: { command: 'mux' } })).toEqual({
+ key: 'result.content_type',
+ detail: 'is not declared, so there is no file type to write',
+ })
+ })
+
+ it('has no reason for a step that declares a file', () => {
+ expect(
+ unsavedReason({ result: { content_type: 'video/mp4', save: true } }),
+ ).toBeNull()
+ })
+})
+
+describe('unsavedSteps', () => {
+ const definition = (steps: Array>) => ({ steps })
+ const steps = definition([
+ { name: 'base', result: { save: false } },
+ { name: 'edit', task: { command: 'mux' } },
+ { name: 'film', result: { content_type: 'video/mp4' } },
+ ])
+
+ it('names every step that ran and wrote nothing, with the reason', () => {
+ const unsaved = unsavedSteps(
+ [
+ { step: 'base', files: [] },
+ { step: 'edit', files: [] },
+ { step: 'film', files: ['final/film.mp4'], subfolder: 'final' },
+ ],
+ [],
+ steps,
+ )
+ expect(unsaved).toEqual([
+ {
+ node: 'base',
+ members: [],
+ reason: {
+ key: 'result.save',
+ detail: 'is false, so the step is kept in memory and never written',
+ },
+ },
+ {
+ node: 'edit',
+ members: [],
+ reason: {
+ key: 'result.content_type',
+ detail: 'is not declared, so there is no file type to write',
+ },
+ },
+ ])
+ })
+
+ it('collapses a for_each group into the one step the workflow declares', () => {
+ const unsaved = unsavedSteps(
+ [
+ { step: 'base@open', files: [] },
+ { step: 'base@reveal', files: [] },
+ ],
+ [],
+ steps,
+ )
+ expect(unsaved).toEqual([
+ {
+ node: 'base',
+ members: ['base@open', 'base@reveal'],
+ reason: {
+ key: 'result.save',
+ detail: 'is false, so the step is kept in memory and never written',
+ },
+ },
+ ])
+ })
+
+ it('lists only steps that have finished, so a running job is not accused', () => {
+ const events = [
+ { seq: 1, event: 'workflow_start', steps: ['base', 'edit', 'film'] },
+ { seq: 2, event: 'step_end', step: 'base', files: [] },
+ ] as unknown as JobEvent[]
+ expect(unsavedSteps(undefined, events, steps).map((s) => s.node)).toEqual([
+ 'base',
+ ])
+ })
+
+ it("ignores a sub-workflow's inner steps, which are no node of this graph", () => {
+ const composed = definition([
+ { name: 'shot1_amnesty', workflow: { path: 'child.json' } },
+ ])
+ const unsaved = unsavedSteps(
+ [
+ { step: 'reference_to_video_audio', files: [] },
+ { step: 'shot1_amnesty', files: ['intermediate/a.mp4'] },
+ ],
+ [],
+ composed,
+ )
+ expect(unsaved).toEqual([])
+ })
+
+ it('has nothing to say without a definition to read the reasons off', () => {
+ expect(unsavedSteps([{ step: 'base', files: [] }], [], null)).toEqual([])
+ })
+})
diff --git a/ui/src/lib/results.ts b/ui/src/lib/results.ts
index c84e946c..0f3db546 100644
--- a/ui/src/lib/results.ts
+++ b/ui/src/lib/results.ts
@@ -1,3 +1,4 @@
+import { flowNodeName } from './runstate'
import type { JobEvent, ManifestEntry, StepEndEvent } from './types'
/** One step's output files, with where in the run directory they landed. */
@@ -86,3 +87,103 @@ export function sectionBySubfolder(groups: StepGroup[]): SubfolderSection[] {
groups: bySubfolder.get(subfolder)!,
}))
}
+
+/** One step of a run that finished without writing a file. */
+export interface UnsavedStep {
+ /** The step's name in the definition. One entry covers a whole for_each
+ * group, since that is the one step the workflow declares. */
+ node: string
+ /** The member names the engine ran it under, one per for_each entry -
+ * empty for a step that has no `for_each`, which is its own name. */
+ members: string[]
+ /** Why nothing was written, read off the definition - null when the
+ * definition does not explain it. */
+ reason: UnsavedReason | null
+}
+
+/** The JSON key a step writes nothing because of, and what that means -
+ * kept apart so the page can set the key in the type the engine resolves
+ * literally. */
+export interface UnsavedReason {
+ key: string
+ detail: string
+}
+
+/** Why a step's result block saves no file: the two ways the engine skips
+ * the save, both in `Result.save` (dw/result.py). */
+export function unsavedReason(step: Record): UnsavedReason | null {
+ const result = step.result
+ if (result && result.save === false) {
+ return {
+ key: 'result.save',
+ detail: 'is false, so the step is kept in memory and never written',
+ }
+ }
+ if (!result || typeof result.content_type !== 'string') {
+ return {
+ key: 'result.content_type',
+ detail: 'is not declared, so there is no file type to write',
+ }
+ }
+ return null
+}
+
+/** The steps that ran and wrote nothing, with the reason the definition
+ * gives for each.
+ *
+ * The page groups a run's files by producing step and drops the empty
+ * groups, which made a workflow whose steps deliberately write nothing
+ * indistinguishable from a run whose outputs had gone missing: the steps
+ * were there, their files were not, and nothing said why. A step counts as
+ * having run when the manifest carries its entry - every step that finished
+ * gets one - or a top-level step_end named it, so a job still in flight
+ * never lists a step it has not reached yet, and a historical job, which
+ * has no events at all, reads correctly off the manifest alone. */
+export function unsavedSteps(
+ manifest: ManifestEntry[] | undefined,
+ events: JobEvent[],
+ definition: Record | null,
+): UnsavedStep[] {
+ const byNode = new Map>()
+ for (const step of (definition?.steps ?? []) as Array>) {
+ if (typeof step?.name === 'string') byNode.set(step.name, step)
+ }
+ if (!byNode.size) return []
+
+ // An inner step of a sub-workflow is no node of the parent's graph, so
+ // the entry it rolls up into the manifest is no evidence about anything
+ // here - only a definition step, or a member of one, names a node
+ const nodeOf = (step: string): string | undefined => {
+ const node = flowNodeName(step)
+ return node && byNode.has(node) ? node : undefined
+ }
+
+ const wrote = new Set()
+ for (const group of groupResultFiles(manifest, events)) {
+ const node = nodeOf(group.step)
+ if (node) wrote.add(node)
+ }
+
+ const ran = new Map()
+ const note = (step: string) => {
+ const node = nodeOf(step)
+ if (!node) return
+ const members = ran.get(node) ?? []
+ if (!members.includes(step)) members.push(step)
+ ran.set(node, members)
+ }
+ for (const entry of manifest ?? []) note(entry.step)
+ for (const event of events) {
+ if (event.event === 'step_end' && !event.parent_step && event.step) {
+ note(event.step as string)
+ }
+ }
+
+ return [...ran]
+ .filter(([node]) => !wrote.has(node))
+ .map(([node, members]) => ({
+ node,
+ members: members.filter((member) => member !== node),
+ reason: unsavedReason(byNode.get(node)!),
+ }))
+}
diff --git a/ui/src/lib/runstate.test.ts b/ui/src/lib/runstate.test.ts
new file mode 100644
index 00000000..43990901
--- /dev/null
+++ b/ui/src/lib/runstate.test.ts
@@ -0,0 +1,54 @@
+import { describe, expect, it } from 'vitest'
+import { finishedNodes, flowNodeName } from './runstate'
+import type { JobEvent } from './types'
+
+const ended = (step: string, parent_step?: string): JobEvent =>
+ ({ seq: 0, event: 'step_end', step, parent_step }) as JobEvent
+
+describe('flowNodeName', () => {
+ it('reduces a for_each member to the group the definition declares', () => {
+ expect(flowNodeName('base@open')).toBe('base')
+ })
+
+ it("reduces a sub-workflow's inner step to the composed step that queued it", () => {
+ expect(flowNodeName('reference_to_video_audio', 'shot1_amnesty')).toBe(
+ 'shot1_amnesty',
+ )
+ })
+
+ it('takes the parent even when the child step is a for_each group itself', () => {
+ expect(flowNodeName('base@open', 'shot@reveal')).toBe('shot')
+ })
+
+ it('leaves an ordinary step alone, and has nothing for a missing name', () => {
+ expect(flowNodeName('film')).toBe('film')
+ expect(flowNodeName(undefined)).toBeUndefined()
+ })
+})
+
+describe('finishedNodes', () => {
+ const members = ['base@open', 'base@reveal', 'film']
+
+ it('finishes a for_each group only once every member has ended', () => {
+ expect(finishedNodes([ended('base@open')], members)).toEqual([])
+ expect(
+ finishedNodes([ended('base@open'), ended('base@reveal')], members),
+ ).toEqual(['base'])
+ })
+
+ it('finishes a plain step on its own end', () => {
+ expect(finishedNodes([ended('film')], members)).toEqual(['film'])
+ })
+
+ it("does not finish a composed step for one of its child's inner steps", () => {
+ const names = ['shot1_amnesty', 'episode']
+ const events = [ended('inner_a', 'shot1_amnesty')]
+ expect(finishedNodes(events, names)).toEqual([])
+ events.push(ended('shot1_amnesty'))
+ expect(finishedNodes(events, names)).toEqual(['shot1_amnesty'])
+ })
+
+ it('has nothing finished before the run has reported anything', () => {
+ expect(finishedNodes([], members)).toEqual([])
+ })
+})
diff --git a/ui/src/lib/runstate.ts b/ui/src/lib/runstate.ts
new file mode 100644
index 00000000..8bd0fc9c
--- /dev/null
+++ b/ui/src/lib/runstate.ts
@@ -0,0 +1,61 @@
+/** Where a run's event stream meets the definition it is running.
+ *
+ * The flow graph is drawn from the definition, where `for_each` is left
+ * unexpanded and a sub-workflow is one step, but the engine *runs* the
+ * definition expanded: it reports `base@open` where the definition says
+ * `base`, and a sub-workflow's inner step under its own name where the
+ * definition says the composed step. Nothing the page highlights matches
+ * until both sides are reduced to the same name - which is what broke the
+ * flow view's run state when list-driven steps arrived. */
+
+import type { JobEvent } from './types'
+
+/** The for_each step/member separator - `dw/for_each.py`'s
+ * MEMBER_SEPARATOR, reserved in every step name, so the first one in a
+ * name always splits a member from its group. */
+const MEMBER_SEPARATOR = '@'
+
+/** The name the definition gives the step the engine ran.
+ *
+ * A for_each member reduces to its group (`base@open` -> `base`); anything
+ * a sub-workflow emitted reduces to the step that queued it, which the
+ * engine puts on every event a child emits as `parent_step`
+ * (`_parent_progress_fields`, dw/workflow.py). A plain step is its own
+ * name, returned as it is. */
+export function flowNodeName(
+ step?: string,
+ parentStep?: string,
+): string | undefined {
+ const name = parentStep || step
+ if (!name) return undefined
+ const at = name.indexOf(MEMBER_SEPARATOR)
+ return at === -1 ? name : name.slice(0, at)
+}
+
+/** The nodes a run has finished, given the step names its `workflow_start`
+ * listed.
+ *
+ * A for_each group is one node, so it is done only once every member is -
+ * greening it on the first would say the group is behind us while its
+ * remaining entries are still queued. A sub-workflow is done when the
+ * composed step itself ends, not when one of the child's inner steps does,
+ * so only the top-level step_end events count: a child's carry the parent's
+ * name in `parent_step`, which is exactly the marker to leave out. */
+export function finishedNodes(
+ events: JobEvent[],
+ stepNames: string[],
+): string[] {
+ const ended = new Set(
+ events
+ .filter((event) => event.event === 'step_end' && !event.parent_step)
+ .map((event) => event.step as string),
+ )
+ const members = new Map()
+ for (const name of stepNames) {
+ const node = flowNodeName(name)
+ if (node) members.set(node, [...(members.get(node) ?? []), name])
+ }
+ return [...members]
+ .filter(([, names]) => names.every((name) => ended.has(name)))
+ .map(([node]) => node)
+}