diff --git a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/out.plan.direct.txt b/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/out.plan.direct.txt
deleted file mode 100644
index 03adb9ae87..0000000000
--- a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/out.plan.direct.txt
+++ /dev/null
@@ -1,3 +0,0 @@
-update jobs.job_with_permissions.permissions
-
-Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged
diff --git a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/out.plan.terraform.txt b/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/out.plan.terraform.txt
deleted file mode 100644
index 045b1c2486..0000000000
--- a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/out.plan.terraform.txt
+++ /dev/null
@@ -1 +0,0 @@
-Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
diff --git a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/output.txt b/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/output.txt
index e229cda82a..b0cd23c833 100644
--- a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/output.txt
+++ b/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/output.txt
@@ -8,6 +8,7 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
 
 === Delete one permission and deploy again
 
diff --git a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/script b/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/script
index d9e45e18bd..cc1d9464e1 100644
--- a/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/script
+++ b/acceptance/bundle/resources/permissions/jobs/delete_one/cloud/script
@@ -33,8 +33,7 @@ trace $CLI bundle deploy
 # although they are semantically the same. We're not doing the same transformation in direct, because permissions get endpoint uses a different order.
 print_requests | sort_acl_requests > out.requests_create.json
 
-# Badness: because of remote reordering we have drift in direct there
-trace $CLI bundle plan > out.plan.$DATABRICKS_BUNDLE_ENGINE.txt
+trace $CLI bundle plan
 
 job_id=$($CLI bundle summary --output json | jq -r '.resources.jobs.job_with_permissions.id')
 
