diff --git a/CHANGELOG.md b/CHANGELOG.md index 71054a6..c005126 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ See [docs/changelog-template.md](docs/changelog-template.md) for formatting your ## [Unreleased] +### Pipelines + +- New pipeline editor: a higher-level visual DAG whose nodes are whole CORAL graphs or standalone executables, connected by ordering dependencies. A central "Pipeline editor" view (toggled from the top-right) lets you add stages (a CORAL graph from a file or the current canvas, or an executable), configure each stage's own resources (MPI nodes/tasks, time limit), and run the whole pipeline. On execution each stage is submitted as its own Slurm job chained after its predecessors, so stages run in the correct order, independent branches run in parallel, and the run continues even if the app is closed. Connections that would create a cycle are rejected. This first version runs in remote (Slurm) mode only and passes ordering between stages (no file hand-off yet). + ## [1.4.1] - 2026-07-14 ### Canvas-graph diff --git a/CLAUDE.md b/CLAUDE.md index d6c31b6..d2e4b1e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -98,9 +98,21 @@ Stores use Svelte 5 runes (`.svelte.ts` files): - `auth.svelte.ts` - JWT token for coral-remote-server API. Methods: `setToken()`, `setUsername()`, `clearToken()`. Persisted in electron-store under `access_token` / `username`. - `settingsStore.svelte.ts` - User settings stored under a single `'settings'` key in electron-store. Exports named key constants (`SSH_PATH`, `URL_VISUALIZER`, `URL_REMOTE_SERVER`, `USE_MPI`) and a `settingsState` object with `getKey(key)` / `setKey(key, value)` methods. - `currentProjectStore.svelte.ts` - Current project metadata (`id`, `name`). Methods: `set()`, `clear()`, `updateName()`. Exports `ApiProject` interface for use in components. -- `jobsStore.svelte.ts` - Slurm job tracking. `jobsState` has `isEmpty`/`oneOrLess` getters, `update()` refreshes from SSH. `jobIdMapState` maps Slurm scheduler IDs → internal incremental job IDs (used in touch-dir paths). +- `jobsStore.svelte.ts` - Slurm job tracking. `jobsState` has `isEmpty`/`oneOrLess` getters, `update()` refreshes from SSH. `jobIdMapState` maps Slurm scheduler IDs → a `JobIdEntry` (`internalId` used in touch-dir paths, `backendKind`, and the per-job `workingDirectory`). The `workingDirectory` is the root for single runs and the per-stage subdirectory for pipeline stages; `getOutFileContent` / `getNodesExecutionStatus` resolve a job's `slurm-.out` and `nodes-exec-status/` from it (via `getJobWorkingDirectory` / `getEntryByInternal`), so logs and node status work uniformly for both. - `toastsStore.svelte.ts` - Toast notifications. Use `toastState.add({ message, type })` to show a toast. Supports `'error'`/`'success'` types with auto-dismissal. - `dndStore.svelte.ts` - Drag-and-drop state (`dndNodeDataState.current`) for tracking which sidebar node is being dragged onto the canvas. +- `viewModeStore.svelte.ts` - Which central editor is shown: `viewModeState.value` is `'graph' | 'pipeline'`, with `toggle()`. Transient (not persisted). `App.svelte` renders `PipelineCanvas` when `'pipeline'`, else the graph/parameters editor. +- `pipeline.svelte.ts` - Transient store for the pipeline editor, holding the pipeline as xyflow `Node[]`/`Edge[]` (one node per stage, one edge per ordering dependency). Mirrors the get/set shape of `nodes.svelte.ts` (`getNodes`/`setNodes`/`getEdges`/`setEdges`) for canvas binding, plus `pipelineState` with `addCoralStage()`, `addExecutableStage()`, `updateStageData()`, `updateStageConfig()`, `setStageParameters()`, `removeStage()`, `clear()`, `toPipeline()`, and a `validation` getter. + +### Pipeline Orchestration + +The pipeline editor composes whole CORAL graphs and standalone executables as nodes of a higher-level DAG, where each edge means "target runs after source" (pure ordering — no data/artifact passing yet). This is separate from the CORAL canvas; coral itself runs one flat graph and cannot orchestrate across jobs. + +- `src/lib/types/pipelineTypes.ts` - `Pipeline`, `PipelineStage` (`coral` | `executable`, each carrying its own `CoralJobConfig`/`ExecutableJobConfig`), `PipelineEdge`, and the `StageData` stored on canvas nodes. +- `src/lib/orchestration/executionOrder.ts` - Pure, unit-tested `resolveExecutionOrder(pipeline)` (Kahn topological sort; throws `PipelineCycleError` on cycles) and `parentsOf(stageId, edges)`. No app/IO dependencies — the contract a future extracted service would reimplement. +- `src/lib/orchestration/pipelineOrchestrator.ts` - `runPipelineRemote(pipeline)`: topo-orders the DAG and submits each stage as its own Slurm job into `/pipeline-/stage-`, chained via `--dependency=afterok` on its parents' Slurm ids, then polls all jobs concurrently for live feedback. Remote (Slurm) only for now. +- Per-stage submit primitives live in `sshMessages.ts`: `submitCoralStageRemote()` / `submitExecutableStageRemote()` build+upload+submit a single stage (with `sbatch --parsable --kill-on-invalid-dep=yes [--dependency=afterok:…]`) and return the Slurm id without polling. The existing single-run remote functions route through these. `withMpiPlugin(network, useMpi)` injects the MPI plugin block; `jobPolling()` is exported for reuse. +- UI: `PipelineCanvas.svelte` (a second `` with a toolbar; an `isValidConnection` acyclicity guard replaces the coral canvas's type validation) plus the stage node components `nodes/CoralStageNode.svelte` and `nodes/ExecutableStageNode.svelte` (the per-stage config card is the node body). ### Node Type System diff --git a/src/App.svelte b/src/App.svelte index 7f80e7f..eacb9a4 100644 --- a/src/App.svelte +++ b/src/App.svelte @@ -2,6 +2,7 @@ import { SvelteFlowProvider } from '@xyflow/svelte' import FlowCanvas from './lib/components/FlowCanvas.svelte' import ParametersView from './lib/components/ParametersView.svelte' + import PipelineCanvas from './lib/components/PipelineCanvas.svelte' import Sidebar from './lib/components/layout/Sidebar.svelte' import SidebarButtons from './lib/components/layout/SidebarButtons.svelte' import JobsTable from './lib/components/layout/JobsTable.svelte' @@ -12,8 +13,10 @@ import ToastsWrapper from './lib/components/ToastsWrapper.svelte' import { settingsState } from './lib/stores/settingsStore.svelte' import { graphHistoryState } from './lib/stores/graphStack.svelte' + import { viewModeState } from './lib/stores/viewModeStore.svelte' let isCoralMode = $derived(settingsState.isCoralMode) + let viewMode = $derived(viewModeState.value) let executionLocation = $derived(settingsState.execution.location) let backendKind = $derived(settingsState.execution.backendKind) @@ -63,7 +66,7 @@
- {#if isCoralMode} + {#if isCoralMode && viewMode === 'single'}
+
- {#if isCoralMode} + {#if viewMode === 'pipeline'} + + {:else if isCoralMode} {:else} @@ -167,4 +181,19 @@ height: 100%; position: relative; } + + .view-toggle { + background: var(--primary-color); + color: var(--ternary-color); + border: 1px solid var(--ternary-color); + border-radius: 8px; + padding: 0.4rem 0.7rem; + font-size: 0.8rem; + font-weight: bold; + cursor: pointer; + } + + .view-toggle:hover { + border-color: var(--border-color-hover); + } diff --git a/src/lib/components/JobConfigModal.svelte b/src/lib/components/JobConfigModal.svelte index 544dbf4..83dee99 100644 --- a/src/lib/components/JobConfigModal.svelte +++ b/src/lib/components/JobConfigModal.svelte @@ -13,6 +13,7 @@ import { parametersState } from '../stores/parametersStore.svelte' import { settingsState } from '../stores/settingsStore.svelte' import { toastState } from '../stores/toastsStore.svelte' + import { isValidSlurmTime, SLURM_TIME_HINT } from '../utils/slurmTime' interface Props { modalId: string @@ -35,15 +36,8 @@ ) let totalProcesses = $derived(nodes * tasksPerNode) - // Slurm --time accepted formats: minutes | minutes:seconds | hours:minutes:seconds - // | days-hours | days-hours:minutes | days-hours:minutes:seconds | 0 - // See: https://slurm.schedmd.com/sbatch.html#OPT_time - const SLURM_TIME_PATTERN = - /^(\d+|\d+:\d{2}(:\d{2})?|\d+-\d{2}(:\d{2}(:\d{2})?)?)$/ let timeLimitError = $derived( - SLURM_TIME_PATTERN.test(timeLimit) - ? '' - : 'Use: minutes, minutes:seconds, HH:MM:SS, D-HH, D-HH:MM, D-HH:MM:SS' + isValidSlurmTime(timeLimit) ? '' : SLURM_TIME_HINT ) const handleConfirm = () => { diff --git a/src/lib/components/PipelineCanvas.svelte b/src/lib/components/PipelineCanvas.svelte new file mode 100644 index 0000000..3b601f3 --- /dev/null +++ b/src/lib/components/PipelineCanvas.svelte @@ -0,0 +1,239 @@ + + +
+ + + + +
+ {#if !isRemote} + + {/if} +
+ + + +
+
+ +
+ {#if validation.issues.length > 0} +
    + {#each validation.issues as issue (issue)} +
  • {issue}
    • + {/each} +
+ {/if} +
+ + + + +
+ + diff --git a/src/lib/components/layout/ExecutionBadge.svelte b/src/lib/components/layout/ExecutionBadge.svelte index 1f6f17e..3b95d94 100644 --- a/src/lib/components/layout/ExecutionBadge.svelte +++ b/src/lib/components/layout/ExecutionBadge.svelte @@ -3,6 +3,7 @@ ExecutionLocation, BackendKind, } from '../../types/settingsTypes' + import { viewModeState } from '../../stores/viewModeStore.svelte' interface Props { location: ExecutionLocation @@ -10,12 +11,16 @@ } let { location, backendKind }: Props = $props() + + let modeLabel = $derived( + viewModeState.value === 'pipeline' ? 'pipeline' : backendKind + )
{location} · - {backendKind} + {modeLabel}
diff --git a/src/lib/components/nodes/ExecutableStageNode.svelte b/src/lib/components/nodes/ExecutableStageNode.svelte new file mode 100644 index 0000000..0d39156 --- /dev/null +++ b/src/lib/components/nodes/ExecutableStageNode.svelte @@ -0,0 +1,208 @@ + + + + + + +
+
+ exe + + pipelineState.updateStageData(id, { + name: (e.currentTarget as HTMLInputElement).value, + })} + /> + +
+ + + +
+ + {data.parameters ? data.parametersFileName : 'no parameters'} + + +
+ + + +
+ + + + diff --git a/src/lib/orchestration/executionOrder.test.ts b/src/lib/orchestration/executionOrder.test.ts new file mode 100644 index 0000000..003f14d --- /dev/null +++ b/src/lib/orchestration/executionOrder.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, it } from 'vitest' +import { + parentsOf, + resolveExecutionOrder, + PipelineCycleError, +} from './executionOrder' +import type { Pipeline, PipelineStage } from '../types/pipelineTypes' + +const stage = (id: string): PipelineStage => ({ + id, + name: id, + kind: 'coral', + graph: {}, + config: { + kind: 'coral', + nodes: 1, + tasksPerNode: 1, + timeLimit: '01:00:00', + useMpi: false, + }, +}) + +const pipeline = (ids: string[], edges: [string, string][]): Pipeline => ({ + stages: ids.map(stage), + edges: edges.map(([source, target]) => ({ source, target })), +}) + +/** Asserts `before` appears earlier than `after` in the ordered id list. */ +const precedes = (order: string[], before: string, after: string) => + order.indexOf(before) < order.indexOf(after) + +describe('resolveExecutionOrder', () => { + it('orders a linear chain a → b → c', () => { + const order = resolveExecutionOrder( + pipeline( + ['a', 'b', 'c'], + [ + ['a', 'b'], + ['b', 'c'], + ] + ) + ).map((s) => s.id) + expect(order).toEqual(['a', 'b', 'c']) + }) + + it('orders a diamond so the fan-in node comes last', () => { + // a → b, a → c, b → d, c → d + const order = resolveExecutionOrder( + pipeline( + ['a', 'b', 'c', 'd'], + [ + ['a', 'b'], + ['a', 'c'], + ['b', 'd'], + ['c', 'd'], + ] + ) + ).map((s) => s.id) + expect(precedes(order, 'a', 'b')).toBe(true) + expect(precedes(order, 'a', 'c')).toBe(true) + expect(precedes(order, 'b', 'd')).toBe(true) + expect(precedes(order, 'c', 'd')).toBe(true) + }) + + it('includes disconnected stages', () => { + const order = resolveExecutionOrder( + pipeline(['a', 'b', 'c'], [['a', 'b']]) + ).map((s) => s.id) + expect(order).toHaveLength(3) + expect(new Set(order)).toEqual(new Set(['a', 'b', 'c'])) + expect(precedes(order, 'a', 'b')).toBe(true) + }) + + it('throws PipelineCycleError on a cycle', () => { + expect(() => + resolveExecutionOrder( + pipeline( + ['a', 'b'], + [ + ['a', 'b'], + ['b', 'a'], + ] + ) + ) + ).toThrow(PipelineCycleError) + }) + + it('ignores edges referencing unknown stages', () => { + const order = resolveExecutionOrder( + pipeline( + ['a', 'b'], + [ + ['a', 'b'], + ['ghost', 'a'], + ] + ) + ).map((s) => s.id) + expect(order).toEqual(['a', 'b']) + }) +}) + +describe('parentsOf', () => { + it('returns the deduplicated direct parents', () => { + const edges = [ + { source: 'a', target: 'c' }, + { source: 'b', target: 'c' }, + { source: 'a', target: 'c' }, + ] + expect(parentsOf('c', edges).sort()).toEqual(['a', 'b']) + expect(parentsOf('a', edges)).toEqual([]) + }) +}) diff --git a/src/lib/orchestration/executionOrder.ts b/src/lib/orchestration/executionOrder.ts new file mode 100644 index 0000000..dc1fc90 --- /dev/null +++ b/src/lib/orchestration/executionOrder.ts @@ -0,0 +1,83 @@ +/** + * Pure dependency-resolution for pipelines. Topologically sorts the stage DAG so + * the orchestrator can submit each stage after its parents, and rejects cyclic + * graphs up front. No app/IO dependencies — trivially unit-testable and portable. + */ + +import type { + Pipeline, + PipelineEdge, + PipelineStage, +} from '../types/pipelineTypes' + +/** Thrown by {@link resolveExecutionOrder} when the pipeline contains a cycle. */ +export class PipelineCycleError extends Error { + constructor(message = 'Pipeline contains a cycle') { + super(message) + this.name = 'PipelineCycleError' + } +} + +/** + * Returns the ids of the direct parents of a stage (the sources of edges that + * target it). Parents must complete before the stage runs. + * @param stageId - The stage whose parents to collect. + * @param edges - All pipeline edges. + * @returns The deduplicated list of parent stage ids. + */ +export const parentsOf = (stageId: string, edges: PipelineEdge[]): string[] => { + const parents = new Set() + for (const edge of edges) { + if (edge.target === stageId) parents.add(edge.source) + } + return [...parents] +} + +/** + * Topologically sorts a pipeline's stages using Kahn's algorithm so that every + * stage appears after all of its parents. + * @param pipeline - The pipeline whose stages and edges to order. + * @returns The stages in a valid execution order. + * @throws {PipelineCycleError} If the dependency graph contains a cycle. + */ +export const resolveExecutionOrder = (pipeline: Pipeline): PipelineStage[] => { + const { stages, edges } = pipeline + const stageById = new Map(stages.map((stage) => [stage.id, stage])) + + // Only count edges between known stages so dangling edges can't corrupt counts. + const validEdges = edges.filter( + (edge) => stageById.has(edge.source) && stageById.has(edge.target) + ) + + const indegree = new Map(stages.map((stage) => [stage.id, 0])) + const children = new Map( + stages.map((stage) => [stage.id, []]) + ) + for (const edge of validEdges) { + indegree.set(edge.target, (indegree.get(edge.target) ?? 0) + 1) + children.get(edge.source)!.push(edge.target) + } + + // Seed the queue with all roots (no unmet dependencies), preserving input order. + const queue = stages + .filter((stage) => indegree.get(stage.id) === 0) + .map((s) => s.id) + const order: PipelineStage[] = [] + + while (queue.length > 0) { + const id = queue.shift()! + order.push(stageById.get(id)!) + for (const child of children.get(id)!) { + const next = indegree.get(child)! - 1 + indegree.set(child, next) + if (next === 0) queue.push(child) + } + } + + // Fewer ordered stages than total ⇒ at least one stage never reached indegree 0. + if (order.length !== stages.length) { + throw new PipelineCycleError() + } + + return order +} diff --git a/src/lib/orchestration/pipelineOrchestrator.ts b/src/lib/orchestration/pipelineOrchestrator.ts new file mode 100644 index 0000000..46355c5 --- /dev/null +++ b/src/lib/orchestration/pipelineOrchestrator.ts @@ -0,0 +1,130 @@ +/** + * In-app pipeline orchestration (remote / Slurm only, MVP). + * + * Topologically orders the stage DAG and submits every stage as its own Slurm job, + * chaining each one after its parents with `--dependency=afterok`. Slurm then + * enforces the order and runs independent branches in parallel — the run survives + * the app being closed; the concurrent polling here is only for live feedback. + */ + +import { + submitCoralStageRemote, + submitExecutableStageRemote, + jobPolling, +} from '../utils/sshMessages' +import { jobsState } from '../stores/jobsStore.svelte' +import { settingsState } from '../stores/settingsStore.svelte' +import { toastState } from '../stores/toastsStore.svelte' +import { JobStatus } from '../types/jobTypes' +import { resolveExecutionOrder, parentsOf } from './executionOrder' +import type { Pipeline } from '../types/pipelineTypes' + +const POLL_INTERVAL_MS = 10 * 1000 +const POLL_TIMEOUT_MS = 24 * 60 * 60 * 1000 + +/** + * Submits every stage of a pipeline to the remote Slurm scheduler in dependency + * order and reports progress via toasts and the jobs table. + * @param pipeline - The pipeline (stages + ordering edges) to execute. + * @returns Resolves once all stages have reached a terminal state. + * @throws {Error} If not in remote mode, or if the pipeline is empty/cyclic, or if + * any stage fails to submit. + */ +export const runPipelineRemote = async (pipeline: Pipeline): Promise => { + if (settingsState.execution.location !== 'remote') { + toastState.add({ + message: 'Pipelines currently run in remote mode only', + type: 'error', + }) + return + } + if (pipeline.stages.length === 0) { + toastState.add({ message: 'Pipeline has no stages', type: 'error' }) + return + } + + try { + // Throws PipelineCycleError if the DAG is cyclic. + const order = resolveExecutionOrder(pipeline) + const pipelineDir = `${settingsState.remote.workingDirectory}/pipeline-${Date.now()}` + + // Map each stage id to its submitted Slurm id so children can depend on parents. + const slurmIdByStage = new Map() + + for (const stage of order) { + const stageDir = `${pipelineDir}/stage-${stage.id}` + const dependencyJobIds = parentsOf(stage.id, pipeline.edges).map( + (parentId) => { + const parentSlurmId = slurmIdByStage.get(parentId) + // Guaranteed present: topo order submits every parent before its children. + if (!parentSlurmId) + throw new Error( + `Missing submitted job id for parent stage ${parentId}` + ) + return parentSlurmId + } + ) + + let slurmId: string + if (stage.kind === 'coral') { + slurmId = await submitCoralStageRemote({ + graph: stage.graph as object, + stageDir, + config: stage.config, + dependencyJobIds, + }) + } else if (stage.kind === 'executable') { + if (!stage.parameters) + throw new Error( + `Executable stage "${stage.name}" has no parameters loaded` + ) + slurmId = await submitExecutableStageRemote({ + executablePath: stage.executablePath, + parameters: stage.parameters, + parametersFileName: stage.parametersFileName, + stageDir, + config: stage.config, + dependencyJobIds, + }) + } else { + throw new Error( + `Unknown stage kind for stage ${(stage as { id: string }).id}` + ) + } + + slurmIdByStage.set(stage.id, slurmId) + } + + toastState.add({ + message: `Submitted ${order.length} stage(s) to Slurm`, + type: 'success', + }) + await jobsState.update() + + // Poll all stages concurrently for live feedback; Slurm runs them in order regardless. + const results = await Promise.all( + order.map(async (stage) => { + const slurmId = slurmIdByStage.get(stage.id)! + const finalState = await jobPolling( + slurmId, + POLL_INTERVAL_MS, + POLL_TIMEOUT_MS + ) + return { stage, slurmId, finalState } + }) + ) + + await jobsState.update() + for (const { stage, slurmId, finalState } of results) { + toastState.add({ + message: `${stage.name} (job ${slurmId}): ${finalState}`, + type: finalState === JobStatus.COMPLETED ? 'success' : 'error', + }) + } + } catch (error) { + toastState.add({ + message: error instanceof Error ? error.message : 'Pipeline run failed', + type: 'error', + }) + } +} diff --git a/src/lib/stores/jobsStore.svelte.ts b/src/lib/stores/jobsStore.svelte.ts index 49f6bca..7bd8345 100644 --- a/src/lib/stores/jobsStore.svelte.ts +++ b/src/lib/stores/jobsStore.svelte.ts @@ -87,13 +87,19 @@ export const jobIdMapState = { * @param jobIdScheduler - Slurm ID for remote jobs; equals internalJobId for local jobs. * @param jobIdInternal - The touch-dir counter used in nodes-exec-status//. * @param backendKind - Backend used for this run. + * @param workingDirectory - Remote dir holding this job's logs and touch-dir. */ async add( jobIdScheduler: number | string, jobIdInternal: number, - backendKind: 'coral' | 'executable' + backendKind: 'coral' | 'executable', + workingDirectory?: string ): Promise { - jobIdMap[jobIdScheduler] = { internalId: jobIdInternal, backendKind } + jobIdMap[jobIdScheduler] = { + internalId: jobIdInternal, + backendKind, + workingDirectory, + } await window.electron?.store?.set('jobIdMap', $state.snapshot(jobIdMap)) }, /** @@ -112,6 +118,24 @@ export const jobIdMapState = { ): 'coral' | 'executable' | undefined { return jobIdMap[jobIdScheduler]?.backendKind }, + /** + * @param jobIdScheduler - The scheduler job ID to look up. + * @returns The job's stored working directory, or undefined if not recorded. + */ + getJobWorkingDirectory(jobIdScheduler: string): string | undefined { + return jobIdMap[jobIdScheduler]?.workingDirectory + }, + /** + * Resolves the full job entry by internal id (used by node-status lookups, which + * only carry the internal id, to recover both backend kind and working directory). + * @param jobIdInternal - The internal job ID. + * @returns The matching entry, or undefined if not found. + */ + getEntryByInternal(jobIdInternal: number): JobIdEntry | undefined { + return Object.values(jobIdMap).find( + (entry) => entry.internalId === jobIdInternal + ) + }, } /** diff --git a/src/lib/stores/pipeline.svelte.ts b/src/lib/stores/pipeline.svelte.ts new file mode 100644 index 0000000..62cf720 --- /dev/null +++ b/src/lib/stores/pipeline.svelte.ts @@ -0,0 +1,244 @@ +/** + * Transient store for the pipeline editor: holds the pipeline as xyflow nodes/edges + * (one node per stage, one edge per ordering dependency). Mirrors the get/set shape + * of `nodes.svelte.ts` so `PipelineCanvas` can bind the same way, and exposes a + * higher-level mutation + validation API plus `toPipeline()` for the orchestrator. + */ + +import type { Edge, Node } from '@xyflow/svelte' +import type { + CoralStageData, + ExecutableStageData, + Pipeline, + PipelineStage, + StageData, +} from '../types/pipelineTypes' +import type { CoralJobConfig, ExecutableJobConfig } from '../utils/sshMessages' +import type { ParameterTree } from '../types/parameterTypes' +import { isValidSlurmTime } from '../utils/slurmTime' + +let nodes = $state.raw([]) +let edges = $state.raw([]) +let counter = 0 + +const DEFAULT_CORAL_CONFIG = (): CoralJobConfig => ({ + kind: 'coral', + nodes: 1, + tasksPerNode: 4, + timeLimit: '01:00:00', + useMpi: false, +}) + +const DEFAULT_EXECUTABLE_CONFIG = (): ExecutableJobConfig => ({ + kind: 'executable', + timeLimit: '01:00:00', +}) + +/** Reactive accessors used to bind the SvelteFlow canvas. */ +export const getNodes = (): Node[] => nodes +export const getEdges = (): Edge[] => edges +export const setNodes = (next: Node[]): void => { + nodes = next +} +export const setEdges = (next: Edge[]): void => { + edges = next +} +export const getNodesSnapshot = (): Node[] => + $state.snapshot(nodes) as unknown as Node[] +export const getEdgesSnapshot = (): Edge[] => + $state.snapshot(edges) as unknown as Edge[] + +export const pipelineState = { + get nodes(): Node[] { + return nodes + }, + get edges(): Edge[] { + return edges + }, + + /** + * Adds a CORAL-graph stage node. + * @param params.name - Display name for the stage. + * @param params.graph - The CORAL network object (from a file or canvas snapshot). + * @param params.position - Optional canvas position; cascades by default. + */ + addCoralStage({ + name, + graph, + position, + }: { + name: string + graph: unknown + position?: { x: number; y: number } + }): void { + const data: CoralStageData = { + name, + kind: 'coral', + graph, + config: DEFAULT_CORAL_CONFIG(), + } + addStageNode('coralStage', data, position) + }, + + /** + * Adds an executable stage node. + * @param params.name - Display name for the stage. + * @param params.executablePath - Remote binary path (default from settings). + * @param params.parametersFileName - Params filename (extension selects JSON/PRM). + * @param params.position - Optional canvas position; cascades by default. + */ + addExecutableStage({ + name, + executablePath, + parametersFileName, + position, + }: { + name: string + executablePath: string + parametersFileName: string + position?: { x: number; y: number } + }): void { + const data: ExecutableStageData = { + name, + kind: 'executable', + executablePath, + parametersFileName, + parameters: null, + config: DEFAULT_EXECUTABLE_CONFIG(), + } + addStageNode('executableStage', data, position) + }, + + /** + * Shallow-merges a patch into a stage node's top-level data (e.g. name, executablePath). + * @param id - The stage node id. + * @param patch - Partial stage data to merge. + */ + updateStageData(id: string, patch: Partial): void { + nodes = nodes.map((node) => + node.id === id + ? { ...node, data: { ...node.data, ...patch } as StageData } + : node + ) + }, + + /** + * Shallow-merges a patch into a stage node's job config. + * @param id - The stage node id. + * @param patch - Partial job config to merge. + */ + updateStageConfig( + id: string, + patch: Partial | Partial + ): void { + nodes = nodes.map((node) => { + if (node.id !== id) return node + const data = node.data as unknown as StageData + return { + ...node, + data: { ...data, config: { ...data.config, ...patch } } as StageData, + } + }) + }, + + /** + * Sets the loaded parameters and filename on an executable stage. + * @param id - The stage node id. + * @param params.parameters - The parsed parameter tree. + * @param params.parametersFileName - The params filename. + */ + setStageParameters( + id: string, + { + parameters, + parametersFileName, + }: { parameters: ParameterTree; parametersFileName: string } + ): void { + this.updateStageData(id, { + parameters, + parametersFileName, + } as Partial) + }, + + /** + * Removes a stage node and any edges connected to it. + * @param id - The stage node id to remove. + */ + removeStage(id: string): void { + nodes = nodes.filter((node) => node.id !== id) + edges = edges.filter((edge) => edge.source !== id && edge.target !== id) + }, + + /** Clears all stages and edges. */ + clear(): void { + nodes = [] + edges = [] + }, + + /** + * Builds the plain {@link Pipeline} for the orchestrator from the current canvas. + * @returns The pipeline (stages + ordering edges). + */ + toPipeline(): Pipeline { + const snapshotNodes = getNodesSnapshot() + const snapshotEdges = getEdgesSnapshot() + const stages: PipelineStage[] = snapshotNodes.map( + (node) => + ({ + id: node.id, + ...(node.data as unknown as StageData), + }) as PipelineStage + ) + return { + stages, + edges: snapshotEdges.map((edge) => ({ + source: edge.source, + target: edge.target, + })), + } + }, + + /** + * Validation summary driving the Run button and inline issue list. + * @returns `runnable` plus a list of human-readable `issues`. + */ + get validation(): { runnable: boolean; issues: string[] } { + const issues: string[] = [] + const { stages } = this.toPipeline() + if (stages.length === 0) issues.push('Add at least one stage') + + for (const stage of stages) { + if (stage.kind === 'coral') { + if (!stage.graph) issues.push(`${stage.name}: no graph loaded`) + if (!isValidSlurmTime(stage.config.timeLimit)) + issues.push(`${stage.name}: invalid time limit`) + } else if (stage.kind === 'executable') { + if (!stage.executablePath.trim()) + issues.push(`${stage.name}: no executable path`) + if (!stage.parameters) + issues.push(`${stage.name}: no parameters loaded`) + if (stage.config.timeLimit && !isValidSlurmTime(stage.config.timeLimit)) + issues.push(`${stage.name}: invalid time limit`) + } + } + return { runnable: issues.length === 0, issues } + }, +} + +// ── Private helpers ── + +/** Appends a new stage node, cascading its position when none is given. */ +const addStageNode = ( + type: 'coralStage' | 'executableStage', + data: StageData, + position?: { x: number; y: number } +): void => { + const index = nodes.length + const node = { + id: `p${counter++}`, + type, + position: position ?? { x: 80 + index * 60, y: 80 + index * 60 }, + data, + } as unknown as Node + nodes = [...nodes, node] +} diff --git a/src/lib/stores/viewModeStore.svelte.ts b/src/lib/stores/viewModeStore.svelte.ts new file mode 100644 index 0000000..2bcf201 --- /dev/null +++ b/src/lib/stores/viewModeStore.svelte.ts @@ -0,0 +1,21 @@ +/** + * Which central view is shown: the graph/parameters editor or the pipeline editor. + * Transient (not persisted) — defaults to the graph editor on each launch. + */ + +export type ViewMode = 'single' | 'pipeline' + +let mode = $state('single') + +export const viewModeState = { + get value(): ViewMode { + return mode + }, + set value(next: ViewMode) { + mode = next + }, + /** Toggles between the single-stage editor and the pipeline editor. */ + toggle(): void { + mode = mode === 'single' ? 'pipeline' : 'single' + }, +} diff --git a/src/lib/types/jobTypes.ts b/src/lib/types/jobTypes.ts index 1e0dbe2..211209c 100644 --- a/src/lib/types/jobTypes.ts +++ b/src/lib/types/jobTypes.ts @@ -6,6 +6,12 @@ export type JobIdEntry = { internalId: number backendKind: 'coral' | 'executable' + /** + * Remote directory holding this job's `slurm-.out` and `nodes-exec-status/`. + * For single runs this is the root working directory; for pipeline stages it is + * the per-stage subdirectory. Optional for backward-compatibility and local runs. + */ + workingDirectory?: string } /** diff --git a/src/lib/types/pipelineTypes.ts b/src/lib/types/pipelineTypes.ts new file mode 100644 index 0000000..b00be7f --- /dev/null +++ b/src/lib/types/pipelineTypes.ts @@ -0,0 +1,50 @@ +/** + * Types for the pipeline orchestrator: a higher-level DAG whose nodes ("stages") + * are whole CORAL graphs or standalone executables. Edges express pure ordering + * ("target runs after source") — no data/artifact passing in this MVP. + */ + +import type { CoralJobConfig, ExecutableJobConfig } from '../utils/sshMessages' +import type { ParameterTree } from './parameterTypes' + +/** A pipeline stage that runs a whole CORAL graph. */ +export type CoralPipelineStage = { + id: string + name: string + kind: 'coral' + /** A Network JSON object (loaded from a file or snapshotted from the canvas). */ + graph: unknown + config: CoralJobConfig +} + +/** A pipeline stage that runs a standalone executable with a parameters file. */ +export type ExecutablePipelineStage = { + id: string + name: string + kind: 'executable' + /** Per-stage binary path (defaults from settings when the stage is created). */ + executablePath: string + parametersFileName: string + /** Parameter tree loaded from a file, serialized to disk before the run. */ + parameters: ParameterTree | null + config: ExecutableJobConfig +} + +export type PipelineStage = CoralPipelineStage | ExecutablePipelineStage + +/** Stage fields stored on a canvas node's `data` (the stage minus its id, which is the node id). */ +export type CoralStageData = Omit +export type ExecutableStageData = Omit +export type StageData = CoralStageData | ExecutableStageData + +/** A dependency edge: `target` runs after `source` completes successfully. */ +export type PipelineEdge = { + source: string + target: string +} + +/** A pipeline: a set of stages plus the ordering edges between them. */ +export type Pipeline = { + stages: PipelineStage[] + edges: PipelineEdge[] +} diff --git a/src/lib/utils/slurmTime.ts b/src/lib/utils/slurmTime.ts new file mode 100644 index 0000000..75c8056 --- /dev/null +++ b/src/lib/utils/slurmTime.ts @@ -0,0 +1,22 @@ +/** + * Validation for the Slurm `--time` limit value, shared by the single-run job + * config modal and the pipeline stage nodes. + */ + +// Slurm --time accepted formats: minutes | minutes:seconds | hours:minutes:seconds +// | days-hours | days-hours:minutes | days-hours:minutes:seconds | 0 +// See: https://slurm.schedmd.com/sbatch.html#OPT_time +const SLURM_TIME_PATTERN = + /^(\d+|\d+:\d{2}(:\d{2})?|\d+-\d{2}(:\d{2}(:\d{2})?)?)$/ + +/** Human-readable hint describing the accepted Slurm `--time` formats. */ +export const SLURM_TIME_HINT = + 'Use: minutes, minutes:seconds, HH:MM:SS, D-HH, D-HH:MM, D-HH:MM:SS' + +/** + * Checks whether a string is a valid Slurm `--time` value. + * @param value - The time-limit string to validate. + * @returns True if the value matches an accepted Slurm `--time` format. + */ +export const isValidSlurmTime = (value: string): boolean => + SLURM_TIME_PATTERN.test(value) diff --git a/src/lib/utils/sshMessages.ts b/src/lib/utils/sshMessages.ts index a404f3f..a4634c4 100644 --- a/src/lib/utils/sshMessages.ts +++ b/src/lib/utils/sshMessages.ts @@ -19,6 +19,7 @@ import { isTerminalStatus, } from '../types/jobTypes' import type { RemoteExecutionSettings } from '../types/settingsTypes' +import type { ParameterTree } from '../types/parameterTypes' // The `?raw` Vite suffix imports the file contents as a plain string at build time. // It works identically in dev, built app, and packaged Electron binaries. // Docs: https://vite.dev/guide/assets#importing-asset-as-string @@ -116,49 +117,65 @@ const exportAndEvalGraphRemote = async ( edges: Edge[], config?: CoralJobConfig ): Promise => { + // Parse the canvas once without MPI; the submit primitive injects the MPI block. + const graph = buildGraphPayload(nodes, edges, false) + const jobId = await submitCoralStageRemote({ + graph, + stageDir: settingsState.remote.workingDirectory, + config, + dependencyJobIds: [], + }) + toastState.add({ message: `Submitted job ${jobId}` }) + + // poll every 10 secs for 1 day, finally display final status + const finalState = await jobPolling(jobId, 10 * 1000, 24 * 60 * 60 * 1000) + toastState.add({ + message: `Job id ${jobId}: ${finalState}`, + type: finalState === JobStatus.COMPLETED ? 'success' : 'error', + }) +} + +/** + * Builds, uploads, and submits a single CORAL-graph stage as one Slurm job, with + * optional `afterok` dependencies, and returns its Slurm job id without polling. + * @param params.graph - The CORAL network object (MPI block injected here from config). + * @param params.stageDir - Remote directory used as the job working directory. + * @param params.config - Per-stage Coral job config (MPI, nodes, time limit). + * @param params.dependencyJobIds - Slurm ids this stage must wait for (afterok). + * @returns The submitted Slurm job id. + * @throws {Error} If sbatch does not return a job id. + */ +export const submitCoralStageRemote = async ({ + graph, + stageDir, + config, + dependencyJobIds, +}: { + graph: object + stageDir: string + config?: CoralJobConfig + dependencyJobIds: string[] +}): Promise => { const useMpi = config?.useMpi ?? false - const remoteSettings = settingsState.remote - const workingDirectory = remoteSettings.workingDirectory + const internalJobId = jobIdMapState.getNextKey() + const stageSettings = { ...settingsState.remote, workingDirectory: stageDir } - // build and upload graph JSON (MPI plugin block injected when MPI is enabled) - const graphPayload = buildGraphPayload(nodes, edges, useMpi) + await ensureRemoteDir(stageDir) await uploadFileSsh( - JSON.stringify(graphPayload), - `${workingDirectory}/graph.json` + JSON.stringify(withMpiPlugin(graph, useMpi)), + `${stageDir}/graph.json` ) - - // build and upload batch script - const internalJobId = jobIdMapState.getNextKey() const batchScript = buildBatchScript( useMpi, internalJobId, - remoteSettings, + stageSettings, config ) - await uploadFileSsh(batchScript, `${workingDirectory}/job.sh`) - - // submit job - const resultExecute = await window.electron.invoke('execute-ssh-with-key', { - command: `sbatch ${shellEscape(`${workingDirectory}/job.sh`)}`, - rejectOnNonZeroCode: true, - }) - console.log('SSH Connection Result:', resultExecute) - toastState.add({ message: resultExecute }) - - // get job id and store mapping - const jobId = resultExecute.match(/\d+/)?.[0] - if (!jobId) - throw new Error( - `sbatch did not return a job ID. Output: "${resultExecute}"` - ) - await jobIdMapState.add(jobId, internalJobId, 'coral') + await uploadFileSsh(batchScript, `${stageDir}/job.sh`) - // poll every 10 secs for 1 day, finally display final status - const finalState = await jobPolling(jobId, 10 * 1000, 24 * 60 * 60 * 1000) - toastState.add({ - message: `Job id ${jobId}: ${finalState}`, - type: finalState === JobStatus.COMPLETED ? 'success' : 'error', - }) + const jobId = await submitSbatch(`${stageDir}/job.sh`, dependencyJobIds) + await jobIdMapState.add(jobId, internalJobId, 'coral', stageDir) + return jobId } const exportAndEvalGraphLocal = async ( @@ -180,7 +197,12 @@ const exportAndEvalGraphLocal = async ( }) const jobId = String(resultExecute.jobId) - await jobIdMapState.add(jobId, internalJobId, 'coral') + await jobIdMapState.add( + jobId, + internalJobId, + 'coral', + localSettings.workingDirectory + ) toastState.add({ message: `Started local Coral run ${jobId}` }) const finalState = await localJobPolling(jobId, 1000, 24 * 60 * 60 * 1000) @@ -216,7 +238,12 @@ const exportAndEvalExecutableLocal = async ( }) // local runs have no external scheduler ID — both keys are the same internalJobId - await jobIdMapState.add(internalJobId, internalJobId, 'executable') + await jobIdMapState.add( + internalJobId, + internalJobId, + 'executable', + localSettings.workingDirectory + ) toastState.add({ message: `Started local executable run ${internalJobId}` }) const finalState = await localJobPolling( @@ -235,38 +262,20 @@ const exportAndEvalExecutableRemote = async ( config?: ExecutableJobConfig ): Promise => { const remoteSettings = settingsState.remote - const internalJobId = jobIdMapState.getNextKey() - const parametersPayload = getExecutableParametersPayload() const parametersFileName = config?.parametersFileName || remoteSettings.parametersFileName || 'parameters.json' - const parametersRemotePath = `${remoteSettings.workingDirectory}/${parametersFileName}` - const jobScriptRemotePath = `${remoteSettings.workingDirectory}/job.sh` - await uploadFileSsh( - serializeParametersFile(parametersPayload, parametersFileName), - parametersRemotePath - ) - - const batchScript = buildExecutableBatchScript( - internalJobId, - remoteSettings.workingDirectory, - remoteSettings.executablePath, + const jobId = await submitExecutableStageRemote({ + executablePath: remoteSettings.executablePath, + parameters: getExecutableParametersPayload(), parametersFileName, - config - ) - await uploadFileSsh(batchScript, jobScriptRemotePath) - - const resultExecute = await window.electron.invoke('execute-ssh-with-key', { - command: `sbatch ${shellEscape(jobScriptRemotePath)}`, - rejectOnNonZeroCode: true, + stageDir: remoteSettings.workingDirectory, + config, + dependencyJobIds: [], }) - toastState.add({ message: resultExecute }) - - const jobId = resultExecute.match(/\d+/)?.[0] - if (!jobId) throw new Error('Job ID not found') - await jobIdMapState.add(jobId, internalJobId, 'executable') + toastState.add({ message: `Submitted job ${jobId}` }) const finalState = await jobPolling(jobId, 10 * 1000, 24 * 60 * 60 * 1000) toastState.add({ @@ -276,22 +285,82 @@ const exportAndEvalExecutableRemote = async ( } /** - * Builds the graph payload to upload, injecting the MPI plugin block when MPI is enabled. + * Builds, uploads, and submits a single executable stage as one Slurm job, with + * optional `afterok` dependencies, and returns its Slurm job id without polling. + * @param params.executablePath - Remote path of the binary to run. + * @param params.parameters - Parameter tree serialized to the params file. + * @param params.parametersFileName - Params filename (extension selects JSON/PRM). + * @param params.stageDir - Remote directory used as the job working directory. + * @param params.config - Per-stage executable job config (time limit). + * @param params.dependencyJobIds - Slurm ids this stage must wait for (afterok). + * @returns The submitted Slurm job id. + * @throws {Error} If sbatch does not return a job id. + */ +export const submitExecutableStageRemote = async ({ + executablePath, + parameters, + parametersFileName, + stageDir, + config, + dependencyJobIds, +}: { + executablePath: string + parameters: ParameterTree + parametersFileName: string + stageDir: string + config?: ExecutableJobConfig + dependencyJobIds: string[] +}): Promise => { + const internalJobId = jobIdMapState.getNextKey() + + await ensureRemoteDir(stageDir) + await uploadFileSsh( + serializeParametersFile(parameters, parametersFileName), + `${stageDir}/${parametersFileName}` + ) + const batchScript = buildExecutableBatchScript( + internalJobId, + stageDir, + executablePath, + parametersFileName, + config + ) + await uploadFileSsh(batchScript, `${stageDir}/job.sh`) + + const jobId = await submitSbatch(`${stageDir}/job.sh`, dependencyJobIds) + await jobIdMapState.add(jobId, internalJobId, 'executable', stageDir) + return jobId +} + +/** + * Injects the MPI plugin block into a CORAL network object when MPI is enabled. * The plugin block tells CORAL to initialise MPI with the given thread cap. + * @param network - A CORAL network object (already in protocol form). + * @param useMpi - Whether to prepend the MPI plugin block. + * @returns The network unchanged, or a copy with the MPI plugin block prepended. */ -export const buildGraphPayload = ( - nodes: Node[], - edges: Edge[], - useMpi: boolean -) => { - const parsedGraph = parseGraphWithQualifiedIds(nodes, edges) - if (!useMpi) return parsedGraph +export const withMpiPlugin = (network: object, useMpi: boolean): object => { + if (!useMpi) return network return { plugin: { MPI: { enabled: true, max_num_threads: 1 } }, - ...parsedGraph, + ...network, } } +/** + * Builds the graph payload to upload from canvas nodes/edges, injecting the MPI + * plugin block when MPI is enabled. + * @param nodes - Canvas nodes (snapshots). + * @param edges - Canvas edges (snapshots). + * @param useMpi - Whether to inject the MPI plugin block. + * @returns The CORAL network object ready to upload. + */ +export const buildGraphPayload = ( + nodes: Node[], + edges: Edge[], + useMpi: boolean +): object => withMpiPlugin(parseGraphWithQualifiedIds(nodes, edges), useMpi) + /** * Builds the batch script content from the appropriate template, replacing all placeholders. * {{WORKING_DIRECTORY}}, {{CORAL_BINARY_PATH}}, {{CORAL_PLUGIN_PATH}}, and {{TIME_LIMIT}} apply to both templates. @@ -338,18 +407,58 @@ const buildExecutableBatchScript = ( config?: ExecutableJobConfig ): string => { const timeLimit = config?.timeLimit ?? '01:00:00' - const parametersPath = `${workingDirectory}/${parametersFileName}` + // Pass the parameters file as a bare name (the job already chdir's into the + // working directory). This keeps the volatile directory path — e.g. the + // pipeline's `pipeline-/stage-` — out of the executable's argv. + // Some deal.II programs (e.g. step-70) sniff their dimension from the file path, + // so a stray digit in the directory name would otherwise change the run. return `#!/bin/bash #SBATCH --chdir=${workingDirectory} #SBATCH --output=${workingDirectory}/slurm-%j.out #SBATCH --job-name=executable-${internalJobId} #SBATCH --time=${timeLimit} -${shellQuoteForScript(executablePath)} ${shellQuoteForScript(parametersPath)} +${shellQuoteForScript(executablePath)} ${shellQuoteForScript(parametersFileName)} ` } +/** Creates a remote directory (and parents) so SFTP uploads into it succeed. */ +const ensureRemoteDir = async (dir: string): Promise => { + await window.electron.invoke('execute-ssh-with-key', { + command: `mkdir -p ${shellEscape(dir)}`, + rejectOnNonZeroCode: true, + }) +} + +/** + * Submits a previously-uploaded job script via `sbatch --parsable`, optionally + * chaining it after other jobs. `--kill-on-invalid-dep=yes` makes Slurm cascade a + * cancellation to this job (and its descendants) if an upstream dependency fails. + * @param jobScriptPath - Remote path of the uploaded sbatch script. + * @param dependencyJobIds - Slurm ids this job must wait for (afterok); empty for none. + * @returns The submitted Slurm job id. + * @throws {Error} If sbatch does not return a job id. + */ +const submitSbatch = async ( + jobScriptPath: string, + dependencyJobIds: string[] +): Promise => { + const dependencyFlag = + dependencyJobIds.length > 0 + ? ` --dependency=afterok:${dependencyJobIds.join(':')}` + : '' + const command = `sbatch --parsable --kill-on-invalid-dep=yes${dependencyFlag} ${shellEscape(jobScriptPath)}` + const result = await window.electron.invoke('execute-ssh-with-key', { + command, + rejectOnNonZeroCode: true, + }) + const jobId = result.match(/\d+/)?.[0] + if (!jobId) + throw new Error(`sbatch did not return a job ID. Output: "${result}"`) + return jobId +} + /** Uploads a string as a file to the remote server via SSH. */ const uploadFileSsh = async ( content: string, @@ -371,7 +480,7 @@ const uploadFileSsh = async ( * @returns {Promise} A promise that resolves to the job status ('COMPLETED' or 'FAILED') when the job is finished. * @throws {Error} Throws an error if there is a polling error or if the job times out. */ -const jobPolling = async ( +export const jobPolling = async ( jobId: string, interval: number, timeout?: number @@ -490,7 +599,10 @@ export const getOutFileContent = async ( return await window.electron.invoke('get-local-run-log', { jobId }) } - const outputDirectory = execution.remote.workingDirectory + // Resolve the job's own directory (per-stage for pipelines, root for single runs). + const outputDirectory = + jobIdMapState.getJobWorkingDirectory(String(jobId)) ?? + execution.remote.workingDirectory const command = `cat ${shellEscape(`${outputDirectory}/slurm-${jobId}.out`)}` return await window.electron.invoke('execute-ssh-with-key', { command: command, @@ -510,7 +622,11 @@ export const getNodesExecutionStatus = async ( jobIdInternal: number ): Promise> => { const execution = settingsState.execution - if (execution.backendKind === 'executable') { + // Use the job's own backend kind / directory (a pipeline can mix kinds and dirs), + // falling back to the global settings for jobs predating per-job tracking. + const entry = jobIdMapState.getEntryByInternal(jobIdInternal) + const backendKind = entry?.backendKind ?? execution.backendKind + if (backendKind === 'executable') { return new Map() } // define the command to list the files in the touch-dir @@ -520,9 +636,11 @@ export const getNodesExecutionStatus = async ( jobIdInternal, }) } else if (execution.location === 'remote') { + const statusDirectory = + entry?.workingDirectory ?? execution.remote.workingDirectory try { output = await window.electron.invoke('execute-ssh-with-key', { - command: `ls -tr ${shellEscape(`${execution.remote.workingDirectory}/nodes-exec-status/${jobIdInternal}`)}`, + command: `ls -tr ${shellEscape(`${statusDirectory}/nodes-exec-status/${jobIdInternal}`)}`, rejectOnNonZeroCode: true, }) } catch {