From 0af2e3933d8118f1970af50883e1ffe081e0faf4 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 a0baa0e2d..cd4685bbb 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 2350de822..c33485732 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 4a999de87..b81aca411 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 183a5067d..3bef9c593 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 396391795..2cc1b0ec6 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 5c00a3a2b..7cc36379c 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 99d95d783..e9670182c 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 656b42e03..00176c908 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 694872f19..652a11e87 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 65b3b2b8d..6b392c9fe 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 1ed909e0a..2a8990e71 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 3c4a4a1c7..52d577827 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 e3c8a9ecc..8492396fa 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 3957b47f8..1bd30028f 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 225365b59..5ad5d5663 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 689a3b89e..56347cc75 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 { @@ -296,6 +303,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) => { @@ -311,7 +336,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 9d2a5da6e..c26558cc1 100644 --- a/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx +++ b/src/routes/v2/shared/nodes/TaskNode/TaskNodeCard.tsx @@ -168,9 +168,9 @@ const ClassicInputHandle = observer(function ClassicInputHandle({ ? "text-gray-100 bg-white/10 hover:bg-white/15" : "text-gray-800 bg-black/5 hover:bg-black/10", )} - title={`${input.name}${input.type ? `: ${input.type}` : ""}`} + title={`${input.label ?? input.name}${input.type ? `: ${input.type}` : ""}`} > - {input.name.replace(/_/g, " ")} + {input.label ?? input.name.replace(/_/g, " ")} {showValueDisplay && ( diff --git a/src/utils/annotations.ts b/src/utils/annotations.ts index ab427e1fd..d79637fce 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 e1fab8b05..c67b9e9e5 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 000000000..52eb3b4f4 --- /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 55706d942c59883e254f7dfed5a0d53a2115013c 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 +++ .../components/ConfigurationSection.tsx | 66 ++++++++++--------- 2 files changed, 44 insertions(+), 30 deletions(-) diff --git a/src/flags.ts b/src/flags.ts index edf0af06e..fa46d8ce1 100644 --- a/src/flags.ts +++ b/src/flags.ts @@ -78,4 +78,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: false, + category: "beta", + }, }; 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 8492396fa..d4ec8cfcf 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 && ( <>