feat(hip-0025): add kstatus and custom JSONPath readiness engine

OrderedWaitStrategy, custom readiness evaluation via .status-scoped
JSONPath expressions (helm.sh/readiness-success / -failure), and the
status reader that layers custom readiness over kstatus.

Refs: HIP-0025
Signed-off-by: Rohit Gudi <50377477+caretak3r@users.noreply.github.com>

pkg/kube/client.go

@@ -32,6 +32,7 @@ import (
 	"sync"
 
 	jsonpatch "github.com/evanphx/json-patch/v5"
+	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/engine"
 	v1 "k8s.io/api/core/v1"
 	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiextv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
@@ -117,6 +118,10 @@ const (
 
 	// HookOnlyStrategy: wait only for hook Pods/Jobs to complete; does not wait for general chart resources.
 	HookOnlyStrategy WaitStrategy = "hookOnly"
+
+	// OrderedWaitStrategy: action-layer sequencing uses ordered batches while the kube layer
+	// reuses status watching for per-batch readiness checks.
+	OrderedWaitStrategy WaitStrategy = "ordered"
 )
 
 type FieldValidationDirective string
@@ -168,6 +173,7 @@ func (c *Client) newStatusWatcher(opts ...WaitOption) (*statusWaiter, error) {
 	if waitContext == nil {
 		waitContext = c.WaitContext
 	}
+	readers := append([]engine.StatusReader(nil), o.statusReaders...)
 	sw := &statusWaiter{
 		restMapper:           restMapper,
 		client:               dynamicClient,
@@ -176,8 +182,9 @@ func (c *Client) newStatusWatcher(opts ...WaitOption) (*statusWaiter, error) {
 		waitCtx:              o.waitCtx,
 		waitWithJobsCtx:      o.waitWithJobsCtx,
 		waitForDeleteCtx:     o.waitForDeleteCtx,
-		readers:              o.statusReaders,
+		readers:              readers,
 		statusComputeWorkers: o.statusComputeWorkers,
+		customReadiness:      o.enableCustomReadinessStatusReader,
 	}
 	sw.SetLogger(c.Logger().Handler())
 	return sw, nil
@@ -203,10 +210,12 @@ func (c *Client) GetWaiterWithOptions(strategy WaitStrategy, opts ...WaitOption)
 			return nil, err
 		}
 		return &hookOnlyWaiter{sw: sw}, nil
+	case OrderedWaitStrategy:
+		return c.newStatusWatcher(opts...)
 	case "":
-		return nil, errors.New("wait strategy not set. Choose one of: " + string(StatusWatcherStrategy) + ", " + string(HookOnlyStrategy) + ", " + string(LegacyStrategy))
+		return nil, errors.New("wait strategy not set. Choose one of: " + string(StatusWatcherStrategy) + ", " + string(HookOnlyStrategy) + ", " + string(LegacyStrategy) + ", " + string(OrderedWaitStrategy))
 	default:
-		return nil, errors.New("unknown wait strategy (s" + string(strategy) + "). Valid values are: " + string(StatusWatcherStrategy) + ", " + string(HookOnlyStrategy) + ", " + string(LegacyStrategy))
+		return nil, errors.New("unknown wait strategy (" + string(strategy) + "). Valid values are: " + string(StatusWatcherStrategy) + ", " + string(HookOnlyStrategy) + ", " + string(LegacyStrategy) + ", " + string(OrderedWaitStrategy))
 	}
 }
 

pkg/kube/client_wait_strategy_test.go

