Add code location tag to runs (#32715)

## Summary & Motivation

This PR automatically adds the `dagster/code_location` tag to runs when
a `remote_job_origin` is available. This enables filtering runs by code
location in the web UI and controlling concurrency by code location
using tag concurrency limits.

## How I Tested These Changes

How I Tested These Changes
Verified that runs created with `remote_job_origin` now include the
`dagster/code_location` tag
Confirmed the tag value matches the code location name from
`remote_job_origin.repository_origin.code_location_origin.location_name`
Tested that runs without remote_job_origin are not affected

## Changelog

Runs now automatically include a `dagster/code_location` tag when
created with a `remote_job_origin`, enabling filtering and concurrency
control by code location.

Author: ssup2

python_modules/dagster/dagster/_core/instance/runs/run_domain.py

@@ -35,6 +35,7 @@
     ASSET_RESUME_RETRY_TAG,
     BACKFILL_ID_TAG,
     BACKFILL_TAGS,
+    CODE_LOCATION_TAG,
     PARENT_RUN_ID_TAG,
     PARTITION_NAME_TAG,
     RESUME_RETRY_TAG,
@@ -222,6 +223,15 @@ def create_run(
                 "must include a remote job origin when creating a run using a remote asset graph",
             )
 
+        # Add the code location tag to the validated tags
+        # This is useful for filtering runs by code location in the web server
+        # or control concurrency by code location with tag concurrency limits.
+        if remote_job_origin:
+            validated_tags = {
+                **validated_tags,
+                CODE_LOCATION_TAG: remote_job_origin.repository_origin.code_location_origin.location_name,
+            }
+
         dagster_run = self.construct_run_with_snapshots(
             job_name=job_name,
             run_id=run_id,  # type: ignore  # (possible none)

python_modules/dagster/dagster/_core/storage/tags.py

@@ -7,6 +7,8 @@
 
 REPOSITORY_LABEL_TAG = f"{HIDDEN_TAG_PREFIX}repository"
 
+CODE_LOCATION_TAG = f"{SYSTEM_TAG_PREFIX}code_location"
+
 SCHEDULE_NAME_TAG = f"{SYSTEM_TAG_PREFIX}schedule_name"
 
 SENSOR_NAME_TAG = f"{SYSTEM_TAG_PREFIX}sensor_name"

python_modules/dagster/dagster_tests/core_tests/instance_tests/test_instance.py

@@ -38,6 +38,7 @@
 from dagster._core.storage.tags import (
     ASSET_PARTITION_RANGE_END_TAG,
     ASSET_PARTITION_RANGE_START_TAG,
+    CODE_LOCATION_TAG,
     PARTITION_NAME_TAG,
 )
 from dagster._core.test_utils import (
@@ -365,6 +366,40 @@ def test_submit_run():
             assert instance.run_coordinator.queue()[0].run_id == run.run_id  # pyright: ignore[reportAttributeAccessIssue]
 
 
+def test_create_run_adds_code_location_tag():
+    """Test that CODE_LOCATION_TAG is automatically added when remote_job_origin is provided."""
+    with dg.instance_for_test() as instance:
+        with get_bar_workspace(instance) as workspace:
+            remote_job = (
+                workspace.get_code_location("bar_code_location")
+                .get_repository("bar_repo")
+                .get_full_job("foo")
+            )
+
+            run = create_run_for_test(
+                instance=instance,
+                job_name=remote_job.name,
+                remote_job_origin=remote_job.get_remote_origin(),
+                job_code_origin=remote_job.get_python_origin(),
+            )
+
+            # Verify CODE_LOCATION_TAG is automatically added
+            assert CODE_LOCATION_TAG in run.tags
+            assert run.tags[CODE_LOCATION_TAG] == "bar_code_location"
+
+
+def test_create_run_without_remote_job_origin_has_no_code_location_tag():
+    """Test that CODE_LOCATION_TAG is not added when remote_job_origin is not provided."""
+    with dg.instance_for_test() as instance:
+        run = create_run_for_test(
+            instance=instance,
+            job_name="foo_job",
+        )
+
+        # Verify CODE_LOCATION_TAG is not present
+        assert CODE_LOCATION_TAG not in run.tags
+
+
 def test_create_run_without_asset_execution_type_on_snapshot():
     # verify that even runs created on older versions of dagster still store the
     # execution type on the execution plan snapshot