diff --git a/go/deployment-operator/go.mod b/go/deployment-operator/go.mod
index 238eb1e2b2..d24e9e5358 100644
--- a/go/deployment-operator/go.mod
+++ b/go/deployment-operator/go.mod
@@ -69,6 +69,7 @@ require (
 	github.com/yuin/gopher-lua v1.1.1
 	gitlab.com/gitlab-org/api/client-go v1.46.0
 	golang.org/x/oauth2 v0.36.0
+	golang.org/x/sync v0.22.0
 	golang.org/x/time v0.15.0
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/gotestsum v1.13.0
@@ -340,7 +341,6 @@ require (
 	golang.org/x/exp/typeparams v0.0.0-20260209203927-2842357ff358 // indirect
 	golang.org/x/mod v0.36.0 // indirect
 	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sync v0.22.0 // indirect
 	golang.org/x/sys v0.46.0 // indirect
 	golang.org/x/term v0.44.0 // indirect
 	golang.org/x/text v0.38.0 // indirect
diff --git a/go/deployment-operator/pkg/cache/discovery/cache.go b/go/deployment-operator/pkg/cache/discovery/cache.go
index a24aacc08f..650eb01bfa 100644
--- a/go/deployment-operator/pkg/cache/discovery/cache.go
+++ b/go/deployment-operator/pkg/cache/discovery/cache.go
@@ -37,6 +37,9 @@ type Cache interface {
 	// MaybeResetRESTMapper resets the RESTMapper if the provided GVKs are CustomResourceDefinitions.
 	MaybeResetRESTMapper(...schema.GroupVersionKind)
 
+	// ResetRESTMapper resets cached discovery used by the RESTMapper.
+	ResetRESTMapper()
+
 	// GroupVersionKind returns the set of GroupVersionKinds in the cache.
 	GroupVersionKind() containers.Set[schema.GroupVersionKind]
 
@@ -151,6 +154,19 @@ func (in *cache) MaybeResetRESTMapper(crds ...schema.GroupVersionKind) {
 	}
 }
 
+func (in *cache) ResetRESTMapper() {
+	in.mu.Lock()
+	defer in.mu.Unlock()
+
+	if in.mapper == nil {
+		klog.V(log.LogLevelVerbose).ErrorS(fmt.Errorf("no RESTMapper provided, cannot reset"), "unable to reset RESTMapper")
+		return
+	}
+
+	meta.MaybeResetRESTMapper(in.mapper)
+	klog.V(log.LogLevelExtended).InfoS("resetting RESTMapper")
+}
+
 func (in *cache) RestMapping(gvk schema.GroupVersionKind) (*meta.RESTMapping, error) {
 	in.mu.Lock()
 	mapping, err := in.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
diff --git a/go/deployment-operator/pkg/cache/discovery/cache_test.go b/go/deployment-operator/pkg/cache/discovery/cache_test.go
new file mode 100644
index 0000000000..d2b1d9ee91
--- /dev/null
+++ b/go/deployment-operator/pkg/cache/discovery/cache_test.go
@@ -0,0 +1,34 @@
+package discovery
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/api/meta"
+)
+
+type resettableMapper struct {
+	meta.RESTMapper
+	resetCalls atomic.Int32
+}
+
+func (in *resettableMapper) Reset() {
+	in.resetCalls.Add(1)
+}
+
+func TestResetRESTMapper(t *testing.T) {
+	t.Run("delegates to resettable mapper", func(t *testing.T) {
+		mapper := &resettableMapper{}
+		cache := NewCache(nil, mapper)
+
+		cache.ResetRESTMapper()
+
+		assert.Equal(t, int32(1), mapper.resetCalls.Load())
+	})
+
+	t.Run("ignores missing mapper", func(t *testing.T) {
+		cache := NewCache(nil, nil)
+		assert.NotPanics(t, cache.ResetRESTMapper)
+	})
+}
diff --git a/go/deployment-operator/pkg/common/crd.go b/go/deployment-operator/pkg/common/crd.go
new file mode 100644
index 0000000000..a5a0f4a1f9
--- /dev/null
+++ b/go/deployment-operator/pkg/common/crd.go
@@ -0,0 +1,54 @@
+package common
+
+import (
+	"github.com/samber/lo"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+var crdGroupKind = apiextensionsv1.SchemeGroupVersion.WithKind(CustomResourceDefinitionKind).GroupKind()
+
+func IsCRD(resource unstructured.Unstructured) bool {
+	return resource.GroupVersionKind().GroupKind() == crdGroupKind
+}
+
+func isStatusConditionTrue(resource unstructured.Unstructured, conditionType string) bool {
+	conditions, _, _ := unstructured.NestedSlice(resource.Object, "status", "conditions")
+	return meta.IsStatusConditionTrue(lo.FilterMap(conditions, func(condition any, _ int) (metav1.Condition, bool) {
+		value, ok := condition.(map[string]any)
+		if !ok {
+			return metav1.Condition{}, false
+		}
+
+		currentType, typeFound, _ := unstructured.NestedString(value, "type")
+		conditionStatus, statusFound, _ := unstructured.NestedString(value, "status")
+		return metav1.Condition{
+			Type:   currentType,
+			Status: metav1.ConditionStatus(conditionStatus),
+		}, typeFound && statusFound
+	}), conditionType)
+}
+
+func CRDEstablished(resource unstructured.Unstructured) bool {
+	return IsCRD(resource) && isStatusConditionTrue(resource, string(apiextensionsv1.Established))
+}
+
+func ServedCRDGVKs(resource unstructured.Unstructured) []schema.GroupVersionKind {
+	if !IsCRD(resource) {
+		return nil
+	}
+
+	crd := new(apiextensionsv1.CustomResourceDefinition)
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(resource.Object, crd); err != nil {
+		return nil
+	}
+
+	return lo.FilterMap(crd.Spec.Versions, func(version apiextensionsv1.CustomResourceDefinitionVersion, _ int) (schema.GroupVersionKind, bool) {
+		gvk := schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name}.WithKind(crd.Spec.Names.Kind)
+		return gvk, version.Served && gvk.Group != "" && gvk.Version != "" && gvk.Kind != ""
+	})
+}
diff --git a/go/deployment-operator/pkg/common/crd_test.go b/go/deployment-operator/pkg/common/crd_test.go
new file mode 100644
index 0000000000..b6a5db08d2
--- /dev/null
+++ b/go/deployment-operator/pkg/common/crd_test.go
@@ -0,0 +1,40 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func TestCRDUtilities(t *testing.T) {
+	crd := unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "apiextensions.k8s.io/v1",
+		"kind":       "CustomResourceDefinition",
+		"spec": map[string]any{
+			"group": "example.com",
+			"names": map[string]any{"kind": "Widget", "plural": "widgets"},
+			"versions": []any{
+				map[string]any{"name": "v1", "served": true, "storage": true},
+				map[string]any{"name": "v2", "served": false, "storage": false},
+			},
+		},
+		"status": map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+		},
+	}}
+
+	assert.True(t, IsCRD(crd))
+	assert.True(t, CRDEstablished(crd))
+	assert.Equal(t, []schema.GroupVersionKind{{Group: "example.com", Version: "v1", Kind: "Widget"}}, ServedCRDGVKs(crd))
+	assert.False(t, CRDEstablished(unstructured.Unstructured{}))
+
+	versions, _, _ := unstructured.NestedSlice(crd.Object, "spec", "versions")
+	versions = append(versions,
+		map[string]any{"name": "v3", "served": false},
+		map[string]any{"name": "", "served": true},
+	)
+	assert.NoError(t, unstructured.SetNestedSlice(crd.Object, versions, "spec", "versions"))
+	assert.Equal(t, []schema.GroupVersionKind{{Group: "example.com", Version: "v1", Kind: "Widget"}}, ServedCRDGVKs(crd))
+}
diff --git a/go/deployment-operator/pkg/manifests/template/common.go b/go/deployment-operator/pkg/manifests/template/common.go
index 9e7e3444a4..4f01c4b356 100644
--- a/go/deployment-operator/pkg/manifests/template/common.go
+++ b/go/deployment-operator/pkg/manifests/template/common.go
@@ -16,13 +16,10 @@ import (
 	"sigs.k8s.io/kustomize/kyaml/kio/kioutil"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
 
+	commonpkg "github.com/pluralsh/console/go/deployment-operator/pkg/common"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
 )
 
