Add for_offers_only parameter to /runs/get_plan API method

src/dstack/_internal/cli/commands/offer.py

@@ -91,6 +91,7 @@ def _list_offers(self, args: argparse.Namespace) -> None:
             max_offers=args.max_offers,
             full_offers=args.full_offers,
             unallocated_resources=args.unallocated,
+            for_offers_only=True,
         )
         job_plan = run_plan.job_plans[0]
         if args.format == "plain":

src/dstack/_internal/cli/services/presets/create.py

@@ -808,7 +808,9 @@ def _print_fleet_offers(api: Client, allowed_fleets: tuple[str, ...]) -> None:
         offer_configuration.fleets = list(allowed_fleets)
         run_spec = RunSpec(configuration=offer_configuration, profile=None)
         with console.status("Getting offers..."):
-            run_plan = api.client.runs.get_plan(api.project, run_spec, max_offers=10)
+            run_plan = api.client.runs.get_plan(
+                api.project, run_spec, max_offers=10, for_offers_only=True
+            )
         props = Table(box=None, show_header=False)
         props.add_column(no_wrap=True)
         props.add_column()

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

@@ -52,3 +52,21 @@ def patch_run_spec(run_spec: RunSpec, client_version: Optional[Version]) -> None
         and isinstance(run_spec.configuration.gateway, EntityReference)
     ):
         run_spec.configuration.gateway = run_spec.configuration.gateway.format()
+
+
+def is_run_plan_for_offers_only(
+    run_spec: RunSpec, for_offers_only: bool, client_version: Optional[Version]
+) -> bool:
+    """
+    Clients < 0.21.0 don't support `for_offers_only` argument and rely on a magic configuration
+    that triggers "offer collection only" path.
+
+    TODO: Drop once clients < 0.21.0 are no longer supported.
+
+    NOTE: A real task with `commands == [":"]` would also match this special `dstack offer` path.
+    """
+    if for_offers_only:
+        return True
+    if client_version is not None and client_version < Version("0.21.0"):
+        return run_spec.configuration.type == "task" and run_spec.configuration.commands == [":"]
+    return False

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

@@ -6,7 +6,11 @@
 
 from dstack._internal.core.errors import ResourceNotExistsError
 from dstack._internal.core.models.runs import Run, RunPlan
