Cross-project fleet references in CLI and YAML (#3677)

Allow to reference imported fleets in the `fleets`
property in run and profile configurations:

```yaml
type: dev-environment
ide: vscode

fleets:
- local
- main/shared-1
- project: main  # verbose syntax
  name: shared-2
```

As well as in the `--fleet` CLI option:

```shell
$ dstack apply -f run.dstack.yml --fleet main/shared-1
```

Fleets specified without `project` are local to
the run's project.

Author: jvstme

docs/docs/concepts/exports.md

@@ -134,6 +134,19 @@ $ dstack fleet list
 
 Imported fleets can be used for runs just like the project's own fleets.
 
+<div editor-title=".dstack.yml">
+    
+```yaml
+type: dev-environment
+ide: vscode
+
+fleets:
+- my-local-fleet
+- team-a/my-fleet
+```
+
+</div>
+
 !!! info "Tenant isolation"
     Exported fleets share the same access model as regular fleets. See [Tenant isolation](fleets.md#tenant-isolation) for details.
 

src/dstack/_internal/core/compatibility/common.py

@@ -0,0 +1,14 @@
+from dstack._internal.core.models.common import EntityReference
+from dstack._internal.core.models.profiles import ProfileParams
+
+
+def patch_profile_params(params: ProfileParams) -> None:
+    # If there are no project-prefixed fleets, replace all EntityReference with str
+    # for compatibility with pre-0.20.14 servers that don't support EntityReference.
+    if params.fleets is not None and all(
+        EntityReference.parse(f).project is None for f in params.fleets
+    ):
+        params.fleets = [
+            fleet_ref.format() if isinstance(fleet_ref, EntityReference) else fleet_ref
+            for fleet_ref in params.fleets
+        ]

src/dstack/_internal/core/compatibility/fleets.py

@@ -1,6 +1,10 @@
 from typing import Optional
 
-from dstack._internal.core.models.common import IncludeExcludeDictType, IncludeExcludeSetType
+from dstack._internal.core.compatibility.common import patch_profile_params
+from dstack._internal.core.models.common import (
+    IncludeExcludeDictType,
+    IncludeExcludeSetType,
+)
 from dstack._internal.core.models.fleets import ApplyFleetPlanInput, FleetSpec
 
 
@@ -56,3 +60,7 @@ def get_fleet_spec_excludes(fleet_spec: FleetSpec) -> Optional[IncludeExcludeDic
     if spec_excludes:
         return spec_excludes
     return None
+
+
+def patch_fleet_spec(spec: FleetSpec) -> None:
+    patch_profile_params(spec.profile)

src/dstack/_internal/core/compatibility/runs.py

@@ -1,6 +1,10 @@
 from typing import Optional
 
-from dstack._internal.core.models.common import IncludeExcludeDictType, IncludeExcludeSetType
+from dstack._internal.core.compatibility.common import patch_profile_params
+from dstack._internal.core.models.common import (
+    IncludeExcludeDictType,
+    IncludeExcludeSetType,
+)
 from dstack._internal.core.models.configurations import ServiceConfiguration
 from dstack._internal.core.models.routers import SGLangServiceRouterConfig
 from dstack._internal.core.models.runs import (
@@ -138,3 +142,9 @@ def get_job_submission_excludes(job_submissions: list[JobSubmission]) -> Include
         submission_excludes["job_runtime_data"] = jrd_excludes
 
     return submission_excludes
+
+
+def patch_run_spec(run_spec: RunSpec) -> None:
+    patch_profile_params(run_spec.configuration)
+    if run_spec.profile is not None:
+        patch_profile_params(run_spec.profile)

src/dstack/_internal/core/models/common.py

@@ -143,3 +143,31 @@ class ApplyAction(str, Enum):
 class NetworkMode(str, Enum):
     HOST = "host"
     BRIDGE = "bridge"
+
+
+class EntityReference(CoreModel):
+    """
+    Cross-project entity reference.
+    """
+
+    project: Annotated[
+        Optional[str],
+        Field(description="The project name. If unspecified, refers to the current project"),
+    ]
+    name: Annotated[str, Field(description="The entity name")]
+
+    @classmethod
+    def parse(cls, v: Union[str, "EntityReference"]) -> "EntityReference":
+        if isinstance(v, EntityReference):
+            return v
+        parts = v.split("/")
+        if len(parts) == 1:
+            return cls(project=None, name=parts[0])
+        if len(parts) == 2:
+            return cls(project=parts[0], name=parts[1])
+        raise ValueError("Invalid entity reference. Only `<project>/<name>` format is allowed")
+
+    def format(self) -> str:
+        if self.project is None:
+            return self.name
+        return f"{self.project}/{self.name}"

src/dstack/_internal/core/models/profiles.py

@@ -10,6 +10,7 @@
     CoreConfig,
     CoreModel,
     Duration,
+    EntityReference,
     generate_dual_core_model,
 )
 from dstack._internal.utils.common import list_enum_values_for_annotation
@@ -360,7 +361,21 @@ class ProfileParams(CoreModel):
         Field(description=("The schedule for starting the run at specified time")),
     ] = None
     fleets: Annotated[
-        Optional[list[str]], Field(description="The fleets considered for reuse")
+        Optional[
+            list[
+                Union[
+                    EntityReference,
+                    str,  # For server response compatibility with pre-0.20.14 clients
+                ]
+            ]
+        ],
+        Field(
+            description=(
+                "The fleets considered for reuse."
+                " For fleets owned by the current project, specify fleet names."
+                " For imported fleets, specify `<project name>/<fleet name>`"
+            ),
+        ),
     ] = None
     tags: Annotated[
         Optional[Dict[str, str]],
@@ -382,6 +397,7 @@ class ProfileParams(CoreModel):
     _validate_idle_duration = validator("idle_duration", pre=True, allow_reuse=True)(
         parse_idle_duration
     )
+    _validate_fleets = validator("fleets", allow_reuse=True, each_item=True)(EntityReference.parse)
     _validate_tags = validator("tags", pre=True, allow_reuse=True)(tags_validator)
 
 

src/dstack/_internal/server/background/scheduled_tasks/submitted_jobs.py

@@ -975,6 +975,7 @@ async def _refetch_fleet_models_with_instances(
 ) -> list[FleetModel]:
     res = await session.execute(
         select(FleetModel)
+        .join(FleetModel.project)  # can be referenced by fleet_filters
         .outerjoin(FleetModel.instances)
         .where(
             FleetModel.id.in_(fleets_ids),

src/dstack/_internal/server/compatibility/common.py

@@ -2,10 +2,23 @@
 
 from packaging.version import Version
 
+from dstack._internal.core.models.common import EntityReference
 from dstack._internal.core.models.instances import (
     InstanceAvailability,
     InstanceOfferWithAvailability,
 )
+from dstack._internal.core.models.profiles import ProfileParams
+
+
+def patch_profile_params(params: ProfileParams, client_version: Optional[Version]) -> None:
+    if client_version is None:
+        return
+    # Clients prior to 0.20.14 only support `list[str]` in `fleets`
+    if client_version < Version("0.20.14") and params.fleets is not None:
+        params.fleets = [
+            fleet_ref.format() if isinstance(fleet_ref, EntityReference) else fleet_ref
+            for fleet_ref in params.fleets
+        ]
 
 
 def patch_offers_list(

src/dstack/_internal/server/compatibility/fleets.py

@@ -0,0 +1,23 @@
+from typing import Optional
+
+from packaging.version import Version
+
+from dstack._internal.core.models.fleets import Fleet, FleetPlan, FleetSpec
+from dstack._internal.server.compatibility.common import patch_offers_list, patch_profile_params
+
+
+def patch_fleet_plan(fleet_plan: FleetPlan, client_version: Optional[Version]) -> None:
+    patch_fleet_spec(fleet_plan.spec, client_version)
+    if fleet_plan.effective_spec is not None:
+        patch_fleet_spec(fleet_plan.effective_spec, client_version)
+    if fleet_plan.current_resource is not None:
+        patch_fleet(fleet_plan.current_resource, client_version)
+    patch_offers_list(fleet_plan.offers, client_version)
+
+
+def patch_fleet(fleet: Fleet, client_version: Optional[Version]) -> None:
+    patch_fleet_spec(fleet.spec, client_version)
+
+
+def patch_fleet_spec(fleet_spec: FleetSpec, client_version: Optional[Version]) -> None:
+    patch_profile_params(fleet_spec.profile, client_version)

src/dstack/_internal/server/compatibility/runs.py

@@ -4,7 +4,7 @@
 
 from dstack._internal.core.models.configurations import SERVICE_HTTPS_DEFAULT, ServiceConfiguration
 from dstack._internal.core.models.runs import Run, RunPlan, RunSpec
-from dstack._internal.server.compatibility.common import patch_offers_list
+from dstack._internal.server.compatibility.common import patch_offers_list, patch_profile_params
 
 
 def patch_run_plan(run_plan: RunPlan, client_version: Optional[Version]) -> None:
@@ -41,3 +41,6 @@ def patch_run_spec(run_spec: RunSpec, client_version: Optional[Version]) -> None
         and run_spec.configuration.https is None
     ):
         run_spec.configuration.https = SERVICE_HTTPS_DEFAULT
+    patch_profile_params(run_spec.configuration, client_version)
+    if run_spec.profile is not None:
+        patch_profile_params(run_spec.profile, client_version)