From 52754964668423d328815dc74ae9a46dc7dede0a Mon Sep 17 00:00:00 2001 From: Alexey Volkov Date: Fri, 24 Jul 2026 01:11:29 -0400 Subject: [PATCH 1/2] feat: Conditional execution Adds support for the conditional execution feature (`TaskSpec.isEnabled`). When user selects a Task node, the Config tab in right sidebar allows the user to configure "Enable task": The user can set it to "True" (default) "False" or "Conditional". * True: Unset TaskSpec.isEnabled * False: Set TaskSpec.isEnabled to "false" * Conditional: Adds a new virtual input named "Is enabled?" to the tasks's node visual representation. The user can connect something (task output, graph input) to that input just like with normal inputs. Internally, the connected argument is assigned to the TaskSpec.isEnabled attribute (similar to TaskSpec.arguments["..."]). --- .../__tests__/actions/createSubgraph.test.ts | 6 +- .../__tests__/actions/unpackSubgraph.test.ts | 2 +- .../entities/componentSpecProxy.test.ts | 4 +- .../__tests__/entities/task.test.ts | 6 +- .../__tests__/integration/roundtrip.test.ts | 52 +++++++++++ .../serialization/jsonSerializer.test.ts | 87 ++++++++++++++++++- .../serialization/yamlDeserializer.test.ts | 54 ++++++++++-- .../componentSpec/actions/createSubgraph.ts | 4 +- src/models/componentSpec/entities/task.ts | 5 +- src/models/componentSpec/entities/types.ts | 1 - .../serialization/jsonSerializer.ts | 80 ++++++++++++----- .../serialization/yamlDeserializer.ts | 40 ++++++++- .../components/ConfigurationSection.tsx | 67 ++++++++++++++ .../components/taskConfig.actions.ts | 34 +++++++- .../components/useTaskConfigActions.ts | 2 + .../v2/shared/nodes/TaskNode/TaskNode.tsx | 27 +++++- .../v2/shared/nodes/TaskNode/TaskNodeCard.tsx | 4 +- src/utils/annotations.ts | 1 + src/utils/componentSpec.ts | 5 +- src/utils/conditionalExecution.ts | 40 +++++++++ 20 files changed, 471 insertions(+), 50 deletions(-) create mode 100644 src/utils/conditionalExecution.ts diff --git a/src/models/componentSpec/__tests__/actions/createSubgraph.test.ts b/src/models/componentSpec/__tests__/actions/createSubgraph.test.ts index a0baa0e2d3..cd4685bbb4 100644 --- a/src/models/componentSpec/__tests__/actions/createSubgraph.test.ts +++ b/src/models/componentSpec/__tests__/actions/createSubgraph.test.ts @@ -189,7 +189,7 @@ describe("createSubgraph", () => { $id: idGen.next("task"), name: "ConfiguredTask", componentRef: { name: "MyComponent" }, - isEnabled: { "==": { op1: "a", op2: "b" } }, + isEnabled: { taskOutput: { taskId: "task1", outputName: "out1" } }, }); task.annotations.add({ key: "note", value: "test" }); spec.addTask(task); @@ -205,7 +205,9 @@ describe("createSubgraph", () => { const movedTask = result!.subgraphSpec.tasks.at(0); expect(movedTask?.name).toBe("ConfiguredTask"); expect(movedTask?.componentRef).toEqual({ name: "MyComponent" }); - expect(movedTask?.isEnabled).toEqual({ "==": { op1: "a", op2: "b" } }); + expect(movedTask?.isEnabled).toEqual({ + taskOutput: { taskId: "task1", outputName: "out1" }, + }); expect(movedTask?.annotations.length).toBe(1); }); diff --git a/src/models/componentSpec/__tests__/actions/unpackSubgraph.test.ts b/src/models/componentSpec/__tests__/actions/unpackSubgraph.test.ts index 2350de8220..c334857329 100644 --- a/src/models/componentSpec/__tests__/actions/unpackSubgraph.test.ts +++ b/src/models/componentSpec/__tests__/actions/unpackSubgraph.test.ts @@ -319,7 +319,7 @@ describe("unpackSubgraph roundtrip", () => { $id: idGen.next("spec"), name: "Main", }); - const predicate = { "==": { op1: "a", op2: "b" } }; + const predicate = { taskOutput: { taskId: "task1", outputName: "out1" } }; const task = makeTask(idGen, "ConditionalTask", { isEnabled: predicate, }); diff --git a/src/models/componentSpec/__tests__/entities/componentSpecProxy.test.ts b/src/models/componentSpec/__tests__/entities/componentSpecProxy.test.ts index 4a999de875..b81aca4117 100644 --- a/src/models/componentSpec/__tests__/entities/componentSpecProxy.test.ts +++ b/src/models/componentSpec/__tests__/entities/componentSpecProxy.test.ts @@ -296,14 +296,14 @@ describe("createComponentSpecProxy", () => { $id: "task_1", name: "T", componentRef: {}, - isEnabled: { "==": { op1: "a", op2: "b" } }, + isEnabled: { taskOutput: { taskId: "task1", outputName: "out1" } }, }), ); const graph = getGraph(createComponentSpecProxy(spec)); expect(graph.tasks["T"].isEnabled).toEqual({ - "==": { op1: "a", op2: "b" }, + taskOutput: { taskId: "task1", outputName: "out1" }, }); }); diff --git a/src/models/componentSpec/__tests__/entities/task.test.ts b/src/models/componentSpec/__tests__/entities/task.test.ts index 183a5067d8..3bef9c5936 100644 --- a/src/models/componentSpec/__tests__/entities/task.test.ts +++ b/src/models/componentSpec/__tests__/entities/task.test.ts @@ -58,10 +58,12 @@ describe("Task", () => { $id: "task_1", name: "T", componentRef: {}, - isEnabled: { "==": { op1: "a", op2: "b" } }, + isEnabled: { taskOutput: { taskId: "task1", outputName: "out1" } }, }); - expect(task.isEnabled).toEqual({ "==": { op1: "a", op2: "b" } }); + expect(task.isEnabled).toEqual({ + taskOutput: { taskId: "task1", outputName: "out1" }, + }); }); it("setName updates name", () => { diff --git a/src/models/componentSpec/__tests__/integration/roundtrip.test.ts b/src/models/componentSpec/__tests__/integration/roundtrip.test.ts index 3963917956..2cc1b0ec60 100644 --- a/src/models/componentSpec/__tests__/integration/roundtrip.test.ts +++ b/src/models/componentSpec/__tests__/integration/roundtrip.test.ts @@ -206,6 +206,58 @@ describe("Serialization Roundtrip", () => { }); }); + it("preserves a conditional (reference) isEnabled across a roundtrip", () => { + const yaml = { + name: "ConditionalTest", + inputs: [{ name: "run_it", type: "Boolean" }], + implementation: { + graph: { + tasks: { + Producer: { componentRef: {} }, + Consumer: { + componentRef: {}, + isEnabled: { + taskOutput: { taskId: "Producer", outputName: "flag" }, + }, + }, + InputGated: { + componentRef: {}, + isEnabled: { graphInput: { inputName: "run_it" } }, + }, + }, + }, + }, + }; + + const json = serializer.serialize(deserializer.deserialize(yaml)); + const tasks = getGraph(json).tasks; + + expect(tasks["Consumer"].isEnabled).toEqual({ + taskOutput: { taskId: "Producer", outputName: "flag" }, + }); + expect(tasks["Consumer"].arguments).toBeUndefined(); + expect(tasks["InputGated"].isEnabled).toEqual({ + graphInput: { inputName: "run_it" }, + }); + }); + + it("preserves literal isEnabled 'false' across a roundtrip", () => { + const yaml = { + name: "DisabledTest", + implementation: { + graph: { + tasks: { + Off: { componentRef: {}, isEnabled: "false" }, + }, + }, + }, + }; + + const json = serializer.serialize(deserializer.deserialize(yaml)); + + expect(getGraph(json).tasks["Off"].isEnabled).toBe("false"); + }); + it("handles empty spec correctly", () => { const yaml = { name: "EmptySpec", diff --git a/src/models/componentSpec/__tests__/serialization/jsonSerializer.test.ts b/src/models/componentSpec/__tests__/serialization/jsonSerializer.test.ts index 5c00a3a2b3..7cc36379c4 100644 --- a/src/models/componentSpec/__tests__/serialization/jsonSerializer.test.ts +++ b/src/models/componentSpec/__tests__/serialization/jsonSerializer.test.ts @@ -1,5 +1,7 @@ import { beforeEach, describe, expect, it } from "vitest"; +import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution"; + import { Binding } from "../../entities/binding"; import { ComponentSpec } from "../../entities/componentSpec"; import { Input } from "../../entities/input"; @@ -82,15 +84,96 @@ describe("JsonSerializer", () => { $id: idGen.next("task"), name: "Process", componentRef: {}, - isEnabled: { "==": { op1: "a", op2: "b" } }, + isEnabled: { taskOutput: { taskId: "task1", outputName: "out1" } }, }); spec.addTask(task); const json = serializer.serialize(spec); expect(getGraph(json).tasks["Process"].isEnabled).toEqual({ - "==": { op1: "a", op2: "b" }, + taskOutput: { taskId: "task1", outputName: "out1" }, + }); + }); + + it("serializes literal isEnabled 'false'", () => { + const spec = new ComponentSpec({ + $id: idGen.next("spec"), + name: "Pipeline", + }); + const task = new Task({ + $id: idGen.next("task"), + name: "Process", + componentRef: {}, + isEnabled: "false", + }); + spec.addTask(task); + + expect( + getGraph(serializer.serialize(spec)).tasks["Process"].isEnabled, + ).toBe("false"); + }); + + it("serializes a connection to the reserved is-enabled port as isEnabled (task output)", () => { + const spec = new ComponentSpec({ + $id: idGen.next("spec"), + name: "Pipeline", + }); + const producer = new Task({ + $id: idGen.next("task"), + name: "Producer", + componentRef: {}, }); + const consumer = new Task({ + $id: idGen.next("task"), + name: "Consumer", + componentRef: {}, + }); + spec.addTask(producer); + spec.addTask(consumer); + spec.addBinding( + new Binding({ + $id: idGen.next("binding"), + sourceEntityId: producer.$id, + sourcePortName: "should_run", + targetEntityId: consumer.$id, + targetPortName: IS_ENABLED_PORT_NAME, + }), + ); + + const consumerSpec = getGraph(serializer.serialize(spec)).tasks["Consumer"]; + expect(consumerSpec.isEnabled).toEqual({ + taskOutput: { taskId: "Producer", outputName: "should_run" }, + }); + // The reserved-port binding must not leak into arguments. + expect(consumerSpec.arguments).toBeUndefined(); + }); + + it("serializes a graph-input connection to the reserved is-enabled port", () => { + const spec = new ComponentSpec({ + $id: idGen.next("spec"), + name: "Pipeline", + }); + const input = new Input({ $id: idGen.next("input"), name: "run_it" }); + const task = new Task({ + $id: idGen.next("task"), + name: "Consumer", + componentRef: {}, + }); + spec.addInput(input); + spec.addTask(task); + spec.addBinding( + new Binding({ + $id: idGen.next("binding"), + sourceEntityId: input.$id, + sourcePortName: "run_it", + targetEntityId: task.$id, + targetPortName: IS_ENABLED_PORT_NAME, + }), + ); + + expect( + getGraph(serializer.serialize(spec)).tasks["Consumer"].isEnabled, + ).toEqual({ graphInput: { inputName: "run_it" } }); }); it("serializes metadata from annotations", () => { diff --git a/src/models/componentSpec/__tests__/serialization/yamlDeserializer.test.ts b/src/models/componentSpec/__tests__/serialization/yamlDeserializer.test.ts index 99d95d783c..e9670182cf 100644 --- a/src/models/componentSpec/__tests__/serialization/yamlDeserializer.test.ts +++ b/src/models/componentSpec/__tests__/serialization/yamlDeserializer.test.ts @@ -1,5 +1,8 @@ import { beforeEach, describe, expect, it } from "vitest"; +import { EDITOR_CONDITIONAL_EXECUTION_ANNOTATION } from "@/utils/annotations"; +import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution"; + import { IncrementingIdGenerator } from "../../factories/idGenerator"; import { YamlDeserializer } from "../../serialization/yamlDeserializer"; @@ -102,15 +105,18 @@ describe("YamlDeserializer", () => { expect(spec.tasks.at(0)?.componentRef).toEqual({ name: "Processor" }); }); - it("deserializes task with isEnabled", () => { + it("deserializes a reference isEnabled into a reserved-port binding", () => { const yaml = { name: "Pipeline", implementation: { graph: { tasks: { - ConditionalTask: { + Producer: { componentRef: {} }, + Consumer: { componentRef: {}, - isEnabled: { "==": { op1: "a", op2: "b" } }, + isEnabled: { + taskOutput: { taskId: "Producer", outputName: "flag" }, + }, }, }, }, @@ -118,10 +124,46 @@ describe("YamlDeserializer", () => { }; const spec = deserializer.deserialize(yaml); + const consumer = spec.tasks.find((t) => t.name === "Consumer"); + const producer = spec.tasks.find((t) => t.name === "Producer"); + + // Conditional mode: entity value cleared, mode annotation set. + expect(consumer?.isEnabled).toBeUndefined(); + expect( + consumer?.annotations.get(EDITOR_CONDITIONAL_EXECUTION_ANNOTATION), + ).toBe("true"); + + // A binding to the reserved port drives the connection. + const binding = spec.bindings.find( + (b) => + b.targetEntityId === consumer?.$id && + b.targetPortName === IS_ENABLED_PORT_NAME, + ); + expect(binding).toBeDefined(); + expect(binding?.sourceEntityId).toBe(producer?.$id); + expect(binding?.sourcePortName).toBe("flag"); + }); - expect(spec.tasks.at(0)?.isEnabled).toEqual({ - "==": { op1: "a", op2: "b" }, - }); + it("keeps literal isEnabled 'false' without a binding", () => { + const yaml = { + name: "Pipeline", + implementation: { + graph: { + tasks: { + T: { componentRef: {}, isEnabled: "false" }, + }, + }, + }, + }; + + const spec = deserializer.deserialize(yaml); + const task = spec.tasks.at(0); + + expect(task?.isEnabled).toBe("false"); + expect( + task?.annotations.get(EDITOR_CONDITIONAL_EXECUTION_ANNOTATION), + ).toBeUndefined(); + expect(spec.bindings.length).toBe(0); }); it("deserializes task annotations", () => { diff --git a/src/models/componentSpec/actions/createSubgraph.ts b/src/models/componentSpec/actions/createSubgraph.ts index 656b42e030..00176c9084 100644 --- a/src/models/componentSpec/actions/createSubgraph.ts +++ b/src/models/componentSpec/actions/createSubgraph.ts @@ -9,9 +9,9 @@ import { Task } from "../entities/task"; import type { Annotation, Argument, + ArgumentType, ComponentReference, ExecutionOptionsSpec, - PredicateType, } from "../entities/types"; import type { IdGenerator } from "../factories/idGenerator"; @@ -31,7 +31,7 @@ interface TaskSnapshot { $id: string; name: string; componentRef: ComponentReference; - isEnabled?: PredicateType; + isEnabled?: ArgumentType; annotations: Annotation[]; arguments: Argument[]; executionOptions?: ExecutionOptionsSpec; diff --git a/src/models/componentSpec/entities/task.ts b/src/models/componentSpec/entities/task.ts index 694872f195..652a11e87f 100644 --- a/src/models/componentSpec/entities/task.ts +++ b/src/models/componentSpec/entities/task.ts @@ -12,7 +12,6 @@ import type { ComponentReference, ComponentSpecJson, ExecutionOptionsSpec, - PredicateType, } from "./types"; import { isGraphImplementation } from "./types"; @@ -22,7 +21,7 @@ export class Task extends Model({ name: prop(), componentRef: prop(), subgraphSpec: prop(undefined), - isEnabled: prop(undefined), + isEnabled: prop(undefined), annotations: prop(() => new Annotations({})), arguments: prop(() => []), executionOptions: prop(undefined), @@ -78,7 +77,7 @@ export class Task extends Model({ } @modelAction - setIsEnabled(predicate: PredicateType | undefined) { + setIsEnabled(predicate: ArgumentType | undefined) { this.isEnabled = predicate; } diff --git a/src/models/componentSpec/entities/types.ts b/src/models/componentSpec/entities/types.ts index 65b3b2b8d4..6b392c9fe9 100644 --- a/src/models/componentSpec/entities/types.ts +++ b/src/models/componentSpec/entities/types.ts @@ -11,7 +11,6 @@ export type { InputSpec, MetadataSpec, OutputSpec, - PredicateType, TaskOutputArgument, TaskSpec, TypeSpecType, diff --git a/src/models/componentSpec/serialization/jsonSerializer.ts b/src/models/componentSpec/serialization/jsonSerializer.ts index 1ed909e0a5..2a8990e71d 100644 --- a/src/models/componentSpec/serialization/jsonSerializer.ts +++ b/src/models/componentSpec/serialization/jsonSerializer.ts @@ -1,3 +1,5 @@ +import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution"; + import type { Annotations } from "../annotations"; import { serializeAnnotationValue } from "../annotations"; import type { Binding } from "../entities/binding"; @@ -83,7 +85,20 @@ export class JsonSerializer { const taskBindings = spec.bindings.filter( (b) => b.targetEntityId === task.$id, ); - const args = this.serializeArguments(task.arguments, taskBindings, spec); + + // A connection to the reserved "Is enabled?" port is serialized to + // `isEnabled` rather than to `arguments`. + const conditionalBinding = taskBindings.find( + (b) => b.targetPortName === IS_ENABLED_PORT_NAME, + ); + const argumentBindings = taskBindings.filter( + (b) => b !== conditionalBinding, + ); + const args = this.serializeArguments( + task.arguments, + argumentBindings, + spec, + ); const componentRef = task.subgraphSpec ? { ...task.componentRef, spec: this.serialize(task.subgraphSpec) } @@ -97,7 +112,12 @@ export class JsonSerializer { result.arguments = args; } - if (task.isEnabled) { + const conditionalArgument = conditionalBinding + ? this.bindingToArgument(conditionalBinding, spec) + : undefined; + if (conditionalArgument !== undefined) { + result.isEnabled = conditionalArgument; + } else if (task.isEnabled !== undefined) { result.isEnabled = task.isEnabled; } @@ -121,26 +141,9 @@ export class JsonSerializer { const result: Record = {}; for (const binding of bindings) { - const sourceTask = spec.tasks.find( - (t) => t.$id === binding.sourceEntityId, - ); - const sourceInput = spec.inputs.find( - (i) => i.$id === binding.sourceEntityId, - ); - - if (sourceTask) { - result[binding.targetPortName] = { - taskOutput: { - taskId: sourceTask.name, - outputName: binding.sourcePortName, - }, - }; - } else if (sourceInput) { - result[binding.targetPortName] = { - graphInput: { - inputName: sourceInput.name, - }, - }; + const argument = this.bindingToArgument(binding, spec); + if (argument !== undefined) { + result[binding.targetPortName] = argument; } } @@ -153,6 +156,39 @@ export class JsonSerializer { return result; } + /** + * Resolve a binding's source into the argument reference it serializes to: + * a task output or a graph input. Returns undefined when the source entity + * cannot be resolved. + */ + private bindingToArgument( + binding: Binding, + spec: ComponentSpec, + ): ArgumentType | undefined { + const sourceTask = spec.tasks.find((t) => t.$id === binding.sourceEntityId); + if (sourceTask) { + return { + taskOutput: { + taskId: sourceTask.name, + outputName: binding.sourcePortName, + }, + }; + } + + const sourceInput = spec.inputs.find( + (i) => i.$id === binding.sourceEntityId, + ); + if (sourceInput) { + return { + graphInput: { + inputName: sourceInput.name, + }, + }; + } + + return undefined; + } + private serializeInput(input: Input): InputSpec { const result: InputSpec = { name: input.name, diff --git a/src/models/componentSpec/serialization/yamlDeserializer.ts b/src/models/componentSpec/serialization/yamlDeserializer.ts index 3c4a4a1c71..52d5778279 100644 --- a/src/models/componentSpec/serialization/yamlDeserializer.ts +++ b/src/models/componentSpec/serialization/yamlDeserializer.ts @@ -1,3 +1,9 @@ +import { + EDITOR_CONDITIONAL_EXECUTION_ANNOTATION, + IS_ENABLED_PORT_NAME, + isConditionalArgument, +} from "@/utils/conditionalExecution"; + import { Annotations, deserializeAnnotationValue } from "../annotations"; import { Binding } from "../entities/binding"; import { ComponentSpec } from "../entities/componentSpec"; @@ -123,6 +129,22 @@ export class YamlDeserializer { } } + // A reference-valued `isEnabled` is the "Conditional" mode: it becomes a + // binding to the reserved port (see buildBindings) and the entity keeps + // `isEnabled` empty. Literal values (e.g. "false") stay on the entity. + const conditionalEnabled = isConditionalArgument(taskJson.isEnabled); + if ( + conditionalEnabled && + !annotationItems.some( + (a) => a.key === EDITOR_CONDITIONAL_EXECUTION_ANNOTATION, + ) + ) { + annotationItems.push({ + key: EDITOR_CONDITIONAL_EXECUTION_ANNOTATION, + value: "true", + }); + } + const args: Argument[] = []; if (taskJson.arguments) { for (const [argName, argValue] of Object.entries(taskJson.arguments)) { @@ -149,7 +171,7 @@ export class YamlDeserializer { name: taskName, componentRef, subgraphSpec, - isEnabled: taskJson.isEnabled, + isEnabled: conditionalEnabled ? undefined : taskJson.isEnabled, executionOptions: taskJson.executionOptions, annotations: Annotations.from(annotationItems), arguments: args, @@ -178,7 +200,21 @@ export class YamlDeserializer { for (const [taskName, taskJson] of Object.entries(graph.tasks)) { const targetTask = tasks.find((t) => t.name === taskName); - if (!targetTask || !taskJson.arguments) continue; + if (!targetTask) continue; + + // A reference-valued `isEnabled` connects to the reserved port. + if (isConditionalArgument(taskJson.isEnabled)) { + const binding = this.createBindingFromArgument( + inputs, + tasks, + targetTask.$id, + IS_ENABLED_PORT_NAME, + taskJson.isEnabled as ArgumentType, + ); + if (binding) bindings.push(binding); + } + + if (!taskJson.arguments) continue; for (const [argName, argValue] of Object.entries(taskJson.arguments)) { const binding = this.createBindingFromArgument( diff --git a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx index e3c8a9ecc9..8492396fa5 100644 --- a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx +++ b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx @@ -10,18 +10,29 @@ import { } from "@/components/shared/ReactFlow/FlowCanvas/TaskNode/AnnotationsEditor/utils"; import { ColorPicker } from "@/components/ui/color"; import { BlockStack, InlineStack } from "@/components/ui/layout"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; import { Separator } from "@/components/ui/separator"; import { Switch } from "@/components/ui/switch"; import { Heading, Paragraph } from "@/components/ui/typography"; import type { Task } from "@/models/componentSpec"; import { useAnalytics } from "@/providers/AnalyticsProvider"; +import { useSpec } from "@/routes/v2/shared/providers/SpecContext"; import type { AnnotationConfig, Annotations } from "@/types/annotations"; import { EDITOR_COLLAPSED_ANNOTATION, + EDITOR_CONDITIONAL_EXECUTION_ANNOTATION, TASK_COLOR_ANNOTATION, } from "@/utils/annotations"; +import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution"; import { ISO8601_DURATION_ZERO_DAYS } from "@/utils/constants"; +import type { EnableTaskMode } from "./taskConfig.actions"; import { useTaskConfigActions } from "./useTaskConfigActions"; interface ConfigurationSectionProps { @@ -32,15 +43,39 @@ export const ConfigurationSection = observer(function ConfigurationSection({ task, }: ConfigurationSectionProps) { const { track } = useAnalytics(); + const spec = useSpec(); const { toggleCacheDisable, saveAnnotation, setTaskColor, clearProviderAnnotations, setCollapsed, + setEnableTaskMode, } = useTaskConfigActions(); const isSubgraph = task.subgraphSpec !== undefined; + const isConditionalConnected = + spec?.bindings.some( + (b) => + b.targetEntityId === task.$id && + b.targetPortName === IS_ENABLED_PORT_NAME, + ) ?? false; + const isConditional = + task.annotations.get(EDITOR_CONDITIONAL_EXECUTION_ANNOTATION) === "true" || + isConditionalConnected; + const enableMode: EnableTaskMode = isConditional + ? "conditional" + : task.isEnabled === "false" + ? "false" + : "true"; + + const handleEnableModeChange = (value: string) => { + if (!spec) return; + const mode = value as EnableTaskMode; + setEnableTaskMode(spec, task, mode); + track("v2.pipeline_editor.task_details.enable_task.change", { mode }); + }; + const cacheDisabled = task.executionOptions?.cachingStrategy?.maxCacheStaleness === ISO8601_DURATION_ZERO_DAYS; @@ -155,6 +190,38 @@ export const ConfigurationSection = observer(function ConfigurationSection({ /> + + + + + + Enable task + + + + {isConditional && !isConditionalConnected && ( + + Connect a task output or pipeline input to the “Is enabled?” port on + the node. + + )} + + {!isSubgraph && ( <> diff --git a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/taskConfig.actions.ts b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/taskConfig.actions.ts index 3957b47f8a..1bd30028f9 100644 --- a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/taskConfig.actions.ts +++ b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/taskConfig.actions.ts @@ -1,12 +1,17 @@ -import type { Task } from "@/models/componentSpec"; +import type { ComponentSpec, Task } from "@/models/componentSpec"; import type { UndoGroupable } from "@/routes/v2/shared/nodes/types"; import type { AnnotationConfig } from "@/types/annotations"; import { EDITOR_COLLAPSED_ANNOTATION, + EDITOR_CONDITIONAL_EXECUTION_ANNOTATION, TASK_COLOR_ANNOTATION, } from "@/utils/annotations"; +import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution"; import { ISO8601_DURATION_ZERO_DAYS } from "@/utils/constants"; +/** The three "Enable task" choices exposed in the Config tab. */ +export type EnableTaskMode = "true" | "false" | "conditional"; + export function toggleCacheDisable( undo: UndoGroupable, task: Task, @@ -56,6 +61,33 @@ export function setCollapsed( }); } +export function setEnableTaskMode( + undo: UndoGroupable, + spec: ComponentSpec, + task: Task, + mode: EnableTaskMode, +) { + undo.withGroup("Set enable task", () => { + if (mode === "conditional") { + // The connection is modelled as a binding the user draws to the virtual + // "Is enabled?" port; here we just enter conditional mode so the port + // shows. `isEnabled` is derived from that binding at serialize time. + task.annotations.set(EDITOR_CONDITIONAL_EXECUTION_ANNOTATION, "true"); + task.setIsEnabled(undefined); + return; + } + + // Leaving conditional mode: drop any connection to the reserved port. + spec.removeAllBindingsBy( + (b) => + b.targetEntityId === task.$id && + b.targetPortName === IS_ENABLED_PORT_NAME, + ); + task.annotations.remove(EDITOR_CONDITIONAL_EXECUTION_ANNOTATION); + task.setIsEnabled(mode === "false" ? "false" : undefined); + }); +} + export function clearProviderAnnotations( undo: UndoGroupable, task: Task, diff --git a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/useTaskConfigActions.ts b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/useTaskConfigActions.ts index 225365b591..5ad5d56637 100644 --- a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/useTaskConfigActions.ts +++ b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/useTaskConfigActions.ts @@ -4,6 +4,7 @@ import { clearProviderAnnotations, saveAnnotation, setCollapsed, + setEnableTaskMode, setTaskColor, toggleCacheDisable, } from "./taskConfig.actions"; @@ -16,6 +17,7 @@ export function useTaskConfigActions() { saveAnnotation: saveAnnotation.bind(null, undo), setTaskColor: setTaskColor.bind(null, undo), setCollapsed: setCollapsed.bind(null, undo), + setEnableTaskMode: setEnableTaskMode.bind(null, undo), clearProviderAnnotations: clearProviderAnnotations.bind(null, undo), }; } diff --git a/src/routes/v2/shared/nodes/TaskNode/TaskNode.tsx b/src/routes/v2/shared/nodes/TaskNode/TaskNode.tsx index 5da5a46ea9..bfadd71c9b 100644 --- a/src/routes/v2/shared/nodes/TaskNode/TaskNode.tsx +++ b/src/routes/v2/shared/nodes/TaskNode/TaskNode.tsx @@ -23,10 +23,15 @@ import { useSharedStores } from "@/routes/v2/shared/store/SharedStoreContext"; import { AggregatorOutputType } from "@/types/aggregator"; import { EDITOR_COLLAPSED_ANNOTATION, + EDITOR_CONDITIONAL_EXECUTION_ANNOTATION, isPipelineAggregator, TASK_COLOR_ANNOTATION, } from "@/utils/annotations"; import { isSecretArgument } from "@/utils/componentSpec"; +import { + IS_ENABLED_INPUT_LABEL, + IS_ENABLED_PORT_NAME, +} from "@/utils/conditionalExecution"; import { ISO8601_DURATION_ZERO_DAYS } from "@/utils/constants"; import type { ExecutionStatusStats } from "@/utils/executionStatus"; @@ -44,6 +49,8 @@ export interface TaskNodeInput { type?: TypeSpecType; optional?: boolean; default?: string; + /** Human-readable label; falls back to the (prettified) name when unset. */ + label?: string; } export interface TaskNodeOutput { @@ -299,6 +306,24 @@ export const TaskNode = observer(function TaskNode({ const connectedPorts = resolveConnectedPortNames(entityId, spec); const inputDisplayData = resolveInputDisplayData(task, entityId, spec); + // In "Conditional" execution mode a virtual "Is enabled?" input is exposed so + // the user can connect an upstream value that gates the task. The connection + // itself is an ordinary binding to the reserved port, so its display value and + // connected state are already resolved above. + const isConditionalExecution = + task.annotations.get(EDITOR_CONDITIONAL_EXECUTION_ANNOTATION) === "true" || + connectedPorts.inputs.has(IS_ENABLED_PORT_NAME); + const displayInputs: TaskNodeInput[] = isConditionalExecution + ? [ + ...inputs, + { + name: IS_ENABLED_PORT_NAME, + label: IS_ENABLED_INPUT_LABEL, + type: "Boolean", + }, + ] + : inputs; + const isSelected = isEditorVisualNodeSelected(editor, id, !!selected); const handleOutputTypeChange = (value: AggregatorOutputType) => { @@ -314,7 +339,7 @@ export const TaskNode = observer(function TaskNode({ isSubgraph: isTaskSubgraph(componentSpec), collapsed: isManuallyCollapsed, description, - inputs, + inputs: displayInputs, outputs, connectedInputNames: connectedPorts.inputs, connectedOutputNames: connectedPorts.outputs, diff --git a/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx b/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx index be5171134b..9d0cfd6a56 100644 --- a/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx +++ b/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx @@ -168,12 +168,12 @@ const ClassicInputHandle = observer(function ClassicInputHandle({ ? "text-gray-100 bg-white/10 hover:bg-white/15 dark:hover:bg-white/15 hover:text-gray-100" : "text-gray-800 bg-black/5 hover:bg-black/10 dark:hover:bg-black/10 hover:text-gray-800", )} - title={`${input.name}${input.type ? `: ${input.type}` : ""}`} + title={`${input.label ?? input.name}${input.type ? `: ${input.type}` : ""}`} onClick={onLabelClick} data-input-control data-testid={`input-label-${input.name}`} > - {input.name.replace(/_/g, " ")} + {input.label ?? input.name.replace(/_/g, " ")} {showValueDisplay && ( diff --git a/src/utils/annotations.ts b/src/utils/annotations.ts index ab427e1fd1..d79637fce3 100644 --- a/src/utils/annotations.ts +++ b/src/utils/annotations.ts @@ -19,6 +19,7 @@ import { import type { ComponentSpec } from "./componentSpec"; export * from "./annotationKeys"; +export { EDITOR_CONDITIONAL_EXECUTION_ANNOTATION } from "./conditionalExecution"; export const DISPLAY_NAME_MAX_LENGTH = 100; const PIPELINE_AGGREGATOR_ANNOTATION = "is_input_aggregator"; diff --git a/src/utils/componentSpec.ts b/src/utils/componentSpec.ts index 47ec053f0a..d222cbcc34 100644 --- a/src/utils/componentSpec.ts +++ b/src/utils/componentSpec.ts @@ -409,6 +409,9 @@ export type PredicateType = not: PredicateType; }; +/** + * Optional configuration that specifies how the task should be retried if it fails. + */ interface RetryStrategySpec { maxRetries?: number; } @@ -430,7 +433,7 @@ export interface TaskSpec { arguments?: { [k: string]: ArgumentType; }; - isEnabled?: PredicateType; + isEnabled?: ArgumentType; executionOptions?: ExecutionOptionsSpec; annotations?: { [k: string]: unknown; diff --git a/src/utils/conditionalExecution.ts b/src/utils/conditionalExecution.ts new file mode 100644 index 0000000000..52eb3b4f4c --- /dev/null +++ b/src/utils/conditionalExecution.ts @@ -0,0 +1,40 @@ +import type { ArgumentType } from "./componentSpec"; +import { isGraphInputArgument, isTaskOutputArgument } from "./componentSpec"; + +/** + * Reserved binding port name used to model the virtual "Is enabled?" input on a + * task node. A connection to this port is serialized to `TaskSpec.isEnabled` + * instead of `TaskSpec.arguments[...]`. The sentinel is intentionally unlikely + * to collide with a real component input name. + */ +export const IS_ENABLED_PORT_NAME = "__is_enabled__"; + +/** Human-readable label shown for the virtual "Is enabled?" input. */ +export const IS_ENABLED_INPUT_LABEL = "Is enabled?"; + +/** + * Annotation key marking that a task is in "Conditional" enable mode. Lives here + * (rather than in the UI-oriented `@/utils/annotations` module) so the model + * serializers can reference it without pulling the React component tree into the + * `@/models/componentSpec` barrel's module graph. + */ +export const EDITOR_CONDITIONAL_EXECUTION_ANNOTATION = + "editor.conditional-execution"; + +const GRAPH_INPUT_REGEX = /^\{\{inputs\.([^}]+)\}\}$/; +const TASK_OUTPUT_REGEX = /^\{\{tasks\.([^.]+)\.outputs\.([^}]+)\}\}$/; + +/** + * True when an `isEnabled` value is a reference to an upstream value (a graph + * input or a sibling task output) — i.e. the "Conditional" mode — rather than a + * plain literal such as `"false"`. + */ +export function isConditionalArgument( + value: ArgumentType | undefined, +): boolean { + if (value === undefined) return false; + if (typeof value === "string") { + return GRAPH_INPUT_REGEX.test(value) || TASK_OUTPUT_REGEX.test(value); + } + return isGraphInputArgument(value) || isTaskOutputArgument(value); +} From 9c91bfcb518f8a6ce28b734442aede057d4455a0 Mon Sep 17 00:00:00 2001 From: Alexey Volkov Date: Thu, 6 Aug 2026 19:32:27 -0700 Subject: [PATCH 2/2] chore: Conditional execution - Put the new configuration UI behind a feature flag --- src/flags.ts | 8 ++ .../utils/comparePipelines.test.ts | 12 ++- .../components/ConfigurationSection.tsx | 66 ++++++++------- .../v2/shared/nodes/TaskNode/TaskNodeCard.tsx | 83 ++++++++++++------- src/utils/componentSpec.ts | 2 +- 5 files changed, 105 insertions(+), 66 deletions(-) diff --git a/src/flags.ts b/src/flags.ts index 794f9afd42..78ec911b3f 100644 --- a/src/flags.ts +++ b/src/flags.ts @@ -86,4 +86,12 @@ export const ExistingFlags: ConfigFlags = { default: false, category: "beta", }, + + ["conditional-execution"]: { + name: "Conditional Task Execution", + description: + 'Enable the "Enable task" setting in the task Config tab. Lets you disable a task or gate it on an upstream value connected to its "Is enabled?" port.', + default: true, //temp for demo purposes + category: "beta", + }, }; diff --git a/src/routes/v2/pages/CompareView/utils/comparePipelines.test.ts b/src/routes/v2/pages/CompareView/utils/comparePipelines.test.ts index f5c6a359d2..3d048cee1e 100644 --- a/src/routes/v2/pages/CompareView/utils/comparePipelines.test.ts +++ b/src/routes/v2/pages/CompareView/utils/comparePipelines.test.ts @@ -5,8 +5,8 @@ import type { ArtifactNodeResponse, } from "@/api/types.gen"; import type { + ArgumentType, ComponentSpec, - PredicateType, TaskSpec, } from "@/utils/componentSpec"; @@ -31,7 +31,9 @@ const containerSpec = (): ComponentSpec => ({ const noStatus = new Map(); -const enabledWhen: PredicateType = { "==": { op1: "mode", op2: "train" } }; +const enabledByFlag: ArgumentType = { + taskOutput: { taskId: "prep", outputName: "should_train" }, +}; const side = ( spec: ComponentSpec | undefined, @@ -209,7 +211,9 @@ describe("buildPipelineComparison()", () => { }); test("flags a task guarded by a condition in one run only", () => { - const specA = graphSpec({ train: task("d1", { isEnabled: enabledWhen }) }); + const specA = graphSpec({ + train: task("d1", { isEnabled: enabledByFlag }), + }); const specB = graphSpec({ train: task("d1") }); const [diff] = buildPipelineComparison(side(specA), side(specB)).taskDiffs; @@ -222,7 +226,7 @@ describe("buildPipelineComparison()", () => { test("reports no setting differences when execution options match", () => { const options: Partial = { - isEnabled: enabledWhen, + isEnabled: enabledByFlag, executionOptions: { retryStrategy: { maxRetries: 2 } }, }; const specA = graphSpec({ train: task("d1", options) }); diff --git a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx index 8492396fa5..d4ec8cfcfb 100644 --- a/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx +++ b/src/routes/v2/pages/Editor/nodes/TaskNode/context/TaskDetails/components/ConfigurationSection.tsx @@ -8,6 +8,7 @@ import { launcherTaskAnnotationSchema, parseSchemaToAnnotationConfig, } from "@/components/shared/ReactFlow/FlowCanvas/TaskNode/AnnotationsEditor/utils"; +import { useFlagValue } from "@/components/shared/Settings/useFlags"; import { ColorPicker } from "@/components/ui/color"; import { BlockStack, InlineStack } from "@/components/ui/layout"; import { @@ -44,6 +45,7 @@ export const ConfigurationSection = observer(function ConfigurationSection({ }: ConfigurationSectionProps) { const { track } = useAnalytics(); const spec = useSpec(); + const conditionalExecutionEnabled = useFlagValue("conditional-execution"); const { toggleCacheDisable, saveAnnotation, @@ -190,37 +192,41 @@ export const ConfigurationSection = observer(function ConfigurationSection({ /> - + {conditionalExecutionEnabled && ( + <> + - - - - Enable task - - - - {isConditional && !isConditionalConnected && ( - - Connect a task output or pipeline input to the “Is enabled?” port on - the node. - - )} - + + + + Enable task + + + + {isConditional && !isConditionalConnected && ( + + Connect a task output or pipeline input to the “Is enabled?” + port on the node. + + )} + + + )} {!isSubgraph && ( <> diff --git a/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx b/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx index 9d0cfd6a56..e02d2898cb 100644 --- a/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx +++ b/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx @@ -20,6 +20,7 @@ import { cn } from "@/lib/utils"; import { useTheme } from "@/providers/ThemeProvider"; import { useSpec } from "@/routes/v2/shared/providers/SpecContext"; import { AGGREGATOR_ADD_INPUT_HANDLE_ID } from "@/utils/aggregatorInputs"; +import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution"; import { pluralize } from "@/utils/string"; import { deriveColorPalette } from "./color.utils"; @@ -31,6 +32,9 @@ const AGGREGATOR_INTERNAL_INPUTS = new Set([ "output_type", ]); +/** The conditional gate stays visible while the rest of the inputs condense. */ +const isGatePort = (inputName: string) => inputName === IS_ENABLED_PORT_NAME; + const cardVariants = cva( "min-w-[300px] max-w-[350px] rounded-2xl border-2 p-0 drop-shadow-none cursor-pointer select-none gap-2 transition-shadow", { @@ -262,15 +266,23 @@ export const TaskNodeCard = observer(function TaskNodeCard({ ? inputs.filter((input) => !AGGREGATOR_INTERNAL_INPUTS.has(input.name)) : inputs; - const condensedInputs = - filteredInputs.length > 0 && connectedInputNames.size > 0 - ? filteredInputs.filter((input) => connectedInputNames.has(input.name)) - : filteredInputs.slice(0, 1); - const hiddenInputCount = filteredInputs.length - condensedInputs.length; + const gateInputs = filteredInputs.filter((input) => isGatePort(input.name)); + const regularInputs = filteredInputs.filter( + (input) => !isGatePort(input.name), + ); + + const condensedInputs = regularInputs.some((input) => + connectedInputNames.has(input.name), + ) + ? regularInputs.filter((input) => connectedInputNames.has(input.name)) + : regularInputs.slice(0, 1); + const hiddenInputCount = regularInputs.length - condensedInputs.length; const inputsSectionToggles = !isAggregator && collapsed && hiddenInputCount > 0; const showCondensedInputs = inputsSectionToggles && !inputsExpanded; - const visibleInputs = showCondensedInputs ? condensedInputs : filteredInputs; + const visibleInputs = showCondensedInputs + ? [...condensedInputs, ...gateInputs] + : filteredInputs; const showInputsSection = filteredInputs.length > 0 || isAggregator; const openInputProperties = @@ -408,31 +420,40 @@ export const TaskNodeCard = observer(function TaskNodeCard({ > {isAggregator && } - {visibleInputs.map((input, index) => ( - onHandleClick(`input_${input.name}`, e)} - /> - ))} + {visibleInputs.map((input, index) => { + const condensedAway = + showCondensedInputs && !isGatePort(input.name); + const hiddenCountLabel = + condensedAway && index === 0 + ? `+${hiddenInputCount} more ${pluralize(hiddenInputCount, "input")}` + : undefined; + + return ( + + onHandleClick(`input_${input.name}`, e) + } + /> + ); + })} {collapsed && inputsExpanded && hiddenInputCount > 0 && (