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/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<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,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<string, ArgumentType> = {};
 
     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/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<string, string>();
 
-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<TaskSpec> = {
-      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 e3c8a9ecc9..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,20 +8,32 @@ 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 {
+  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 +44,40 @@ export const ConfigurationSection = observer(function ConfigurationSection({
   task,
 }: ConfigurationSectionProps) {
   const { track } = useAnalytics();
+  const spec = useSpec();
+  const conditionalExecutionEnabled = useFlagValue("conditional-execution");
   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 +192,42 @@ export const ConfigurationSection = observer(function ConfigurationSection({
         />
       </InlineStack>
 
+      {conditionalExecutionEnabled && (
+        <>
+          <Separator />
+
+          <BlockStack gap="1">
+            <InlineStack align="space-between" gap="2" className="w-full">
+              <Paragraph size="xs" tone="subdued">
+                Enable task
+              </Paragraph>
+              <Select value={enableMode} onValueChange={handleEnableModeChange}>
+                <SelectTrigger className="h-6 text-xs px-2 py-0 min-w-25">
+                  <SelectValue />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem value="true" className="text-xs">
+                    True
+                  </SelectItem>
+                  <SelectItem value="false" className="text-xs">
+                    False
+                  </SelectItem>
+                  <SelectItem value="conditional" className="text-xs">
+                    Conditional
+                  </SelectItem>
+                </SelectContent>
+              </Select>
+            </InlineStack>
+            {isConditional && !isConditionalConnected && (
+              <Paragraph size="xs" tone="subdued">
+                Connect a task output or pipeline input to the “Is enabled?”
+                port on the node.
+              </Paragraph>
+            )}
+          </BlockStack>
+        </>
+      )}
+
       {!isSubgraph && (
         <>
           <Separator />
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..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",
   {
@@ -168,12 +172,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, " ")}
           </Button>
         </div>
         {showValueDisplay && (
@@ -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({
           >
             <BlockStack gap="3">
               {isAggregator && <InputAggregatorHandle />}
-              {visibleInputs.map((input, index) => (
-                <ClassicInputHandle
-                  key={input.name}
-                  input={input}
-                  entityId={entityId}
-                  themed={!palette}
-                  isDark={isDark}
-                  displayValue={
-                    showCondensedInputs
-                      ? index === 0
-                        ? `+${hiddenInputCount} more ${pluralize(hiddenInputCount, "input")}`
-                        : undefined
-                      : inputDisplayValues[input.name]
-                  }
-                  hideValue={showCondensedInputs && index !== 0}
-                  isSecret={secretInputNames.has(input.name)}
-                  onRowClick={
-                    inputsSectionToggles
-                      ? undefined
-                      : openInputProperties(input.name)
-                  }
-                  onLabelClick={openInputProperties(input.name)}
-                  onHandleClick={(e) => 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 (
+                  <ClassicInputHandle
+                    key={input.name}
+                    input={input}
+                    entityId={entityId}
+                    themed={!palette}
+                    isDark={isDark}
+                    displayValue={
+                      condensedAway
+                        ? hiddenCountLabel
+                        : inputDisplayValues[input.name]
+                    }
+                    hideValue={condensedAway && !hiddenCountLabel}
+                    isSecret={secretInputNames.has(input.name)}
+                    onRowClick={
+                      inputsSectionToggles
+                        ? undefined
+                        : openInputProperties(input.name)
+                    }
+                    onLabelClick={openInputProperties(input.name)}
+                    onHandleClick={(e) =>
+                      onHandleClick(`input_${input.name}`, e)
+                    }
+                  />
+                );
+              })}
               {collapsed && inputsExpanded && hiddenInputCount > 0 && (
                 <span
                   className={cn(
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..3a1ca4aa8c 100644
--- a/src/utils/componentSpec.ts
+++ b/src/utils/componentSpec.ts
@@ -380,7 +380,7 @@ interface TwoLogicalOperands {
   op1: PredicateType;
   op2: PredicateType;
 }
-export type PredicateType =
+type PredicateType =
   | {
       "==": TwoArgumentOperands;
     }
@@ -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);
+}