@@ -0,0 +1,54 @@
+/*
+Copyright The Helm Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kube
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestClientGetWaiterWithOptions_OrderedWaitStrategy(t *testing.T) {
+	client := newTestClient(t)
+
+	waiter, err := client.GetWaiterWithOptions(OrderedWaitStrategy)
+	require.NoError(t, err)
+	assert.IsType(t, &statusWaiter{}, waiter)
+}
+
+func TestClientGetWaiterWithOptions_UnknownStrategyListsOrdered(t *testing.T) {
+	client := newTestClient(t)
+
+	waiter, err := client.GetWaiterWithOptions(WaitStrategy("bogus"))
+	require.Error(t, err)
+	assert.Nil(t, waiter)
+	assert.Contains(t, err.Error(), string(OrderedWaitStrategy))
+}
+
+func TestNewStatusWatcherCustomReadinessFlag(t *testing.T) {
+	client := newTestClient(t)
+
+	w, err := client.newStatusWatcher(WithCustomReadinessStatusReader())
+	require.NoError(t, err)
+	assert.True(t, w.customReadiness)
+	assert.Empty(t, w.readers, "custom readiness must not inject a reader into the shared reader list")
+
+	w, err = client.newStatusWatcher()
+	require.NoError(t, err)
+	assert.False(t, w.customReadiness)
+}

pkg/kube/custom_readiness_status_reader.go

@@ -0,0 +1,141 @@
+/*
+Copyright The Helm Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kube
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+
+	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/engine"
+	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/event"
+	"github.com/fluxcd/cli-utils/pkg/kstatus/status"
+	"github.com/fluxcd/cli-utils/pkg/object"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type customReadinessStatusReader struct {
+	logger *slog.Logger
+	// fallback is the reader chain the enclosing wait method would use if
+	// custom readiness were disabled. Resources that do not carry both
+	// readiness annotations are delegated to it, so built-in readers (the
+	// composite Deployment/ReplicaSet/StatefulSet readers, the Job reader in
+	// WaitWithJobs) and caller-supplied readers keep their semantics.
+	fallback engine.StatusReader
+	// warnedExpressions dedups authoring-mistake warnings so a bad expression
+	// logs once per resource rather than on every watch event. Keys are
+	// "<ObjMetadata.String()>|<expression>". The reader is invoked
+	// concurrently by the status watcher, hence sync.Map.
+	warnedExpressions sync.Map
+}
+
+// newCustomReadinessStatusReader wraps fallback with per-resource custom
+// readiness evaluation. fallback must be non-nil and must support every
+// GroupKind this reader can receive (every chain built in statuswait.go
+// terminates in a generic reader that supports all GroupKinds).
+func newCustomReadinessStatusReader(logger *slog.Logger, fallback engine.StatusReader) engine.StatusReader {
+	if logger == nil {
+		logger = slog.New(slog.DiscardHandler)
+	}
+	return &customReadinessStatusReader{logger: logger, fallback: fallback}
+}
+
+// Supports returns true for every GroupKind: annotated resources of any kind
+// are evaluated here, and everything else is delegated to the fallback chain.
+func (*customReadinessStatusReader) Supports(schema.GroupKind) bool {
+	return true
+}
+
+func (r *customReadinessStatusReader) ReadStatus(ctx context.Context, reader engine.ClusterReader, resource object.ObjMetadata) (*event.ResourceStatus, error) {
+	u := &unstructured.Unstructured{}
+	u.SetGroupVersionKind(schema.GroupVersionKind{
+		Group: resource.GroupKind.Group,
+		Kind:  resource.GroupKind.Kind,
+	})
+	if err := reader.Get(ctx, client.ObjectKey{Namespace: resource.Namespace, Name: resource.Name}, u); err != nil {
+		return nil, err
+	}
+	return r.readStatus(ctx, reader, resource, u)
+}
+
+func (r *customReadinessStatusReader) ReadStatusForObject(ctx context.Context, reader engine.ClusterReader, resource *unstructured.Unstructured) (*event.ResourceStatus, error) {
+	identifier, err := object.RuntimeToObjMeta(resource)
+	if err != nil {
+		return nil, err
+	}
+	return r.readStatus(ctx, reader, identifier, resource)
+}
+
+func (r *customReadinessStatusReader) readStatus(ctx context.Context, reader engine.ClusterReader, identifier object.ObjMetadata, resource *unstructured.Unstructured) (*event.ResourceStatus, error) {
+	annotations := resource.GetAnnotations()
+
+	successExprs, err := parseReadinessExpressions(annotations[AnnotationReadinessSuccess])
+	if err != nil {
+		return nil, fmt.Errorf("parsing %s for %s/%s: %w", AnnotationReadinessSuccess, identifier.Namespace, identifier.Name, err)
+	}
+	failureExprs, err := parseReadinessExpressions(annotations[AnnotationReadinessFailure])
+	if err != nil {
+		return nil, fmt.Errorf("parsing %s for %s/%s: %w", AnnotationReadinessFailure, identifier.Namespace, identifier.Name, err)
+	}
+
+	result, useKstatus, warnings, err := EvaluateCustomReadiness(resource, successExprs, failureExprs)
+	if err != nil {
+		return nil, err
+	}
+	if useKstatus {
+		// Missing or partial annotations: this resource does not opt in to
+		// custom readiness, so the default reader chain decides its status.
+		return r.fallback.ReadStatusForObject(ctx, reader, resource)
+	}
+
+	for _, w := range warnings {
+		key := identifier.String() + "|" + w.Expression
+		if _, alreadyWarned := r.warnedExpressions.LoadOrStore(key, struct{}{}); alreadyWarned {
+			continue
+		}
+		r.logger.Warn("custom readiness expression cannot be evaluated; treating condition as not met",
+			"kind", identifier.GroupKind.Kind,
+			"namespace", identifier.Namespace,
+			"name", identifier.Name,
+			"expression", w.Expression,
+			"detail", w.Detail,
+		)
+	}
+
+	st, message := status.InProgressStatus, "waiting for custom readiness conditions"
+	switch result {
+	case ReadinessReady:
+		st, message = status.CurrentStatus, "custom readiness conditions met"
+	case ReadinessFailed:
+		st, message = status.FailedStatus, "custom readiness failure condition met"
+	default:
+		// ReadinessPending (and any future status) keep the in-progress
+		// defaults; name any skipped expression so an eventual wait timeout
+		// error is self-explanatory.
+		if len(warnings) > 0 {
+			message = fmt.Sprintf("waiting for custom readiness conditions (%d expression(s) skipped: %s)", len(warnings), warnings[0].Detail)
+		}
+	}
+	return &event.ResourceStatus{
+		Identifier: identifier,
+		Status:     st,
+		Message:    message,
+	}, nil
+}

pkg/kube/custom_readiness_status_reader_test.go

@@ -0,0 +1,617 @@
+/*
+Copyright The Helm Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kube
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/clusterreader"
+	"github.com/fluxcd/cli-utils/pkg/kstatus/polling/statusreaders"
+	"github.com/fluxcd/cli-utils/pkg/kstatus/status"
+	"github.com/fluxcd/cli-utils/pkg/object"
+	"github.com/fluxcd/cli-utils/pkg/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/dynamic/fake"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+func TestStatusWaitWithCustomReadinessReader(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name             string
+		manifest         string
+		expectErrStrs    []string
+		notExpectErrStrs []string
+	}{
+		{
+			name: "custom readiness makes resource current",
+			manifest: `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: ready-config
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} == \"Ready\""]'
+    helm.sh/readiness-failure: '["{.phase} == \"Failed\""]'
+status:
+  phase: Ready
+`,
+		},
+		{
+			name: "custom readiness failure stops wait",
+			manifest: `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: failed-config
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} == \"Ready\""]'
+    helm.sh/readiness-failure: '["{.phase} == \"Failed\""]'
+status:
+  phase: Failed
+`,
+			expectErrStrs: []string{"resource ConfigMap/default/failed-config not ready. status: Failed"},
+		},
+		{
+			name: "resources without annotations fall back to kstatus",
+			manifest: `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: current-config
+  namespace: default
+`,
+		},
+		{
+			// END-TO-END pin for hip-0025-gy6: an ordering operator on a
+			// non-numeric value must leave the resource InProgress (and let
+			// the wait time out per-resource) — NOT kill the status reporter
+			// and abort the whole batch.
+			name: "ordering operator on non-numeric value does not abort the wait",
+			manifest: `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: bad-ordering
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} > \"Ready\""]'
+    helm.sh/readiness-failure: '["{.phase} == \"Failed\""]'
+status:
+  phase: Running
+`,
+			expectErrStrs:    []string{"resource ConfigMap/default/bad-ordering not ready. status: InProgress", "skipped"},
+			notExpectErrStrs: []string{"failed to compute object status"},
+		},
+		{
+			name: "ordering on numeric string value compares numerically",
+			manifest: `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: numeric-string
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} >= 5"]'
+    helm.sh/readiness-failure: '["{.failed} >= 1"]'
+status:
+  phase: "5"
+`,
+		},
+		{
+			name: "incomparable expression skipped when another success condition is met",
+			manifest: `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: mixed-success
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} > \"Ready\"", "{.phase} == \"Running\""]'
+    helm.sh/readiness-failure: '["{.phase} == \"Failed\""]'
+status:
+  phase: Running
+`,
+		},
+		{
+			// Partial annotations (only one of success/failure): custom
+			// evaluation must NOT engage; the resource is judged by the
+			// default chain (a bare ConfigMap is Current), even though the
+			// success expression is unmet.
+			name: "partial annotation falls back to default readiness",
+			manifest: `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: partial-config
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} == \"Ready\""]'
+status:
+  phase: NotReady
+`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			c := newTestClient(t)
+			fakeClient := fake.NewSimpleDynamicClient(scheme.Scheme)
+			fakeMapper := testutil.NewFakeRESTMapper(v1.SchemeGroupVersion.WithKind("ConfigMap"))
+			waiter := statusWaiter{
+				client:          fakeClient,
+				restMapper:      fakeMapper,
+				customReadiness: true,
+			}
+
+			objs := getRuntimeObjFromManifests(t, []string{tt.manifest})
+			for _, obj := range objs {
+				u := obj.(*unstructured.Unstructured)
+				gvr := getGVR(t, fakeMapper, u)
+				require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+			}
+
+			resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+			err := waiter.Wait(resourceList, time.Second)
+			if tt.expectErrStrs == nil {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			for _, expected := range tt.expectErrStrs {
+				assert.Contains(t, err.Error(), expected)
+			}
+			for _, unexpected := range tt.notExpectErrStrs {
+				assert.NotContains(t, err.Error(), unexpected)
+			}
+		})
+	}
+}
+
+func TestWithCustomReadinessStatusReader(t *testing.T) {
+	opts := &waitOptions{}
+	WithCustomReadinessStatusReader()(opts)
+	assert.True(t, opts.enableCustomReadinessStatusReader)
+}
+
+func TestCustomReadinessStatusReaderReadStatusForObject(t *testing.T) {
+	reader := newCustomReadinessStatusReader(nil, statusreaders.NewDefaultStatusReader(testutil.NewFakeRESTMapper(v1.SchemeGroupVersion.WithKind("ConfigMap"))))
+	u := &unstructured.Unstructured{}
+	u.SetAPIVersion("v1")
+	u.SetKind("ConfigMap")
+	u.SetName("ready-config")
+	u.SetNamespace("default")
+	u.SetAnnotations(map[string]string{
+		AnnotationReadinessSuccess: `["{.phase} == \"Ready\""]`,
+		AnnotationReadinessFailure: `["{.phase} == \"Failed\""]`,
+	})
+	require.NoError(t, unstructured.SetNestedField(u.Object, "Ready", "status", "phase"))
+
+	result, err := reader.ReadStatusForObject(context.Background(), nil, u)
+	require.NoError(t, err)
+	assert.Equal(t, status.CurrentStatus, result.Status)
+}
+
+func TestCustomReadinessStatusReaderWarnsOnceForIncomparableExpression(t *testing.T) {
+	var buf bytes.Buffer
+	reader := newCustomReadinessStatusReader(slog.New(slog.NewTextHandler(&buf, nil)), statusreaders.NewDefaultStatusReader(testutil.NewFakeRESTMapper(v1.SchemeGroupVersion.WithKind("ConfigMap"))))
+
+	u := &unstructured.Unstructured{}
+	u.SetAPIVersion("v1")
+	u.SetKind("ConfigMap")
+	u.SetName("bad-ordering")
+	u.SetNamespace("default")
+	u.SetAnnotations(map[string]string{
+		AnnotationReadinessSuccess: `["{.phase} > \"Ready\""]`,
+		AnnotationReadinessFailure: `["{.failed} >= 1"]`,
+	})
+	require.NoError(t, unstructured.SetNestedField(u.Object, "Running", "status", "phase"))
+
+	for i := 0; i < 2; i++ {
+		result, err := reader.ReadStatusForObject(context.Background(), nil, u)
+		require.NoError(t, err)
+		assert.Equal(t, status.InProgressStatus, result.Status)
+		assert.Contains(t, result.Message, "skipped")
+		assert.Contains(t, result.Message, "ordering operators")
+	}
+
+	logged := buf.String()
+	assert.Equal(t, 1, strings.Count(logged, "treating condition as not met"),
+		"warning must be deduplicated to once per resource+expression")
+	assert.Contains(t, logged, "bad-ordering")
+	assert.Contains(t, logged, "ordering operators")
+}
+
+// The composite Deployment reader is the fallback observable: it populates
+// GeneratedResources with the child ReplicaSets it aggregated, which neither
+// the generic kstatus reader nor a bare status.Compute ever does. (In
+// cli-utils v1.2.2 the composite readers compute the same top-level status as
+// status.Compute, so GeneratedResources is the distinguishing signal that the
+// default chain — not an internal kstatus shortcut — handled the resource.)
+func TestCustomReadinessReaderDelegatesToFallbackChain(t *testing.T) {
+	t.Parallel()
+
+	deploymentYAML := func(annotations string) string {
+		return `
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: web
+  namespace: default
+  generation: 1` + annotations + `
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: web
+status:
+  observedGeneration: 1
+  replicas: 1
+  updatedReplicas: 1
+  readyReplicas: 1
+  availableReplicas: 1
+  conditions:
+  - type: Available
+    status: "True"
+    reason: MinimumReplicasAvailable
+`
+	}
+	replicaSetManifest := `
+apiVersion: apps/v1
+kind: ReplicaSet
+metadata:
+  name: web-abc123
+  namespace: default
+  generation: 1
+  labels:
+    app: web
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: web
+status:
+  observedGeneration: 1
+  replicas: 1
+  readyReplicas: 1
+  availableReplicas: 1
+`
+
+	tests := []struct {
+		name                   string
+		annotations            string
+		expectStatus           status.Status
+		expectGeneratedResults bool
+	}{
+		{
+			// No annotations: must be delegated to the composite chain.
+			name:                   "non-annotated deployment handled by composite reader",
+			annotations:            "",
+			expectStatus:           status.CurrentStatus,
+			expectGeneratedResults: true,
+		},
+		{
+			// One annotation only: partial config must also delegate.
+			name: "partial-annotated deployment handled by composite reader",
+			annotations: `
+  annotations:
+    helm.sh/readiness-success: '["{.readyReplicas} >= 1"]'`,
+			expectStatus:           status.CurrentStatus,
+			expectGeneratedResults: true,
+		},
+		{
+			// Both annotations: custom evaluation engages (unmet expression
+			// keeps it InProgress) and the fallback chain is bypassed.
+			name: "fully annotated deployment evaluated by expressions",
+			annotations: `
+  annotations:
+    helm.sh/readiness-success: '["{.readyReplicas} >= 2"]'
+    helm.sh/readiness-failure: '["{.failed} >= 1"]'`,
+			expectStatus:           status.InProgressStatus,
+			expectGeneratedResults: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			fakeMapper := testutil.NewFakeRESTMapper(
+				appsv1.SchemeGroupVersion.WithKind("Deployment"),
+				appsv1.SchemeGroupVersion.WithKind("ReplicaSet"),
+				v1.SchemeGroupVersion.WithKind("Pod"),
+			)
+			fakeClient := fake.NewSimpleDynamicClient(scheme.Scheme)
+			objs := getRuntimeObjFromManifests(t, []string{deploymentYAML(tt.annotations), replicaSetManifest})
+			for _, obj := range objs {
+				u := obj.(*unstructured.Unstructured)
+				gvr := getGVR(t, fakeMapper, u)
+				require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+			}
+			clusterReader := &clusterreader.DynamicClusterReader{
+				DynamicClient: fakeClient,
+				Mapper:        fakeMapper,
+			}
+			reader := newCustomReadinessStatusReader(nil, statusreaders.NewStatusReader(fakeMapper))
+
+			deployment := objs[0].(*unstructured.Unstructured)
+			result, err := reader.ReadStatusForObject(context.Background(), clusterReader, deployment)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectStatus, result.Status)
+			if tt.expectGeneratedResults {
+				require.NotEmpty(t, result.GeneratedResources,
+					"fallback composite reader must aggregate the child ReplicaSet")
+				assert.Equal(t, "web-abc123", result.GeneratedResources[0].Identifier.Name)
+			} else {
+				assert.Empty(t, result.GeneratedResources)
+				assert.Contains(t, result.Message, "custom readiness")
+			}
+
+			// The identifier-based entry point must route identically.
+			id, err := object.RuntimeToObjMeta(deployment)
+			require.NoError(t, err)
+			byID, err := reader.ReadStatus(context.Background(), clusterReader, id)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectStatus, byID.Status)
+			assert.Equal(t, tt.expectGeneratedResults, len(byID.GeneratedResources) > 0)
+		})
+	}
+}
+
+// A plain (non-annotated) Job in a batch that enables custom readiness must
+// keep helm's completion gating: helm's job reader holds a started Job
+// InProgress until a Complete/Failed condition, while plain kstatus reports a
+// started Job as Current. Before the fix the custom readiness reader shadowed
+// the job reader and the wait returned early.
+func TestStatusWaitWithJobsMixedCustomReadiness(t *testing.T) {
+	t.Parallel()
+
+	startedJobManifest := `
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: slow-job
+  namespace: default
+  generation: 1
+status:
+  startTime: 2025-02-06T16:34:20-05:00
+  active: 1
+  ready: 1
+`
+	completeJobManifest := `
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: slow-job
+  namespace: default
+  generation: 1
+status:
+  succeeded: 1
+  conditions:
+  - type: Complete
+    status: "True"
+`
+	readyConfigMapManifest := `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: gated-config
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} == \"Ready\""]'
+    helm.sh/readiness-failure: '["{.phase} == \"Failed\""]'
+status:
+  phase: Ready
+`
+	pendingConfigMapManifest := `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: gated-config
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} == \"Ready\""]'
+    helm.sh/readiness-failure: '["{.phase} == \"Failed\""]'
+status:
+  phase: Pending
+`
+
+	tests := []struct {
+		name             string
+		manifests        []string
+		expectErrStrs    []string
+		notExpectErrStrs []string
+	}{
+		{
+			name:      "plain job keeps completion gating in mixed batch",
+			manifests: []string{startedJobManifest, readyConfigMapManifest},
+			expectErrStrs: []string{
+				"resource Job/default/slow-job not ready. status: InProgress",
+			},
+			notExpectErrStrs: []string{"ConfigMap"},
+		},
+		{
+			name:      "annotated resource keeps expression gating in mixed batch",
+			manifests: []string{completeJobManifest, pendingConfigMapManifest},
+			expectErrStrs: []string{
+				"resource ConfigMap/default/gated-config not ready. status: InProgress",
+			},
+			notExpectErrStrs: []string{"Job/default/slow-job"},
+		},
+		{
+			name:      "batch completes when job is complete and expression met",
+			manifests: []string{completeJobManifest, readyConfigMapManifest},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			c := newTestClient(t)
+			fakeClient := fake.NewSimpleDynamicClient(scheme.Scheme)
+			fakeMapper := testutil.NewFakeRESTMapper(
+				batchv1.SchemeGroupVersion.WithKind("Job"),
+				v1.SchemeGroupVersion.WithKind("ConfigMap"),
+			)
+			waiter := statusWaiter{
+				client:          fakeClient,
+				restMapper:      fakeMapper,
+				customReadiness: true,
+			}
+			objs := getRuntimeObjFromManifests(t, tt.manifests)
+			for _, obj := range objs {
+				u := obj.(*unstructured.Unstructured)
+				gvr := getGVR(t, fakeMapper, u)
+				require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+			}
+			resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+			err := waiter.WaitWithJobs(resourceList, 2*time.Second)
+			if tt.expectErrStrs == nil {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			for _, expected := range tt.expectErrStrs {
+				assert.Contains(t, err.Error(), expected)
+			}
+			for _, unexpected := range tt.notExpectErrStrs {
+				assert.NotContains(t, err.Error(), unexpected)
+			}
+		})
+	}
+}
+
+// WatchUntilReady is the hook path: hook readiness must be governed by the
+// hook readers (helm's job/pod readers + always-ready for everything else),
+// never by the readiness annotations, even when custom readiness is enabled
+// on the waiter. Two directions:
+//   - a started-but-incomplete hook Job must still time out (kstatus would
+//     wrongly report a started Job as Current);
+//   - an annotated non-Job hook must complete even though its success
+//     expression is unmet (annotations are ignored for hooks).
+func TestWatchUntilReadyIgnoresCustomReadiness(t *testing.T) {
+	t.Parallel()
+
+	startedJobManifest := `
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: hook-job
+  namespace: default
+  generation: 1
+status:
+  startTime: 2025-02-06T16:34:20-05:00
+  active: 1
+`
+	completeJobManifest := `
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: hook-job
+  namespace: default
+  generation: 1
+status:
+  succeeded: 1
+  conditions:
+  - type: Complete
+    status: "True"
+`
+	annotatedConfigMapManifest := `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: hook-config
+  namespace: default
+  annotations:
+    helm.sh/readiness-success: '["{.phase} == \"Ready\""]'
+    helm.sh/readiness-failure: '["{.phase} == \"Failed\""]'
+status:
+  phase: Pending
+`
+
+	tests := []struct {
+		name          string
+		manifests     []string
+		expectErrStrs []string
+	}{
+		{
+			name:      "incomplete hook job still gated by hook job reader",
+			manifests: []string{startedJobManifest},
+			expectErrStrs: []string{
+				"resource Job/default/hook-job not ready. status: InProgress",
+			},
+		},
+		{
+			name:      "annotated hook resource ignores readiness annotations",
+			manifests: []string{annotatedConfigMapManifest},
+		},
+		{
+			name:      "complete hook job succeeds",
+			manifests: []string{completeJobManifest},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			c := newTestClient(t)
+			fakeClient := fake.NewSimpleDynamicClient(scheme.Scheme)
+			fakeMapper := testutil.NewFakeRESTMapper(
+				batchv1.SchemeGroupVersion.WithKind("Job"),
+				v1.SchemeGroupVersion.WithKind("ConfigMap"),
+			)
+			waiter := statusWaiter{
+				client:          fakeClient,
+				restMapper:      fakeMapper,
+				customReadiness: true,
+			}
+			objs := getRuntimeObjFromManifests(t, tt.manifests)
+			for _, obj := range objs {
+				u := obj.(*unstructured.Unstructured)
+				gvr := getGVR(t, fakeMapper, u)
+				require.NoError(t, fakeClient.Tracker().Create(gvr, u, u.GetNamespace()))
+			}
+			resourceList := getResourceListFromRuntimeObjs(t, c, objs)
+			err := waiter.WatchUntilReady(resourceList, 2*time.Second)
+			if tt.expectErrStrs == nil {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			for _, expected := range tt.expectErrStrs {
+				assert.Contains(t, err.Error(), expected)
+			}
+		})
+	}
+}

pkg/kube/options.go

@@ -94,12 +94,20 @@ func WithStatusComputeWorkers(n int) WaitOption {
 	}
 }
 
+// WithCustomReadinessStatusReader enables custom readiness evaluation while waiting.
+func WithCustomReadinessStatusReader() WaitOption {
+	return func(wo *waitOptions) {
+		wo.enableCustomReadinessStatusReader = true
+	}
+}
+
 type waitOptions struct {
-	ctx                  context.Context
+	ctx                               context.Context
-	watchUntilReadyCtx   context.Context
+	watchUntilReadyCtx                context.Context
-	waitCtx              context.Context
+	waitCtx                           context.Context
-	waitWithJobsCtx      context.Context
+	waitWithJobsCtx                   context.Context
-	waitForDeleteCtx     context.Context
+	waitForDeleteCtx                  context.Context
-	statusReaders        []engine.StatusReader
+	statusReaders                     []engine.StatusReader
-	statusComputeWorkers int
+	statusComputeWorkers              int
+	enableCustomReadinessStatusReader bool
 }

pkg/kube/readiness.go

@@ -0,0 +1,376 @@
+/*
+Copyright The Helm Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kube
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/util/jsonpath"
+
+	releaseutil "helm.sh/helm/v4/pkg/release/v1/util"
+)
+
+// The custom-readiness annotation keys' canonical home is
+// pkg/release/v1/util, next to the other sequencing annotations, so the pure
+// plan builder (pkg/release/v1/sequence) can read them without importing
+// pkg/kube. These value-identical aliases keep existing kube callers working.
+const (
+	// AnnotationReadinessSuccess declares custom readiness success conditions.
+	AnnotationReadinessSuccess = releaseutil.AnnotationReadinessSuccess
+
+	// AnnotationReadinessFailure declares custom readiness failure conditions.
+	// Failure conditions take precedence over success conditions.
+	AnnotationReadinessFailure = releaseutil.AnnotationReadinessFailure
+)
+
+// ReadinessStatus represents the evaluated readiness of a resource.
+type ReadinessStatus int
+
+const (
+	// ReadinessPending means neither success nor failure conditions are met.
+	ReadinessPending ReadinessStatus = iota
+	// ReadinessReady means at least one success condition is true.
+	ReadinessReady
+	// ReadinessFailed means at least one failure condition is true.
+	ReadinessFailed
+)
+
+func (r ReadinessStatus) String() string {
+	switch r {
+	case ReadinessPending:
+		return "Pending"
+	case ReadinessReady:
+		return "Ready"
+	case ReadinessFailed:
+		return "Failed"
+	default:
+		return "Unknown"
+	}
+}
+
+// ExpressionWarning describes a readiness expression that could not be
+// evaluated against the observed object state and was therefore treated as
+// "condition not met" rather than failing the whole wait.
+type ExpressionWarning struct {
+	// Expression is the original annotation expression, e.g. `{.phase} > "Ready"`.
+	Expression string
+	// Detail is a human-readable reason, including the observed value.
+	Detail string
+}
+
+// errIncomparableOrdering marks a comparison that cannot succeed against the
+// observed values: an ordering operator (<, <=, >, >=) applied to operands
+// that are not both numeric. The actual value's type is only known at runtime,
+// so lint cannot always catch this. EvaluateCustomReadiness downgrades it to
+// an ExpressionWarning — the expression counts as "condition not met",
+// mirroring how a missing status key is treated — instead of returning an
+// error that would abort the entire wait.
+var errIncomparableOrdering = errors.New("ordering operators (<, <=, >, >=) require numeric values")
+
+// EvaluateCustomReadiness evaluates custom readiness expressions against a
+// resource's .status field.
+//
+// When both successExprs and failureExprs are empty or nil, or when only one
+// side is provided, the caller should fall back to kstatus and useKstatus will
+// be returned as true.
+//
+// Expressions that cannot be evaluated against the observed values (an
+// ordering operator applied to non-numeric operands) are treated as
+// "condition not met" and reported in the returned warnings rather than as an
+// error. Expressions after the first met condition are not evaluated. All
+// other evaluation problems (invalid JSONPath, unsupported operator) are
+// returned as errors.
+func EvaluateCustomReadiness(obj *unstructured.Unstructured, successExprs, failureExprs []string) (ReadinessStatus, bool, []ExpressionWarning, error) {
+	hasSuccess := len(successExprs) > 0
+	hasFailure := len(failureExprs) > 0
+
+	if !hasSuccess || !hasFailure {
+		// Partial annotations: fall back to kstatus. The warning for this
+		// case is emitted once per batch by warnIfPartialReadinessAnnotations
+		// in the sequencing layer, not here (which runs on every poll tick).
+		return ReadinessPending, true, nil, nil
+	}
+
+	statusObj, found, err := unstructured.NestedMap(obj.Object, "status")
+	if err != nil || !found {
+		return ReadinessPending, false, nil, nil
+	}
+
+	statusWrapper := map[string]any{"status": statusObj}
+
+	var warnings []ExpressionWarning
+
+	for _, expr := range failureExprs {
+		met, err := evaluateExpression(statusWrapper, expr)
+		if err != nil {
+			if errors.Is(err, errIncomparableOrdering) {
+				warnings = append(warnings, ExpressionWarning{Expression: expr, Detail: err.Error()})
+				continue
+			}
+			return ReadinessPending, false, warnings, fmt.Errorf("evaluating failure expression %q: %w", expr, err)
+		}
+		if met {
+			return ReadinessFailed, false, warnings, nil
+		}
+	}
+
+	for _, expr := range successExprs {
+		met, err := evaluateExpression(statusWrapper, expr)
+		if err != nil {
+			if errors.Is(err, errIncomparableOrdering) {
+				warnings = append(warnings, ExpressionWarning{Expression: expr, Detail: err.Error()})
+				continue
+			}
+			return ReadinessPending, false, warnings, fmt.Errorf("evaluating success expression %q: %w", expr, err)
+		}
+		if met {
+			return ReadinessReady, false, warnings, nil
+		}
+	}
+
+	return ReadinessPending, false, warnings, nil
+}
+
+func parseReadinessExpressions(annotation string) ([]string, error) {
+	annotation = strings.TrimSpace(annotation)
+	if annotation == "" {
+		return nil, nil
+	}
+
+	var exprs []string
+	if err := json.Unmarshal([]byte(annotation), &exprs); err != nil {
+		return nil, fmt.Errorf("parsing readiness annotation JSON: %w", err)
+	}
+
+	for i, expr := range exprs {
+		exprs[i] = strings.TrimSpace(expr)
+	}
+
+	return exprs, nil
+}
+
+// ValidateReadinessExpressions parses a readiness annotation value (a JSON
+// array of "{<jsonpath>} <op> <value>" expressions) and verifies that every
+// expression is well-formed and compiles to a valid JSONPath. It does not
+// evaluate anything against a live object — it is the static-validation entry
+// point used by `helm lint` so malformed expressions are caught at author time
+// rather than silently failing readiness at install time. An empty or absent
+// annotation is valid (returns nil). It also rejects ordering operators
+// (<, <=, >, >=) whose comparison value is not numeric, since such a
+// comparison can never evaluate at runtime.
+func ValidateReadinessExpressions(annotation string) error {
+	exprs, err := parseReadinessExpressions(annotation)
+	if err != nil {
+		return err
+	}
+
+	for _, expr := range exprs {
+		path, op, val, err := parseReadinessExpression(expr)
+		if err != nil {
+			return err
+		}
+		template, err := readinessJSONPath(path)
+		if err != nil {
+			return err
+		}
+		if err := jsonpath.New("readiness").Parse(template); err != nil {
+			return fmt.Errorf("invalid JSONPath %q: %w", template, err)
+		}
+		// An ordering comparison can only ever succeed numerically, and the
+		// literal's type is statically known: a non-numeric comparison value
+		// with <, <=, >, >= is a definite authoring error.
+		switch op {
+		case "<", "<=", ">", ">=":
+			if _, ok := tryParseFloat(trimReadinessValue(val)); !ok {
+				return fmt.Errorf("expression %q: ordering operator %q requires a numeric comparison value, got %s", expr, op, val)
+			}
+		}
+	}
+
+	return nil
+}
+
+func evaluateExpression(obj map[string]any, expr string) (bool, error) {
+	path, op, rawVal, err := parseReadinessExpression(expr)
+	if err != nil {
+		return false, err
+	}
+
+	template, err := readinessJSONPath(path)
+	if err != nil {
+		return false, err
+	}
+
+	jp := jsonpath.New("readiness")
+	if err := jp.Parse(template); err != nil {
+		return false, fmt.Errorf("invalid JSONPath %q: %w", template, err)
+	}
+
+	var buf bytes.Buffer
+	if err := jp.Execute(&buf, obj); err != nil {
+		return false, nil
+	}
+
+	return compareValues(strings.TrimSpace(buf.String()), op, rawVal)
+}
+
+func readinessJSONPath(path string) (string, error) {
+	switch {
+	case strings.HasPrefix(path, ".status.") || strings.HasPrefix(path, ".status["):
+		// Already rooted at .status — avoid double-prefixing
+		return "{" + path + "}", nil
+	case strings.HasPrefix(path, "."):
+		return "{.status" + path + "}", nil
+	case strings.HasPrefix(path, "["):
+		return "{.status" + path + "}", nil
+	default:
+		return "", fmt.Errorf("invalid JSONPath %q: path must start with . or [", path)
+	}
+}
+
+// parseReadinessExpression parses "{<jsonpath>} <operator> <value>" into its parts.
+func parseReadinessExpression(expr string) (path, op, val string, err error) {
+	expr = strings.TrimSpace(expr)
+	if expr == "" {
+		return "", "", "", errors.New("expression cannot be empty")
+	}
+	if !strings.HasPrefix(expr, "{") {
+		return "", "", "", fmt.Errorf("expression must start with {<jsonpath>}: %q", expr)
+	}
+
+	closeBrace := strings.Index(expr, "}")
+	if closeBrace < 0 {
+		return "", "", "", fmt.Errorf("expression missing closing }: %q", expr)
+	}
+
+	path = strings.TrimSpace(expr[1:closeBrace])
+	if path == "" {
+		return "", "", "", fmt.Errorf("expression missing JSONPath: %q", expr)
+	}
+
+	rest := strings.TrimSpace(expr[closeBrace+1:])
+	if rest == "" {
+		return "", "", "", fmt.Errorf("expression missing operator and value: %q", expr)
+	}
+
+	parts := strings.Fields(rest)
+	if len(parts) == 0 {
+		return "", "", "", fmt.Errorf("expression missing operator: %q", expr)
+	}
+
+	switch parts[0] {
+	case "==", "!=", "<", "<=", ">", ">=":
+		op = parts[0]
+	default:
+		return "", "", "", fmt.Errorf("unsupported operator %q in expression %q", parts[0], expr)
+	}
+
+	val = strings.TrimSpace(rest[len(op):])
+	if val == "" {
+		return "", "", "", fmt.Errorf("expression missing comparison value: %q", expr)
+	}
+
+	return path, op, val, nil
+}
+
+func compareValues(actual, op, expected string) (bool, error) {
+	actual = strings.TrimSpace(actual)
+	expected = trimReadinessValue(strings.TrimSpace(expected))
+
+	if actualFloat, ok := tryParseFloat(actual); ok {
+		if expectedFloat, ok := tryParseFloat(expected); ok {
+			return compareNumeric(actualFloat, op, expectedFloat)
+		}
+	}
+
+	if actualBool, actualIsBool := tryParseBool(actual); actualIsBool {
+		if expectedBool, expectedIsBool := tryParseBool(expected); expectedIsBool {
+			switch op {
+			case "==":
+				return actualBool == expectedBool, nil
+			case "!=":
+				return actualBool != expectedBool, nil
+			default:
+				return false, fmt.Errorf("cannot compare %q %s %q: %w", actual, op, expected, errIncomparableOrdering)
+			}
+		}
+	}
+
+	switch op {
+	case "==":
+		return actual == expected, nil
+	case "!=":
+		return actual != expected, nil
+	default:
+		return false, fmt.Errorf("cannot compare %q %s %q: %w", actual, op, expected, errIncomparableOrdering)
+	}
+}
+
+func trimReadinessValue(value string) string {
+	if len(value) >= 2 {
+		if value[0] == '"' && value[len(value)-1] == '"' {
+			return value[1 : len(value)-1]
+		}
+		if value[0] == '\'' && value[len(value)-1] == '\'' {
+			return value[1 : len(value)-1]
+		}
+	}
+	return value
+}
+
+func tryParseFloat(value string) (float64, bool) {
+	parsed, err := strconv.ParseFloat(value, 64)
+	return parsed, err == nil
+}
+
+// tryParseBool accepts only the literals "true" and "false", unlike
+// strconv.ParseBool, so numeric strings such as "1" fall through to numeric
+// or string comparison instead of matching a boolean literal.
+func tryParseBool(value string) (bool, bool) {
+	switch value {
+	case "true":
+		return true, true
+	case "false":
+		return false, true
+	}
+	return false, false
+}
+
+func compareNumeric(actual float64, op string, expected float64) (bool, error) {
+	switch op {
+	case "==":
+		return actual == expected, nil
+	case "!=":
+		return actual != expected, nil
+	case "<":
+		return actual < expected, nil
+	case "<=":
+		return actual <= expected, nil
+	case ">":
+		return actual > expected, nil
+	case ">=":
+		return actual >= expected, nil
+	default:
+		return false, fmt.Errorf("unknown operator %q", op)
+	}
+}

pkg/kube/readiness_test.go

@@ -0,0 +1,545 @@
+/*
+Copyright The Helm Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kube
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+func makeUnstructuredWithStatus(t *testing.T, statusFields map[string]any) *unstructured.Unstructured {
+	t.Helper()
+
+	obj := &unstructured.Unstructured{}
+	obj.SetAPIVersion("v1")
+	obj.SetKind("ConfigMap")
+	obj.SetName("test-resource")
+
+	if statusFields != nil {
+		require.NoError(t, unstructured.SetNestedField(obj.Object, statusFields, "status"))
+	}
+
+	return obj
+}
+
+func TestEvaluateReadiness_BothNil_UsesKstatus(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, nil)
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(obj, nil, nil)
+
+	require.NoError(t, err)
+	assert.True(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+
+	result, useKstatus, _, err = EvaluateCustomReadiness(obj, []string{}, []string{})
+
+	require.NoError(t, err)
+	assert.True(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+}
+
+func TestEvaluateReadiness_OnlySuccess_FallsBackToKstatus(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, nil)
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(obj, []string{`{.ready} == true`}, nil)
+
+	require.NoError(t, err)
+	assert.True(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+}
+
+func TestEvaluateReadiness_OnlyFailure_FallsBackToKstatus(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, nil)
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(obj, nil, []string{`{.failed} == true`})
+
+	require.NoError(t, err)
+	assert.True(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+}
+
+func TestEvaluateReadiness_SuccessTrue(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{
+		"succeeded": int64(1),
+	})
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.succeeded} == 1`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessReady, result)
+}
+
+func TestEvaluateReadiness_FailurePrecedesSuccess(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{
+		"succeeded": int64(1),
+		"failed":    int64(1),
+	})
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.succeeded} == 1`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessFailed, result)
+}
+
+func TestEvaluateReadiness_ORSemantics_AnySuccessTrue(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{
+		"phase":   "Succeeded",
+		"another": "nope",
+	})
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.another} == "yes"`, `{.phase} == "Succeeded"`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessReady, result)
+}
+
+func TestEvaluateReadiness_NeitherConditionMet(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{
+		"succeeded": int64(0),
+	})
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.succeeded} == 1`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+}
+
+func TestEvaluateReadiness_MissingStatusField(t *testing.T) {
+	t.Run("missing status object", func(t *testing.T) {
+		obj := makeUnstructuredWithStatus(t, nil)
+
+		result, useKstatus, _, err := EvaluateCustomReadiness(
+			obj,
+			[]string{`{.phase} == "Running"`},
+			[]string{`{.failed} >= 1`},
+		)
+
+		require.NoError(t, err)
+		assert.False(t, useKstatus)
+		assert.Equal(t, ReadinessPending, result)
+	})
+
+	t.Run("missing status field", func(t *testing.T) {
+		obj := makeUnstructuredWithStatus(t, map[string]any{})
+
+		result, useKstatus, _, err := EvaluateCustomReadiness(
+			obj,
+			[]string{`{.phase} == "Running"`},
+			[]string{`{.failed} >= 1`},
+		)
+
+		require.NoError(t, err)
+		assert.False(t, useKstatus)
+		assert.Equal(t, ReadinessPending, result)
+	})
+}
+
+func TestEvaluateReadiness_NumericComparisonOperators(t *testing.T) {
+	tests := []struct {
+		name   string
+		expr   string
+		status map[string]any
+	}{
+		{name: "equal", expr: `{.succeeded} == 1`, status: map[string]any{"succeeded": int64(1)}},
+		{name: "not equal", expr: `{.succeeded} != 2`, status: map[string]any{"succeeded": int64(1)}},
+		{name: "less than", expr: `{.succeeded} < 2`, status: map[string]any{"succeeded": int64(1)}},
+		{name: "less than or equal", expr: `{.succeeded} <= 1`, status: map[string]any{"succeeded": int64(1)}},
+		{name: "greater than", expr: `{.succeeded} > 0`, status: map[string]any{"succeeded": int64(1)}},
+		{name: "greater than or equal", expr: `{.succeeded} >= 1`, status: map[string]any{"succeeded": int64(1)}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			obj := makeUnstructuredWithStatus(t, tt.status)
+
+			result, useKstatus, warnings, err := EvaluateCustomReadiness(
+				obj,
+				[]string{tt.expr},
+				[]string{`{.failed} >= 1`},
+			)
+
+			require.NoError(t, err)
+			assert.False(t, useKstatus)
+			assert.Empty(t, warnings)
+			assert.Equal(t, ReadinessReady, result)
+		})
+	}
+}
+
+func TestEvaluateReadiness_BooleanComparison(t *testing.T) {
+	tests := []struct {
+		name   string
+		expr   string
+		status map[string]any
+	}{
+		{name: "boolean equals", expr: `{.ready} == true`, status: map[string]any{"ready": true}},
+		{name: "boolean not equals", expr: `{.ready} != false`, status: map[string]any{"ready": true}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			obj := makeUnstructuredWithStatus(t, tt.status)
+
+			result, useKstatus, _, err := EvaluateCustomReadiness(
+				obj,
+				[]string{tt.expr},
+				[]string{`{.failed} == true`},
+			)
+
+			require.NoError(t, err)
+			assert.False(t, useKstatus)
+			assert.Equal(t, ReadinessReady, result)
+		})
+	}
+}
+
+func TestEvaluateReadiness_StringComparison(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{
+		"phase": "Running",
+	})
+
+	result, useKstatus, warnings, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.phase} == "Running"`},
+		[]string{`{.phase} == "Failed"`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Empty(t, warnings)
+	assert.Equal(t, ReadinessReady, result)
+
+	result, useKstatus, warnings, err = EvaluateCustomReadiness(
+		obj,
+		[]string{`{.phase} != "Failed"`},
+		[]string{`{.phase} == "Failed"`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Empty(t, warnings)
+	assert.Equal(t, ReadinessReady, result)
+}
+
+func TestEvaluateReadiness_InvalidJSONPath(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{
+		"phase": "Running",
+	})
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.phase[} == "Running"`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.ErrorContains(t, err, "invalid JSONPath")
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+}
+
+func TestEvaluateReadiness_InvalidOperator(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{
+		"phase": "Running",
+	})
+
+	result, useKstatus, _, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.phase} <> "Running"`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.ErrorContains(t, err, "unsupported operator")
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+}
+
+func TestParseReadinessExpression(t *testing.T) {
+	t.Run("valid expressions", func(t *testing.T) {
+		tests := []struct {
+			name     string
+			expr     string
+			wantPath string
+			wantOp   string
+			wantVal  string
+		}{
+			{name: "string equality", expr: `{.phase} == "Running"`, wantPath: ".phase", wantOp: "==", wantVal: `"Running"`},
+			{name: "numeric greater than or equal", expr: `{.count} >= 3`, wantPath: ".count", wantOp: ">=", wantVal: "3"},
+			{name: "boolean not equal", expr: `{.ready} != false`, wantPath: ".ready", wantOp: "!=", wantVal: "false"},
+		}
+
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				path, op, val, err := parseReadinessExpression(tt.expr)
+
+				require.NoError(t, err)
+				assert.Equal(t, tt.wantPath, path)
+				assert.Equal(t, tt.wantOp, op)
+				assert.Equal(t, tt.wantVal, val)
+			})
+		}
+	})
+
+	t.Run("invalid expressions", func(t *testing.T) {
+		tests := []struct {
+			name        string
+			expr        string
+			wantErrPart string
+		}{
+			{name: "missing opening brace", expr: `.phase == "Running"`, wantErrPart: "must start"},
+			{name: "missing closing brace", expr: `{.phase == "Running"`, wantErrPart: "missing closing"},
+			{name: "missing operator", expr: `{.phase} "Running"`, wantErrPart: "unsupported operator"},
+			{name: "unsupported operator", expr: `{.phase} <> "Running"`, wantErrPart: "unsupported operator"},
+			{name: "missing value", expr: `{.phase} ==`, wantErrPart: "missing comparison value"},
+		}
+
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				_, _, _, err := parseReadinessExpression(tt.expr)
+				assert.ErrorContains(t, err, tt.wantErrPart)
+			})
+		}
+	})
+}
+
+func TestReadinessJSONPath(t *testing.T) {
+	// Locks the contract that paths already rooted at .status do NOT get
+	// .status double-prepended. Chart authors commonly write
+	// `{.status.phase}` (the canonical form in K8s docs); without the
+	// guard, the effective JSONPath becomes `.status.status.phase` and
+	// the readiness gate silently fails to find any value.
+	tests := []struct {
+		name    string
+		path    string
+		want    string
+		wantErr string
+	}{
+		{name: "already rooted at .status. — no double-prepend", path: ".status.phase", want: "{.status.phase}"},
+		{name: "already rooted at .status[ — no double-prepend", path: `.status['observedGeneration']`, want: `{.status['observedGeneration']}`},
+		{name: "non-status dot path gets .status prefix", path: ".phase", want: "{.status.phase}"},
+		{name: "bracket-only path gets .status prefix", path: "[0]", want: "{.status[0]}"},
+		{name: "invalid path rejected", path: "phase", wantErr: "must start with . or ["},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := readinessJSONPath(tt.path)
+			if tt.wantErr != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestValidateReadinessExpressions(t *testing.T) {
+	tests := []struct {
+		name       string
+		annotation string
+		wantErr    string
+	}{
+		{name: "empty annotation is valid", annotation: ""},
+		{name: "well-formed single expression", annotation: `["{.status.phase} == \"Ready\""]`},
+		{name: "well-formed multiple expressions", annotation: `["{.status.succeeded} >= 1", "{.status.ready} == true"]`},
+		{name: "not a JSON array", annotation: `not-json`, wantErr: "parsing readiness annotation JSON"},
+		{name: "missing closing brace", annotation: `["{.status.succeeded >= 1"]`, wantErr: "missing closing"},
+		{name: "path not rooted at . or [", annotation: `["{status.phase} == \"Ready\""]`, wantErr: "must start with . or ["},
+		{name: "ordering operator with quoted string value", annotation: `["{.status.phase} > \"Running\""]`, wantErr: "requires a numeric comparison value"},
+		{name: "ordering operator with boolean value", annotation: `["{.status.ready} >= true"]`, wantErr: "requires a numeric comparison value"},
+		{name: "ordering operator with quoted numeric value is valid", annotation: `["{.status.replicas} >= \"3\""]`},
+		{name: "ordering operator with unquoted non-numeric value", annotation: `["{.status.phase} > Running"]`, wantErr: "requires a numeric comparison value"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ValidateReadinessExpressions(tt.annotation)
+			if tt.wantErr != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+		})
+	}
+}
+
+func TestReadinessExpressionsJSON(t *testing.T) {
+	t.Run("empty list", func(t *testing.T) {
+		exprs, err := parseReadinessExpressions(`[]`)
+
+		require.NoError(t, err)
+		assert.Empty(t, exprs)
+	})
+
+	t.Run("malformed annotation JSON", func(t *testing.T) {
+		_, err := parseReadinessExpressions(`["{.ready} == true"`)
+		assert.ErrorContains(t, err, "parsing readiness annotation JSON")
+	})
+}
+
+func TestEvaluateReadiness_OrderingOnNonNumericString_TreatedAsNotMet(t *testing.T) {
+	for _, op := range []string{"<", "<=", ">", ">="} {
+		t.Run(op, func(t *testing.T) {
+			obj := makeUnstructuredWithStatus(t, map[string]any{"phase": "Running"})
+			expr := fmt.Sprintf(`{.phase} %s "Ready"`, op)
+
+			result, useKstatus, warnings, err := EvaluateCustomReadiness(
+				obj,
+				[]string{expr},
+				[]string{`{.failed} >= 1`},
+			)
+
+			require.NoError(t, err)
+			assert.False(t, useKstatus)
+			assert.Equal(t, ReadinessPending, result)
+			require.Len(t, warnings, 1)
+			assert.Equal(t, expr, warnings[0].Expression)
+			assert.Contains(t, warnings[0].Detail, "ordering operators")
+			assert.Contains(t, warnings[0].Detail, `"Running"`)
+		})
+	}
+}
+
+func TestEvaluateReadiness_IncomparableFailureExpression_Symmetric(t *testing.T) {
+	// A type-mismatched FAILURE expression must mean "failure condition not
+	// met" — not "failed" and not an error — so a satisfied success
+	// expression still marks the resource ready, with the mistake surfaced
+	// as a warning.
+	obj := makeUnstructuredWithStatus(t, map[string]any{"phase": "Running"})
+
+	result, useKstatus, warnings, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.phase} == "Running"`},
+		[]string{`{.phase} > "Failed"`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessReady, result)
+	require.Len(t, warnings, 1)
+	assert.Equal(t, `{.phase} > "Failed"`, warnings[0].Expression)
+}
+
+func TestEvaluateReadiness_NumericStringCoercionPreserved(t *testing.T) {
+	// A string status value that parses as a number keeps comparing
+	// numerically — only genuinely non-numeric values changed behavior.
+	tests := []struct{ name, expr string }{
+		{name: "unquoted numeric literal", expr: `{.phase} >= 5`},
+		{name: "quoted numeric literal", expr: `{.phase} >= "5"`},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			obj := makeUnstructuredWithStatus(t, map[string]any{"phase": "5"})
+
+			result, useKstatus, warnings, err := EvaluateCustomReadiness(
+				obj,
+				[]string{tt.expr},
+				[]string{`{.failed} >= 1`},
+			)
+
+			require.NoError(t, err)
+			assert.False(t, useKstatus)
+			assert.Empty(t, warnings)
+			assert.Equal(t, ReadinessReady, result)
+		})
+	}
+}
+
+func TestEvaluateReadiness_OrderingOnBooleans_TreatedAsNotMet(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{"ready": true})
+
+	result, useKstatus, warnings, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.ready} >= true`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.NoError(t, err)
+	assert.False(t, useKstatus)
+	assert.Equal(t, ReadinessPending, result)
+	require.Len(t, warnings, 1)
+}
+
+func TestEvaluateReadiness_AllSuccessExpressionsIncomparable_Pending(t *testing.T) {
+	obj := makeUnstructuredWithStatus(t, map[string]any{"phase": "Running"})
+
+	result, _, warnings, err := EvaluateCustomReadiness(
+		obj,
+		[]string{`{.phase} > "Ready"`, `{.phase} < "Ready"`},
+		[]string{`{.failed} >= 1`},
+	)
+
+	require.NoError(t, err)
+	assert.Equal(t, ReadinessPending, result)
+	assert.Len(t, warnings, 2)
+}
+
+func TestCompareValues_OrderingOnNonNumeric_ReturnsSentinel(t *testing.T) {
+	met, err := compareValues("Running", ">", `"Ready"`)
+	require.ErrorIs(t, err, errIncomparableOrdering)
+	assert.False(t, met)
+
+	// A numeric actual against a non-numeric literal is equally incomparable.
+	met, err = compareValues("5", ">", `"high"`)
+	require.ErrorIs(t, err, errIncomparableOrdering)
+	assert.False(t, met)
+}
+
+func TestCompareValues_NumericStringIsNotBoolean(t *testing.T) {
+	// "1" must not satisfy a boolean literal via strconv.ParseBool semantics.
+	met, err := compareValues("1", "==", "true")
+	require.NoError(t, err)
+	assert.False(t, met)
+
+	met, err = compareValues("0", "!=", "false")
+	require.NoError(t, err)
+	assert.True(t, met)
+
+	met, err = compareValues("true", "==", "true")
+	require.NoError(t, err)
+	assert.True(t, met)
+
+	met, err = compareValues("false", "!=", "true")
+	require.NoError(t, err)
+	assert.True(t, met)
+}

pkg/kube/statuswait.go

@@ -52,6 +52,12 @@ type statusWaiter struct {
 	waitForDeleteCtx     context.Context
 	readers              []engine.StatusReader
 	statusComputeWorkers int
+	// customReadiness enables per-resource evaluation of the
+	// helm.sh/readiness-success|failure annotations in Wait and WaitWithJobs.
+	// WatchUntilReady deliberately ignores it: hook readiness is defined by
+	// hook semantics, and the partial-annotation warning only runs for
+	// sequenced release resources.
+	customReadiness bool
 	logging.LogHolder
 }
 
@@ -75,6 +81,17 @@ func getStatusWatcher(dynamicClient dynamic.Interface, mapper meta.RESTMapper) *
 	return sw
 }
 
+// wrapCustomReadiness layers custom readiness evaluation on top of the reader
+// chain a wait method would otherwise use. It must wrap the complete chain so
+// that resources without the annotations are delegated to exactly the readers
+// they would get with custom readiness disabled.
+func (w *statusWaiter) wrapCustomReadiness(fallback engine.StatusReader) engine.StatusReader {
+	if !w.customReadiness {
+		return fallback
+	}
+	return newCustomReadinessStatusReader(w.Logger(), fallback)
+}
+
 func (w *statusWaiter) WatchUntilReady(resourceList ResourceList, timeout time.Duration) error {
 	if timeout == 0 {
 		timeout = DefaultStatusWatcherTimeout
@@ -89,6 +106,7 @@ func (w *statusWaiter) WatchUntilReady(resourceList ResourceList, timeout time.D
 	// We don't want to wait on any other resources as watchUntilReady is only for Helm hooks.
 	// If custom readers are defined they can be used as Helm hooks support any resource.
 	// We put them in front since the DelegatingStatusReader uses the first reader that matches.
+	// Custom readiness annotations (w.customReadiness) are deliberately not applied to hooks.
 	genericSR := statusreaders.NewGenericStatusReader(w.restMapper, alwaysReady)
 
 	sr := &statusreaders.DelegatingStatusReader{
@@ -107,7 +125,7 @@ func (w *statusWaiter) Wait(resourceList ResourceList, timeout time.Duration) er
 	w.Logger().Debug("waiting for resources", "count", len(resourceList), "timeout", timeout)
 	sw := getStatusWatcher(w.client, w.restMapper)
 	sw.StatusComputeWorkers = w.statusComputeWorkers
-	sw.StatusReader = statusreaders.NewStatusReader(w.restMapper, w.readers...)
+	sw.StatusReader = w.wrapCustomReadiness(statusreaders.NewStatusReader(w.restMapper, w.readers...))
 	return w.wait(ctx, resourceList, sw)
 }
 
@@ -123,8 +141,7 @@ func (w *statusWaiter) WaitWithJobs(resourceList ResourceList, timeout time.Dura
 	newCustomJobStatusReader := helmStatusReaders.NewCustomJobStatusReader(w.restMapper)
 	readers := append([]engine.StatusReader(nil), w.readers...)
 	readers = append(readers, newCustomJobStatusReader)
-	customSR := statusreaders.NewStatusReader(w.restMapper, readers...)
+	sw.StatusReader = w.wrapCustomReadiness(statusreaders.NewStatusReader(w.restMapper, readers...))
-	sw.StatusReader = customSR
 	return w.wait(ctx, resourceList, sw)
 }