diff --git a/acceptance/bundle/resources/permissions/jobs/delete_one/local/out.plan_update.direct.json b/acceptance/bundle/resources/permissions/jobs/delete_one/local/out.plan_update.direct.json
index 6109bdb49a..1366b05c71 100644
--- a/acceptance/bundle/resources/permissions/jobs/delete_one/local/out.plan_update.direct.json
+++ b/acceptance/bundle/resources/permissions/jobs/delete_one/local/out.plan_update.direct.json
@@ -100,7 +100,7 @@
       },
       "changes": {
         "local": {
-          "permissions": {
+          "permissions[user_name='test-dabs-1@databricks.com']": {
             "action": "update"
           }
         }
diff --git a/acceptance/bundle/resources/permissions/jobs/deleted_remotely/out.plan_restore.direct.json b/acceptance/bundle/resources/permissions/jobs/deleted_remotely/out.plan_restore.direct.json
index 0c6b2fcac3..f79a3b3e73 100644
--- a/acceptance/bundle/resources/permissions/jobs/deleted_remotely/out.plan_restore.direct.json
+++ b/acceptance/bundle/resources/permissions/jobs/deleted_remotely/out.plan_restore.direct.json
@@ -88,7 +88,10 @@
       },
       "changes": {
         "remote": {
-          "permissions": {
+          "permissions[group_name='data-team']": {
+            "action": "update"
+          },
+          "permissions[user_name='viewer@example.com']": {
             "action": "update"
           }
         }
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_locally/databricks.yml b/acceptance/bundle/resources/permissions/jobs/reorder_locally/databricks.yml
new file mode 100644
index 0000000000..4a5270fe68
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_locally/databricks.yml
@@ -0,0 +1,19 @@
+bundle:
+  name: test-bundle
+
+resources:
+  jobs:
+    foo:
+      name: job1
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/Users/user@example.com/notebook
+            source: WORKSPACE
+      permissions:
+        - {"level": "CAN_VIEW", "user_name": "viewer@example.com"}
+        - level: CAN_MANAGE
+          group_name: data-team
+        - level: CAN_MANAGE
+          service_principal_name: f37d18cd-98a8-4db5-8112-12dd0a6bfe38
+          # PLACEHOLDER
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_locally/out.test.toml b/acceptance/bundle/resources/permissions/jobs/reorder_locally/out.test.toml
new file mode 100644
index 0000000000..d560f1de04
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_locally/out.test.toml
@@ -0,0 +1,5 @@
+Local = true
+Cloud = false
+
+[EnvMatrix]
+  DATABRICKS_BUNDLE_ENGINE = ["terraform", "direct"]
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_locally/output.txt b/acceptance/bundle/resources/permissions/jobs/reorder_locally/output.txt
new file mode 100644
index 0000000000..73bedc7d33
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_locally/output.txt
@@ -0,0 +1,18 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_locally/script b/acceptance/bundle/resources/permissions/jobs/reorder_locally/script
new file mode 100644
index 0000000000..be6a8540ec
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_locally/script
@@ -0,0 +1,7 @@
+trace $CLI bundle deploy
+update_file.py databricks.yml '- {"level": "CAN_VIEW", "user_name": "viewer@example.com"}' ''
+update_file.py databricks.yml '  # PLACEHOLDER' '- {"level": "CAN_VIEW", "user_name": "viewer@example.com"}'
+
+trace $CLI bundle plan
+trace $CLI bundle deploy
+trace $CLI bundle plan
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_locally/test.toml b/acceptance/bundle/resources/permissions/jobs/reorder_locally/test.toml
new file mode 100644
index 0000000000..a030353d57
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_locally/test.toml
@@ -0,0 +1 @@
+RecordRequests = false
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_remotely/databricks.yml b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/databricks.yml
new file mode 100644
index 0000000000..1ec2f87780
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/databricks.yml
@@ -0,0 +1,18 @@
+bundle:
+  name: test-bundle
+
+resources:
+  jobs:
+    foo:
+      name: job1
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/Users/user@example.com/notebook
+            source: WORKSPACE
+      permissions:
+        - {"level": "CAN_VIEW", "user_name": "viewer@example.com"}
+        - level: CAN_MANAGE
+          group_name: data-team
+        - level: CAN_MANAGE
+          service_principal_name: f37d18cd-98a8-4db5-8112-12dd0a6bfe38
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_remotely/out.test.toml b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/out.test.toml
new file mode 100644
index 0000000000..d560f1de04
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/out.test.toml
@@ -0,0 +1,5 @@
+Local = true
+Cloud = false
+
+[EnvMatrix]
+  DATABRICKS_BUNDLE_ENGINE = ["terraform", "direct"]
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_remotely/output.txt b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/output.txt
new file mode 100644
index 0000000000..b6052360bf
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/output.txt
@@ -0,0 +1,14 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> [CLI] permissions set jobs [NUMID] --json @reorder.json
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_remotely/reorder.json b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/reorder.json
new file mode 100644
index 0000000000..f826a97da0
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/reorder.json
@@ -0,0 +1,8 @@
+{
+  "access_control_list": [
+      {"permission_level": "CAN_MANAGE", "group_name": "data-team"},
+      {"permission_level": "CAN_VIEW", "user_name": "viewer@example.com"},
+      {"permission_level": "IS_OWNER", "user_name": "tester@databricks.com"},
+      {"permission_level": "CAN_MANAGE", "service_principal_name": "f37d18cd-98a8-4db5-8112-12dd0a6bfe38"}
+  ]
+}
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_remotely/script b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/script
new file mode 100644
index 0000000000..71389de5c9
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/script
@@ -0,0 +1,8 @@
+trace $CLI bundle deploy
+trace $CLI bundle plan | contains.py "2 unchanged"
+
+job_id=$($CLI bundle summary --output json | jq -r '.resources.jobs.foo.id')
+
+trace $CLI permissions set jobs $job_id --json @reorder.json > /dev/null
+
+trace $CLI bundle plan | contains.py "2 unchanged"
diff --git a/acceptance/bundle/resources/permissions/jobs/reorder_remotely/test.toml b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/test.toml
new file mode 100644
index 0000000000..a030353d57
--- /dev/null
+++ b/acceptance/bundle/resources/permissions/jobs/reorder_remotely/test.toml
@@ -0,0 +1 @@
+RecordRequests = false
diff --git a/acceptance/bundle/resources/permissions/jobs/update/out.plan_delete_one.direct.json b/acceptance/bundle/resources/permissions/jobs/update/out.plan_delete_one.direct.json
index b26a8e0320..4cd774c170 100644
--- a/acceptance/bundle/resources/permissions/jobs/update/out.plan_delete_one.direct.json
+++ b/acceptance/bundle/resources/permissions/jobs/update/out.plan_delete_one.direct.json
@@ -92,7 +92,7 @@
       },
       "changes": {
         "local": {
-          "permissions": {
+          "permissions[group_name='data-team']": {
             "action": "update"
           }
         }
diff --git a/acceptance/bundle/resources/permissions/jobs/update/out.plan_update.direct.json b/acceptance/bundle/resources/permissions/jobs/update/out.plan_update.direct.json
index 25f5ffc8c3..b407988457 100644
--- a/acceptance/bundle/resources/permissions/jobs/update/out.plan_update.direct.json
+++ b/acceptance/bundle/resources/permissions/jobs/update/out.plan_update.direct.json
@@ -96,7 +96,7 @@
       },
       "changes": {
         "local": {
-          "permissions[0].permission_level": {
+          "permissions[user_name='viewer@example.com'].permission_level": {
             "action": "update"
           }
         }
diff --git a/acceptance/bundle/resources/permissions/pipelines/update/out.plan_delete_one.direct.json b/acceptance/bundle/resources/permissions/pipelines/update/out.plan_delete_one.direct.json
index 4cc6f95d91..76c46e7a58 100644
--- a/acceptance/bundle/resources/permissions/pipelines/update/out.plan_delete_one.direct.json
+++ b/acceptance/bundle/resources/permissions/pipelines/update/out.plan_delete_one.direct.json
@@ -72,7 +72,7 @@
       },
       "changes": {
         "local": {
-          "permissions": {
+          "permissions[group_name='data-team']": {
             "action": "update"
           }
         }
diff --git a/acceptance/bundle/resources/permissions/pipelines/update/out.plan_update.direct.json b/acceptance/bundle/resources/permissions/pipelines/update/out.plan_update.direct.json
index 66f025be6d..cb547063ee 100644
--- a/acceptance/bundle/resources/permissions/pipelines/update/out.plan_update.direct.json
+++ b/acceptance/bundle/resources/permissions/pipelines/update/out.plan_update.direct.json
@@ -76,7 +76,7 @@
       },
       "changes": {
         "local": {
-          "permissions[0].permission_level": {
+          "permissions[user_name='viewer@example.com'].permission_level": {
             "action": "update"
           }
         }
diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go
index 8103fde30f..5cd389417b 100644
--- a/bundle/direct/bundle_plan.go
+++ b/bundle/direct/bundle_plan.go
@@ -144,7 +144,7 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 		// for integers: compare 0 with actual object ID. As long as real object IDs are never 0 we're good.
 		// Once we add non-id fields or add per-field details to "bundle plan", we must read dynamic data and deal with references as first class citizen.
 		// This means distinguishing between 0 that are actually object ids and 0 that are there because typed struct integer cannot contain ${...} string.
-		localDiff, err := structdiff.GetStructDiff(savedState, entry.NewState.Value)
+		localDiff, err := structdiff.GetStructDiff(savedState, entry.NewState.Value, adapter.KeyedSlices())
 		if err != nil {
 			logdiag.LogError(ctx, fmt.Errorf("%s: diffing local state: %w", errorPrefix, err))
 			return false
@@ -187,7 +187,7 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 				return false
 			}
 
-			remoteDiff, err := structdiff.GetStructDiff(savedState, remoteStateComparable)
+			remoteDiff, err := structdiff.GetStructDiff(savedState, remoteStateComparable, adapter.KeyedSlices())
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: diffing remote state: %w", errorPrefix, err))
 				return false
diff --git a/bundle/direct/dresources/adapter.go b/bundle/direct/dresources/adapter.go
index 3e57bb80d2..e747cb6fca 100644
--- a/bundle/direct/dresources/adapter.go
+++ b/bundle/direct/dresources/adapter.go
@@ -83,6 +83,10 @@ type IResource interface {
 
 	// [Optional] WaitAfterUpdate waits for the resource to become ready after update. Returns optionally updated remote state.
 	WaitAfterUpdate(ctx context.Context, newState any) (remoteState any, e error)
+
+	// [Optional] KeyedSlices returns a map from path patterns to KeyFunc for comparing slices by key instead of by index.
+	// Example: func (*ResourcePermissions) KeyedSlices(state *PermissionsState) map[string]any
+	KeyedSlices() map[string]any
 }
 
 // Adapter wraps resource implementation, validates signatures and type consistency across methods
@@ -106,6 +110,7 @@ type Adapter struct {
 
 	fieldTriggersLocal  map[string]deployplan.ActionType
 	fieldTriggersRemote map[string]deployplan.ActionType
+	keyedSlices         map[string]any
 }
 
 func NewAdapter(typedNil any, client *databricks.WorkspaceClient) (*Adapter, error) {
@@ -136,6 +141,7 @@ func NewAdapter(typedNil any, client *databricks.WorkspaceClient) (*Adapter, err
 		classifyChange:      nil,
 		fieldTriggersLocal:  map[string]deployplan.ActionType{},
 		fieldTriggersRemote: map[string]deployplan.ActionType{},
+		keyedSlices:         nil,
 	}
 
 	err = adapter.initMethods(impl)
@@ -194,6 +200,16 @@ func loadFieldTriggers(triggerCall *calladapt.BoundCaller, isLocal bool) (map[st
 	return result, nil
 }
 
+// loadKeyedSlices validates and calls KeyedSlices method, returning the resulting map.
+func loadKeyedSlices(call *calladapt.BoundCaller) (map[string]any, error) {
+	outs, err := call.Call()
+	if err != nil {
+		return nil, fmt.Errorf("failed to call KeyedSlices: %w", err)
+	}
+	result := outs[0].(map[string]any)
+	return result, nil
+}
+
 func (a *Adapter) initMethods(resource any) error {
 	err := calladapt.EnsureNoExtraMethods(resource, calladapt.TypeOf[IResource]())
 	if err != nil {
@@ -262,6 +278,17 @@ func (a *Adapter) initMethods(resource any) error {
 		return err
 	}
 
+	keyedSlicesCall, err := calladapt.PrepareCall(resource, calladapt.TypeOf[IResource](), "KeyedSlices")
+	if err != nil {
+		return err
+	}
+	if keyedSlicesCall != nil {
+		a.keyedSlices, err = loadKeyedSlices(keyedSlicesCall)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -610,6 +637,12 @@ func (a *Adapter) ClassifyChange(change structdiff.Change, remoteState any, isLo
 	return actionType, nil
 }
 
+// KeyedSlices returns a map from path patterns to KeyFunc for comparing slices by key.
+// If the resource doesn't implement KeyedSlices, returns nil.
+func (a *Adapter) KeyedSlices() map[string]any {
+	return a.keyedSlices
+}
+
 // prepareCallRequired prepares a call and ensures the method is found.
 func prepareCallRequired(resource any, methodName string) (*calladapt.BoundCaller, error) {
 	caller, err := calladapt.PrepareCall(resource, calladapt.TypeOf[IResource](), methodName)
diff --git a/bundle/direct/dresources/all_test.go b/bundle/direct/dresources/all_test.go
index b9c115768d..9ed83ab68f 100644
--- a/bundle/direct/dresources/all_test.go
+++ b/bundle/direct/dresources/all_test.go
@@ -470,7 +470,7 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 	if remoteStateFromUpdate != nil {
 		remappedStateFromUpdate, err := adapter.RemapState(remoteStateFromUpdate)
 		require.NoError(t, err)
-		changes, err := structdiff.GetStructDiff(remappedState, remappedStateFromUpdate)
+		changes, err := structdiff.GetStructDiff(remappedState, remappedStateFromUpdate, nil)
 		require.NoError(t, err)
 		// Filter out timestamp fields that are expected to differ in value
 		var relevantChanges []structdiff.Change
diff --git a/bundle/direct/dresources/permissions.go b/bundle/direct/dresources/permissions.go
index c6da6851d8..9fb407749e 100644
--- a/bundle/direct/dresources/permissions.go
+++ b/bundle/direct/dresources/permissions.go
@@ -85,6 +85,25 @@ func (*ResourcePermissions) PrepareState(s *PermissionsState) *PermissionsState
 	return s
 }
 
+func accessControlRequestKey(x iam.AccessControlRequest) (string, string) {
+	if x.UserName != "" {
+		return "user_name", x.UserName
+	}
+	if x.ServicePrincipalName != "" {
+		return "service_principal_name", x.ServicePrincipalName
+	}
+	if x.GroupName != "" {
+		return "group_name", x.GroupName
+	}
+	return "", ""
+}
+
+func (*ResourcePermissions) KeyedSlices() map[string]any {
+	return map[string]any{
+		"permissions": accessControlRequestKey,
+	}
+}
+
 // parsePermissionsID extracts the object type and ID from a permissions ID string.
 // Handles both 3-part IDs ("/jobs/123") and 4-part IDs ("/sql/warehouses/uuid").
 func parsePermissionsID(id string) (extractedType, extractedID string, err error) {
diff --git a/libs/structs/structaccess/set_test.go b/libs/structs/structaccess/set_test.go
index f2348618c6..0d29bea822 100644
--- a/libs/structs/structaccess/set_test.go
+++ b/libs/structs/structaccess/set_test.go
@@ -460,7 +460,7 @@ func TestSet(t *testing.T) {
 			require.NoError(t, err)
 
 			// Compare the actual changes using structdiff
-			changes, err := structdiff.GetStructDiff(original, target)
+			changes, err := structdiff.GetStructDiff(original, target, nil)
 			require.NoError(t, err)
 			assert.Equal(t, tt.expectedChanges, changes)
 		})
diff --git a/libs/structs/structdiff/bench_test.go b/libs/structs/structdiff/bench_test.go
index 30988d00fc..f040604783 100644
--- a/libs/structs/structdiff/bench_test.go
+++ b/libs/structs/structdiff/bench_test.go
@@ -19,7 +19,7 @@ func bench(b *testing.B, job1, job2 string) {
 
 	b.ResetTimer()
 	for range b.N {
-		changes, err := GetStructDiff(&x, &y)
+		changes, err := GetStructDiff(&x, &y, nil)
 		if err != nil {
 			b.Fatalf("error: %s", err)
 		}
diff --git a/libs/structs/structdiff/diff.go b/libs/structs/structdiff/diff.go
index 03e6423f04..160f14b649 100644
--- a/libs/structs/structdiff/diff.go
+++ b/libs/structs/structdiff/diff.go
@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"slices"
 	"sort"
+	"strings"
 
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/structs/structtag"
@@ -16,10 +17,64 @@ type Change struct {
 	New  any
 }
 
+// KeyFunc extracts a key field name and value from a slice element.
+// It can be either:
+//   - func(T) (string, string) - typed function for specific element type T
+//   - func(any) (string, string) - generic function accepting any element
+//
+// The function returns (keyField, keyValue). The keyField is typically a field name
+// like "task_key", and keyValue is the value that uniquely identifies the element.
+type KeyFunc = any
+
+// keyFuncCaller wraps a KeyFunc and provides a type-checked Call method.
+type keyFuncCaller struct {
+	fn      reflect.Value
+	argType reflect.Type
+}
+
+func newKeyFuncCaller(fn any) (*keyFuncCaller, error) {
+	v := reflect.ValueOf(fn)
+	if v.Kind() != reflect.Func {
+		return nil, fmt.Errorf("KeyFunc must be a function, got %T", fn)
+	}
+	t := v.Type()
+	if t.NumIn() != 1 {
+		return nil, fmt.Errorf("KeyFunc must have exactly 1 parameter, got %d", t.NumIn())
+	}
+	if t.NumOut() != 2 {
+		return nil, fmt.Errorf("KeyFunc must return exactly 2 values, got %d", t.NumOut())
+	}
+	if t.Out(0).Kind() != reflect.String || t.Out(1).Kind() != reflect.String {
+		return nil, fmt.Errorf("KeyFunc must return (string, string), got (%v, %v)", t.Out(0), t.Out(1))
+	}
+	return &keyFuncCaller{fn: v, argType: t.In(0)}, nil
+}
+
+func (c *keyFuncCaller) call(elem any) (string, string) {
+	elemValue := reflect.ValueOf(elem)
+	out := c.fn.Call([]reflect.Value{elemValue})
+	keyField := out[0].String()
+	keyValue := out[1].String()
+	return keyField, keyValue
+}
+
+// diffContext holds configuration for the diff operation.
+type diffContext struct {
+	sliceKeys map[string]KeyFunc
+}
+
 // GetStructDiff compares two Go structs and returns a list of Changes or an error.
 // Respects ForceSendFields if present.
 // Types of a and b must match exactly, otherwise returns an error.
-func GetStructDiff(a, b any) ([]Change, error) {
+//
+// The sliceKeys parameter maps path patterns to functions that extract
+// key field/value pairs from slice elements. When provided, slices at matching
+// paths are compared as maps keyed by (keyField, keyValue) instead of by index.
+// Path patterns use dot notation (e.g., "tasks" or "job.tasks").
+// The [*] wildcard matches any slice index in the path.
+// Note, key wildcard is not supported yet ("a.*.c")
+// Pass nil if no slice key functions are needed.
+func GetStructDiff(a, b any, sliceKeys map[string]KeyFunc) ([]Change, error) {
 	v1 := reflect.ValueOf(a)
 	v2 := reflect.ValueOf(b)
 
@@ -38,24 +93,27 @@ func GetStructDiff(a, b any) ([]Change, error) {
 		return nil, fmt.Errorf("type mismatch: %v vs %v", v1.Type(), v2.Type())
 	}
 
-	diffValues(nil, v1, v2, &changes)
+	ctx := &diffContext{sliceKeys: sliceKeys}
+	if err := diffValues(ctx, nil, v1, v2, &changes); err != nil {
+		return nil, err
+	}
 	return changes, nil
 }
 
 // diffValues appends changes between v1 and v2 to the slice.  path is the current
 // JSON-style path (dot + brackets).  At the root path is "".
-func diffValues(path *structpath.PathNode, v1, v2 reflect.Value, changes *[]Change) {
+func diffValues(ctx *diffContext, path *structpath.PathNode, v1, v2 reflect.Value, changes *[]Change) error {
 	if !v1.IsValid() {
 		if !v2.IsValid() {
-			return
+			return nil
 		}
 
 		*changes = append(*changes, Change{Path: path, Old: nil, New: v2.Interface()})
-		return
+		return nil
 	} else if !v2.IsValid() {
 		// v1 is valid
 		*changes = append(*changes, Change{Path: path, Old: v1.Interface(), New: nil})
-		return
+		return nil
 	}
 
 	v1Type := v1.Type()
@@ -63,7 +121,7 @@ func diffValues(path *structpath.PathNode, v1, v2 reflect.Value, changes *[]Chan
 	// This should not happen; if it does, record this a full change
 	if v1Type != v2.Type() {
 		*changes = append(*changes, Change{Path: path, Old: v1.Interface(), New: v2.Interface()})
-		return
+		return nil
 	}
 
 	kind := v1.Kind()
@@ -74,11 +132,11 @@ func diffValues(path *structpath.PathNode, v1, v2 reflect.Value, changes *[]Chan
 		v1Nil := v1.IsNil()
 		v2Nil := v2.IsNil()
 		if v1Nil && v2Nil {
-			return
+			return nil
 		}
 		if v1Nil || v2Nil {
 			*changes = append(*changes, Change{Path: path, Old: v1.Interface(), New: v2.Interface()})
-			return
+			return nil
 		}
 	default:
 		// Not a nilable type.
@@ -87,27 +145,32 @@ func diffValues(path *structpath.PathNode, v1, v2 reflect.Value, changes *[]Chan
 
 	switch kind {
 	case reflect.Pointer:
-		diffValues(path, v1.Elem(), v2.Elem(), changes)
+		return diffValues(ctx, path, v1.Elem(), v2.Elem(), changes)
 	case reflect.Struct:
-		diffStruct(path, v1, v2, changes)
+		return diffStruct(ctx, path, v1, v2, changes)
 	case reflect.Slice, reflect.Array:
-		if v1.Len() != v2.Len() {
+		if keyFunc := ctx.findKeyFunc(path); keyFunc != nil {
+			return diffSliceByKey(ctx, path, v1, v2, keyFunc, changes)
+		} else if v1.Len() != v2.Len() {
 			*changes = append(*changes, Change{Path: path, Old: v1.Interface(), New: v2.Interface()})
 		} else {
 			for i := range v1.Len() {
 				node := structpath.NewIndex(path, i)
-				diffValues(node, v1.Index(i), v2.Index(i), changes)
+				if err := diffValues(ctx, node, v1.Index(i), v2.Index(i), changes); err != nil {
+					return err
+				}
 			}
 		}
 	case reflect.Map:
 		if v1Type.Key().Kind() == reflect.String {
-			diffMapStringKey(path, v1, v2, changes)
+			return diffMapStringKey(ctx, path, v1, v2, changes)
 		} else {
 			deepEqualValues(path, v1, v2, changes)
 		}
 	default:
 		deepEqualValues(path, v1, v2, changes)
 	}
+	return nil
 }
 
 func deepEqualValues(path *structpath.PathNode, v1, v2 reflect.Value, changes *[]Change) {
@@ -116,7 +179,7 @@ func deepEqualValues(path *structpath.PathNode, v1, v2 reflect.Value, changes *[
 	}
 }
 
-func diffStruct(path *structpath.PathNode, s1, s2 reflect.Value, changes *[]Change) {
+func diffStruct(ctx *diffContext, path *structpath.PathNode, s1, s2 reflect.Value, changes *[]Change) error {
 	t := s1.Type()
 	forced1 := getForceSendFields(s1)
 	forced2 := getForceSendFields(s2)
@@ -129,7 +192,9 @@ func diffStruct(path *structpath.PathNode, s1, s2 reflect.Value, changes *[]Chan
 
 		// Continue traversing embedded structs. Do not add the key to the path though.
 		if sf.Anonymous {
-			diffValues(path, s1.Field(i), s2.Field(i), changes)
+			if err := diffValues(ctx, path, s1.Field(i), s2.Field(i), changes); err != nil {
+				return err
+			}
 			continue
 		}
 
@@ -163,11 +228,14 @@ func diffStruct(path *structpath.PathNode, s1, s2 reflect.Value, changes *[]Chan
 			}
 		}
 
-		diffValues(node, v1Field, v2Field, changes)
+		if err := diffValues(ctx, node, v1Field, v2Field, changes); err != nil {
+			return err
+		}
 	}
+	return nil
 }
 
-func diffMapStringKey(path *structpath.PathNode, m1, m2 reflect.Value, changes *[]Change) {
+func diffMapStringKey(ctx *diffContext, path *structpath.PathNode, m1, m2 reflect.Value, changes *[]Change) error {
 	keySet := map[string]reflect.Value{}
 	for _, k := range m1.MapKeys() {
 		// Key is always string at this point
@@ -190,8 +258,11 @@ func diffMapStringKey(path *structpath.PathNode, m1, m2 reflect.Value, changes *
 		v1 := m1.MapIndex(k)
 		v2 := m2.MapIndex(k)
 		node := structpath.NewStringKey(path, ks)
-		diffValues(node, v1, v2, changes)
+		if err := diffValues(ctx, node, v1, v2, changes); err != nil {
+			return err
+		}
 	}
+	return nil
 }
 
 func getForceSendFields(v reflect.Value) []string {
@@ -208,3 +279,147 @@ func getForceSendFields(v reflect.Value) []string {
 	}
 	return nil
 }
+
+// findKeyFunc returns the KeyFunc for the given path, or nil if none matches.
+// Path patterns support [*] to match any slice index.
+func (ctx *diffContext) findKeyFunc(path *structpath.PathNode) KeyFunc {
+	if ctx.sliceKeys == nil {
+		return nil
+	}
+	pathStr := pathToPattern(path)
+	return ctx.sliceKeys[pathStr]
+}
+
+// pathToPattern converts a PathNode to a pattern string for matching.
+// Slice indices are converted to [*] wildcard.
+func pathToPattern(path *structpath.PathNode) string {
+	if path == nil {
+		return ""
+	}
+
+	components := path.AsSlice()
+	var result strings.Builder
+
+	for i, node := range components {
+		if idx, ok := node.Index(); ok {
+			// Convert numeric index to wildcard
+			_ = idx
+			result.WriteString("[*]")
+		} else if key, value, ok := node.KeyValue(); ok {
+			// Key-value syntax
+			result.WriteString("[")
+			result.WriteString(key)
+			result.WriteString("=")
+			result.WriteString(structpath.EncodeMapKey(value))
+			result.WriteString("]")
+		} else if key, ok := node.StringKey(); ok {
+			if i != 0 {
+				result.WriteString(".")
+			}
+			result.WriteString(key)
+		}
+	}
+
+	return result.String()
+}
+
+// sliceElement holds a slice element with its key information.
+type sliceElement struct {
+	keyField string
+	keyValue string
+	value    reflect.Value
+}
+
+// validateKeyFuncElementType verifies that the first element type in the sequence
+// is assignable to the expected type. If the sequence is empty, it succeeds.
+func validateKeyFuncElementType(seq reflect.Value, expected reflect.Type) error {
+	if seq.Len() == 0 {
+		return nil
+	}
+	elem := seq.Index(0)
+	if !elem.Type().AssignableTo(expected) {
+		return fmt.Errorf("KeyFunc expects %v, got %v", expected, elem.Type())
+	}
+	return nil
+}
+
+// diffSliceByKey compares two slices using the provided key function.
+// Elements are matched by their (keyField, keyValue) pairs instead of by index.
+// Duplicate keys are allowed and matched in order.
+func diffSliceByKey(ctx *diffContext, path *structpath.PathNode, v1, v2 reflect.Value, keyFunc KeyFunc, changes *[]Change) error {
+	caller, err := newKeyFuncCaller(keyFunc)
+	if err != nil {
+		return err
+	}
+
+	// Validate element types up-front to avoid runtime panics and to return a clear error.
+	if err := validateKeyFuncElementType(v1, caller.argType); err != nil {
+		return err
+	}
+	if err := validateKeyFuncElementType(v2, caller.argType); err != nil {
+		return err
+	}
+
+	// Build lists of elements grouped by key, preserving order within each key
+	elements1 := make(map[string][]sliceElement)
+	elements2 := make(map[string][]sliceElement)
+	seen := make(map[string]bool)
+	var orderedKeys []string
+
+	// Build from first slice
+	for i := range v1.Len() {
+		elem := v1.Index(i)
+		keyField, keyValue := caller.call(elem.Interface())
+		elements1[keyValue] = append(elements1[keyValue], sliceElement{keyField: keyField, keyValue: keyValue, value: elem})
+		if !seen[keyValue] {
+			seen[keyValue] = true
+			orderedKeys = append(orderedKeys, keyValue)
+		}
+	}
+
+	// Build from second slice
+	for i := range v2.Len() {
+		elem := v2.Index(i)
+		keyField, keyValue := caller.call(elem.Interface())
+		elements2[keyValue] = append(elements2[keyValue], sliceElement{keyField: keyField, keyValue: keyValue, value: elem})
+		if !seen[keyValue] {
+			seen[keyValue] = true
+			orderedKeys = append(orderedKeys, keyValue)
+		}
+	}
+
+	// Compare elements by key in original order
+	for _, keyValue := range orderedKeys {
+		list1 := elements1[keyValue]
+		list2 := elements2[keyValue]
+
+		var keyField string
+		if len(list1) > 0 {
+			keyField = list1[0].keyField
+		} else {
+			keyField = list2[0].keyField
+		}
+
+		// Match elements in order
+		minLen := min(len(list1), len(list2))
+		for i := range minLen {
+			node := structpath.NewKeyValue(path, keyField, keyValue)
+			if err := diffValues(ctx, node, list1[i].value, list2[i].value, changes); err != nil {
+				return err
+			}
+		}
+
+		// Handle extra elements in old (deleted)
+		for i := minLen; i < len(list1); i++ {
+			node := structpath.NewKeyValue(path, keyField, keyValue)
+			*changes = append(*changes, Change{Path: node, Old: list1[i].value.Interface(), New: nil})
+		}
+
+		// Handle extra elements in new (added)
+		for i := minLen; i < len(list2); i++ {
+			node := structpath.NewKeyValue(path, keyField, keyValue)
+			*changes = append(*changes, Change{Path: node, Old: nil, New: list2[i].value.Interface()})
+		}
+	}
+	return nil
+}
diff --git a/libs/structs/structdiff/diff_test.go b/libs/structs/structdiff/diff_test.go
index fb15ecb03c..9a216e0c4f 100644
--- a/libs/structs/structdiff/diff_test.go
+++ b/libs/structs/structdiff/diff_test.go
@@ -391,7 +391,7 @@ func TestGetStructDiff(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := GetStructDiff(tt.a, tt.b)
+			got, err := GetStructDiff(tt.a, tt.b, nil)
 
 			assert.Equal(t, tt.want, resolveChanges(got))
 
@@ -403,7 +403,7 @@ func TestGetStructDiff(t *testing.T) {
 		})
 
 		t.Run(tt.name+" mirror", func(t *testing.T) {
-			got, err := GetStructDiff(tt.b, tt.a)
+			got, err := GetStructDiff(tt.b, tt.a, nil)
 
 			var mirrorWant []ResolvedChange
 			for _, ch := range tt.want {
@@ -424,15 +424,271 @@ func TestGetStructDiff(t *testing.T) {
 		})
 
 		t.Run(tt.name+" equal A", func(t *testing.T) {
-			got, err := GetStructDiff(tt.a, tt.a)
+			got, err := GetStructDiff(tt.a, tt.a, nil)
 			assert.NoError(t, err)
 			assert.Nil(t, got)
 		})
 
 		t.Run(tt.name+" equal B", func(t *testing.T) {
-			got, err := GetStructDiff(tt.b, tt.b)
+			got, err := GetStructDiff(tt.b, tt.b, nil)
 			assert.NoError(t, err)
 			assert.Nil(t, got)
 		})
 	}
 }
+
+type Task struct {
+	TaskKey     string `json:"task_key,omitempty"`
+	Description string `json:"description,omitempty"`
+	Timeout     int    `json:"timeout,omitempty"`
+}
+
+type Job struct {
+	Name  string `json:"name,omitempty"`
+	Tasks []Task `json:"tasks,omitempty"`
+}
+
+func taskKeyFunc(task Task) (string, string) {
+	return "task_key", task.TaskKey
+}
+
+func TestGetStructDiffSliceKeys(t *testing.T) {
+	sliceKeys := map[string]KeyFunc{
+		"tasks": taskKeyFunc,
+	}
+
+	tests := []struct {
+		name string
+		a, b any
+		want []ResolvedChange
+	}{
+		{
+			name: "slice with same keys same order",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}, {TaskKey: "b", Description: "two"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}, {TaskKey: "b", Description: "two"}}},
+			want: nil,
+		},
+		{
+			name: "slice with same keys different order",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}, {TaskKey: "b", Description: "two"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "b", Description: "two"}, {TaskKey: "a", Description: "one"}}},
+			want: nil,
+		},
+		{
+			name: "slice with same keys field change",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "changed"}}},
+			want: []ResolvedChange{{Field: "tasks[task_key='a'].description", Old: "one", New: "changed"}},
+		},
+		{
+			name: "slice element added",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}, {TaskKey: "b", Description: "two"}}},
+			want: []ResolvedChange{{Field: "tasks[task_key='b']", Old: nil, New: Task{TaskKey: "b", Description: "two"}}},
+		},
+		{
+			name: "slice element removed",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}, {TaskKey: "b", Description: "two"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}}},
+			want: []ResolvedChange{{Field: "tasks[task_key='b']", Old: Task{TaskKey: "b", Description: "two"}, New: nil}},
+		},
+		{
+			name: "slice element replaced",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "b", Description: "two"}}},
+			want: []ResolvedChange{
+				{Field: "tasks[task_key='a']", Old: Task{TaskKey: "a", Description: "one"}, New: nil},
+				{Field: "tasks[task_key='b']", Old: nil, New: Task{TaskKey: "b", Description: "two"}},
+			},
+		},
+		{
+			name: "multiple changes with reorder",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "one"}, {TaskKey: "b", Description: "two"}, {TaskKey: "c", Description: "three"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "c", Description: "changed"}, {TaskKey: "a", Description: "one"}}},
+			want: []ResolvedChange{
+				{Field: "tasks[task_key='b']", Old: Task{TaskKey: "b", Description: "two"}, New: nil},
+				{Field: "tasks[task_key='c'].description", Old: "three", New: "changed"},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := GetStructDiff(tt.a, tt.b, sliceKeys)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.want, resolveChanges(got))
+		})
+	}
+}
+
+type Nested struct {
+	Items []Item `json:"items,omitempty"`
+}
+
+type Item struct {
+	ID    string `json:"id,omitempty"`
+	Value int    `json:"value,omitempty"`
+}
+
+type Root struct {
+	Nested []Nested `json:"nested,omitempty"`
+}
+
+func itemKeyFunc(item Item) (string, string) {
+	return "id", item.ID
+}
+
+func TestGetStructDiffNestedSliceKeys(t *testing.T) {
+	sliceKeys := map[string]KeyFunc{
+		"nested[*].items": itemKeyFunc,
+	}
+
+	tests := []struct {
+		name string
+		a, b any
+		want []ResolvedChange
+	}{
+		{
+			name: "nested slice with same keys different order",
+			a:    Root{Nested: []Nested{{Items: []Item{{ID: "x", Value: 1}, {ID: "y", Value: 2}}}}},
+			b:    Root{Nested: []Nested{{Items: []Item{{ID: "y", Value: 2}, {ID: "x", Value: 1}}}}},
+			want: nil,
+		},
+		{
+			name: "nested slice field change",
+			a:    Root{Nested: []Nested{{Items: []Item{{ID: "x", Value: 1}}}}},
+			b:    Root{Nested: []Nested{{Items: []Item{{ID: "x", Value: 99}}}}},
+			want: []ResolvedChange{{Field: "nested[0].items[id='x'].value", Old: 1, New: 99}},
+		},
+		{
+			name: "nested slice element added",
+			a:    Root{Nested: []Nested{{Items: []Item{{ID: "x", Value: 1}}}}},
+			b:    Root{Nested: []Nested{{Items: []Item{{ID: "x", Value: 1}, {ID: "y", Value: 2}}}}},
+			want: []ResolvedChange{{Field: "nested[0].items[id='y']", Old: nil, New: Item{ID: "y", Value: 2}}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := GetStructDiff(tt.a, tt.b, sliceKeys)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.want, resolveChanges(got))
+		})
+	}
+}
+
+func TestGetStructDiffSliceKeysInvalidFunc(t *testing.T) {
+	tests := []struct {
+		name    string
+		keyFunc any
+		errMsg  string
+	}{
+		{
+			name:    "not a function",
+			keyFunc: "not a function",
+			errMsg:  "KeyFunc must be a function, got string",
+		},
+		{
+			name:    "wrong number of parameters",
+			keyFunc: func() (string, string) { return "", "" },
+			errMsg:  "KeyFunc must have exactly 1 parameter, got 0",
+		},
+		{
+			name:    "too many parameters",
+			keyFunc: func(a, b Task) (string, string) { return "", "" },
+			errMsg:  "KeyFunc must have exactly 1 parameter, got 2",
+		},
+		{
+			name:    "wrong number of returns",
+			keyFunc: func(t Task) string { return "" },
+			errMsg:  "KeyFunc must return exactly 2 values, got 1",
+		},
+		{
+			name:    "wrong first return type",
+			keyFunc: func(t Task) (int, string) { return 0, "" },
+			errMsg:  "KeyFunc must return (string, string), got (int, string)",
+		},
+		{
+			name:    "wrong second return type",
+			keyFunc: func(t Task) (string, int) { return "", 0 },
+			errMsg:  "KeyFunc must return (string, string), got (string, int)",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sliceKeys := map[string]KeyFunc{"tasks": tt.keyFunc}
+			a := Job{Tasks: []Task{{TaskKey: "a"}}}
+			b := Job{Tasks: []Task{{TaskKey: "a"}}}
+			_, err := GetStructDiff(a, b, sliceKeys)
+			assert.EqualError(t, err, tt.errMsg)
+		})
+	}
+}
+
+func TestGetStructDiffSliceKeysWrongArgType(t *testing.T) {
+	// Function expects Item but slice contains Task
+	sliceKeys := map[string]KeyFunc{
+		"tasks": func(item Item) (string, string) {
+			return "id", item.ID
+		},
+	}
+	a := Job{Tasks: []Task{{TaskKey: "a"}}}
+	b := Job{Tasks: []Task{{TaskKey: "b"}}}
+	_, err := GetStructDiff(a, b, sliceKeys)
+	assert.EqualError(t, err, "KeyFunc expects structdiff.Item, got structdiff.Task")
+}
+
+func TestGetStructDiffSliceKeysDuplicates(t *testing.T) {
+	sliceKeys := map[string]KeyFunc{
+		"tasks": taskKeyFunc,
+	}
+
+	tests := []struct {
+		name string
+		a, b Job
+		want []ResolvedChange
+	}{
+		{
+			name: "same duplicates no change",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}, {TaskKey: "a", Description: "2"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}, {TaskKey: "a", Description: "2"}}},
+			want: nil,
+		},
+		{
+			name: "duplicates with field change",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}, {TaskKey: "a", Description: "2"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}, {TaskKey: "a", Description: "changed"}}},
+			want: []ResolvedChange{{Field: "tasks[task_key='a'].description", Old: "2", New: "changed"}},
+		},
+		{
+			name: "extra in old is deleted",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}, {TaskKey: "a", Description: "2"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}}},
+			want: []ResolvedChange{{Field: "tasks[task_key='a']", Old: Task{TaskKey: "a", Description: "2"}, New: nil}},
+		},
+		{
+			name: "extra in new is added",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}, {TaskKey: "a", Description: "2"}}},
+			want: []ResolvedChange{{Field: "tasks[task_key='a']", Old: nil, New: Task{TaskKey: "a", Description: "2"}}},
+		},
+		{
+			name: "two in old one in new with change",
+			a:    Job{Tasks: []Task{{TaskKey: "a", Description: "1"}, {TaskKey: "a", Description: "2"}}},
+			b:    Job{Tasks: []Task{{TaskKey: "a", Description: "changed"}}},
+			want: []ResolvedChange{
+				{Field: "tasks[task_key='a'].description", Old: "1", New: "changed"},
+				{Field: "tasks[task_key='a']", Old: Task{TaskKey: "a", Description: "2"}, New: nil},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := GetStructDiff(tt.a, tt.b, sliceKeys)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.want, resolveChanges(got))
+		})
+	}
+}
diff --git a/libs/structs/structdiff/jobsettings_test.go b/libs/structs/structdiff/jobsettings_test.go
index beeb1c5878..198cd34510 100644
--- a/libs/structs/structdiff/jobsettings_test.go
+++ b/libs/structs/structdiff/jobsettings_test.go
@@ -444,16 +444,16 @@ const jobExampleResponseNils = `
 func testEqual(t *testing.T, input string) {
 	var x, y jobs.JobSettings
 	require.NoError(t, json.Unmarshal([]byte(input), &x))
-	changes, err := GetStructDiff(x, x)
+	changes, err := GetStructDiff(x, x, nil)
 	require.NoError(t, err)
 	require.Nil(t, changes)
 
 	require.NoError(t, json.Unmarshal([]byte(input), &y))
-	changes, err = GetStructDiff(x, y)
+	changes, err = GetStructDiff(x, y, nil)
 	require.NoError(t, err)
 	require.Nil(t, changes)
 
-	changes, err = GetStructDiff(&x, &y)
+	changes, err = GetStructDiff(&x, &y, nil)
 	require.NoError(t, err)
 	require.Nil(t, changes)
 }
@@ -471,7 +471,7 @@ func TestJobDiff(t *testing.T) {
 	require.NoError(t, json.Unmarshal([]byte(jobExampleResponseZeroes), &zero))
 	require.NoError(t, json.Unmarshal([]byte(jobExampleResponseNils), &nils))
 
-	changes, err := GetStructDiff(job, zero)
+	changes, err := GetStructDiff(job, zero, nil)
 	require.NoError(t, err)
 	require.GreaterOrEqual(t, len(changes), 75)
 	assert.Equal(t, "budget_policy_id", changes[0].Path.String())
@@ -488,7 +488,7 @@ func TestJobDiff(t *testing.T) {
 	assert.Equal(t, "string", changes[3].Old)
 	assert.Equal(t, "", changes[3].New)
 
-	changes, err = GetStructDiff(job, nils)
+	changes, err = GetStructDiff(job, nils, nil)
 	require.NoError(t, err)
 	require.GreaterOrEqual(t, len(changes), 77)
 	assert.Equal(t, "budget_policy_id", changes[0].Path.String())
@@ -509,7 +509,7 @@ func TestJobDiff(t *testing.T) {
 	assert.Equal(t, "string", changes[3].Old)
 	assert.Nil(t, changes[3].New)
 
-	changes, err = GetStructDiff(zero, nils)
+	changes, err = GetStructDiff(zero, nils, nil)
 	require.NoError(t, err)
 	assert.GreaterOrEqual(t, len(changes), 58)
 	assert.Equal(t, "budget_policy_id", changes[0].Path.String())
diff --git a/libs/structs/structpath/path.go b/libs/structs/structpath/path.go
index 55ab1e8d15..1627ae6f7d 100644
--- a/libs/structs/structpath/path.go
+++ b/libs/structs/structpath/path.go
@@ -11,7 +11,7 @@ import (
 )
 
 const (
-	// Encodes string key, which is encoded as .field or as ['spark.conf']x
+	// Encodes string key, which is encoded as .field or as ['spark.conf']
 	tagStringKey = -1
 
 	// Encodes wildcard after a dot: foo.*
@@ -19,6 +19,9 @@ const (
 
 	// Encodes wildcard in brackets: foo[*]
 	tagBracketStar = -3
+
+	// Encodes key/value index, which is encoded as [key='value'] or [key="value"]
+	tagKeyValue = -4
 )
 
 // PathNode represents a node in a path for struct diffing.
@@ -29,6 +32,7 @@ type PathNode struct {
 	// If index >= 0, the node specifies a slice/array index in index.
 	// If index < 0, this describes the type of node
 	index int
+	value string // Used for tagKeyValue: stores the value part of [key='value']
 }
 
 func (p *PathNode) IsRoot() bool {
@@ -59,6 +63,16 @@ func (p *PathNode) BracketStar() bool {
 	return p.index == tagBracketStar
 }
 
+func (p *PathNode) KeyValue() (key, value string, ok bool) {
+	if p == nil {
+		return "", "", false
+	}
+	if p.index == tagKeyValue {
+		return p.key, p.value, true
+	}
+	return "", "", false
+}
+
 // StringKey returns either Field() or MapKey() if either is available
 func (p *PathNode) StringKey() (string, bool) {
 	if p == nil {
@@ -128,6 +142,15 @@ func NewBracketStar(prev *PathNode) *PathNode {
 	}
 }
 
+func NewKeyValue(prev *PathNode, key, value string) *PathNode {
+	return &PathNode{
+		prev:  prev,
+		key:   key,
+		index: tagKeyValue,
+		value: value,
+	}
+}
+
 // String returns the string representation of the path.
 // The string keys are encoded in dot syntax (foo.bar) if they don't have any reserved characters (so can be parsed as fields).
 // Otherwise they are encoded in brackets + single quotes: tags['name']. Single quote can escaped by placing two single quotes.
@@ -160,6 +183,12 @@ func (p *PathNode) String() string {
 			}
 		} else if node.index == tagBracketStar {
 			result.WriteString("[*]")
+		} else if node.index == tagKeyValue {
+			result.WriteString("[")
+			result.WriteString(node.key)
+			result.WriteString("=")
+			result.WriteString(EncodeMapKey(node.value))
+			result.WriteString("]")
 		} else if isValidField(node.key) {
 			// Valid field name
 			if i != 0 {
@@ -191,11 +220,15 @@ func EncodeMapKey(s string) string {
 //   - FIELD_START: After a dot, expects field name or "*"
 //   - FIELD: Reading field name characters
 //   - DOT_STAR: Encountered "*" (at start or after dot), expects ".", "[", or EOF
-//   - BRACKET_OPEN: Just encountered "[", expects digit, "'" or "*"
+//   - BRACKET_OPEN: Just encountered "[", expects digit, "'", "*", or identifier for key-value
 //   - INDEX: Reading array index digits, expects more digits or "]"
 //   - MAP_KEY: Reading map key content, expects any char or "'"
 //   - MAP_KEY_QUOTE: Encountered "'" in map key, expects "'" (escape) or "]" (end)
 //   - WILDCARD: Reading "*" in brackets, expects "]"
+//   - KEYVALUE_KEY: Reading key part of [key='value'], expects identifier chars or "="
+//   - KEYVALUE_EQUALS: After key, expecting "'" to start value
+//   - KEYVALUE_VALUE: Reading value content, expects any char or quote
+//   - KEYVALUE_VALUE_QUOTE: Encountered quote in value, expects same quote (escape) or "]" (end)
 //   - EXPECT_DOT_OR_END: After bracket close, expects ".", "[" or end of string
 //   - END: Successfully completed parsing
 //
@@ -204,11 +237,15 @@ func EncodeMapKey(s string) string {
 //   - FIELD_START: [a-zA-Z_-] -> FIELD, "*" -> DOT_STAR, other -> ERROR
 //   - FIELD: [a-zA-Z0-9_-] -> FIELD, "." -> FIELD_START, "[" -> BRACKET_OPEN, EOF -> END
 //   - DOT_STAR: "." -> FIELD_START, "[" -> BRACKET_OPEN, EOF -> END, other -> ERROR
-//   - BRACKET_OPEN: [0-9] -> INDEX, "'" -> MAP_KEY, "*" -> WILDCARD
+//   - BRACKET_OPEN: [0-9] -> INDEX, "'" -> MAP_KEY, "*" -> WILDCARD, identifier -> KEYVALUE_KEY
 //   - INDEX: [0-9] -> INDEX, "]" -> EXPECT_DOT_OR_END
 //   - MAP_KEY: (any except "'") -> MAP_KEY, "'" -> MAP_KEY_QUOTE
 //   - MAP_KEY_QUOTE: "'" -> MAP_KEY (escape), "]" -> EXPECT_DOT_OR_END (end key)
 //   - WILDCARD: "]" -> EXPECT_DOT_OR_END
+//   - KEYVALUE_KEY: identifier -> KEYVALUE_KEY, "=" -> KEYVALUE_EQUALS
+//   - KEYVALUE_EQUALS: "'" or '"' -> KEYVALUE_VALUE
+//   - KEYVALUE_VALUE: (any except quote) -> KEYVALUE_VALUE, quote -> KEYVALUE_VALUE_QUOTE
+//   - KEYVALUE_VALUE_QUOTE: quote -> KEYVALUE_VALUE (escape), "]" -> EXPECT_DOT_OR_END
 //   - EXPECT_DOT_OR_END: "." -> FIELD_START, "[" -> BRACKET_OPEN, EOF -> END
 func Parse(s string) (*PathNode, error) {
 	if s == "" {
@@ -226,6 +263,10 @@ func Parse(s string) (*PathNode, error) {
 		stateMapKey
 		stateMapKeyQuote
 		stateWildcard
+		stateKeyValueKey
+		stateKeyValueEquals
+		stateKeyValueValue
+		stateKeyValueValueQuote
 		stateExpectDotOrEnd
 		stateEnd
 	)
@@ -233,6 +274,7 @@ func Parse(s string) (*PathNode, error) {
 	state := stateStart
 	var result *PathNode
 	var currentToken strings.Builder
+	var keyValueKey string // Stores the key part of [key='value']
 	pos := 0
 
 	for pos < len(s) {
@@ -296,6 +338,9 @@ func Parse(s string) (*PathNode, error) {
 				state = stateMapKey
 			} else if ch == '*' {
 				state = stateWildcard
+			} else if !isReservedFieldChar(ch) {
+				currentToken.WriteByte(ch)
+				state = stateKeyValueKey
 			} else {
 				return nil, fmt.Errorf("unexpected character '%c' after '[' at position %d", ch, pos)
 			}
@@ -346,6 +391,47 @@ func Parse(s string) (*PathNode, error) {
 				return nil, fmt.Errorf("unexpected character '%c' after '*' at position %d", ch, pos)
 			}
 
+		case stateKeyValueKey:
+			if ch == '=' {
+				keyValueKey = currentToken.String()
+				currentToken.Reset()
+				state = stateKeyValueEquals
+			} else if !isReservedFieldChar(ch) {
+				currentToken.WriteByte(ch)
+			} else {
+				return nil, fmt.Errorf("unexpected character '%c' in key-value key at position %d", ch, pos)
+			}
+
+		case stateKeyValueEquals:
+			if ch == '\'' {
+				state = stateKeyValueValue
+			} else {
+				return nil, fmt.Errorf("expected quote after '=' but got '%c' at position %d", ch, pos)
+			}
+
+		case stateKeyValueValue:
+			if ch == '\'' {
+				state = stateKeyValueValueQuote
+			} else {
+				currentToken.WriteByte(ch)
+			}
+
+		case stateKeyValueValueQuote:
+			switch ch {
+			case '\'':
+				// Escaped quote - add single quote to value and continue
+				currentToken.WriteByte(ch)
+				state = stateKeyValueValue
+			case ']':
+				// End of key-value
+				result = NewKeyValue(result, keyValueKey, currentToken.String())
+				currentToken.Reset()
+				keyValueKey = ""
+				state = stateExpectDotOrEnd
+			default:
+				return nil, fmt.Errorf("unexpected character '%c' after quote in key-value at position %d", ch, pos)
+			}
+
 		case stateExpectDotOrEnd:
 			switch ch {
 			case '.':
@@ -390,6 +476,14 @@ func Parse(s string) (*PathNode, error) {
 		return nil, errors.New("unexpected end of input after quote in map key")
 	case stateWildcard:
 		return nil, errors.New("unexpected end of input after wildcard '*'")
+	case stateKeyValueKey:
+		return nil, errors.New("unexpected end of input while parsing key-value key")
+	case stateKeyValueEquals:
+		return nil, errors.New("unexpected end of input after '=' in key-value")
+	case stateKeyValueValue:
+		return nil, errors.New("unexpected end of input while parsing key-value value")
+	case stateKeyValueValueQuote:
+		return nil, errors.New("unexpected end of input after quote in key-value value")
 	case stateEnd:
 		return result, nil
 	default:
diff --git a/libs/structs/structpath/path_test.go b/libs/structs/structpath/path_test.go
index a9e19501dc..c4e427666d 100644
--- a/libs/structs/structpath/path_test.go
+++ b/libs/structs/structpath/path_test.go
@@ -13,6 +13,7 @@ func TestPathNode(t *testing.T) {
 		String      string
 		Index       any
 		StringKey   any
+		KeyValue    []string // [key, value] or nil
 		Root        any
 		DotStar     bool
 		BracketStar bool
@@ -48,6 +49,12 @@ func TestPathNode(t *testing.T) {
 			String:      "[*]",
 			BracketStar: true,
 		},
+		{
+			name:     "key value",
+			node:     NewKeyValue(nil, "name", "foo"),
+			String:   "[name='foo']",
+			KeyValue: []string{"name", "foo"},
+		},
 
 		// Two node tests
 		{
@@ -175,6 +182,38 @@ func TestPathNode(t *testing.T) {
 			String:      "bla.*[*]",
 			BracketStar: true,
 		},
+
+		// Key-value tests
+		{
+			name:     "key value with parent",
+			node:     NewKeyValue(NewStringKey(nil, "tasks"), "task_key", "my_task"),
+			String:   "tasks[task_key='my_task']",
+			KeyValue: []string{"task_key", "my_task"},
+		},
+		{
+			name:      "key value then field",
+			node:      NewStringKey(NewKeyValue(nil, "name", "foo"), "id"),
+			String:    "[name='foo'].id",
+			StringKey: "id",
+		},
+		{
+			name:     "key value with quote in value",
+			node:     NewKeyValue(nil, "name", "it's"),
+			String:   "[name='it''s']",
+			KeyValue: []string{"name", "it's"},
+		},
+		{
+			name:     "key value with empty value",
+			node:     NewKeyValue(nil, "key", ""),
+			String:   "[key='']",
+			KeyValue: []string{"key", ""},
+		},
+		{
+			name:      "complex path with key value",
+			node:      NewStringKey(NewKeyValue(NewStringKey(NewStringKey(nil, "resources"), "jobs"), "task_key", "my_task"), "notebook_task"),
+			String:    "resources.jobs[task_key='my_task'].notebook_task",
+			StringKey: "notebook_task",
+		},
 	}
 
 	for _, tt := range tests {
@@ -212,6 +251,18 @@ func TestPathNode(t *testing.T) {
 				assert.True(t, isStringKey)
 			}
 
+			// KeyValue
+			gotKey, gotValue, isKeyValue := tt.node.KeyValue()
+			if tt.KeyValue == nil {
+				assert.Equal(t, "", gotKey)
+				assert.Equal(t, "", gotValue)
+				assert.False(t, isKeyValue)
+			} else {
+				assert.Equal(t, tt.KeyValue[0], gotKey)
+				assert.Equal(t, tt.KeyValue[1], gotValue)
+				assert.True(t, isKeyValue)
+			}
+
 			// IsRoot
 			isRoot := tt.node.IsRoot()
 			if tt.Root == nil {
@@ -266,12 +317,12 @@ func TestParseErrors(t *testing.T) {
 		{
 			name:  "invalid array index",
 			input: "[abc]",
-			error: "unexpected character 'a' after '[' at position 1",
+			error: "unexpected character ']' in key-value key at position 4",
 		},
 		{
 			name:  "negative array index",
 			input: "[-1]",
-			error: "unexpected character '-' after '[' at position 1",
+			error: "unexpected character ']' in key-value key at position 3",
 		},
 		{
 			name:  "unclosed map key quote",
@@ -331,7 +382,7 @@ func TestParseErrors(t *testing.T) {
 		{
 			name:  "invalid character in bracket",
 			input: "field[name",
-			error: "unexpected character 'n' after '[' at position 6",
+			error: "unexpected end of input while parsing key-value key",
 		},
 		{
 			name:  "unexpected character after valid path",
@@ -380,6 +431,53 @@ func TestParseErrors(t *testing.T) {
 			input: "bla.*foo",
 			error: "unexpected character 'f' after '.*' at position 5",
 		},
+
+		// Invalid key-value patterns
+		{
+			name:  "key-value missing equals",
+			input: "[name'value']",
+			error: "unexpected character ''' in key-value key at position 5",
+		},
+		{
+			name:  "key-value missing value quote",
+			input: "[name=value]",
+			error: "expected quote after '=' but got 'v' at position 6",
+		},
+		{
+			name:  "key-value incomplete key",
+			input: "[name",
+			error: "unexpected end of input while parsing key-value key",
+		},
+		{
+			name:  "key-value incomplete after equals",
+			input: "[name=",
+			error: "unexpected end of input after '=' in key-value",
+		},
+		{
+			name:  "key-value incomplete value",
+			input: "[name='value",
+			error: "unexpected end of input while parsing key-value value",
+		},
+		{
+			name:  "key-value incomplete after value quote",
+			input: "[name='value'",
+			error: "unexpected end of input after quote in key-value value",
+		},
+		{
+			name:  "key-value invalid char after value quote",
+			input: "[name='value'x]",
+			error: "unexpected character 'x' after quote in key-value at position 13",
+		},
+		{
+			name:  "double quotes are not supported a.t.m",
+			input: "[name=\"value\"]",
+			error: "expected quote after '=' but got '\"' at position 6",
+		},
+		{
+			name:  "mixed quotes never going to be supported",
+			input: "[name='value\"]",
+			error: "unexpected end of input while parsing key-value value",
+		},
 	}
 
 	for _, tt := range tests {
diff --git a/libs/structs/structvar/structvar_test.go b/libs/structs/structvar/structvar_test.go
index 60705a4fb6..9d8c1d2d8d 100644
--- a/libs/structs/structvar/structvar_test.go
+++ b/libs/structs/structvar/structvar_test.go
@@ -110,7 +110,7 @@ func TestResolveRef(t *testing.T) {
 			},
 			reference: "${var.test}",
 			value:     "value",
-			errorMsg:  "unexpected character",
+			errorMsg:  "invalid path",
 		},
 	}