Add fail-closed apply for named runs

mkdocs/docs/reference/cli/dstack/apply.md

@@ -17,6 +17,34 @@ $ dstack apply --help
 
 </div>
 
+## Preventing implicit run recreation
+
+Use `--no-recreate` when a script should ensure that an exact named run is active without
+silently changing, stopping, or replacing an existing run:
+
+```shell
+dstack apply -f dev.dstack.yml --no-recreate -y -d
+```
+
+The option supports named dev environment, task, and service configurations. It rejects unnamed
+runs and non-run configurations, and it is mutually exclusive with `--force`.
+
+If the named run is absent or finished, the command follows the normal apply flow. If the run is
+active, the command succeeds as a no-op only when all of the following are true:
+
+- the run is owned by the user requesting the plan;
+- the plan requires an in-place update action but the effective run specification has no changes;
+- the run is `submitted`, `provisioning`, or `running`.
+
+Any configuration change is rejected, including one that could normally be updated in place.
+Recreation actions and runs in `pending` or `terminating` state are also rejected. These failures
+happen before the CLI sends a stop or apply request. Stop the run explicitly before applying a
+configuration that needs to replace it.
+
+For absent or finished runs, the server still validates that the resource observed by the plan has
+not changed before applying it. A concurrent same-name run therefore causes the apply to fail
+instead of being overwritten.
+
 ## User SSH key
 
 By default, `dstack` uses its own SSH key to attach to runs (`~/.dstack/ssh/id_rsa`).

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

@@ -11,9 +11,25 @@
 )
 from dstack._internal.cli.utils.common import console
 from dstack._internal.core.errors import CLIError
-from dstack._internal.core.models.configurations import ApplyConfigurationType
+from dstack._internal.core.models.configurations import (
+    AnyApplyConfiguration,
+    ApplyConfigurationType,
+)
 
 NOTSET = object()
+_RUN_CONFIGURATION_TYPES = {
+    ApplyConfigurationType.DEV_ENVIRONMENT,
+    ApplyConfigurationType.TASK,
+    ApplyConfigurationType.SERVICE,
+}
+
+
+def validate_no_recreate_configuration(configuration: AnyApplyConfiguration) -> None:
+    configuration_type = ApplyConfigurationType(configuration.type)
+    if configuration_type not in _RUN_CONFIGURATION_TYPES:
+        raise CLIError("--no-recreate is only supported for run configurations")
+    if configuration.name is None:
+        raise CLIError("--no-recreate requires a named run")
 
 
 class ApplyCommand(APIBaseCommand):
@@ -51,11 +67,17 @@ def _register(self):
             help="Do not ask for confirmation",
             action="store_true",
         )