-var (
-	crdGK = schema.GroupKind{Group: "apiextensions.k8s.io", Kind: "CustomResourceDefinition"}
-)
-
 func setNamespaces(mapper meta.RESTMapper, objs []unstructured.Unstructured,
 	defaultNamespace string, enforceNamespace bool) ([]unstructured.Unstructured, error) {
 	// find any crds in the set of resources.
@@ -134,8 +131,7 @@ func IsCRD(u *unstructured.Unstructured) bool {
 	if u == nil {
 		return false
 	}
-	gvk := u.GroupVersionKind()
-	return crdGK == gvk.GroupKind()
+	return commonpkg.IsCRD(*u)
 }
 
 // LookupResourceScope tries to look up the scope of the type of the provided
diff --git a/go/deployment-operator/pkg/streamline/applier/applier.go b/go/deployment-operator/pkg/streamline/applier/applier.go
index 0db67c0840..6e6fdc29b6 100644
--- a/go/deployment-operator/pkg/streamline/applier/applier.go
+++ b/go/deployment-operator/pkg/streamline/applier/applier.go
@@ -4,8 +4,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/pluralsh/console/go/client"
-	"github.com/pluralsh/console/go/polly/containers"
 	"github.com/samber/lo"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -13,11 +11,13 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/klog/v2"
 
+	"github.com/pluralsh/console/go/client"
 	"github.com/pluralsh/console/go/deployment-operator/internal/helpers"
 	discoverycache "github.com/pluralsh/console/go/deployment-operator/pkg/cache/discovery"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
 	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
 	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline/store"
+	"github.com/pluralsh/console/go/polly/containers"
 )
 
 type Applier struct {
@@ -41,6 +41,8 @@ func (in *Applier) Apply(ctx context.Context,
 	opts ...WaveProcessorOption,
 ) ([]client.ComponentAttributes, []client.ServiceErrorAttributes, error) {
 	resources = in.ensureServiceAnnotation(resources, service.ID)
+	gates := []Gate{newCRDGate(in.store, in.discoveryCache, withCRDGateResources(resources, lo.FromPtr(service.DryRun)))}
+	opts = append(opts, WithWaveGates(lo.Filter(gates, func(g Gate, _ int) bool { return g.Enabled() })...))
 
 	if err := in.store.SyncServiceComponents(service.ID, resources); err != nil {
 		return nil, nil, err
@@ -72,7 +74,7 @@ func (in *Applier) Apply(ctx context.Context,
 		}
 
 		now := time.Now()
-		syncPhase = lo.ToPtr(phase.Name())
+		syncPhase = new(phase.Name())
 
 		if !phase.HasWaves() {
 			klog.V(log.LogLevelDefault).InfoS(
@@ -134,7 +136,7 @@ func (in *Applier) Apply(ctx context.Context,
 			serviceErrorList = append(serviceErrorList, client.ServiceErrorAttributes{
 				Source:  string(phase.Name()),
 				Message: "waiting for resources to be ready",
-				Warning: lo.ToPtr(true),
+				Warning: new(true),
 			})
 			klog.V(log.LogLevelTrace).InfoS("waiting for resources to be ready", "phase", phase.Name())
 			break
@@ -198,8 +200,8 @@ func (in *Applier) Destroy(ctx context.Context, serviceID string) ([]client.Comp
 		err = in.client.
 			Resource(helpers.GVRFromGVK(live.GroupVersionKind())).
 			Namespace(live.GetNamespace()).Delete(ctx, live.GetName(), metav1.DeleteOptions{
-			GracePeriodSeconds: lo.ToPtr(int64(0)),
-			PropagationPolicy:  lo.ToPtr(metav1.DeletePropagationBackground),
+			GracePeriodSeconds: new(int64(0)),
+			PropagationPolicy:  new(metav1.DeletePropagationBackground),
 		})
 		if errors.IsNotFound(err) {
 			if err := in.store.DeleteComponent(smcommon.NewStoreKeyFromUnstructured(lo.FromPtr(live))); err != nil {
diff --git a/go/deployment-operator/pkg/streamline/applier/crd_gate.go b/go/deployment-operator/pkg/streamline/applier/crd_gate.go
new file mode 100644
index 0000000000..5e5349919f
--- /dev/null
+++ b/go/deployment-operator/pkg/streamline/applier/crd_gate.go
@@ -0,0 +1,219 @@
+package applier
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"golang.org/x/sync/singleflight"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
+
+	discoverycache "github.com/pluralsh/console/go/deployment-operator/pkg/cache/discovery"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/common"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
+	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline/store"
+)
+
+const crdGateWaitTimeout = 30 * time.Second
+
+var crdGateBackoff = wait.Backoff{
+	Duration: 100 * time.Millisecond,
+	Factor:   1.5,
+	Jitter:   0.1,
+	Steps:    100,
+	Cap:      2 * time.Second,
+}
+
+type crdGate struct {
+	store          store.Store
+	discoveryCache discoverycache.Cache
+	waitTimeout    time.Duration
+	backoff        wait.Backoff
+
+	mu    sync.RWMutex
+	crds  map[schema.GroupVersionKind]unstructured.Unstructured
+	waits singleflight.Group
+}
+
+type CRDGateOption func(*crdGate)
+
+func withCRDGateResources(resources []unstructured.Unstructured, dryRun bool) CRDGateOption {
+	return func(gate *crdGate) {
+		if dryRun {
+			return
+		}
+
+		candidates := make(map[schema.GroupVersionKind]unstructured.Unstructured)
+		for _, resource := range resources {
+			if !common.IsCRD(resource) {
+				continue
+			}
+
+			component, err := gate.store.GetAppliedComponent(resource)
+			if err != nil {
+				klog.V(log.LogLevelExtended).ErrorS(err, "could not classify CRD for apply-scoped readiness", "crd", resource.GetName())
+				continue
+			}
+			if component != nil {
+				continue
+			}
+
+			for _, gvk := range common.ServedCRDGVKs(resource) {
+				candidates[gvk] = resource
+				klog.V(log.LogLevelDebug).InfoS("registered CRD gate candidate", "crd", resource.GetName(), "gvk", gvk.String())
+			}
+		}
+		if len(candidates) == 0 {
+			klog.V(log.LogLevelTrace).InfoS("no CRD gate candidates registered")
+			return
+		}
+
+		gate.mu.Lock()
+		defer gate.mu.Unlock()
+		for gvk, crd := range candidates {
+			gate.crds[gvk] = crd
+		}
+	}
+}
+
+func newCRDGate(store store.Store, discoveryCache discoverycache.Cache, opts ...CRDGateOption) *crdGate {
+	result := &crdGate{
+		store:          store,
+		discoveryCache: discoveryCache,
+		waitTimeout:    crdGateWaitTimeout,
+		backoff:        crdGateBackoff,
+		crds:           make(map[schema.GroupVersionKind]unstructured.Unstructured),
+	}
+
+	for _, opt := range opts {
+		opt(result)
+	}
+
+	return result
+}
+
+func (in *crdGate) Run(ctx context.Context, condition GateCondition, resource unstructured.Unstructured) error {
+	switch condition {
+	case GateConditionPreApply:
+		return in.waitForMapping(ctx, resource.GroupVersionKind())
+	case GateConditionPostApply:
+		in.markApplied(resource)
+		return nil
+	default:
+		return nil
+	}
+}
+
+func (in *crdGate) Enabled() bool {
+	return len(in.crds) > 0
+}
+
+func (in *crdGate) markApplied(resource unstructured.Unstructured) {
+	if !common.IsCRD(resource) || resource.GetUID() == "" {
+		klog.V(log.LogLevelTrace).InfoS("skipping CRD gate mark-applied for non-CRD or missing UID", "resource", resource.GetName(), "gvk", resource.GroupVersionKind().String(), "uid", resource.GetUID())
+		return
+	}
+
+	key := smcommon.NewStoreKeyFromUnstructured(resource)
+	in.mu.Lock()
+	defer in.mu.Unlock()
+
+	for gvk, candidate := range in.crds {
+		if smcommon.NewStoreKeyFromUnstructured(candidate) != key {
+			continue
+		}
+
+		in.crds[gvk] = resource
+		klog.V(log.LogLevelDebug).InfoS("marked CRD gate candidate as applied", "crd", resource.GetName(), "gvk", gvk.String(), "uid", resource.GetUID())
+	}
+}
+
+func (in *crdGate) waitForMapping(ctx context.Context, gvk schema.GroupVersionKind) error {
+	crd, eligible := in.appliedCRD(gvk)
+	if !eligible {
+		klog.V(log.LogLevelTrace).InfoS("skipping CRD gate wait for ineligible GVK", "gvk", gvk.String())
+		return nil
+	}
+	if err := ctx.Err(); err != nil {
+		klog.V(log.LogLevelDebug).ErrorS(err, "CRD gate wait context already cancelled", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+		return err
+	}
+
+	klog.V(log.LogLevelDebug).InfoS("waiting for CRD gate REST mapping", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID(), "timeout", in.waitTimeout.String())
+	resultCh := in.waits.DoChan(gvk.String(), func() (any, error) {
+		return in.resolve(ctx, gvk, crd)
+	})
+
+	select {
+	case <-ctx.Done():
+		klog.V(log.LogLevelDebug).ErrorS(ctx.Err(), "CRD gate wait cancelled", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+		return ctx.Err()
+	case result := <-resultCh:
+		if err := ctx.Err(); err != nil {
+			klog.V(log.LogLevelDebug).ErrorS(err, "CRD gate wait context cancelled after result", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+			return err
+		}
+		if result.Err != nil {
+			klog.V(log.LogLevelDefault).ErrorS(result.Err, "CRD gate wait failed", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+			return result.Err
+		}
+
+		klog.V(log.LogLevelDebug).InfoS("CRD gate REST mapping is ready", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+		return nil
+	}
+}
+
+func (in *crdGate) appliedCRD(gvk schema.GroupVersionKind) (unstructured.Unstructured, bool) {
+	in.mu.RLock()
+	defer in.mu.RUnlock()
+
+	crd, ok := in.crds[gvk]
+	return crd, ok && crd.GetUID() != ""
+}
+
+func (in *crdGate) resolve(ctx context.Context, gvk schema.GroupVersionKind, crd unstructured.Unstructured) (any, error) {
+	ctx, cancel := context.WithTimeout(ctx, in.waitTimeout)
+	defer cancel()
+
+	klog.V(log.LogLevelDebug).InfoS("resolving CRD gate REST mapping", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+	var lastErr error
+	err := wait.ExponentialBackoffWithContext(ctx, in.backoff, func(ctx context.Context) (bool, error) {
+		established, err := in.store.IsCRDEstablished(crd)
+		if err != nil {
+			lastErr = err
+			klog.V(log.LogLevelTrace).ErrorS(err, "CRD gate establishment check failed", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+			return false, nil
+		}
+		if !established {
+			lastErr = fmt.Errorf("CRD %s is not established", crd.GetName())
+			klog.V(log.LogLevelTrace).InfoS("CRD gate waiting for CRD establishment", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+			return false, nil
+		}
+
+		in.discoveryCache.ResetRESTMapper()
+		_, err = in.discoveryCache.RestMapping(gvk)
+		if err != nil {
+			lastErr = err
+			klog.V(log.LogLevelTrace).ErrorS(err, "CRD gate REST mapping not ready", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+			return false, nil
+		}
+
+		in.discoveryCache.Add(gvk)
+		klog.V(log.LogLevelDebug).InfoS("resolved CRD gate REST mapping", "crd", crd.GetName(), "gvk", gvk.String(), "uid", crd.GetUID())
+		return true, nil
+	})
+	if err == nil {
+		return nil, nil
+	}
+	if lastErr == nil {
+		lastErr = err
+	}
+
+	return nil, fmt.Errorf("timed out waiting for CRD %s to establish REST mapping for %s after %s: %w: %w",
+		crd.GetName(), gvk.String(), in.waitTimeout, lastErr, err)
+}
diff --git a/go/deployment-operator/pkg/streamline/applier/crd_gate_test.go b/go/deployment-operator/pkg/streamline/applier/crd_gate_test.go
new file mode 100644
index 0000000000..ac0d844202
--- /dev/null
+++ b/go/deployment-operator/pkg/streamline/applier/crd_gate_test.go
@@ -0,0 +1,612 @@
+package applier
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/pluralsh/console/go/client"
+	"github.com/pluralsh/console/go/polly/containers"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/version"
+	"k8s.io/client-go/dynamic"
+
+	discoverycache "github.com/pluralsh/console/go/deployment-operator/pkg/cache/discovery"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline"
+	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline/store"
+)
+
+type fakeDynamicClient struct {
+	resource dynamic.NamespaceableResourceInterface
+}
+
+type failingAppliedStore struct {
+	store.Store
+	err error
+}
+
+func (in *failingAppliedStore) GetAppliedComponent(unstructured.Unstructured) (*smcommon.Component, error) {
+	return nil, in.err
+}
+
+func (in *fakeDynamicClient) Resource(schema.GroupVersionResource) dynamic.NamespaceableResourceInterface {
+	return in.resource
+}
+
+type fakeCRDDiscoveryCache struct {
+	mu               sync.Mutex
+	mapping          *meta.RESTMapping
+	restMappingErr   error
+	restMappingFunc  func(schema.GroupVersionKind) (*meta.RESTMapping, error)
+	restMappingCalls int
+	resetCalls       int
+	addedGVKs        []schema.GroupVersionKind
+}
+
+type doneObservedContext struct {
+	context.Context
+	once     sync.Once
+	observed chan struct{}
+}
+
+func (in *doneObservedContext) Done() <-chan struct{} {
+	in.once.Do(func() { close(in.observed) })
+	return in.Context.Done()
+}
+
+func (in *fakeCRDDiscoveryCache) Add(gvks ...schema.GroupVersionKind) {
+	in.mu.Lock()
+	defer in.mu.Unlock()
+	in.addedGVKs = append(in.addedGVKs, gvks...)
+}
+
+func (in *fakeCRDDiscoveryCache) Delete(...schema.GroupVersionKind)               {}
+func (in *fakeCRDDiscoveryCache) Refresh() error                                  { return nil }
+func (in *fakeCRDDiscoveryCache) MaybeResetRESTMapper(...schema.GroupVersionKind) {}
+
+func (in *fakeCRDDiscoveryCache) ResetRESTMapper() {
+	in.mu.Lock()
+	defer in.mu.Unlock()
+	in.resetCalls++
+}
+
+func (in *fakeCRDDiscoveryCache) GroupVersionKind() containers.Set[schema.GroupVersionKind] {
+	return containers.NewSet[schema.GroupVersionKind]()
+}
+
+func (in *fakeCRDDiscoveryCache) GroupVersionResource() containers.Set[schema.GroupVersionResource] {
+	return containers.NewSet[schema.GroupVersionResource]()
+}
+
+func (in *fakeCRDDiscoveryCache) GroupVersion() containers.Set[schema.GroupVersion] {
+	return containers.NewSet[schema.GroupVersion]()
+}
+
+func (in *fakeCRDDiscoveryCache) ServerVersion() *version.Info { return &version.Info{} }
+
+func (in *fakeCRDDiscoveryCache) KindFor(gvr schema.GroupVersionResource) (schema.GroupVersionKind, error) {
+	return schema.GroupVersionKind{Group: gvr.Group, Version: gvr.Version}, nil
+}
+
+func (in *fakeCRDDiscoveryCache) RestMapping(gvk schema.GroupVersionKind) (*meta.RESTMapping, error) {
+	in.mu.Lock()
+	in.restMappingCalls++
+	fn := in.restMappingFunc
+	mapping, err := in.mapping, in.restMappingErr
+	in.mu.Unlock()
+	if fn != nil {
+		return fn(gvk)
+	}
+	return mapping, err
+}
+
+func (in *fakeCRDDiscoveryCache) OnGroupVersionAdded(discoverycache.GroupVersionUpdateFunc)         {}
+func (in *fakeCRDDiscoveryCache) OnGroupVersionDeleted(discoverycache.GroupVersionUpdateFunc)       {}
+func (in *fakeCRDDiscoveryCache) OnGroupVersionKindAdded(discoverycache.GroupVersionKindUpdateFunc) {}
+func (in *fakeCRDDiscoveryCache) OnGroupVersionKindDeleted(discoverycache.GroupVersionKindUpdateFunc) {
+}
+func (in *fakeCRDDiscoveryCache) OnGroupVersionResourceAdded(discoverycache.GroupVersionResourceUpdateFunc) {
+}
+func (in *fakeCRDDiscoveryCache) OnGroupVersionResourceDeleted(discoverycache.GroupVersionResourceUpdateFunc) {
+}
+
+func TestCRDGateEligibility(t *testing.T) {
+	t.Run("activates only a newly and successfully applied CRD", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		cache := &fakeCRDDiscoveryCache{}
+		crd := testCRD("widgets.example.com", "", "")
+		require.NoError(t, storeInstance.SyncServiceComponents("service", []unstructured.Unstructured{crd}))
+
+		gate := newTestCRDGate(storeInstance, cache, withCRDGateResources([]unstructured.Unstructured{crd}, false))
+		assert.True(t, gate.Enabled())
+
+		_, eligible := gate.appliedCRD(widgetGVK())
+		assert.False(t, eligible)
+		require.NoError(t, runPreApplyGate(gate, context.Background(), widgetGVK()))
+		assert.Equal(t, 0, cache.restMappingCalls)
+
+		applied := testCRD("widgets.example.com", "uid-new", "True")
+		markTestGate(t, gate, applied)
+		actual, eligible := gate.appliedCRD(widgetGVK())
+		assert.True(t, eligible)
+		assert.Equal(t, applied.GetUID(), actual.GetUID())
+	})
+
+	t.Run("does not register an existing CRD", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		cache := &fakeCRDDiscoveryCache{}
+		crd := testCRD("widgets.example.com", "uid-existing", "True")
+		require.NoError(t, storeInstance.SaveComponent(crd))
+		require.NoError(t, storeInstance.SyncServiceComponents("service", []unstructured.Unstructured{crd}))
+
+		gate := newTestCRDGate(storeInstance, cache, withCRDGateResources([]unstructured.Unstructured{crd}, false))
+		assert.False(t, gate.Enabled())
+		markTestGate(t, gate, crd)
+
+		_, eligible := gate.appliedCRD(widgetGVK())
+		assert.False(t, eligible)
+	})
+
+	t.Run("does not register dry run resources", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		crd := testCRD("widgets.example.com", "", "")
+		require.NoError(t, storeInstance.SyncServiceComponents("service", []unstructured.Unstructured{crd}))
+
+		gate := newTestCRDGate(storeInstance, &fakeCRDDiscoveryCache{}, withCRDGateResources([]unstructured.Unstructured{crd}, true))
+		assert.False(t, gate.Enabled())
+		markTestGate(t, gate, testCRD("widgets.example.com", "uid-dry", "True"))
+
+		_, eligible := gate.appliedCRD(widgetGVK())
+		assert.False(t, eligible)
+	})
+
+	t.Run("fails open when candidate classification fails", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		classificationErr := errors.New("database unavailable")
+		crd := testCRD("widgets.example.com", "", "")
+		gate := newTestCRDGate(&failingAppliedStore{Store: storeInstance, err: classificationErr}, &fakeCRDDiscoveryCache{}, withCRDGateResources([]unstructured.Unstructured{crd}, false))
+
+		assert.False(t, gate.Enabled())
+		markTestGate(t, gate, testCRD("widgets.example.com", "uid-new", "True"))
+
+		_, eligible := gate.appliedCRD(widgetGVK())
+		assert.False(t, eligible)
+	})
+}
+
+func TestCRDGateWaitForMapping(t *testing.T) {
+	t.Run("uses persisted establishment and refreshes discovery", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		mapping := widgetMapping()
+		cache := &fakeCRDDiscoveryCache{mapping: mapping}
+		gate, applied := activeTestCRDGate(t, storeInstance, cache)
+		require.NoError(t, storeInstance.SaveComponent(applied))
+
+		require.NoError(t, runPreApplyGate(gate, context.Background(), widgetGVK()))
+		assert.Equal(t, 1, cache.resetCalls)
+		assert.Equal(t, 1, cache.restMappingCalls)
+		assert.Contains(t, cache.addedGVKs, widgetGVK())
+
+		require.NoError(t, runPreApplyGate(gate, context.Background(), widgetGVK()))
+		assert.Equal(t, 2, cache.resetCalls)
+		assert.Equal(t, 2, cache.restMappingCalls)
+	})
+
+	t.Run("waits for watch persistence and delayed discovery", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		var mappingCalls int
+		cache := &fakeCRDDiscoveryCache{restMappingFunc: func(gvk schema.GroupVersionKind) (*meta.RESTMapping, error) {
+			mappingCalls++
+			if mappingCalls == 1 {
+				return nil, &meta.NoKindMatchError{GroupKind: gvk.GroupKind(), SearchedVersions: []string{gvk.Version}}
+			}
+			return widgetMapping(), nil
+		}}
+		gate, applied := activeTestCRDGate(t, storeInstance, cache)
+		pending := applied.DeepCopy()
+		pending.Object["status"] = map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "False"}},
+		}
+		require.NoError(t, storeInstance.SaveComponent(*pending))
+
+		resultCh := make(chan error, 1)
+		go func() {
+			resultCh <- runPreApplyGate(gate, context.Background(), widgetGVK())
+		}()
+		require.NoError(t, storeInstance.SaveComponent(applied))
+
+		require.NoError(t, <-resultCh)
+		assert.Equal(t, 2, cache.resetCalls)
+		assert.Equal(t, 2, cache.restMappingCalls)
+	})
+
+	t.Run("requires the applied UID", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		cache := &fakeCRDDiscoveryCache{mapping: widgetMapping()}
+		gate, _ := activeTestCRDGate(t, storeInstance, cache)
+		require.NoError(t, storeInstance.SaveComponent(testCRD("widgets.example.com", "different-uid", "True")))
+		gate.waitTimeout = 15 * time.Millisecond
+
+		assert.Error(t, runPreApplyGate(gate, context.Background(), widgetGVK()))
+		assert.Equal(t, 0, cache.resetCalls)
+	})
+
+	t.Run("deduplicates concurrent exact GVK waits", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		started := make(chan struct{})
+		release := make(chan struct{})
+		var once sync.Once
+		cache := &fakeCRDDiscoveryCache{restMappingFunc: func(schema.GroupVersionKind) (*meta.RESTMapping, error) {
+			once.Do(func() { close(started) })
+			<-release
+			return widgetMapping(), nil
+		}}
+		gate, applied := activeTestCRDGate(t, storeInstance, cache)
+		require.NoError(t, storeInstance.SaveComponent(applied))
+
+		errCh := make(chan error, 2)
+		go func() {
+			errCh <- runPreApplyGate(gate, context.Background(), widgetGVK())
+		}()
+		<-started
+
+		secondCtx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		observedCtx := &doneObservedContext{Context: secondCtx, observed: make(chan struct{})}
+		go func() {
+			errCh <- runPreApplyGate(gate, observedCtx, widgetGVK())
+		}()
+		<-observedCtx.observed
+		close(release)
+
+		require.NoError(t, <-errCh)
+		require.NoError(t, <-errCh)
+		assert.Equal(t, 1, cache.resetCalls)
+		assert.Equal(t, 1, cache.restMappingCalls)
+	})
+
+	t.Run("resolves different served GVKs independently", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		v1 := widgetGVK()
+		v2 := schema.GroupVersionKind{Group: v1.Group, Version: "v2", Kind: v1.Kind}
+		blocked := make(chan struct{})
+		release := make(chan struct{})
+		cache := &fakeCRDDiscoveryCache{restMappingFunc: func(gvk schema.GroupVersionKind) (*meta.RESTMapping, error) {
+			if gvk == v1 {
+				close(blocked)
+				<-release
+			}
+			return &meta.RESTMapping{
+				Resource:         schema.GroupVersionResource{Group: gvk.Group, Version: gvk.Version, Resource: "widgets"},
+				GroupVersionKind: gvk,
+				Scope:            meta.RESTScopeNamespace,
+			}, nil
+		}}
+		manifest := testCRD("widgets.example.com", "", "")
+		versions, _, _ := unstructured.NestedSlice(manifest.Object, "spec", "versions")
+		versions = append(versions, map[string]any{"name": "v2", "served": true, "storage": false})
+		require.NoError(t, unstructured.SetNestedSlice(manifest.Object, versions, "spec", "versions"))
+		require.NoError(t, storeInstance.SyncServiceComponents("service", []unstructured.Unstructured{manifest}))
+		gate := newTestCRDGate(storeInstance, cache, withCRDGateResources([]unstructured.Unstructured{manifest}, false))
+		applied := manifest.DeepCopy()
+		applied.SetUID("uid-multiversion")
+		applied.Object["status"] = map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+		}
+		markTestGate(t, gate, *applied)
+		require.NoError(t, storeInstance.SaveComponent(*applied))
+
+		v1Result := make(chan error, 1)
+		go func() {
+			v1Result <- runPreApplyGate(gate, context.Background(), v1)
+		}()
+		<-blocked
+
+		require.NoError(t, runPreApplyGate(gate, context.Background(), v2))
+		assert.Contains(t, cache.addedGVKs, v2)
+
+		close(release)
+		require.NoError(t, <-v1Result)
+	})
+
+	t.Run("returns immediately for unrelated GVK", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		cache := &fakeCRDDiscoveryCache{}
+		gate, _ := activeTestCRDGate(t, storeInstance, cache)
+
+		require.NoError(t, runPreApplyGate(gate, context.Background(), schema.GroupVersionKind{Group: "other.io", Version: "v1", Kind: "Other"}))
+		assert.Equal(t, 0, cache.restMappingCalls)
+	})
+
+	t.Run("honors cancellation", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		gate, _ := activeTestCRDGate(t, storeInstance, &fakeCRDDiscoveryCache{})
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		assert.ErrorIs(t, runPreApplyGate(gate, ctx, widgetGVK()), context.Canceled)
+	})
+
+	t.Run("times out while establishment is absent", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		gate, _ := activeTestCRDGate(t, storeInstance, &fakeCRDDiscoveryCache{})
+		gate.waitTimeout = 15 * time.Millisecond
+
+		err := runPreApplyGate(gate, context.Background(), widgetGVK())
+		assert.Error(t, err)
+		assert.True(t, errors.Is(err, context.DeadlineExceeded) || errors.Is(err, wait.ErrWaitTimeout))
+	})
+}
+
+func TestOnApplyCRDGateIsolation(t *testing.T) {
+	noMatch := func(gvk schema.GroupVersionKind) error {
+		return &meta.NoKindMatchError{GroupKind: gvk.GroupKind(), SearchedVersions: []string{gvk.Version}}
+	}
+
+	t.Run("runs matching introduced GVK gate before client lookup", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		streamline.ResetGlobalStore()
+		streamline.InitGlobalStore(storeInstance)
+		t.Cleanup(streamline.ResetGlobalStore)
+
+		cache := &fakeCRDDiscoveryCache{mapping: widgetMapping()}
+		gate, applied := activeTestCRDGate(t, storeInstance, cache)
+		require.NoError(t, storeInstance.SaveComponent(applied))
+		resource := &fakeResourceInterface{applyFn: func(_ context.Context, _ string, obj *unstructured.Unstructured, _ metav1.ApplyOptions, _ ...string) (*unstructured.Unstructured, error) {
+			applied := obj.DeepCopy()
+			applied.SetUID("widget-uid")
+			return applied, nil
+		}}
+		processor := NewWaveProcessor(
+			&fakeDynamicClient{resource: resource},
+			cache,
+			smcommon.SyncPhaseSync,
+			NewWave([]unstructured.Unstructured{testResource(widgetGVK())}, ApplyWave),
+			WithWaveGates(gate),
+			WithWaveMaxConcurrentApplies(1),
+		)
+
+		_, serviceErrors := processor.Run(context.Background())
+
+		assert.Empty(t, serviceErrors)
+		assert.Equal(t, 1, cache.resetCalls)
+		assert.Equal(t, []string{"apply"}, resource.calls)
+	})
+
+	t.Run("runs post-apply gate when client lookup succeeds before pre-apply is eligible", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		streamline.ResetGlobalStore()
+		streamline.InitGlobalStore(storeInstance)
+		t.Cleanup(streamline.ResetGlobalStore)
+
+		crd := testCRD("widgets.example.com", "", "")
+		require.NoError(t, storeInstance.SyncServiceComponents("service", []unstructured.Unstructured{crd}))
+		cache := &fakeCRDDiscoveryCache{mapping: widgetMapping()}
+		gate := newTestCRDGate(storeInstance, cache, withCRDGateResources([]unstructured.Unstructured{crd}, false))
+		resource := &fakeResourceInterface{applyFn: func(_ context.Context, _ string, obj *unstructured.Unstructured, _ metav1.ApplyOptions, _ ...string) (*unstructured.Unstructured, error) {
+			applied := obj.DeepCopy()
+			if obj.GetKind() == "CustomResourceDefinition" {
+				applied.SetUID("crd-uid")
+				applied.Object["status"] = map[string]any{
+					"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+				}
+			}
+			return applied, nil
+		}}
+		processor := NewWaveProcessor(
+			&fakeDynamicClient{resource: resource},
+			cache,
+			smcommon.SyncPhaseSync,
+			NewWave([]unstructured.Unstructured{crd}, ApplyWave),
+			WithWaveGates(gate),
+			WithWaveMaxConcurrentApplies(1),
+		)
+
+		_, serviceErrors := processor.Run(context.Background())
+
+		require.Empty(t, serviceErrors)
+		applied, eligible := gate.appliedCRD(widgetGVK())
+		assert.True(t, eligible)
+		assert.Equal(t, types.UID("crd-uid"), applied.GetUID())
+		assert.Equal(t, []string{"apply"}, resource.calls)
+	})
+
+	t.Run("preserves unrelated no-match", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		streamline.ResetGlobalStore()
+		streamline.InitGlobalStore(storeInstance)
+		t.Cleanup(streamline.ResetGlobalStore)
+
+		other := schema.GroupVersionKind{Group: "other.io", Version: "v1", Kind: "Other"}
+		cache := &fakeCRDDiscoveryCache{restMappingFunc: func(gvk schema.GroupVersionKind) (*meta.RESTMapping, error) {
+			return nil, noMatch(gvk)
+		}}
+		gate, _ := activeTestCRDGate(t, storeInstance, cache)
+		processor := NewWaveProcessor(
+			nil,
+			cache,
+			smcommon.SyncPhaseSync,
+			NewWave([]unstructured.Unstructured{testResource(other)}, ApplyWave),
+			WithWaveGates(gate),
+			WithWaveMaxConcurrentApplies(1),
+		)
+
+		_, serviceErrors := processor.Run(context.Background())
+
+		assert.Len(t, serviceErrors, 1)
+		assert.Equal(t, 0, cache.resetCalls)
+		assert.Equal(t, 1, cache.restMappingCalls)
+	})
+
+	t.Run("preserves non-no-match errors", func(t *testing.T) {
+		storeInstance := newCRDGateTestStore(t)
+		streamline.ResetGlobalStore()
+		streamline.InitGlobalStore(storeInstance)
+		t.Cleanup(streamline.ResetGlobalStore)
+
+		other := schema.GroupVersionKind{Group: "other.io", Version: "v1", Kind: "Other"}
+		mappingErr := errors.New("discovery unavailable")
+		cache := &fakeCRDDiscoveryCache{restMappingErr: mappingErr}
+		gate, _ := activeTestCRDGate(t, storeInstance, cache)
+		processor := NewWaveProcessor(
+			nil,
+			cache,
+			smcommon.SyncPhaseSync,
+			NewWave([]unstructured.Unstructured{testResource(other)}, ApplyWave),
+			WithWaveGates(gate),
+			WithWaveMaxConcurrentApplies(1),
+		)
+
+		_, serviceErrors := processor.Run(context.Background())
+
+		assert.Len(t, serviceErrors, 1)
+		assert.Equal(t, 0, cache.resetCalls)
+		assert.Equal(t, 1, cache.restMappingCalls)
+	})
+}
+
+func TestApplierCreatesCRDAwareResourcesInOneApply(t *testing.T) {
+	storeInstance := newCRDGateTestStore(t)
+	streamline.ResetGlobalStore()
+	streamline.InitGlobalStore(storeInstance)
+	t.Cleanup(streamline.ResetGlobalStore)
+
+	crd := testCRD("widgets.example.com", "", "")
+	widget := testResource(widgetGVK())
+	crdMapping := &meta.RESTMapping{
+		Resource: schema.GroupVersionResource{
+			Group: "apiextensions.k8s.io", Version: "v1", Resource: "customresourcedefinitions",
+		},
+		GroupVersionKind: crd.GroupVersionKind(),
+		Scope:            meta.RESTScopeRoot,
+	}
+	var widgetMappingCalls int
+	cache := &fakeCRDDiscoveryCache{restMappingFunc: func(gvk schema.GroupVersionKind) (*meta.RESTMapping, error) {
+		if gvk == crd.GroupVersionKind() {
+			return crdMapping, nil
+		}
+
+		widgetMappingCalls++
+		if widgetMappingCalls == 1 {
+			return nil, &meta.NoKindMatchError{GroupKind: gvk.GroupKind(), SearchedVersions: []string{gvk.Version}}
+		}
+		return widgetMapping(), nil
+	}}
+	resource := &fakeResourceInterface{applyFn: func(_ context.Context, _ string, obj *unstructured.Unstructured, _ metav1.ApplyOptions, _ ...string) (*unstructured.Unstructured, error) {
+		applied := obj.DeepCopy()
+		if obj.GetKind() == "CustomResourceDefinition" {
+			applied.SetUID("crd-uid")
+			applied.Object["status"] = map[string]any{
+				"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+			}
+			if err := storeInstance.SaveComponent(*applied); err != nil {
+				return nil, err
+			}
+		} else {
+			applied.SetUID("widget-uid")
+		}
+		return applied, nil
+	}}
+	applier := NewApplier(&fakeDynamicClient{resource: resource}, cache, storeInstance, WithWaveDelay(time.Millisecond))
+	dryRun := false
+
+	components, serviceErrors, err := applier.Apply(context.Background(), client.ServiceDeploymentForAgent{
+		ID: "service", Name: "service", Namespace: "default", DryRun: &dryRun,
+	}, []unstructured.Unstructured{crd, widget})
+
+	require.NoError(t, err)
+	assert.Empty(t, serviceErrors)
+	assert.Len(t, components, 2)
+	assert.Equal(t, 2, cache.resetCalls)
+	assert.Equal(t, 3, widgetMappingCalls)
+	assert.Contains(t, cache.addedGVKs, widgetGVK())
+	assert.Equal(t, []string{"apply", "apply"}, resource.calls)
+}
+
+func newCRDGateTestStore(t *testing.T) store.Store {
+	t.Helper()
+	result, err := store.NewDatabaseStore(context.Background())
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, result.Shutdown()) })
+	return result
+}
+
+func newTestCRDGate(store store.Store, cache discoverycache.Cache, opts ...CRDGateOption) *crdGate {
+	gate := newCRDGate(store, cache, opts...)
+	gate.waitTimeout = time.Second
+	gate.backoff = wait.Backoff{Duration: time.Millisecond, Factor: 1, Steps: 100, Cap: time.Millisecond}
+	return gate
+}
+
+func activeTestCRDGate(t *testing.T, storeInstance store.Store, cache discoverycache.Cache) (*crdGate, unstructured.Unstructured) {
+	t.Helper()
+	manifest := testCRD("widgets.example.com", "", "")
+	require.NoError(t, storeInstance.SyncServiceComponents("service", []unstructured.Unstructured{manifest}))
+	gate := newTestCRDGate(storeInstance, cache, withCRDGateResources([]unstructured.Unstructured{manifest}, false))
+	applied := testCRD("widgets.example.com", "uid-new", "True")
+	markTestGate(t, gate, applied)
+	return gate, applied
+}
+
+func markTestGate(t *testing.T, gate *crdGate, resource unstructured.Unstructured) {
+	t.Helper()
+	require.NoError(t, gate.Run(context.Background(), GateConditionPostApply, resource))
+}
+
+func runPreApplyGate(gate *crdGate, ctx context.Context, gvk schema.GroupVersionKind) error {
+	return gate.Run(ctx, GateConditionPreApply, testResource(gvk))
+}
+
+func testCRD(name, uid, established string) unstructured.Unstructured {
+	result := unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "apiextensions.k8s.io/v1",
+		"kind":       "CustomResourceDefinition",
+		"metadata": map[string]any{
+			"name": name,
+		},
+		"spec": map[string]any{
+			"group": "example.com",
+			"names": map[string]any{"kind": "Widget", "plural": "widgets"},
+			"versions": []any{map[string]any{
+				"name": "v1", "served": true, "storage": true,
+			}},
+		},
+	}}
+	result.SetUID(types.UID(uid))
+	if established != "" {
+		result.Object["status"] = map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": established}},
+		}
+	}
+	return result
+}
+
+func widgetGVK() schema.GroupVersionKind {
+	return schema.GroupVersionKind{Group: "example.com", Version: "v1", Kind: "Widget"}
+}
+
+func widgetMapping() *meta.RESTMapping {
+	return &meta.RESTMapping{
+		Resource:         schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"},
+		GroupVersionKind: widgetGVK(),
+		Scope:            meta.RESTScopeNamespace,
+	}
+}
+
+func testResource(gvk schema.GroupVersionKind) unstructured.Unstructured {
+	result := unstructured.Unstructured{}
+	result.SetGroupVersionKind(gvk)
+	result.SetName("example")
+	result.SetNamespace("default")
+	return result
+}
diff --git a/go/deployment-operator/pkg/streamline/applier/gate.go b/go/deployment-operator/pkg/streamline/applier/gate.go
new file mode 100644
index 0000000000..44d97cb62b
--- /dev/null
+++ b/go/deployment-operator/pkg/streamline/applier/gate.go
@@ -0,0 +1,23 @@
+package applier
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+type GateCondition string
+
+const (
+	// GateConditionPreApply runs before resolving the apply client and applying the resource.
+	GateConditionPreApply GateCondition = "pre-apply"
+	// GateConditionPostApply runs after a resource is successfully applied to the cluster.
+	GateConditionPostApply GateCondition = "post-apply"
+)
+
+// Gate adds conditional resource processing to a wave without coupling WaveProcessor to a concrete gate.
+// Implementations must be safe for concurrent Run calls within a wave.
+type Gate interface {
+	Run(ctx context.Context, condition GateCondition, resource unstructured.Unstructured) error
+	Enabled() bool
+}
diff --git a/go/deployment-operator/pkg/streamline/applier/gate_test.go b/go/deployment-operator/pkg/streamline/applier/gate_test.go
new file mode 100644
index 0000000000..37377bd87d
--- /dev/null
+++ b/go/deployment-operator/pkg/streamline/applier/gate_test.go
@@ -0,0 +1,53 @@
+package applier
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+type recordingGate struct {
+	err        error
+	conditions []GateCondition
+}
+
+func (in *recordingGate) Run(_ context.Context, condition GateCondition, _ unstructured.Unstructured) error {
+	in.conditions = append(in.conditions, condition)
+	return in.err
+}
+
+func (in *recordingGate) Enabled() bool {
+	return true
+}
+
+func TestWaveProcessorGates(t *testing.T) {
+	t.Run("runs injected gates in order", func(t *testing.T) {
+		first := &recordingGate{}
+		second := &recordingGate{}
+		processor := NewWaveProcessor(nil, nil, "", Wave{}, WithWaveGates(first), WithWaveGates(second))
+
+		err := processor.runGates(context.Background(), GateConditionPreApply, unstructured.Unstructured{})
+
+		require.NoError(t, err)
+		assert.Equal(t, []GateCondition{GateConditionPreApply}, first.conditions)
+		assert.Equal(t, []GateCondition{GateConditionPreApply}, second.conditions)
+		assert.Len(t, processor.gates, 2)
+	})
+
+	t.Run("stops on gate error", func(t *testing.T) {
+		gateErr := errors.New("gate failed")
+		first := &recordingGate{}
+		second := &recordingGate{err: gateErr}
+		third := &recordingGate{}
+		processor := NewWaveProcessor(nil, nil, "", Wave{}, WithWaveGates(first, second, third))
+
+		err := processor.runGates(context.Background(), GateConditionPostApply, unstructured.Unstructured{})
+
+		assert.ErrorIs(t, err, gateErr)
+		assert.Empty(t, third.conditions)
+	})
+}
diff --git a/go/deployment-operator/pkg/streamline/applier/wave.go b/go/deployment-operator/pkg/streamline/applier/wave.go
index c54e1cfddb..ed901121ae 100644
--- a/go/deployment-operator/pkg/streamline/applier/wave.go
+++ b/go/deployment-operator/pkg/streamline/applier/wave.go
@@ -6,17 +6,6 @@ import (
 	"sync"
 	"time"
 
-	console "github.com/pluralsh/console/go/client"
-	"github.com/pluralsh/console/go/deployment-operator/internal/errors"
-	"github.com/pluralsh/console/go/deployment-operator/internal/helpers"
-	"github.com/pluralsh/console/go/deployment-operator/internal/utils"
-	discoverycache "github.com/pluralsh/console/go/deployment-operator/pkg/cache/discovery"
-	"github.com/pluralsh/console/go/deployment-operator/pkg/common"
-	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
-	"github.com/pluralsh/console/go/deployment-operator/pkg/manifests/template"
-	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline"
-	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
-	"github.com/pluralsh/console/go/polly/cache"
 	"github.com/samber/lo"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -28,6 +17,18 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
+
+	console "github.com/pluralsh/console/go/client"
+	"github.com/pluralsh/console/go/deployment-operator/internal/errors"
+	"github.com/pluralsh/console/go/deployment-operator/internal/helpers"
+	"github.com/pluralsh/console/go/deployment-operator/internal/utils"
+	discoverycache "github.com/pluralsh/console/go/deployment-operator/pkg/cache/discovery"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/common"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/manifests/template"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/streamline"
+	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
+	"github.com/pluralsh/console/go/polly/cache"
 )
 
 type WaveType string