-from dstack._internal.server.compatibility.runs import patch_run, patch_run_plan
+from dstack._internal.server.compatibility.runs import (
+    is_run_plan_for_offers_only,
+    patch_run,
+    patch_run_plan,
+)
 from dstack._internal.server.db import get_session
 from dstack._internal.server.models import ProjectModel, UserModel
 from dstack._internal.server.schemas.runs import (
@@ -133,6 +137,10 @@ async def get_plan(
     user, project = user_project
     if not user.ssh_public_key and not body.run_spec.ssh_key_pub:
         await users.refresh_ssh_key(session=session, actor=user)
+    # TODO: Use body.for_offers_only directly once clients < 0.21.0 are no longer supported
+    for_offers_only = is_run_plan_for_offers_only(
+        run_spec=body.run_spec, for_offers_only=body.for_offers_only, client_version=client_version
+    )
     run_plan = await runs.get_plan(
         session=session,
         project=project,
@@ -142,6 +150,7 @@ async def get_plan(
         full_offers=body.full_offers,
         unallocated_resources=body.unallocated_resources,
         legacy_repo_dir=legacy_repo_dir,
+        for_offers_only=for_offers_only,
     )
     patch_run_plan(run_plan, client_version)
     return CustomJSONResponse(run_plan)

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

@@ -52,6 +52,12 @@ class GetRunPlanRequest(CoreModel):
         bool,
         Field(description="Subtract allocated resources to return only unallocated resources"),
     ] = False
+    for_offers_only: Annotated[
+        bool,
+        Field(
+            description="Set to True if the run plan is requested for offer collection only, not a real run submission"
+        ),
+    ] = False
 
 
 class SubmitRunRequest(CoreModel):

src/dstack/_internal/server/services/runs/__init__.py

@@ -534,6 +534,7 @@ async def get_plan(
     max_offers: Optional[int],
     full_offers: bool,
     unallocated_resources: bool,
+    for_offers_only: bool,
     legacy_repo_dir: bool = False,
 ) -> RunPlan:
     effective_run_spec = RunSpec.model_validate(run_spec.model_dump())
@@ -574,6 +575,7 @@ async def get_plan(
         max_offers=max_offers,
         full_offers=full_offers,
         unallocated_resources=unallocated_resources,
+        for_offers_only=for_offers_only,
     )
     run_plan = RunPlan(
         project_name=project.name,

src/dstack/_internal/server/services/runs/plan.py

@@ -89,14 +89,14 @@ async def get_job_plans(
     max_offers: Optional[int],
     full_offers: bool,
     unallocated_resources: bool,
+    for_offers_only: bool,
 ) -> list[JobPlan]:
     """
     Returns job plans for the given run spec.
 
     Normal run planning (`dstack apply`) selects the best fleet candidate for each planned job
     and builds offers from that path. `dstack offer` without `--group-by` uses the same
-    `/runs/get_plan` API, but its synthetic run spec is detected by
-    `_should_select_best_fleet_candidate()`. In that case, planning skips
+    `/runs/get_plan` API but with `for_offers_only=True`. In that case, planning skips
     best-fleet-candidate selection and collects offers directly: global offers when no fleets
     are specified, or offers from the selected fleets when `--fleet` is used.
 
@@ -119,7 +119,7 @@ async def get_job_plans(
         job_num=0,
     )
 
-    if _should_select_best_fleet_candidate(run_spec) and run_spec.merged_profile.instances is None:
+    if not for_offers_only and run_spec.merged_profile.instances is None:
         candidate_fleet_models = await _select_candidate_fleet_models(
             session=session,
             project=project,
@@ -1016,27 +1016,6 @@ def _get_job_plan(
     )
 
 
-def _should_select_best_fleet_candidate(run_spec: RunSpec) -> bool:
-    """
-    Returns ``True`` for normal run planning and ``False`` for `dstack offer` without
-    `--group-by`.
-
-    Both `dstack apply` and `dstack offer` without `--group-by` call `/runs/get_plan`. The
-    current way to recognize `dstack offer` without `--group-by` is the synthetic task spec
-    that the CLI sends with `type == "task"` and `commands == [":"]`.
-    TODO: Replace this command-shape hack with an explicit request/API signal for
-    `dstack offer` without `--group-by`.
-
-    When this function returns ``False``, the planner skips best-fleet-candidate selection
-    and goes directly to the special `dstack offer` collection path:
-    global offers when no fleets are specified, or offers from the selected fleets when
-    `--fleet` is used.
-
-    A real task with `commands == [":"]` would also match this special `dstack offer` path.
-    """
-    return not (run_spec.configuration.type == "task" and run_spec.configuration.commands == [":"])
-
-
 def _get_offers_from_instances(
     instances: list[InstanceModel],
 ) -> list[tuple[InstanceModel, InstanceOfferWithAvailability]]:

src/dstack/api/_public/runs.py

@@ -478,6 +478,7 @@ def get_run_plan(
         max_offers: Optional[int] = None,
         full_offers: bool = False,
         unallocated_resources: bool = False,
+        for_offers_only: bool = False,
     ) -> RunPlan:
         """
         Get a run plan.
@@ -498,6 +499,8 @@ def get_run_plan(
             full_offers: Return full offers not adjusted by requirements.
             unallocated_resources: Subtract allocated resources to return only unallocated
                 resources.
+            for_offers_only: Set to True if the run plan is requested for offer collection only,
+                not a real run submission.
 
         Returns:
             Run plan.
@@ -554,6 +557,7 @@ def get_run_plan(
             max_offers=max_offers,
             full_offers=full_offers,
             unallocated_resources=unallocated_resources,
+            for_offers_only=for_offers_only,
         )
         return run_plan
 

src/dstack/api/server/_runs.py

@@ -77,12 +77,14 @@ def get_plan(
         max_offers: Optional[int] = None,
         full_offers: bool = False,
         unallocated_resources: bool = False,
+        for_offers_only: bool = False,
     ) -> RunPlan:
         body = GetRunPlanRequest(
             run_spec=run_spec,
             max_offers=max_offers,
             full_offers=full_offers,
             unallocated_resources=unallocated_resources,
+            for_offers_only=for_offers_only,
         )
         body = copy.deepcopy(body)
         patch_run_spec(body.run_spec)

src/tests/_internal/server/compatibility/test_runs.py

@@ -0,0 +1,81 @@
+from typing import Optional
+
+import pytest
+from packaging.version import Version
+
+from dstack._internal.core.models.configurations import (
+    AnyRunConfiguration,
+    DevEnvironmentConfiguration,
+    TaskConfiguration,
+)
+from dstack._internal.server.compatibility.runs import is_run_plan_for_offers_only
+from dstack._internal.server.testing.common import get_run_spec
+
+_OFFER_CLI_CONFIGURATION = TaskConfiguration(commands=[":"], image="scratch", user="root")
+
+
+class TestIsRunPlanForOffersOnly:
+    @pytest.mark.parametrize(
+        ("configuration", "for_offers_only", "client_version", "expected"),
+        [
+            pytest.param(_OFFER_CLI_CONFIGURATION, True, Version("0.21.0"), True, id="flag-set"),
+            pytest.param(
+                DevEnvironmentConfiguration(),
+                True,
+                Version("0.21.0"),
+                True,
+                id="flag-set-for-any-configuration",
+            ),
+            pytest.param(
+                _OFFER_CLI_CONFIGURATION,
+                False,
+                Version("0.20.30"),
+                True,
+                id="old-client-sends-offer-cli-configuration",
+            ),
+            pytest.param(
+                TaskConfiguration(commands=["echo"], image="scratch"),
+                False,
+                Version("0.20.30"),
+                False,
+                id="old-client-sends-regular-task",
+            ),
+            pytest.param(
+                DevEnvironmentConfiguration(),
+                False,
+                Version("0.20.30"),
+                False,
+                id="old-client-sends-configuration-without-commands",
+            ),
+            pytest.param(
+                _OFFER_CLI_CONFIGURATION,
+                False,
+                Version("0.21.0"),
+                False,
+                id="new-client-does-not-rely-on-offer-cli-configuration",
+            ),
+            pytest.param(
+                _OFFER_CLI_CONFIGURATION,
+                False,
+                None,
+                False,
+                id="dev-client-does-not-rely-on-offer-cli-configuration",
+            ),
+        ],
+    )
+    def test_returns_expected(
+        self,
+        configuration: AnyRunConfiguration,
+        for_offers_only: bool,
+        client_version: Optional[Version],
+        expected: bool,
+    ) -> None:
+        run_spec = get_run_spec(repo_id="test-repo", configuration=configuration)
+        assert (
+            is_run_plan_for_offers_only(
+                run_spec=run_spec,
+                for_offers_only=for_offers_only,
+                client_version=client_version,
+            )
+            is expected
+        )