Restrict sandbox plugin to services

 * Reject compiled operations that enable plugins.sandbox on non-service runs.
 * Allow disabled sandbox configs to keep existing job specs valid.

Author: md-mq

cli/polyaxon/_flow/operations/compiled_operation.py

@@ -1,10 +1,13 @@
 from typing import Dict, List, Optional
 from typing_extensions import Literal
 
+from clipped.compact.pydantic import model_validator, validation_after
+from clipped.config.schema import skip_partial
+
 from polyaxon._flow.io import V1IO
 from polyaxon._flow.operations.base import BaseOp
 from polyaxon._flow.params import ParamSpec, ops_params
-from polyaxon._flow.run import RunMixin, V1Runtime
+from polyaxon._flow.run import RunMixin, V1RunKind, V1Runtime
 from polyaxon.exceptions import PolyaxonSchemaError
 
 
@@ -21,6 +24,23 @@ class V1CompiledOperation(BaseOp, RunMixin):
     def get_run_kind(self):
         return self.run.kind if self.run else None
 
+    @model_validator(**validation_after)
+    @skip_partial
+    def validate_sandbox_kind(cls, values):
+        plugins = cls.get_value_for_key("plugins", values)
+        sandbox = cls.get_value_for_key("sandbox", plugins) if plugins else None
+        if sandbox is None or sandbox is False:
+            return values
+
+        run = cls.get_value_for_key("run", values)
+        if not run or run.kind != V1RunKind.SERVICE:
+            got = run.kind if run else "unset"
+            raise ValueError(
+                "plugins.sandbox is only supported on kind: service in this version. "
+                f"Got kind: {got}."
+            )
+        return values
+
     def validate_params(
         self,
         params: Optional[Dict] = None,

cli/tests/test_polyflow/test_ops/test_op_runs.py

@@ -864,6 +864,35 @@ def test_job_refs_params(self):
                 params=params, inputs=config.inputs, outputs=None, is_template=False
             )
 
+    def test_sandbox_requires_service(self):
+        config_dict = {
+            "plugins": {"sandbox": True},
+            "run": {"kind": V1RunKind.JOB, "container": {"image": "test"}},
+        }
+        with self.assertRaises(ValidationError) as ctx:
+            V1CompiledOperation.from_dict(config_dict)
+        assert "plugins.sandbox is only supported" in str(ctx.exception)
+
+        config_dict = {
+            "plugins": {"sandbox": "{{ inputs.sandbox }}"},
+            "run": {"kind": V1RunKind.JOB, "container": {"image": "test"}},
+        }
+        with self.assertRaises(ValidationError) as ctx:
+            V1CompiledOperation.from_dict(config_dict)
+        assert "plugins.sandbox is only supported" in str(ctx.exception)
+
+        config_dict = {
+            "plugins": {"sandbox": False},
+            "run": {"kind": V1RunKind.JOB, "container": {"image": "test"}},
+        }
+        V1CompiledOperation.from_dict(config_dict)
+
+        config_dict = {
+            "plugins": {"sandbox": True},
+            "run": {"kind": V1RunKind.SERVICE, "container": {"image": "test"}},
+        }
+        V1CompiledOperation.from_dict(config_dict)
+
     def test_executable(self):
         config_dict = {
             "startAt": "foo",

cli/tests/test_polyflow/test_plugins/test_plugins.py

@@ -97,7 +97,7 @@ def test_get_from_spec(self):
                     "externalHost": True,
                     "sandbox": True,
                 },
-                "run": {"kind": V1RunKind.JOB, "container": {"image": "test"}},
+                "run": {"kind": V1RunKind.SERVICE, "container": {"image": "test"}},
             }
         )
         plugins = compiled_operation.plugins