From 846b6305d5609d376dba20b57ced88feab621873 Mon Sep 17 00:00:00 2001
From: Morgan Wowk <morgan.wowk@shopify.com>
Date: Wed, 29 Jul 2026 02:06:34 -0700
Subject: [PATCH] Add optional targetCluster to pipeline submission

submitPipelineRun gains an optional targetCluster; when set it adds the
tangleml.com/orchestration/target_cluster run annotation the backend
router uses to pick the execution cluster. When unset the run payload is
byte-identical to today, so default routing is unaffected. The annotation
key is centralized in src/utils/annotations.ts.
---
 src/utils/annotations.ts         |  2 ++
 src/utils/submitPipeline.test.ts | 37 ++++++++++++++++++++++++++++++++
 src/utils/submitPipeline.ts      | 14 +++++++++++-
 3 files changed, 52 insertions(+), 1 deletion(-)

diff --git a/src/utils/annotations.ts b/src/utils/annotations.ts
index 8ee238d2f6..b45c900b13 100644
--- a/src/utils/annotations.ts
+++ b/src/utils/annotations.ts
@@ -13,6 +13,8 @@ export const PIPELINE_TAGS_ANNOTATION = "tags";
 export const PIPELINE_CANONICAL_NAME_ANNOTATION = "canonical-pipeline-name";
 export const RUN_NAME_TEMPLATE_ANNOTATION = "run-name-template";
 export const RUN_SOURCE_ANNOTATION = "source";
+export const ORCHESTRATION_TARGET_CLUSTER_ANNOTATION =
+  "tangleml.com/orchestration/target_cluster";
 export const EDITOR_POSITION_ANNOTATION = "editor.position";
 export const EDITOR_COLLAPSED_ANNOTATION = "editor.collapsed";
 export const EDITOR_FLOW_DIRECTION_ANNOTATION = "editor.flow-direction";
diff --git a/src/utils/submitPipeline.test.ts b/src/utils/submitPipeline.test.ts
index 8c99bd146e..f5a3fdb89c 100644
--- a/src/utils/submitPipeline.test.ts
+++ b/src/utils/submitPipeline.test.ts
@@ -227,6 +227,43 @@ describe("submitPipelineRun", () => {
     });
   });
 
+  describe("targetCluster handling", () => {
+    const componentSpec: ComponentSpec = {
+      name: "simple-component",
+      implementation: {
+        container: {
+          image: "test:latest",
+        },
+      },
+    };
+
+    it("adds the target_cluster annotation when targetCluster is set", async () => {
+      await submitPipelineRun(componentSpec, mockBackendUrl, {
+        targetCluster: "ml-offline-us-ce1-eo9",
+      });
+
+      expect(pipelineRunService.createPipelineRun).toHaveBeenCalledWith(
+        expect.objectContaining({
+          annotations: {
+            source: "web-app",
+            "tangleml.com/orchestration/target_cluster":
+              "ml-offline-us-ce1-eo9",
+          },
+        }),
+        mockBackendUrl,
+        undefined,
+      );
+    });
+
+    it("leaves run annotations unchanged when targetCluster is unset", async () => {
+      await submitPipelineRun(componentSpec, mockBackendUrl, {});
+
+      const payload = vi.mocked(pipelineRunService.createPipelineRun).mock
+        .calls[0][0];
+      expect(payload.annotations).toEqual({ source: "web-app" });
+    });
+  });
+
   describe("taskArguments handling", () => {
     it("should include taskArguments in payload when provided", async () => {
       // Arrange
diff --git a/src/utils/submitPipeline.ts b/src/utils/submitPipeline.ts
index 55bed2d17e..4573193e40 100644
--- a/src/utils/submitPipeline.ts
+++ b/src/utils/submitPipeline.ts
@@ -12,7 +12,10 @@ import {
 import type { PipelineRun } from "@/types/pipelineRun";
 
 import { transformAggregatorComponentSpec } from "./aggregatorTransform";
-import { RUN_SOURCE_ANNOTATION } from "./annotations";
+import {
+  ORCHESTRATION_TARGET_CLUSTER_ANNOTATION,
+  RUN_SOURCE_ANNOTATION,
+} from "./annotations";
 import { buildAnnotationsWithCanonicalName } from "./canonicalPipelineName";
 import type {
   ArgumentType,
@@ -30,6 +33,7 @@ export async function submitPipelineRun(
     authorizationToken?: string;
     runNameOverride?: boolean;
     canonicalName?: string;
+    targetCluster?: string;
     onSuccess?: (data: PipelineRun) => void;
     onError?: (error: Error) => void;
   },
@@ -91,6 +95,14 @@ export async function submitPipelineRun(
     const payload = {
       annotations: {
         [RUN_SOURCE_ANNOTATION]: "web-app",
+        // Opt-in override selecting which execution cluster the orchestrator
+        // routes this run to. Omitted entirely when unset, so the default
+        // routing (and the run payload) is unchanged.
+        ...(options?.targetCluster
+          ? {
+              [ORCHESTRATION_TARGET_CLUSTER_ANNOTATION]: options.targetCluster,
+            }
+          : {}),
       },
       root_task: {
         componentRef: {