feat: add conditional execution to the pipeline editor

src/models/componentSpec/__tests__/actions/createSubgraph.test.ts

@@ -1,5 +1,7 @@
 import { beforeEach, describe, expect, it } from "vitest";
 
+import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution";
+
 import { createSubgraph } from "../../actions/createSubgraph";
 import { Binding } from "../../entities/binding";
 import { ComponentSpec } from "../../entities/componentSpec";
@@ -88,6 +90,51 @@ describe("createSubgraph", () => {
     expect(result!.subgraphSpec.inputs.at(0)?.name).toBe("data");
   });
 
+  it("uses a regular subgraph input for an external condition", () => {
+    const spec = new ComponentSpec({
+      $id: idGen.next("spec"),
+      name: "Main",
+    });
+    const condition = new Input({
+      $id: idGen.next("input"),
+      name: "condition",
+    });
+    const task = new Task({
+      $id: idGen.next("task"),
+      name: "InnerTask",
+      componentRef: {},
+      isEnabled: "true",
+    });
+    spec.addInput(condition);
+    spec.addTask(task);
+    spec.addBinding(
+      new Binding({
+        $id: idGen.next("binding"),
+        sourceEntityId: condition.$id,
+        sourcePortName: "condition",
+        targetEntityId: task.$id,
+        targetPortName: IS_ENABLED_PORT_NAME,
+      }),
+    );
+
+    const result = createSubgraph({
+      spec,
+      selectedTaskIds: [task.$id],
+      subgraphName: "Sub",
+      idGen,
+    });
+
+    expect(result?.subgraphSpec.inputs.at(0)?.name).toBe("run_when");
+    expect(
+      spec.bindings.find(
+        (binding) => binding.targetEntityId === result?.replacementTask.$id,
+      )?.targetPortName,
+    ).toBe("run_when");
+    expect(result?.subgraphSpec.bindings.at(0)?.targetPortName).toBe(
+      IS_ENABLED_PORT_NAME,
+    );
+  });
+
   it("moves internal bindings to subgraph", () => {
     const spec = new ComponentSpec({
       $id: idGen.next("spec"),
@@ -189,7 +236,9 @@ describe("createSubgraph", () => {
       $id: idGen.next("task"),
       name: "ConfiguredTask",
       componentRef: { name: "MyComponent" },
-      isEnabled: { "==": { op1: "a", op2: "b" } },
+      isEnabled: {
+        taskOutput: { taskId: "condition", outputName: "result" },
+      },
     });
     task.annotations.add({ key: "note", value: "test" });
     spec.addTask(task);
@@ -205,7 +254,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: "condition", outputName: "result" },
+    });
     expect(movedTask?.annotations.length).toBe(1);
   });
 

src/models/componentSpec/__tests__/actions/unpackSubgraph.test.ts

@@ -314,22 +314,21 @@ describe("unpackSubgraph roundtrip", () => {
     });
   });
 