@@ -110,9 +111,13 @@ type WaveProcessor struct {
 	// client is the dynamic client used to apply the resources.
 	client dynamic.Interface
 
-	// discoveryCache is the discovery discoveryCache used to get information about the API resources.
+	// discoveryCache is the discovery cache used to get information about the API resources.
 	discoveryCache discoverycache.Cache
 
+	// gates is a list of gates executed during the wave processing.
+	// Each gate is executed in order and can block the processing of the wave until it is satisfied.
+	gates []Gate
+
 	phase smcommon.SyncPhase
 
 	// wave to be processed. It contains the resources to be applied or deleted.
@@ -282,6 +287,24 @@ func (in *WaveProcessor) clientForResource(resource unstructured.Unstructured) (
 		return nil, err
 	}
 
+	return in.clientForMapping(resource, mapping), nil
+}
+
+func (in *WaveProcessor) runGates(ctx context.Context, condition GateCondition, resource unstructured.Unstructured) error {
+	for _, gate := range in.gates {
+		if gate == nil {
+			continue
+		}
+
+		if err := gate.Run(ctx, condition, resource); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (in *WaveProcessor) clientForMapping(resource unstructured.Unstructured, mapping *meta.RESTMapping) dynamic.ResourceInterface {
 	gvr := schema.GroupVersionResource{
 		Group:    mapping.Resource.Group,
 		Version:  mapping.Resource.Version,
@@ -290,10 +313,10 @@ func (in *WaveProcessor) clientForResource(resource unstructured.Unstructured) (
 
 	if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
 		namespace := lo.Ternary(len(resource.GetNamespace()) == 0, "default", resource.GetNamespace())
-		return in.client.Resource(gvr).Namespace(namespace), nil
+		return in.client.Resource(gvr).Namespace(namespace)
 	}
 
-	return in.client.Resource(gvr), nil
+	return in.client.Resource(gvr)
 }
 
 func (in *WaveProcessor) onDelete(ctx context.Context, resource unstructured.Unstructured) {
@@ -359,7 +382,7 @@ func (in *WaveProcessor) onApply(ctx context.Context, resource unstructured.Unst
 		warning := fmt.Sprintf("resource %s/%s is already managed by another service %s", resource.GetKind(), resource.GetName(), entry.ServiceID)
 		klog.V(log.LogLevelDebug).Info(warning)
 		if !template.IsCRD(&resource) {
-			in.errorsChan <- console.ServiceErrorAttributes{Source: "apply", Message: warning, Warning: lo.ToPtr(true)}
+			in.errorsChan <- console.ServiceErrorAttributes{Source: "apply", Message: warning, Warning: new(true)}
 		}
 
 		resource.SetUID(types.UID(entry.UID))
@@ -367,6 +390,16 @@ func (in *WaveProcessor) onApply(ctx context.Context, resource unstructured.Unst
 		return
 	}
 
+	if in.gatesEnabled() {
+		if gateErr := in.runGates(ctx, GateConditionPreApply, resource); gateErr != nil {
+			in.errorsChan <- console.ServiceErrorAttributes{
+				Source:  in.phase.String(),
+				Message: fmt.Sprintf("pre-apply gate failed for %s/%s: %s", resource.GetNamespace(), resource.GetName(), gateErr.Error()),
+			}
+			return
+		}
+	}
+
 	c, err := in.clientForResource(resource)
 	if err != nil {
 		in.errorsChan <- console.ServiceErrorAttributes{
@@ -395,6 +428,14 @@ func (in *WaveProcessor) onApply(ctx context.Context, resource unstructured.Unst
 	}
 
 	in.waveStatistics.applied++
+	if in.gatesEnabled() {
+		if gateErr := in.runGates(ctx, GateConditionPostApply, lo.FromPtr(appliedResource)); gateErr != nil {
+			in.errorsChan <- console.ServiceErrorAttributes{
+				Source:  in.phase.String(),
+				Message: fmt.Sprintf("post-apply gate failed for %s/%s: %s", resource.GetNamespace(), resource.GetName(), gateErr.Error()),
+			}
+		}
+	}
 
 	if in.onApplyCallback != nil {
 		in.onApplyCallback(resource)
@@ -425,6 +466,10 @@ func (in *WaveProcessor) onApply(ctx context.Context, resource unstructured.Unst
 	in.componentChan <- lo.FromPtr(common.ToComponentAttributes(appliedResource))
 }
 
+func (in *WaveProcessor) gatesEnabled() bool {
+	return !in.dryRun && len(in.gates) > 0
+}
+
 // doReplace uses full PUT instead of SSA, creating the resource if it is missing.
 func (in *WaveProcessor) doReplace(ctx context.Context, c dynamic.ResourceInterface, u unstructured.Unstructured) (*unstructured.Unstructured, error) {
 	dryRunOptions := lo.Ternary(in.dryRun, []string{metav1.DryRunAll}, []string{})
@@ -490,8 +535,8 @@ func (in *WaveProcessor) forceRecreate(ctx context.Context, c dynamic.ResourceIn
 
 	// Force sync by deleting the resource and letting it recreate next.
 	if err := c.Delete(ctx, u.GetName(), metav1.DeleteOptions{
-		GracePeriodSeconds: lo.ToPtr(int64(0)),
-		PropagationPolicy:  lo.ToPtr(metav1.DeletePropagationForeground),
+		GracePeriodSeconds: new(int64(0)),
+		PropagationPolicy:  new(metav1.DeletePropagationForeground),
 	}); err != nil {
 		return nil, err
 	}
@@ -607,6 +652,13 @@ func WithWaveSvcCache(c *cache.Cache[console.ServiceDeploymentForAgent]) WavePro
 	}
 }
 
+// WithWaveGates adds gates to a wave processor in execution order.
+func WithWaveGates(gates ...Gate) WaveProcessorOption {
+	return func(w *WaveProcessor) {
+		w.gates = append(w.gates, gates...)
+	}
+}
+
 func NewWaveProcessor(dynamicClient dynamic.Interface, cache discoverycache.Cache, phase smcommon.SyncPhase, wave Wave, opts ...WaveProcessorOption) *WaveProcessor {
 	result := &WaveProcessor{
 		mu:                   sync.Mutex{},
diff --git a/go/deployment-operator/pkg/streamline/applier/wave_test.go b/go/deployment-operator/pkg/streamline/applier/wave_test.go
index fde055a5ab..5623871d07 100644
--- a/go/deployment-operator/pkg/streamline/applier/wave_test.go
+++ b/go/deployment-operator/pkg/streamline/applier/wave_test.go
@@ -13,6 +13,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/dynamic"
 )
 
 type fakeResourceInterface struct {
@@ -31,6 +32,10 @@ type fakeResourceInterface struct {
 	applyStatusFn      func(context.Context, string, *unstructured.Unstructured, metav1.ApplyOptions) (*unstructured.Unstructured, error)
 }
 
+func (f *fakeResourceInterface) Namespace(string) dynamic.ResourceInterface {
+	return f
+}
+
 func (f *fakeResourceInterface) Create(ctx context.Context, obj *unstructured.Unstructured, options metav1.CreateOptions, subresources ...string) (*unstructured.Unstructured, error) {
 	f.calls = append(f.calls, "create")
 	if f.createFn != nil {
diff --git a/go/deployment-operator/pkg/streamline/store/db_queries.go b/go/deployment-operator/pkg/streamline/store/db_queries.go
index f4b4a3e58b..f8cb490def 100644
--- a/go/deployment-operator/pkg/streamline/store/db_queries.go
+++ b/go/deployment-operator/pkg/streamline/store/db_queries.go
@@ -22,7 +22,8 @@ const (
 			apply_sha TEXT,
 			server_sha TEXT,
 			manifest BOOLEAN DEFAULT 0, -- Indicates if the component was created from an original manifest set of a service
-			applied BOOLEAN DEFAULT 0 -- Indicates if the component was already applied to the cluster
+			applied BOOLEAN DEFAULT 0, -- Indicates if the component was already applied to the cluster
+			established BOOLEAN NOT NULL DEFAULT 0 -- Indicates if a CRD has been established
 		);
 		CREATE UNIQUE INDEX IF NOT EXISTS idx_unique_component ON component("group", version, kind, namespace, name);
 		CREATE INDEX IF NOT EXISTS idx_parent ON component(parent_uid);
@@ -85,6 +86,14 @@ const (
 		WHERE "group" = ? AND version = ? AND kind = ? AND applied = 1
 	`
 
+	getCRDEstablished = `
+		SELECT EXISTS(
+			SELECT 1
+			FROM component
+			WHERE name = ? AND namespace = ? AND "group" = ? AND version = ? AND kind = ? AND uid = ? AND established = 1
+		)
+	`
+
 	setComponentWithSHA = `
 		INSERT INTO component (
 			uid,
@@ -100,7 +109,8 @@ const (
 		    service_id,
 		    delete_phase,
 		    server_sha,
-		    applied
+		    applied,
+		    established
 		) VALUES (
 			?,
 			?,
@@ -115,6 +125,7 @@ const (
 		    ?,
 		    ?,
 		    ?,
+		    ?,
 		    ?
 		) ON CONFLICT("group", version, kind, namespace, name) DO UPDATE SET
 			uid = excluded.uid,
@@ -125,12 +136,14 @@ const (
 			service_id = excluded.service_id,
 			delete_phase = excluded.delete_phase,
 			server_sha = excluded.server_sha,
-		    applied = excluded.applied
+		    applied = excluded.applied,
+		    established = excluded.established
 	`
 
 	setComponentUnsynced = `
 		UPDATE component
 		SET applied = 0,
+		    established = 0,
 		    uid = '',
 		    health = 1,
 		    created_at = NULL,
diff --git a/go/deployment-operator/pkg/streamline/store/db_store.go b/go/deployment-operator/pkg/streamline/store/db_store.go
index 65d0285817..d6103a3d50 100644
--- a/go/deployment-operator/pkg/streamline/store/db_store.go
+++ b/go/deployment-operator/pkg/streamline/store/db_store.go
@@ -276,6 +276,7 @@ func (in *DatabaseStore) SaveComponent(obj unstructured.Unstructured) error {
 			smcommon.GetDeletePhase(obj),
 			serverSHA,
 			true,
+			common.CRDEstablished(obj),
 		},
 	})
 }
@@ -315,7 +316,8 @@ func (in *DatabaseStore) SaveComponents(objects []unstructured.Unstructured) err
 		  service_id,
 		  delete_phase,
 		  server_sha,
-		  applied
+		  applied,
+		  established
 		) VALUES `)
 
 	valueStrings := make([]string, 0, len(objects))
@@ -357,7 +359,7 @@ func (in *DatabaseStore) SaveComponents(objects []unstructured.Unstructured) err
 			continue
 		}
 
-		valueStrings = append(valueStrings, fmt.Sprintf("('%s','%s','%s','%s','%s','%s','%s',%d,'%s',%d,'%s','%s','%s', 1)",
+		valueStrings = append(valueStrings, fmt.Sprintf("('%s','%s','%s','%s','%s','%s','%s',%d,'%s',%d,'%s','%s','%s',1,%d)",
 			obj.GetUID(),
 			lo.FromPtr(ownerRef),
 			gvk.Group,
@@ -371,6 +373,7 @@ func (in *DatabaseStore) SaveComponents(objects []unstructured.Unstructured) err
 			serviceID,
 			smcommon.GetDeletePhase(obj),
 			serverSHA,
+			lo.Ternary(common.CRDEstablished(obj), 1, 0),
 		))
 	}
 
@@ -388,7 +391,8 @@ func (in *DatabaseStore) SaveComponents(objects []unstructured.Unstructured) err
 	  service_id = excluded.service_id,
       delete_phase = excluded.delete_phase,
 	  server_sha = excluded.server_sha,
-	  applied = excluded.applied
+	  applied = excluded.applied,
+	  established = excluded.established
 	`)
 
 	in.maybeSaveHookComponents(conn, objects)
@@ -721,6 +725,32 @@ func (in *DatabaseStore) GetAppliedComponent(obj unstructured.Unstructured) (res
 	return result, err
 }
 
+func (in *DatabaseStore) IsCRDEstablished(obj unstructured.Unstructured) (established bool, err error) {
+	if !common.IsCRD(obj) || obj.GetUID() == "" {
+		return false, nil
+	}
+
+	conn, cancelFunc, err := in.take()
+	if err != nil {
+		return false, err
+	}
+	defer func() {
+		in.pool.Put(conn)
+		cancelFunc()
+	}()
+
+	gvk := obj.GroupVersionKind()
+	err = sqlitex.ExecuteTransient(conn, getCRDEstablished, &sqlitex.ExecOptions{
+		Args: []any{obj.GetName(), obj.GetNamespace(), gvk.Group, gvk.Version, gvk.Kind, obj.GetUID()},
+		ResultFunc: func(stmt *sqlite.Stmt) error {
+			established = stmt.ColumnBool(0)
+			return nil
+		},
+	})
+
+	return established, err
+}
+
 func (in *DatabaseStore) GetAppliedComponentByUID(uid types.UID) (result *client.ComponentChildAttributes, err error) {
 	conn, cancelFunc, err := in.take()
 	if err != nil {
diff --git a/go/deployment-operator/pkg/streamline/store/db_store_profiled.go b/go/deployment-operator/pkg/streamline/store/db_store_profiled.go
index d40753ba1f..7c5ce28554 100644
--- a/go/deployment-operator/pkg/streamline/store/db_store_profiled.go
+++ b/go/deployment-operator/pkg/streamline/store/db_store_profiled.go
@@ -5,12 +5,13 @@ import (
 	"time"
 
 	"github.com/DataDog/dd-trace-go/v2/ddtrace/tracer"
-	"github.com/pluralsh/console/go/polly/containers"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/v2"
 
+	"github.com/pluralsh/console/go/polly/containers"
+
 	"github.com/pluralsh/console/go/client"
 
 	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
@@ -101,6 +102,14 @@ func (p *ProfiledStore) GetAppliedComponent(obj unstructured.Unstructured) (*smc
 	return res, err
 }
 
+func (p *ProfiledStore) IsCRDEstablished(obj unstructured.Unstructured) (established bool, err error) {
+	_ = trace(context.Background(), "IsCRDEstablished", func() error {
+		established, err = p.inner.IsCRDEstablished(obj)
+		return err
+	})
+	return
+}
+
 // GetAppliedComponentByUID wraps Store.GetAppliedComponentByUID with tracing.
 func (p *ProfiledStore) GetAppliedComponentByUID(uid types.UID) (*client.ComponentChildAttributes, error) {
 	var (
@@ -307,7 +316,7 @@ func (p *ProfiledStore) GetComponentInsights() ([]client.ClusterInsightComponent
 func (p *ProfiledStore) SaveComponentAttributes(obj client.ComponentChildAttributes, args ...any) error {
 	var err error
 	_ = trace(context.Background(), "SaveComponentAttributes", func() error {
-		err = p.inner.SaveComponentAttributes(obj, args)
+		err = p.inner.SaveComponentAttributes(obj, args...)
 		return err
 	})
 	return err
diff --git a/go/deployment-operator/pkg/streamline/store/db_store_profiled_local.go b/go/deployment-operator/pkg/streamline/store/db_store_profiled_local.go
index 32c0b0dc62..08d8400c8c 100644
--- a/go/deployment-operator/pkg/streamline/store/db_store_profiled_local.go
+++ b/go/deployment-operator/pkg/streamline/store/db_store_profiled_local.go
@@ -4,14 +4,15 @@ import (
 	"context"
 	"time"
 
-	"github.com/pluralsh/console/go/client"
-	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
-	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
-	"github.com/pluralsh/console/go/polly/containers"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/v2"
+
+	"github.com/pluralsh/console/go/client"
+	"github.com/pluralsh/console/go/deployment-operator/pkg/log"
+	smcommon "github.com/pluralsh/console/go/deployment-operator/pkg/streamline/common"
+	"github.com/pluralsh/console/go/polly/containers"
 )
 
 var storeTimeout = 5 * time.Second
@@ -96,6 +97,14 @@ func (p *ProfiledStoreLocal) GetAppliedComponent(obj unstructured.Unstructured)
 	return res, err
 }
 
+func (p *ProfiledStoreLocal) IsCRDEstablished(obj unstructured.Unstructured) (established bool, err error) {
+	_ = traceLocal(context.Background(), "IsCRDEstablished", func() error {
+		established, err = p.inner.IsCRDEstablished(obj)
+		return err
+	})
+	return
+}
+
 // GetAppliedComponentByUID wraps Store.GetAppliedComponentByUID with tracing.
 func (p *ProfiledStoreLocal) GetAppliedComponentByUID(uid types.UID) (*client.ComponentChildAttributes, error) {
 	var (
@@ -329,7 +338,7 @@ func (p *ProfiledStoreLocal) SetComponentUnsynced(obj unstructured.Unstructured)
 func (p *ProfiledStoreLocal) SaveComponentAttributes(obj client.ComponentChildAttributes, args ...any) error {
 	var err error
 	_ = traceLocal(context.Background(), "SaveComponentAttributes", func() error {
-		err = p.inner.SaveComponentAttributes(obj, args)
+		err = p.inner.SaveComponentAttributes(obj, args...)
 		return err
 	})
 	return err
diff --git a/go/deployment-operator/pkg/streamline/store/db_store_test.go b/go/deployment-operator/pkg/streamline/store/db_store_test.go
index 7eae0af5be..6e993edb03 100644
--- a/go/deployment-operator/pkg/streamline/store/db_store_test.go
+++ b/go/deployment-operator/pkg/streamline/store/db_store_test.go
@@ -123,6 +123,104 @@ func TestComponentCache_Init(t *testing.T) {
 	})
 }
 
+func TestComponentCache_CRDEstablished(t *testing.T) {
+	newStore := func(t *testing.T) store.Store {
+		t.Helper()
+		result, err := store.NewDatabaseStore(context.Background())
+		require.NoError(t, err)
+		t.Cleanup(func() { require.NoError(t, result.Shutdown()) })
+		return result
+	}
+
+	makeCRD := func(name, uid, status string) unstructured.Unstructured {
+		result := createUnstructuredResource("apiextensions.k8s.io", "v1", "CustomResourceDefinition", "", name)
+		result.SetUID(types.UID(uid))
+		if status != "" {
+			result.Object["status"] = map[string]any{
+				"conditions": []any{map[string]any{"type": "Established", "status": status}},
+			}
+		}
+		return result
+	}
+
+	t.Run("tracks watch status transitions and exact UID", func(t *testing.T) {
+		storeInstance := newStore(t)
+		crd := makeCRD("widgets.example.com", "uid-1", "False")
+
+		require.NoError(t, storeInstance.SaveComponent(crd))
+		established, err := storeInstance.IsCRDEstablished(crd)
+		require.NoError(t, err)
+		assert.False(t, established)
+
+		crd.Object["status"] = map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+		}
+		require.NoError(t, storeInstance.SaveComponent(crd))
+		established, err = storeInstance.IsCRDEstablished(crd)
+		require.NoError(t, err)
+		assert.True(t, established)
+
+		stale := crd.DeepCopy()
+		stale.SetUID("uid-2")
+		established, err = storeInstance.IsCRDEstablished(*stale)
+		require.NoError(t, err)
+		assert.False(t, established)
+
+		crd.Object["status"] = map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "Unknown"}},
+		}
+		require.NoError(t, storeInstance.SaveComponent(crd))
+		established, err = storeInstance.IsCRDEstablished(crd)
+		require.NoError(t, err)
+		assert.False(t, established)
+	})
+
+	t.Run("defaults missing status and clears readiness when unsynced", func(t *testing.T) {
+		storeInstance := newStore(t)
+		crd := makeCRD("missing.example.com", "uid-missing", "")
+		require.NoError(t, storeInstance.SaveComponent(crd))
+
+		established, err := storeInstance.IsCRDEstablished(crd)
+		require.NoError(t, err)
+		assert.False(t, established)
+
+		crd.Object["status"] = map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+		}
+		require.NoError(t, storeInstance.SaveComponent(crd))
+		require.NoError(t, storeInstance.SetComponentUnsynced(crd))
+
+		established, err = storeInstance.IsCRDEstablished(crd)
+		require.NoError(t, err)
+		assert.False(t, established)
+	})
+
+	t.Run("persists batch status and preserves it when syncing apply metadata", func(t *testing.T) {
+		storeInstance := newStore(t)
+		crd := makeCRD("gadgets.example.com", "uid-batch", "True")
+
+		require.NoError(t, storeInstance.SaveComponents([]unstructured.Unstructured{crd}))
+		require.NoError(t, storeInstance.SyncAppliedResource(crd))
+
+		established, err := storeInstance.IsCRDEstablished(crd)
+		require.NoError(t, err)
+		assert.True(t, established)
+	})
+
+	t.Run("ignores established-like conditions on non-CRDs", func(t *testing.T) {
+		storeInstance := newStore(t)
+		resource := createComponent("uid-other")
+		resource.Object["status"] = map[string]any{
+			"conditions": []any{map[string]any{"type": "Established", "status": "True"}},
+		}
+
+		require.NoError(t, storeInstance.SaveComponent(resource))
+		established, err := storeInstance.IsCRDEstablished(resource)
+		require.NoError(t, err)
+		assert.False(t, established)
+	})
+}
+
 func TestComponentCache_SetComponent(t *testing.T) {
 	t.Run("cache should save and return simple parent and child structure", func(t *testing.T) {
 		storeInstance, err := store.NewDatabaseStore(context.Background(), store.WithStorage(api.StorageFile))
diff --git a/go/deployment-operator/pkg/streamline/store/store.go b/go/deployment-operator/pkg/streamline/store/store.go
index aedad721cf..aff1db6d99 100644
--- a/go/deployment-operator/pkg/streamline/store/store.go
+++ b/go/deployment-operator/pkg/streamline/store/store.go
@@ -36,6 +36,9 @@ type Store interface {
 
 	GetAppliedComponent(obj unstructured.Unstructured) (*smcommon.Component, error)
 
+	// IsCRDEstablished returns whether the exact CRD object observed by the watch is established.
+	IsCRDEstablished(obj unstructured.Unstructured) (bool, error)
+
 	GetAppliedComponentByUID(uid types.UID) (*client.ComponentChildAttributes, error)
 
 	// GetAppliedComponentsByGVK returns all applied components matching provided GVK.