Fix #5682: add IS_OWNER for permissions added by Python mutators

NEXT_CHANGELOG.md

@@ -11,6 +11,7 @@
 ### Bundles
 
  * `bundle generate job` now downloads workspace files referenced by `spark_python_task`, rewriting them to a relative path like it already does for notebooks. Git-sourced files and cloud URIs are left untouched ([#5799](https://github.com/databricks/cli/pull/5799)).
+* Fix permissions added to a job or pipeline by a Python (PyDABs) mutator failing to deploy with "must have exactly one owner"; the deploying identity is now set as owner, matching resources whose permissions are declared in YAML ([#5821](https://github.com/databricks/cli/pull/5821)).
 
 ### Dependency updates
 

acceptance/bundle/python/mutator-permissions-owner-5682/databricks.yml.tmpl

@@ -0,0 +1,21 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+variables:
+  grantee_group:
+    default: some-group
+
+python:
+  mutators:
+    - "mutators:add_pipeline_permission"
+
+resources:
+  pipelines:
+    my_pipeline:
+      name: test-pipeline-$UNIQUE_NAME
+      catalog: main
+      schema: default
+      serverless: true
+      libraries:
+        - notebook:
+            path: ./nb.py

acceptance/bundle/python/mutator-permissions-owner-5682/mutators.py

@@ -0,0 +1,20 @@
+from dataclasses import replace
+from databricks.bundles.pipelines import Pipeline, PipelinePermission
+from databricks.bundles.core import pipeline_mutator, Bundle
+
+
+# Regression test for https://github.com/databricks/cli/issues/5682.
+# Adds a permission to a pipeline that is already defined in YAML (mirrors the
+# reporter's mutator, which reads a bundle variable). Resources updated by a
+# PythonMutator go through NormalizeResources; that path now runs FixPermissions,
+# so the deploying user is added as IS_OWNER and the permissions PUT succeeds.
+# Before the fix the ACL had no owner and the backend rejected the PUT with
+# "The pipeline must have exactly one owner".
+@pipeline_mutator
+def add_pipeline_permission(bundle: Bundle, pipeline: Pipeline) -> Pipeline:
+    group = bundle.resolve_variable(bundle.variables["grantee_group"])
+    permissions = [
+        *pipeline.permissions,
+        PipelinePermission.from_dict({"group_name": group, "level": "CAN_RUN"}),
+    ]
+    return replace(pipeline, permissions=permissions)

acceptance/bundle/python/mutator-permissions-owner-5682/nb.py

@@ -0,0 +1 @@
+# Databricks notebook source

acceptance/bundle/python/mutator-permissions-owner-5682/output.txt

@@ -0,0 +1,12 @@
+
+>>> uv run [UV_ARGS] -q [CLI] bundle validate --output json
+[
+  "CAN_RUN",
+  "IS_OWNER"
+]
+
+>>> uv run [UV_ARGS] -q [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!

acceptance/bundle/python/mutator-permissions-owner-5682/script

@@ -0,0 +1,22 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+if [ -n "$CLOUD_ENV" ]; then
+    # Unique group so parallel cloud runs don't collide; must exist for the PUT to
+    # reach the owner check rather than failing on an unknown principal.
+    export BUNDLE_VAR_grantee_group="dabs-5682-$UNIQUE_NAME"
+    $CLI groups create --display-name "$BUNDLE_VAR_grantee_group" &> /dev/null || true
+fi
+
+cleanup() {
+    uv run $UV_ARGS -q $CLI bundle destroy --auto-approve &> LOG.destroy
+    rm -fr .databricks __pycache__
+}
+trap cleanup EXIT
+
+# The Python mutator adds a CAN_RUN permission; FixPermissions then adds IS_OWNER for
+# the current user (issue #5682 - previously it did not run on Python-updated resources).
+trace uv run $UV_ARGS -q $CLI bundle validate --output json | \
+  jq ".resources.pipelines.my_pipeline.permissions | map(.level)"
+
+# With the owner present, the deploy succeeds on both engines.
+trace uv run $UV_ARGS -q $CLI bundle deploy

acceptance/bundle/python/mutator-permissions-owner-5682/test.toml

@@ -0,0 +1,2 @@
+Cloud = true
+RequiresUnityCatalog = true

bundle/config/mutator/resourcemutator/fix_permissions.go

@@ -211,6 +211,12 @@ func createPermissionFromPrincipal(principal, level string) dyn.Value {
 }
 
 func (m *fixPermissions) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	// CurrentUser is populated by PopulateCurrentUser early in the initialize phase.
+	// It can be nil when this mutator runs outside that phase (e.g. NormalizeResources
+	// after PythonMutator); there is no user to add as owner, so skip.
+	if b.Config.Workspace.CurrentUser == nil {
+		return nil
+	}
 	currentUser := b.Config.Workspace.CurrentUser.UserName
 
 	err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {

bundle/config/mutator/resourcemutator/resource_mutator.go

@@ -295,6 +295,17 @@ func NormalizeResources(
 		return
 	}
 
+	// Permissions added to an existing resource by a Python mutator must still get the
+	// deploying user as IS_OWNER, otherwise the Permissions API rejects the PUT with
+	// "must have exactly one owner" (#5682). FixPermissions is idempotent, so re-running
+	// it on resources that already have an owner is a no-op. ApplyBundlePermissions is
+	// intentionally not re-run here: it is not idempotent (it appends bundle-level
+	// permissions) and already ran for these resources in ProcessStaticResources.
+	bundle.ApplyContext(ctx, b, FixPermissions())
+	if logdiag.HasError(ctx) {
+		return
+	}
+
 	// after mutators, we merge updated resources back to snapshot to preserve non-selected resources
 	err = b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) {
 		return mergeResources(root, snapshot)

bundle/direct/dresources/all_test.go

@@ -358,7 +358,8 @@ var testDeps = map[string]prepareWorkspace{
 		return &PermissionsState{
 			ObjectID: "/pipelines/" + resp.PipelineId,
 			EmbeddedSlice: []StatePermission{{
-				Level:    "CAN_MANAGE",
+				// Pipelines require exactly one owner, like jobs.
+				Level:    "IS_OWNER",
 				UserName: "user@example.com",
 			}},
 		}, nil

libs/testserver/permissions.go

@@ -314,25 +314,27 @@ func (s *FakeWorkspace) SetPermissions(req Request) any {
 		})
 	}
 
-	// Validate job ownership requirements
-	if requestObjectType == "jobs" {
-		hasOwner := false
+	// Jobs and pipelines require exactly one owner. The real Permissions API rejects
+	// a PUT with zero owners OR more than one owner (verified against the backend);
+	// both cases return "The <type> must have exactly one owner."
+	ownerNoun := map[string]string{"jobs": "job", "pipelines": "pipeline"}[requestObjectType]
+	if ownerNoun != "" {
+		owners := 0
 		for _, acl := range existingPermissions.AccessControlList {
 			for _, perm := range acl.AllPermissions {
 				if perm.PermissionLevel == "IS_OWNER" {
-					hasOwner = true
-					break
+					owners++
 				}
 			}
-			if hasOwner {
-				break
-			}
 		}
 
-		if !hasOwner {
+		if owners != 1 {
 			return Response{
 				StatusCode: 400,
-				Body:       map[string]string{"message": "The job must have exactly one owner."},
+				Body: map[string]string{
+					"error_code": "INVALID_PARAMETER_VALUE",
+					"message":    "The " + ownerNoun + " must have exactly one owner.",
+				},
 			}
 		}
 	}