-  it("isEnabled predicate preserved after roundtrip", () => {
+  it("isEnabled value is preserved after roundtrip", () => {
     const spec = new ComponentSpec({
       $id: idGen.next("spec"),
       name: "Main",
     });
-    const predicate = { "==": { op1: "a", op2: "b" } };
     const task = makeTask(idGen, "ConditionalTask", {
-      isEnabled: predicate,
+      isEnabled: "false",
     });
     spec.addTask(task);
 
     expect(roundtrip(spec, [task.$id], idGen)).toBe(true);
 
     const restored = spec.tasks.find((t) => t.name === "ConditionalTask");
     expect(restored).toBeDefined();
-    expect(restored?.isEnabled).toEqual(predicate);
+    expect(restored?.isEnabled).toBe("false");
   });
 
   it("static arguments preserved after roundtrip", () => {

src/models/componentSpec/__tests__/entities/componentSpecProxy.test.ts

@@ -296,14 +296,16 @@ describe("createComponentSpecProxy", () => {
           $id: "task_1",
           name: "T",
           componentRef: {},
-          isEnabled: { "==": { op1: "a", op2: "b" } },
+          isEnabled: {
+            taskOutput: { taskId: "condition", outputName: "result" },
+          },
         }),
       );
 
       const graph = getGraph(createComponentSpecProxy(spec));
 
       expect(graph.tasks["T"].isEnabled).toEqual({
-        "==": { op1: "a", op2: "b" },
+        taskOutput: { taskId: "condition", outputName: "result" },
       });
     });
 

src/models/componentSpec/__tests__/entities/task.test.ts

@@ -53,15 +53,18 @@ describe("Task", () => {
     expect(task.arguments[0]).toEqual({ name: "input", value: "test" });
   });
 
-  it("can set isEnabled predicate", () => {
+  it("can set an isEnabled reference", () => {
+    const isEnabled = {
+      taskOutput: { taskId: "condition", outputName: "result" },
+    };
     const task = new Task({
       $id: "task_1",
       name: "T",
       componentRef: {},
-      isEnabled: { "==": { op1: "a", op2: "b" } },
+      isEnabled,
     });
 
-    expect(task.isEnabled).toEqual({ "==": { op1: "a", op2: "b" } });
+    expect(task.isEnabled).toEqual(isEnabled);
   });
 
   it("setName updates name", () => {

src/models/componentSpec/__tests__/integration/roundtrip.test.ts

@@ -183,15 +183,18 @@ describe("Serialization Roundtrip", () => {
     });
   });
 
-  it("preserves isEnabled predicate", () => {
+  it("preserves an isEnabled reference", () => {
     const yaml = {
       name: "ConditionalTest",
       implementation: {
         graph: {
           tasks: {
+            Condition: { componentRef: {} },
             ConditionalTask: {
               componentRef: {},
-              isEnabled: { "!=": { op1: "a", op2: "b" } },
+              isEnabled: {
+                taskOutput: { taskId: "Condition", outputName: "result" },
+              },
             },
           },
         },
@@ -202,7 +205,7 @@ describe("Serialization Roundtrip", () => {
     const json = serializer.serialize(spec);
 
     expect(getGraph(json).tasks["ConditionalTask"].isEnabled).toEqual({
-      "!=": { op1: "a", op2: "b" },
+      taskOutput: { taskId: "Condition", outputName: "result" },
     });
   });
 

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,53 @@ describe("JsonSerializer", () => {
       $id: idGen.next("task"),
       name: "Process",
       componentRef: {},
-      isEnabled: { "==": { op1: "a", op2: "b" } },
+      isEnabled: {
+        taskOutput: { taskId: "condition", outputName: "result" },
+      },
     });
     spec.addTask(task);
 
     const json = serializer.serialize(spec);
 
     expect(getGraph(json).tasks["Process"].isEnabled).toEqual({
-      "==": { op1: "a", op2: "b" },
+      taskOutput: { taskId: "condition", outputName: "result" },
+    });
+  });
+
+  it("serializes a conditional binding as isEnabled", () => {
+    const spec = new ComponentSpec({
+      $id: idGen.next("spec"),
+      name: "Pipeline",
+    });
+    const condition = new Task({
+      $id: idGen.next("task"),
+      name: "Condition",
+      componentRef: {},
+    });
+    const conditionalTask = new Task({
+      $id: idGen.next("task"),
+      name: "ConditionalTask",
+      componentRef: {},
+      isEnabled: "true",
+    });
+    spec.addTask(condition);
+    spec.addTask(conditionalTask);
+    spec.addBinding(
+      new Binding({
+        $id: idGen.next("binding"),
+        sourceEntityId: condition.$id,
+        sourcePortName: "result",
+        targetEntityId: conditionalTask.$id,
+        targetPortName: IS_ENABLED_PORT_NAME,
+      }),
+    );
+
+    const taskSpec = getGraph(serializer.serialize(spec)).tasks.ConditionalTask;
+
+    expect(taskSpec.isEnabled).toEqual({
+      taskOutput: { taskId: "Condition", outputName: "result" },
     });
+    expect(taskSpec.arguments).toBeUndefined();
   });
 
   it("serializes metadata from annotations", () => {

src/models/componentSpec/__tests__/serialization/yamlDeserializer.test.ts

@@ -1,5 +1,7 @@
 import { beforeEach, describe, expect, it } from "vitest";
 
+import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution";
+
 import { IncrementingIdGenerator } from "../../factories/idGenerator";
 import { YamlDeserializer } from "../../serialization/yamlDeserializer";
 
@@ -102,26 +104,35 @@ describe("YamlDeserializer", () => {
     expect(spec.tasks.at(0)?.componentRef).toEqual({ name: "Processor" });
   });
 
-  it("deserializes task with isEnabled", () => {
+  it("deserializes an isEnabled reference as a conditional binding", () => {
     const yaml = {
       name: "Pipeline",
       implementation: {
         graph: {
           tasks: {
+            Condition: { componentRef: {} },
             ConditionalTask: {
               componentRef: {},
-              isEnabled: { "==": { op1: "a", op2: "b" } },
+              isEnabled: {
+                taskOutput: { taskId: "Condition", outputName: "result" },
+              },
             },
           },
         },
       },
     };
 
     const spec = deserializer.deserialize(yaml);
+    const task = spec.tasks.find(
+      (candidate) => candidate.name === "ConditionalTask",
+    );
+    const binding = spec.bindings.find(
+      (candidate) => candidate.targetEntityId === task?.$id,
+    );
 
-    expect(spec.tasks.at(0)?.isEnabled).toEqual({
-      "==": { op1: "a", op2: "b" },
-    });
+    expect(task?.isEnabled).toBe("true");
+    expect(binding?.targetPortName).toBe(IS_ENABLED_PORT_NAME);
+    expect(binding?.sourcePortName).toBe("result");
   });
 
   it("deserializes task annotations", () => {

src/models/componentSpec/actions/createSubgraph.ts

@@ -1,3 +1,4 @@
+import { IS_ENABLED_PORT_NAME } from "@/utils/conditionalExecution";
 import { deepClone } from "@/utils/deepClone";
 
 import { Annotations } from "../annotations";
@@ -9,9 +10,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 +32,7 @@ interface TaskSnapshot {
   $id: string;
   name: string;
   componentRef: ComponentReference;
-  isEnabled?: PredicateType;
+  isEnabled?: ArgumentType;
   annotations: Annotation[];
   arguments: Argument[];
   executionOptions?: ExecutionOptionsSpec;
@@ -128,7 +129,11 @@ export function createSubgraph({
   const usedInputNames = new Set<string>();
   for (const [, bindings] of Object.entries(incomingBySource)) {
     const first = bindings[0];
-    const inputName = deduplicatePortName(first.targetPortName, usedInputNames);
+    const preferredInputName =
+      first.targetPortName === IS_ENABLED_PORT_NAME
+        ? "run_when"
+        : first.targetPortName;
+    const inputName = deduplicatePortName(preferredInputName, usedInputNames);
     const input = new Input({
       $id: idGen.next("input"),
       name: inputName,

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<string>(),
   componentRef: prop<ComponentReference>(),
   subgraphSpec: prop<ComponentSpec | undefined>(undefined),
-  isEnabled: prop<PredicateType | undefined>(undefined),
+  isEnabled: prop<ArgumentType | undefined>(undefined),
   annotations: prop<Annotations>(() => new Annotations({})),
   arguments: prop<Argument[]>(() => []),
   executionOptions: prop<ExecutionOptionsSpec | undefined>(undefined),
@@ -78,8 +77,8 @@ export class Task extends Model({
   }
 
   @modelAction
-  setIsEnabled(predicate: PredicateType | undefined) {
-    this.isEnabled = predicate;
+  setIsEnabled(condition: ArgumentType | undefined) {
+    this.isEnabled = condition;
   }
 
   @modelAction

src/models/componentSpec/entities/types.ts

@@ -11,7 +11,6 @@ export type {
   InputSpec,
   MetadataSpec,
   OutputSpec,
-  PredicateType,
   TaskOutputArgument,
   TaskSpec,
   TypeSpecType,