-        self._parser.add_argument(
+        apply_safety = self._parser.add_mutually_exclusive_group()
+        apply_safety.add_argument(
             "--force",
             help="Force apply when no changes detected",
             action="store_true",
         )
+        apply_safety.add_argument(
+            "--no-recreate",
+            help="Fail instead of changing an active run; unchanged active runs are no-op",
+            action="store_true",
+        )
         self._parser.add_argument(
             "-d",
             "--detach",
@@ -89,6 +111,8 @@ def _command(self, args: argparse.Namespace):
             if not args.yes and args.configuration_file == APPLY_STDIN_NAME:
                 raise CLIError("Cannot read configuration from stdin if -y/--yes is not specified")
             configuration_path, configuration = load_apply_configuration(args.configuration_file)
+            if args.no_recreate:
+                validate_no_recreate_configuration(configuration)
             configurator_class = get_apply_configurator_class(configuration.type)
             configurator = configurator_class(api_client=self.api)
             configurator_parser = configurator.get_parser()

src/dstack/_internal/cli/services/configurators/run.py

@@ -6,6 +6,8 @@
 import subprocess
 import sys
 import time
+from dataclasses import dataclass
+from enum import Enum
 from pathlib import Path
 from typing import Dict, List, Optional, TypeVar
 
@@ -65,6 +67,9 @@
     RunSpec,
     RunStatus,
 )
+from dstack._internal.core.models.runs import (
+    Run as RunModel,
+)
 from dstack._internal.core.services.diff import diff_models
 from dstack._internal.core.services.repos import get_repo_creds_and_default_branch
 from dstack._internal.core.services.ssh.ports import PortUsedError
@@ -77,12 +82,70 @@
 from dstack.api._public.runs import Run
 
 _BIND_ADDRESS_ARG = "bind_address"
+_NO_RECREATE_NOOP_STATUSES = {
+    RunStatus.SUBMITTED,
+    RunStatus.PROVISIONING,
+    RunStatus.RUNNING,
+}
 
 logger = get_logger(__name__)
 
 RunConfigurationT = TypeVar("RunConfigurationT", bound=AnyRunConfiguration)
 
 
+class ApplyPlanResult(str, Enum):
+    """Outcome of the non-streaming part of applying a run plan."""
+
+    CANCELLED = "cancelled"
+    NOOP = "noop"
+    SUBMITTED = "submitted"
+
+
+@dataclass(frozen=True)
+class RunApplyFence:
+    run_id: str
+    deployment_num: int
+    user: str
+
+
+@dataclass(frozen=True)
+class ApplyPlanOutcome:
+    result: ApplyPlanResult
+    fence: Optional[RunApplyFence] = None
+
+
+def get_run_apply_fence(run: RunModel) -> RunApplyFence:
+    return RunApplyFence(
+        run_id=str(run.id),
+        deployment_num=run.deployment_num,
+        user=run.user,
+    )
+
+
+def validate_no_recreate_plan(run_plan: RunPlan) -> Optional[RunApplyFence]:
+    """Validate an active plan using the fail-closed no-recreate contract."""
+    current = run_plan.current_resource
+    if current is None or current.status.is_finished():
+        return None
+    run_name = run_plan.run_spec.run_name
+    if current.user != run_plan.user:
+        raise CLIError(
+            f"Refusing to use active run {run_name} with --no-recreate;"
+            " the run is owned by another user"
+        )
+    diff = render_run_spec_diff(run_plan.get_effective_run_spec(), current.run_spec)
+    if (
+        run_plan.action == ApplyAction.UPDATE
+        and diff is None
+        and current.status in _NO_RECREATE_NOOP_STATUSES
+    ):
+        return get_run_apply_fence(current)
+    raise CLIError(
+        f"Refusing to change active run {run_name} with --no-recreate;"
+        " stop it explicitly before applying this configuration"
+    )
+
+
 class BaseRunConfigurator(
     ApplyEnvVarsConfiguratorMixin,
     BaseApplyConfigurator[RunConfigurationT],
@@ -99,12 +162,15 @@ def apply_configuration(
             configuration_path=configuration_path,
             configurator_args=configurator_args,
         )
-        return self.apply_plan(
+        self.apply_plan(
             run_plan=run_plan,
             repo=repo,
             command_args=command_args,
             configurator_args=configurator_args,
         )
+        # Preserve the historical apply_configuration() return contract. Callers that need a
+        # structured outcome can invoke apply_plan() directly.
+        return None
 
     def get_plan(
         self,
@@ -145,9 +211,16 @@ def apply_plan(
         command_args: argparse.Namespace,
         configurator_args: argparse.Namespace,
         plan_properties: Optional[Dict[str, str]] = None,
-    ):
-        """Apply a run plan using the standard CLI behavior."""
+    ) -> Optional[ApplyPlanOutcome]:
+        """Apply a run plan using the standard CLI behavior.
+
+        A structured outcome is returned for cancellation, no-op, and detached submission.
+        Foreground apply retains its existing attach and interrupt behavior and may return ``None``.
+        """
         run_name = run_plan.run_spec.run_name
+        no_recreate = getattr(command_args, "no_recreate", False)
+        if no_recreate and run_name is None:
+            raise CLIError("--no-recreate requires a named run")
 
         no_fleets = False
         if len(run_plan.job_plans[0].offers) == 0:
@@ -171,6 +244,14 @@ def apply_plan(
                 run_plan.get_effective_run_spec(),
                 run_plan.current_resource.run_spec,
             )
+            if no_recreate and not run_plan.current_resource.status.is_finished():
+                fence = validate_no_recreate_plan(run_plan)
+                assert fence is not None
+                console.print(f"Run [code]{run_name}[/] is already up to date.")
+                return ApplyPlanOutcome(
+                    result=ApplyPlanResult.NOOP,
+                    fence=fence,
+                )
             if run_plan.action == ApplyAction.UPDATE and diff is not None:
                 console.print(
                     f"Active run [code]{run_name}[/] already exists."
@@ -184,7 +265,10 @@ def apply_plan(
                 )
                 if command_args.yes and not command_args.force:
                     console.print("Use --force to apply anyway.")
-                    return
+                    return ApplyPlanOutcome(
+                        result=ApplyPlanResult.NOOP,
+                        fence=get_run_apply_fence(run_plan.current_resource),
+                    )
                 confirm_message = "Stop and override the run?"
             elif not run_plan.current_resource.status.is_finished():
                 stop_run_name = run_plan.current_resource.run_spec.run_name
@@ -198,7 +282,7 @@ def apply_plan(
 
         if not command_args.yes and not confirm_ask(confirm_message):
             console.print("\nExiting...")
-            return
+            return ApplyPlanOutcome(result=ApplyPlanResult.CANCELLED)
 
         if stop_run_name is not None:
             with console.status("Stopping run..."):
@@ -229,7 +313,10 @@ def apply_plan(
             if run_plan.action == ApplyAction.UPDATE:
                 detach_message = f"Run [code]{run.name}[/] updated, detaching..."
             console.print(detach_message)
-            return
+            return ApplyPlanOutcome(
+                result=ApplyPlanResult.SUBMITTED,
+                fence=get_run_apply_fence(run._run),
+            )
 
         abort_at_exit = False
         try:

src/tests/_internal/cli/commands/test_apply.py

@@ -0,0 +1,82 @@
+from types import SimpleNamespace
+from typing import cast
+
+import pytest
+
+from dstack._internal.cli.commands.apply import validate_no_recreate_configuration
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.configurations import (
+    AnyApplyConfiguration,
+    ApplyConfigurationType,
+)
+from tests._internal.cli.common import run_dstack_cli
+
+
+def _configuration(
+    configuration_type: ApplyConfigurationType,
+    *,
+    name: str | None,
+) -> AnyApplyConfiguration:
+    return cast(
+        AnyApplyConfiguration,
+        SimpleNamespace(type=configuration_type.value, name=name),
+    )
+
+
+def test_help_documents_no_recreate(capsys: pytest.CaptureFixture[str]) -> None:
+    assert run_dstack_cli(["apply", "--help"]) == 0
+
+    normalized_output = " ".join(capsys.readouterr().out.split())
+    assert "[--force | --no-recreate]" in normalized_output
+    assert (
+        "Fail instead of changing an active run; unchanged active runs are no-op"
+        in normalized_output
+    )
+
+
+def test_no_recreate_is_mutually_exclusive_with_force(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    assert run_dstack_cli(["apply", "--force", "--no-recreate"]) == 2
+
+    assert "not allowed with argument --force" in capsys.readouterr().err
+
+
+@pytest.mark.parametrize(
+    "configuration_type",
+    [
+        ApplyConfigurationType.FLEET,
+        ApplyConfigurationType.GATEWAY,
+        ApplyConfigurationType.VOLUME,
+    ],
+)
+def test_no_recreate_rejects_non_run_configurations(
+    configuration_type: ApplyConfigurationType,
+) -> None:
+    configuration = _configuration(configuration_type, name="resource")
+
+    with pytest.raises(CLIError, match="only supported for run configurations"):
+        validate_no_recreate_configuration(configuration)
+
+
+@pytest.mark.parametrize(
+    "configuration_type",
+    [
+        ApplyConfigurationType.DEV_ENVIRONMENT,
+        ApplyConfigurationType.TASK,
+        ApplyConfigurationType.SERVICE,
+    ],
+)
+def test_no_recreate_accepts_named_run_configurations(
+    configuration_type: ApplyConfigurationType,
+) -> None:
+    configuration = _configuration(configuration_type, name="dev-run")
+
+    validate_no_recreate_configuration(configuration)
+
+
+def test_no_recreate_rejects_unnamed_run_configuration() -> None:
+    configuration = _configuration(ApplyConfigurationType.DEV_ENVIRONMENT, name=None)
+
+    with pytest.raises(CLIError, match="requires a named run"):
+        validate_no_recreate_configuration(configuration)

src/tests/_internal/cli/services/configurators/test_run.py

@@ -113,15 +113,16 @@ class TestApplyConfiguration:
     def test_composes_get_plan_and_apply_plan(self, monkeypatch):
         run_plan, repo = Mock(), Mock()
         get_plan = Mock(return_value=(run_plan, repo))
-        apply_plan = Mock()
+        apply_plan = Mock(return_value=object())
         monkeypatch.setattr(ServiceConfigurator, "get_plan", get_plan)
         monkeypatch.setattr(ServiceConfigurator, "apply_plan", apply_plan)
         conf, command_args, configurator_args = Mock(), Mock(), Mock()
 
-        ServiceConfigurator(api_client=Mock()).apply_configuration(
+        result = ServiceConfigurator(api_client=Mock()).apply_configuration(
             conf, "svc.dstack.yml", command_args, configurator_args
         )
 
+        assert result is None
         get_plan.assert_called_once_with(
             conf=conf,
             configuration_path="svc.dstack.yml",