Apply Capacity Reservation tenancy to AWS instance launch

src/dstack/_internal/core/backends/aws/compute.py

@@ -277,6 +277,7 @@ def create_instance(
         )
         enable_efa = max_efa_interfaces > 0
         is_capacity_block = False
+        reservation_tenancy = None
         try:
             vpc_id, subnets_ids = self._get_vpc_id_subnets_ids_or_error(
                 ec2_client=ec2_client,
@@ -297,6 +298,7 @@ def create_instance(
                     instance_count=1,
                 )
                 if reservation is not None:
+                    reservation_tenancy = reservation.get("Tenancy")
                     # Filter out az different from capacity reservation
                     subnet_id_to_az_map = {
                         k: v
@@ -355,6 +357,7 @@ def create_instance(
                         max_efa_interfaces=max_efa_interfaces,
                         reservation_id=instance_config.reservation,
                         is_capacity_block=is_capacity_block,
+                        tenancy=reservation_tenancy,
                     )
                 )
             except botocore.exceptions.ClientError as e:

src/dstack/_internal/core/backends/aws/resources.py

@@ -152,6 +152,7 @@ def create_instances_struct(
     max_efa_interfaces: int = 0,
     reservation_id: Optional[str] = None,
     is_capacity_block: bool = False,
+    tenancy: Optional[str] = None,
 ) -> Dict[str, Any]:
     struct: Dict[str, Any] = dict(
         BlockDeviceMappings=[
@@ -213,6 +214,12 @@ def create_instances_struct(
             "CapacityReservationTarget": {"CapacityReservationId": reservation_id}
         }
 
+    # A Capacity Reservation created with non-default tenancy (e.g. `dedicated`) only
+    # accepts instances launched with a matching `Placement.Tenancy`. Apply it
+    # automatically so users don't have to configure tenancy explicitly.
+    if tenancy is not None and tenancy != "default":
+        struct.setdefault("Placement", {})["Tenancy"] = tenancy
+
     return struct
 
 

src/tests/_internal/core/backends/aws/test_resources.py

@@ -8,6 +8,7 @@
     _create_network_interfaces_struct,
     _is_valid_tag_key,
     _is_valid_tag_value,
+    create_instances_struct,
     get_image_id_and_username,
     validate_tags,
 )
@@ -238,6 +239,43 @@ def test_raises_resource_not_found_if_image_config_property_not_set(
         assert "cpu image not configured" in caplog.text
 
 
+class TestCreateInstancesStruct:
+    def _struct(self, **kwargs):
+        return create_instances_struct(
+            disk_size=100,
+            image_id="ami-1",
+            instance_type="m5.large",
+            iam_instance_profile=None,
+            user_data="",
+            tags=[],
+            security_group_id="sg-1",
+            spot=False,
+            **kwargs,
+        )
+
+    def test_no_tenancy_by_default(self):
+        struct = self._struct(reservation_id="cr-1")
+        assert "Placement" not in struct
+
+    def test_default_tenancy_not_set(self):
+        # `default` is the implicit AWS behavior, so it should not be sent explicitly
+        struct = self._struct(reservation_id="cr-1", tenancy="default")
+        assert "Placement" not in struct
+
+    def test_dedicated_tenancy_applied(self):
+        struct = self._struct(reservation_id="cr-1", tenancy="dedicated")
+        assert struct["Placement"]["Tenancy"] == "dedicated"
+
+    def test_tenancy_merged_with_placement_group(self):
+        struct = self._struct(
+            reservation_id="cr-1",
+            tenancy="dedicated",
+            placement_group_name="pg-1",
+        )
+        assert struct["Placement"]["GroupName"] == "pg-1"
+        assert struct["Placement"]["Tenancy"] == "dedicated"
+
+
 class TestCreateNetworkInterfacesStruct:
     def test_non_efa_instance_single_interface(self):
         interfaces = _create_network_interfaces_struct(