feat: Pipeline editor - Conditional execution

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);
   });
 

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,
     });

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" },
       });
     });
 

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

@@ -57,10 +57,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", () => {

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",

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", () => {

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,26 +105,65 @@ 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" },
+              },
             },
           },
         },
       },
     };
 
     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", () => {

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;

src/models/componentSpec/entities/task.ts

@@ -11,7 +11,6 @@ import type {
   ComponentReference,
   ComponentSpecJson,
   ExecutionOptionsSpec,
-  PredicateType,
 } from "./types";
 import { isGraphImplementation } from "./types";
 
@@ -21,7 +20,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),
@@ -65,7 +64,7 @@ export class Task extends Model({
   }
 
   @modelAction
-  setIsEnabled(predicate: PredicateType | undefined) {
+  setIsEnabled(predicate: ArgumentType | undefined) {
     this.isEnabled = predicate;
   }
 

src/models/componentSpec/entities/types.ts

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