diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go
index 2272eedbe..081e646e9 100644
--- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go
+++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go
@@ -119,6 +119,13 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error {
 		return err
 	}
 
+	// OR the CLI flag into the YAML value so either source can enable the watcher.
+	prometheusCREnabled, err := getPrometheusCREnabled(flagSet)
+	if err != nil {
+		return err
+	}
+	target.PrometheusCR.Enabled = target.PrometheusCR.Enabled || prometheusCREnabled
+
 	target.HTTPS.Enabled, err = getHttpsEnabled(flagSet)
 	if err != nil {
 		return err
diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go
index caf639ed9..294f9bfe6 100644
--- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go
+++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go
@@ -34,6 +34,7 @@ func getFlagSet(errorHandling pflag.ErrorHandling) *pflag.FlagSet {
 	flagSet := pflag.NewFlagSet(targetAllocatorName, errorHandling)
 	flagSet.String(configFilePathFlagName, DefaultConfigFilePath, "The path to the config file.")
 	flagSet.String(kubeConfigPathFlagName, filepath.Join(homedir.HomeDir(), ".kube", "config"), "absolute path to the KubeconfigPath file")
+	flagSet.Bool(prometheusCREnabledFlagName, false, "Enable watching of Prometheus Operator custom resources (ServiceMonitor/PodMonitor) to dynamically generate scrape configs.")
 	flagSet.Bool(reloadConfigFlagName, false, "Enable automatic configuration reloading. This functionality is deprecated and will be removed in a future release.")
 	flagSet.Bool(httpsEnabledFlagName, true, "Enable HTTPS additional server")
 	flagSet.String(listenAddrHttpsFlagName, ":8443", "The address where this service serves over HTTPS.")
@@ -58,6 +59,10 @@ func getConfigReloadEnabled(flagSet *pflag.FlagSet) (bool, error) {
 	return flagSet.GetBool(reloadConfigFlagName)
 }
 
+func getPrometheusCREnabled(flagSet *pflag.FlagSet) (bool, error) {
+	return flagSet.GetBool(prometheusCREnabledFlagName)
+}
+
 func getHttpsListenAddr(flagSet *pflag.FlagSet) (string, error) {
 	return flagSet.GetString(listenAddrHttpsFlagName)
 }
diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go
index 8beed29d0..66c71a268 100644
--- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go
+++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go
@@ -17,6 +17,7 @@ func TestGetFlagSet(t *testing.T) {
 	// Check if each flag exists
 	assert.NotNil(t, fs.Lookup(configFilePathFlagName), "Flag %s not found", configFilePathFlagName)
 	assert.NotNil(t, fs.Lookup(kubeConfigPathFlagName), "Flag %s not found", kubeConfigPathFlagName)
+	assert.NotNil(t, fs.Lookup(prometheusCREnabledFlagName), "Flag %s not found", prometheusCREnabledFlagName)
 }
 
 func TestFlagGetters(t *testing.T) {
@@ -45,6 +46,18 @@ func TestFlagGetters(t *testing.T) {
 			expectedValue: true,
 			getterFunc:    func(fs *pflag.FlagSet) (interface{}, error) { return getConfigReloadEnabled(fs) },
 		},
+		{
+			name:          "GetPrometheusCREnabled",
+			flagArgs:      []string{"--" + prometheusCREnabledFlagName},
+			expectedValue: true,
+			getterFunc:    func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) },
+		},
+		{
+			name:          "GetPrometheusCREnabledDefault",
+			flagArgs:      []string{},
+			expectedValue: false,
+			getterFunc:    func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) },
+		},
 		{
 			name:        "InvalidFlag",
 			flagArgs:    []string{"--invalid-flag", "value"},
diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go
new file mode 100644
index 000000000..9c432d7ee
--- /dev/null
+++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go
@@ -0,0 +1,36 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestScrapeProtocolsDefaultedOnLoad guards against "scrape_protocols cannot be empty" regression.
+func TestScrapeProtocolsDefaultedOnLoad(t *testing.T) {
+	got := CreateDefaultConfig()
+	err := LoadFromFile("./testdata/scrape_protocols_omitted_test.yaml", &got)
+	require.NoError(t, err)
+
+	require.NotNil(t, got.PromConfig)
+	require.NotEmpty(t, got.PromConfig.ScrapeConfigs)
+
+	for _, sc := range got.PromConfig.ScrapeConfigs {
+		assert.NotEmpty(t, sc.ScrapeProtocols,
+			"scrape_protocols must be defaulted for job %q", sc.JobName)
+	}
+
+	// Spot-check the specific static job from the reproduction by name.
+	found := false
+	for _, sc := range got.PromConfig.ScrapeConfigs {
+		if sc.JobName == "prometheus-sample-app" {
+			found = true
+			assert.Greater(t, len(sc.ScrapeProtocols), 0)
+		}
+	}
+	require.True(t, found, "expected the 'prometheus-sample-app' job to be present")
+}
diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml
new file mode 100644
index 000000000..9405f61e8
--- /dev/null
+++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml
@@ -0,0 +1,14 @@
+# Regression fixture: a scrape job that omits scrape_protocols.
+# The TA config-load path must default scrape_protocols so the agent does not
+# reject the config with "scrape_protocols cannot be empty".
+config:
+  scrape_configs:
+    - job_name: prometheus-sample-app
+      kubernetes_sd_configs:
+        - role: pod
+      relabel_configs:
+        - source_labels: [__meta_kubernetes_pod_label_app]
+          regex: prometheus-sample-app
+          action: keep
+        - target_label: label1
+          replacement: value1
diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go
index 2651e8d29..47be05a3d 100644
--- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go
+++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"log/slog"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -20,16 +21,40 @@ import (
 	promconfig "github.com/prometheus/prometheus/config"
 	kubeDiscovery "github.com/prometheus/prometheus/discovery/kubernetes"
 	"gopkg.in/yaml.v2"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	apiextensionsinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions"
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 
 	allocatorconfig "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/config"
 )
 
+const defaultCollectorNamespace = "amazon-cloudwatch"
+
 const minEventInterval = time.Second * 5
 
+// monitoringResources maps the prometheus-operator resource name to the
+// GroupVersionResource used to build its informer.
+var monitoringResources = map[string]schema.GroupVersionResource{
+	monitoringv1.ServiceMonitorName: monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.ServiceMonitorName),
+	monitoringv1.PodMonitorName:     monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.PodMonitorName),
+}
+
+// crdNameToResource maps a CustomResourceDefinition object name to the
+// prometheus-operator resource key whose informer it backs. The TA watches
+// these CRDs so it can start/stop each informer independently as the CRD
+// appears or disappears, rather than requiring both CRDs to exist at startup.
+var crdNameToResource = map[string]string{
+	monitoringv1.ServiceMonitorName + "." + monitoringv1.SchemeGroupVersion.Group: monitoringv1.ServiceMonitorName,
+	monitoringv1.PodMonitorName + "." + monitoringv1.SchemeGroupVersion.Group:     monitoringv1.PodMonitorName,
+}
+
 func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*PrometheusCRWatcher, error) {
 	mClient, err := monitoringclient.NewForConfig(cfg.ClusterConfig)
 	if err != nil {
@@ -41,19 +66,32 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr
 		return nil, err
 	}
 
-	factory := informers.NewMonitoringInformerFactories(map[string]struct{}{v1.NamespaceAll: {}}, map[string]struct{}{}, mClient, allocatorconfig.DefaultResyncTime, nil) //TODO decide what strategy to use regarding namespaces
-
-	monitoringInformers, err := getInformers(factory)
+	crdClient, err := apiextensionsclient.NewForConfig(cfg.ClusterConfig)
 	if err != nil {
 		return nil, err
 	}
 
 	// TODO: We should make these durations configurable
+	// Namespace must be non-empty; the config generator panics otherwise.
+	collectorNamespace := os.Getenv("OTELCOL_NAMESPACE")
+	if collectorNamespace == "" {
+		if ns, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil && len(ns) > 0 {
+			collectorNamespace = string(ns)
+		} else {
+			collectorNamespace = defaultCollectorNamespace
+		}
+		logger.Info("OTELCOL_NAMESPACE not set, resolved namespace", "namespace", collectorNamespace)
+	}
 	prom := &monitoringv1.Prometheus{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: collectorNamespace,
+		},
 		Spec: monitoringv1.PrometheusSpec{
 			CommonPrometheusFields: monitoringv1.CommonPrometheusFields{
 				ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()),
 			},
+			// Must be non-empty; default to scrape interval.
+			EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()),
 		},
 	}
 
@@ -68,11 +106,17 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr
 
 	podMonSelector := getSelector(cfg.PodMonitorSelector)
 
+	// Informers are NOT built here. Each ServiceMonitor/PodMonitor informer is
+	// started lazily, only once its CRD is observed to exist (see Watch). This
+	// lets the TA start and run healthily whether or not the CRDs are present,
+	// and pick up each type independently when its CRD appears.
 	return &PrometheusCRWatcher{
 		logger:                 logger,
 		kubeMonitoringClient:   mClient,
 		k8sClient:              clientset,
-		informers:              monitoringInformers,
+		crdClient:              crdClient,
+		informers:              map[string]*informers.ForResource{},
+		informerStopChannels:   map[string]chan struct{}{},
 		stopChannel:            make(chan struct{}),
 		eventInterval:          minEventInterval,
 		configGenerator:        generator,
@@ -87,13 +131,19 @@ type PrometheusCRWatcher struct {
 	logger               logr.Logger
 	kubeMonitoringClient monitoringclient.Interface
 	k8sClient            kubernetes.Interface
-	informers            map[string]*informers.ForResource
+	crdClient            apiextensionsclient.Interface
 	eventInterval        time.Duration
 	stopChannel          chan struct{}
 	configGenerator      *prometheus.ConfigGenerator
 	prom                 *monitoringv1.Prometheus
 	kubeConfigPath       string
 
+	// informersMtx guards informers and informerStopChannels, which are mutated
+	// from CRD-watch event handlers (separate goroutines) and read by LoadConfig.
+	informersMtx         sync.RWMutex
+	informers            map[string]*informers.ForResource
+	informerStopChannels map[string]chan struct{}
+
 	serviceMonitorSelector labels.Selector
 	podMonitorSelector     labels.Selector
 }
@@ -105,65 +155,198 @@ func getSelector(s map[string]string) labels.Selector {
 	return labels.SelectorFromSet(s)
 }
 
-// getInformers returns a map of informers for the given resources.
-func getInformers(factory informers.FactoriesForNamespaces) (map[string]*informers.ForResource, error) {
-	serviceMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.ServiceMonitorName))
+// newMonitoringFactory builds a fresh monitoring informer factory. A new factory
+// is created per informer start so that a stopped informer (its CRD was deleted)
+// can be cleanly restarted later if the CRD is recreated — shared informers
+// cannot be restarted once their stop channel is closed.
+func (w *PrometheusCRWatcher) newMonitoringFactory() informers.FactoriesForNamespaces {
+	return informers.NewMonitoringInformerFactories(
+		map[string]struct{}{v1.NamespaceAll: {}},
+		map[string]struct{}{},
+		w.kubeMonitoringClient,
+		allocatorconfig.DefaultResyncTime,
+		nil,
+	) //TODO decide what strategy to use regarding namespaces
+}
+
+// crdExists reports whether the named CustomResourceDefinition is currently
+// registered in the cluster.
+func (w *PrometheusCRWatcher) crdExists(ctx context.Context, crdName string) (bool, error) {
+	_, err := w.crdClient.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, crdName, metav1.GetOptions{})
 	if err != nil {
-		return nil, err
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+	return true, nil
+}
+
+// startMonitorInformer builds and starts the informer for a single monitoring
+// resource (ServiceMonitor or PodMonitor), waits for its cache to sync, wires up
+// the notification handler, and records it for use by LoadConfig. It is
+// idempotent: calling it for an already-running informer is a no-op.
+func (w *PrometheusCRWatcher) startMonitorInformer(resourceName string, notifyEvents chan struct{}) error {
+	w.informersMtx.Lock()
+	defer w.informersMtx.Unlock()
+
+	if _, running := w.informers[resourceName]; running {
+		return nil
+	}
+
+	gvr, ok := monitoringResources[resourceName]
+	if !ok {
+		return fmt.Errorf("unknown monitoring resource %q", resourceName)
 	}
 
-	podMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.PodMonitorName))
+	informer, err := informers.NewInformersForResource(w.newMonitoringFactory(), gvr)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	return map[string]*informers.ForResource{
-		monitoringv1.ServiceMonitorName: serviceMonitorInformers,
-		monitoringv1.PodMonitorName:     podMonitorInformers,
-	}, nil
+	stopCh := make(chan struct{})
+	informer.Start(stopCh)
+	if ok := cache.WaitForNamedCacheSync(resourceName, stopCh, informer.HasSynced); !ok {
+		close(stopCh)
+		return fmt.Errorf("failed to sync %s informer cache", resourceName)
+	}
+
+	informer.AddEventHandler(notifyHandler(notifyEvents))
+
+	w.informers[resourceName] = informer
+	w.informerStopChannels[resourceName] = stopCh
+
+	// A new resource type just became available; trigger a config reload.
+	notify(notifyEvents)
+	return nil
 }
 
-// Watch wrapped informers and wait for an initial sync.
-func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors chan error) error {
-	success := true
-	// this channel needs to be buffered because notifications are asynchronous and neither producers nor consumers wait
-	notifyEvents := make(chan struct{}, 1)
+// stopMonitorInformer stops the informer for a single monitoring resource (its
+// CRD was removed) and drops it from the active set so its targets are no longer
+// generated. It is idempotent.
+func (w *PrometheusCRWatcher) stopMonitorInformer(resourceName string, notifyEvents chan struct{}) {
+	w.informersMtx.Lock()
+	defer w.informersMtx.Unlock()
+
+	stopCh, running := w.informerStopChannels[resourceName]
+	if !running {
+		return
+	}
+	close(stopCh)
+	delete(w.informerStopChannels, resourceName)
+	delete(w.informers, resourceName)
 
-	for name, resource := range w.informers {
-		resource.Start(w.stopChannel)
+	// The type went away; trigger a config reload so its targets are dropped.
+	notify(notifyEvents)
+}
 
-		if ok := cache.WaitForNamedCacheSync(name, w.stopChannel, resource.HasSynced); !ok {
-			success = false
+// crdObjectName extracts the CustomResourceDefinition name from an informer
+// event object, tolerating the tombstone wrapper delivered on some deletes.
+func crdObjectName(obj interface{}) (string, bool) {
+	switch t := obj.(type) {
+	case *apiextensionsv1.CustomResourceDefinition:
+		return t.Name, true
+	case cache.DeletedFinalStateUnknown:
+		if crd, ok := t.Obj.(*apiextensionsv1.CustomResourceDefinition); ok {
+			return crd.Name, true
 		}
+	}
+	return "", false
+}
 
-		// only send an event notification if there isn't one already
-		resource.AddEventHandler(cache.ResourceEventHandlerFuncs{
-			// these functions only write to the notification channel if it's empty to avoid blocking
-			// if scrape config updates are being rate-limited
-			AddFunc: func(obj interface{}) {
-				select {
-				case notifyEvents <- struct{}{}:
-				default:
-				}
-			},
-			UpdateFunc: func(oldObj, newObj interface{}) {
-				select {
-				case notifyEvents <- struct{}{}:
-				default:
-				}
-			},
-			DeleteFunc: func(obj interface{}) {
-				select {
-				case notifyEvents <- struct{}{}:
-				default:
-				}
-			},
-		})
+// watchCRDs starts a CustomResourceDefinition informer that starts/stops the
+// per-type monitoring informers as the ServiceMonitor/PodMonitor CRDs are
+// created or deleted at runtime — no TA restart required. The informer replays
+// existing CRDs on its initial sync, so CRDs present at startup are handled here
+// too (startMonitorInformer is idempotent).
+func (w *PrometheusCRWatcher) watchCRDs(notifyEvents chan struct{}) {
+	factory := apiextensionsinformers.NewSharedInformerFactory(w.crdClient, allocatorconfig.DefaultResyncTime)
+	crdInformer := factory.Apiextensions().V1().CustomResourceDefinitions().Informer()
+	_, _ = crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			name, ok := crdObjectName(obj)
+			if !ok {
+				return
+			}
+			resourceName, tracked := crdNameToResource[name]
+			if !tracked {
+				return
+			}
+			if err := w.startMonitorInformer(resourceName, notifyEvents); err != nil {
+				w.logger.Error(err, "prometheus-cr: failed to start informer after CRD became available", "crd", name)
+				return
+			}
+			w.logger.Info("prometheus-cr: CRD available, started informer", "crd", name, "resource", resourceName)
+		},
+		DeleteFunc: func(obj interface{}) {
+			name, ok := crdObjectName(obj)
+			if !ok {
+				return
+			}
+			resourceName, tracked := crdNameToResource[name]
+			if !tracked {
+				return
+			}
+			w.stopMonitorInformer(resourceName, notifyEvents)
+			w.logger.Info("prometheus-cr: CRD removed, stopped informer and dropped targets", "crd", name, "resource", resourceName)
+		},
+	})
+	factory.Start(w.stopChannel)
+}
+
+// notify performs a non-blocking send on the buffered notification channel.
+func notify(notifyEvents chan struct{}) {
+	select {
+	case notifyEvents <- struct{}{}:
+	default:
+	}
+}
+
+// notifyHandler returns event handlers that coalesce ServiceMonitor/PodMonitor
+// changes into the notification channel (non-blocking so rate-limiting upstream
+// is never starved).
+func notifyHandler(notifyEvents chan struct{}) cache.ResourceEventHandlerFuncs {
+	return cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) { notify(notifyEvents) },
+		UpdateFunc: func(oldObj, newObj interface{}) { notify(notifyEvents) },
+		DeleteFunc: func(obj interface{}) { notify(notifyEvents) },
 	}
-	if !success {
-		return fmt.Errorf("failed to sync cache")
+}
+
+// Watch starts watching for ServiceMonitor/PodMonitor changes. It is resilient
+// to either CRD being absent: it starts the informer for each CRD that exists
+// now, defers the rest until their CRDs appear (via a CustomResourceDefinition
+// watch), and never fails startup just because a CRD is missing. This means the
+// TA starts and stays healthy regardless of CRD install ordering, and begins
+// (or stops) watching each type automatically as its CRD is created or deleted.
+func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors chan error) error {
+	// this channel needs to be buffered because notifications are asynchronous and neither producers nor consumers wait
+	notifyEvents := make(chan struct{}, 1)
+
+	// Start informers for CRDs that already exist. Absent CRDs are simply
+	// skipped here; the CRD watch below starts them if/when they appear.
+	for crdName, resourceName := range crdNameToResource {
+		exists, err := w.crdExists(context.Background(), crdName)
+		if err != nil {
+			// Surface the error but keep going — a transient API error must not
+			// take down the allocator. The CRD watch will recover the informer.
+			w.logger.Error(err, "prometheus-cr: failed to check for CRD, deferring to CRD watch", "crd", crdName)
+			continue
+		}
+		if !exists {
+			w.logger.Info("prometheus-cr: CRD not present at startup, deferring informer until it is created", "crd", crdName)
+			continue
+		}
+		if startErr := w.startMonitorInformer(resourceName, notifyEvents); startErr != nil {
+			w.logger.Error(startErr, "prometheus-cr: failed to start informer for present CRD, deferring to CRD watch", "crd", crdName)
+			continue
+		}
+		w.logger.Info("prometheus-cr: started informer for present CRD", "crd", crdName, "resource", resourceName)
 	}
 
+	// React to CRDs being created/deleted at runtime with no restart.
+	w.watchCRDs(notifyEvents)
+
 	// limit the rate of outgoing events
 	w.rateLimitedEventSender(upstreamEvents, notifyEvents)
 
@@ -204,32 +387,54 @@ func (w *PrometheusCRWatcher) rateLimitedEventSender(upstreamEvents chan Event,
 }
 
 func (w *PrometheusCRWatcher) Close() error {
+	// Stop any per-type monitoring informers (each has its own stop channel).
+	w.informersMtx.Lock()
+	for name, stopCh := range w.informerStopChannels {
+		close(stopCh)
+		delete(w.informerStopChannels, name)
+		delete(w.informers, name)
+	}
+	w.informersMtx.Unlock()
+
+	// Stop the CRD watch and release Watch's blocking read.
 	close(w.stopChannel)
 	return nil
 }
 
 func (w *PrometheusCRWatcher) LoadConfig(ctx context.Context) (*promconfig.Config, error) {
+	// Snapshot the currently-running informers. Either may be absent if its CRD
+	// is not (yet) installed; in that case its monitors are simply skipped and
+	// no scrape jobs are generated for that type.
+	w.informersMtx.RLock()
+	smInformer := w.informers[monitoringv1.ServiceMonitorName]
+	pmInformer := w.informers[monitoringv1.PodMonitorName]
+	w.informersMtx.RUnlock()
+
 	store := assets.NewStoreBuilder(w.k8sClient.CoreV1(), w.k8sClient.CoreV1())
 	serviceMonitorInstances := make(map[string]*monitoringv1.ServiceMonitor)
-	smRetrieveErr := w.informers[monitoringv1.ServiceMonitorName].ListAll(w.serviceMonitorSelector, func(sm interface{}) {
-		monitor := sm.(*monitoringv1.ServiceMonitor)
-		key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
-		w.addStoreAssetsForServiceMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.Endpoints, store)
-		serviceMonitorInstances[key] = monitor
-	})
-	if smRetrieveErr != nil {
-		return nil, smRetrieveErr
+	if smInformer != nil {
+		smRetrieveErr := smInformer.ListAll(w.serviceMonitorSelector, func(sm interface{}) {
+			monitor := sm.(*monitoringv1.ServiceMonitor)
+			key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
+			w.addStoreAssetsForServiceMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.Endpoints, store)
+			serviceMonitorInstances[key] = monitor
+		})
+		if smRetrieveErr != nil {
+			return nil, smRetrieveErr
+		}
 	}
 
 	podMonitorInstances := make(map[string]*monitoringv1.PodMonitor)
-	pmRetrieveErr := w.informers[monitoringv1.PodMonitorName].ListAll(w.podMonitorSelector, func(pm interface{}) {
-		monitor := pm.(*monitoringv1.PodMonitor)
-		key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
-		w.addStoreAssetsForPodMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.PodMetricsEndpoints, store)
-		podMonitorInstances[key] = monitor
-	})
-	if pmRetrieveErr != nil {
-		return nil, pmRetrieveErr
+	if pmInformer != nil {
+		pmRetrieveErr := pmInformer.ListAll(w.podMonitorSelector, func(pm interface{}) {
+			monitor := pm.(*monitoringv1.PodMonitor)
+			key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
+			w.addStoreAssetsForPodMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.PodMetricsEndpoints, store)
+			podMonitorInstances[key] = monitor
+		})
+		if pmRetrieveErr != nil {
+			return nil, pmRetrieveErr
+		}
 	}
 
 	generatedConfig, err := w.configGenerator.GenerateServerConfiguration(
diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go
index 7ad07f717..662800508 100644
--- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go
+++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go
@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-logr/logr"
 	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	fakemonitoringclient "github.com/prometheus-operator/prometheus-operator/pkg/client/versioned/fake"
 	"github.com/prometheus-operator/prometheus-operator/pkg/informers"
@@ -21,7 +22,10 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apiextensionsfake "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/ptr"
@@ -252,17 +256,12 @@ func TestLoadConfig(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			w := getTestPrometheusCRWatcher(t, tt.serviceMonitor, tt.podMonitor)
-			for _, informer := range w.informers {
-				// Start informers in order to populate cache.
-				informer.Start(w.stopChannel)
-			}
-
-			// Wait for informers to sync.
-			for _, informer := range w.informers {
-				for !informer.HasSynced() {
-					time.Sleep(50 * time.Millisecond)
-				}
-			}
+			defer func() { _ = w.Close() }()
+
+			// Start both informers via the per-type lifecycle (both CRDs present).
+			notifyEvents := make(chan struct{}, 1)
+			require.NoError(t, w.startMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents))
+			require.NoError(t, w.startMonitorInformer(monitoringv1.PodMonitorName, notifyEvents))
 
 			got, err := w.LoadConfig(context.Background())
 			require.NoError(t, err)
@@ -305,11 +304,15 @@ func TestRateLimit(t *testing.T) {
 	_, err = w.kubeMonitoringClient.MonitoringV1().ServiceMonitors("test").Create(context.Background(), serviceMonitor, metav1.CreateOptions{})
 	require.NoError(t, err)
 
-	// wait for cache sync first
-	for _, informer := range w.informers {
-		success := cache.WaitForCacheSync(w.stopChannel, informer.HasSynced)
-		require.True(t, success)
-	}
+	// wait for the watcher to start the ServiceMonitor informer and finish its
+	// startup loop (both CRDs present => both informers running) so the
+	// rate-limited event sender is active before we measure event timing.
+	require.Eventually(t, func() bool {
+		w.informersMtx.RLock()
+		defer w.informersMtx.RUnlock()
+		sm, ok := w.informers[monitoringv1.ServiceMonitorName]
+		return ok && sm.HasSynced() && len(w.informers) == 2
+	}, 5*time.Second, 10*time.Millisecond)
 
 	require.Eventually(t, func() bool {
 		_, createErr := w.kubeMonitoringClient.MonitoringV1().ServiceMonitors("test").Update(context.Background(), serviceMonitor, metav1.UpdateOptions{})
@@ -353,9 +356,26 @@ func TestRateLimit(t *testing.T) {
 
 }
 
-// getTestPrometheuCRWatcher creates a test instance of PrometheusCRWatcher with fake clients
-// and test secrets.
+// getTestPrometheusCRWatcher creates a test instance of PrometheusCRWatcher with
+// fake clients and test secrets. Both the ServiceMonitor and PodMonitor CRDs are
+// registered in the fake apiextensions client, so the watcher behaves as if both
+// CRDs are present. Use getTestPrometheusCRWatcherWithCRDs to control CRD presence.
 func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, pm *monitoringv1.PodMonitor) *PrometheusCRWatcher {
+	return getTestPrometheusCRWatcherWithCRDs(t, sm, pm, true, true)
+}
+
+// crdFor builds a minimal CustomResourceDefinition object for the given
+// monitoring resource (e.g. "servicemonitors"), matching the names the watcher
+// keys on.
+func crdFor(resourceName string) *apiextensionsv1.CustomResourceDefinition {
+	return &apiextensionsv1.CustomResourceDefinition{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: resourceName + "." + monitoringv1.SchemeGroupVersion.Group,
+		},
+	}
+}
+
+func getTestPrometheusCRWatcherWithCRDs(t *testing.T, sm *monitoringv1.ServiceMonitor, pm *monitoringv1.PodMonitor, smCRD, pmCRD bool) *PrometheusCRWatcher {
 	mClient := fakemonitoringclient.NewSimpleClientset() //nolint:staticcheck // NewClientset causes structured merge diff schema errors in tests
 	if sm != nil {
 		_, err := mClient.MonitoringV1().ServiceMonitors("test").Create(context.Background(), sm, metav1.CreateOptions{})
@@ -370,6 +390,15 @@ func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, p
 		}
 	}
 
+	var crdObjects []runtime.Object
+	if smCRD {
+		crdObjects = append(crdObjects, crdFor(monitoringv1.ServiceMonitorName))
+	}
+	if pmCRD {
+		crdObjects = append(crdObjects, crdFor(monitoringv1.PodMonitorName))
+	}
+	crdClient := apiextensionsfake.NewSimpleClientset(crdObjects...)
+
 	k8sClient := fake.NewSimpleClientset()
 	_, err := k8sClient.CoreV1().Secrets("test").Create(context.Background(), &v1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
@@ -392,12 +421,6 @@ func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, p
 		t.Fatal(t, err)
 	}
 
-	factory := informers.NewMonitoringInformerFactories(map[string]struct{}{v1.NamespaceAll: {}}, map[string]struct{}{}, mClient, 0, nil)
-	informers, err := getInformers(factory)
-	if err != nil {
-		t.Fatal(t, err)
-	}
-
 	prom := &monitoringv1.Prometheus{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test",
@@ -417,9 +440,12 @@ func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, p
 	}
 
 	return &PrometheusCRWatcher{
+		logger:                 logr.Discard(),
 		kubeMonitoringClient:   mClient,
 		k8sClient:              k8sClient,
-		informers:              informers,
+		crdClient:              crdClient,
+		informers:              map[string]*informers.ForResource{},
+		informerStopChannels:   map[string]chan struct{}{},
 		configGenerator:        generator,
 		prom:                   prom,
 		serviceMonitorSelector: getSelector(nil),
@@ -445,3 +471,139 @@ func sanitizeScrapeConfigsForTest(scs []*promconfig.ScrapeConfig) {
 		sc.ExtraScrapeMetrics = nil
 	}
 }
+
+// runningInformers returns the set of monitoring resource names whose informers
+// are currently active.
+func runningInformers(w *PrometheusCRWatcher) map[string]bool {
+	w.informersMtx.RLock()
+	defer w.informersMtx.RUnlock()
+	out := map[string]bool{}
+	for name := range w.informers {
+		out[name] = true
+	}
+	return out
+}
+
+// TestWatchStartsHealthyWithoutCRDs verifies the TA does not error when neither
+// the ServiceMonitor nor PodMonitor CRD exists: Watch returns no error, no
+// informers are started, and LoadConfig yields a config with no scrape jobs.
+func TestWatchStartsHealthyWithoutCRDs(t *testing.T) {
+	w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false)
+	w.eventInterval = 5 * time.Millisecond
+	defer func() { _ = w.Close() }()
+
+	watchDone := make(chan error, 1)
+	go func() { watchDone <- w.Watch(make(chan Event, 1), make(chan error, 1)) }()
+
+	// Give the CRD watch time to sync and (not) start anything.
+	require.Never(t, func() bool {
+		return len(runningInformers(w)) > 0
+	}, 200*time.Millisecond, 20*time.Millisecond)
+
+	cfg, err := w.LoadConfig(context.Background())
+	require.NoError(t, err)
+	assert.Empty(t, cfg.ScrapeConfigs)
+
+	// Watch must still be running (resilient), not have returned an error.
+	select {
+	case err := <-watchDone:
+		t.Fatalf("Watch exited unexpectedly with: %v", err)
+	default:
+	}
+}
+
+// TestWatchPerTypeIndependence verifies that with only the ServiceMonitor CRD
+// present, the SM informer starts and the PM informer does not.
+func TestWatchPerTypeIndependence(t *testing.T) {
+	w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, false)
+	w.eventInterval = 5 * time.Millisecond
+	defer func() { _ = w.Close() }()
+
+	go func() { _ = w.Watch(make(chan Event, 1), make(chan error, 1)) }()
+
+	require.Eventually(t, func() bool {
+		return runningInformers(w)[monitoringv1.ServiceMonitorName]
+	}, 5*time.Second, 10*time.Millisecond)
+
+	// PodMonitor CRD is absent, so its informer must never start.
+	require.Never(t, func() bool {
+		return runningInformers(w)[monitoringv1.PodMonitorName]
+	}, 200*time.Millisecond, 20*time.Millisecond)
+}
+
+// TestWatchStartsInformerWhenCRDCreated verifies the TA begins watching a type
+// when its CRD is created at runtime — no restart required.
+func TestWatchStartsInformerWhenCRDCreated(t *testing.T) {
+	w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false)
+	w.eventInterval = 5 * time.Millisecond
+	defer func() { _ = w.Close() }()
+
+	go func() { _ = w.Watch(make(chan Event, 1), make(chan error, 1)) }()
+
+	// Nothing running initially.
+	require.Never(t, func() bool {
+		return len(runningInformers(w)) > 0
+	}, 200*time.Millisecond, 20*time.Millisecond)
+
+	// Create the ServiceMonitor CRD after startup.
+	_, err := w.crdClient.ApiextensionsV1().CustomResourceDefinitions().Create(
+		context.Background(), crdFor(monitoringv1.ServiceMonitorName), metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return runningInformers(w)[monitoringv1.ServiceMonitorName]
+	}, 5*time.Second, 10*time.Millisecond)
+}
+
+// TestStopMonitorInformerDropsType verifies the stop path used when a CRD is
+// removed at runtime: the informer is stopped, dropped from the active set (so
+// LoadConfig no longer emits its targets), and a reload is signalled. This is
+// the logic invoked by the CRD watch's DeleteFunc; it is exercised directly so
+// the assertion does not depend on fake-clientset delete-watch delivery.
+func TestStopMonitorInformerDropsType(t *testing.T) {
+	w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, false)
+	defer func() { _ = w.Close() }()
+
+	notifyEvents := make(chan struct{}, 1)
+	require.NoError(t, w.startMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents))
+	require.True(t, runningInformers(w)[monitoringv1.ServiceMonitorName])
+	// drain the start notification
+	select {
+	case <-notifyEvents:
+	default:
+	}
+
+	w.stopMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents)
+
+	assert.False(t, runningInformers(w)[monitoringv1.ServiceMonitorName], "informer should be stopped and dropped")
+	// a reload must be signalled so the dropped type's targets are removed
+	select {
+	case <-notifyEvents:
+	default:
+		t.Fatal("expected a reload notification after stopping the informer")
+	}
+
+	// LoadConfig must succeed and emit no scrape jobs once the type is dropped.
+	cfg, err := w.LoadConfig(context.Background())
+	require.NoError(t, err)
+	assert.Empty(t, cfg.ScrapeConfigs)
+}
+
+// TestCRDObjectName verifies the CRD-name extraction used by the CRD watch,
+// including the delete tombstone wrapper.
+func TestCRDObjectName(t *testing.T) {
+	crd := crdFor(monitoringv1.ServiceMonitorName)
+	name, ok := crdObjectName(crd)
+	require.True(t, ok)
+	assert.Equal(t, "servicemonitors.monitoring.coreos.com", name)
+	if _, tracked := crdNameToResource[name]; !tracked {
+		t.Fatalf("CRD name %q is not mapped to a monitoring resource", name)
+	}
+
+	name, ok = crdObjectName(cache.DeletedFinalStateUnknown{Key: "k", Obj: crd})
+	require.True(t, ok)
+	assert.Equal(t, "servicemonitors.monitoring.coreos.com", name)
+
+	_, ok = crdObjectName("not-a-crd")
+	assert.False(t, ok)
+}
diff --git a/internal/manifests/collector/annotations.go b/internal/manifests/collector/annotations.go
index a665e465e..364072330 100644
--- a/internal/manifests/collector/annotations.go
+++ b/internal/manifests/collector/annotations.go
@@ -6,6 +6,7 @@ package collector
 import (
 	"crypto/sha256"
 	"fmt"
+	"log/slog"
 
 	"github.com/aws/amazon-cloudwatch-agent-operator/apis/v1alpha1"
 )
@@ -28,7 +29,7 @@ func Annotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string {
 	}
 
 	// make sure sha256 for configMap is always calculated
-	annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config)
+	annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance))
 
 	return annotations
 }
@@ -51,11 +52,27 @@ func PodAnnotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string {
 	}
 
 	// make sure sha256 for configMap is always calculated
-	podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config)
+	podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance))
 
 	return podAnnotations
 }
 
+// configHashInput returns the combined config string used for the pod-template restart hash.
+func configHashInput(instance v1alpha1.AmazonCloudWatchAgent) string {
+	config := instance.Spec.Config
+	if !instance.Spec.Prometheus.IsEmpty() {
+		promYaml, err := instance.Spec.Prometheus.Yaml()
+		if err != nil {
+			// Static sentinel; Yaml() over map[string]interface{} rarely fails in practice.
+			slog.Warn("failed to serialize Spec.Prometheus for config hash", "error", err)
+			config += "\x00prometheus-serialize-error"
+		} else {
+			config += "\x00" + promYaml // null byte prevents collision between config suffix and promYAML prefix
+		}
+	}
+	return config
+}
+
 func getConfigMapSHA(config string) string {
 	h := sha256.Sum256([]byte(config))
 	return fmt.Sprintf("%x", h)
diff --git a/internal/manifests/collector/annotations_test.go b/internal/manifests/collector/annotations_test.go
index e02ef1678..1ab8e3bd7 100644
--- a/internal/manifests/collector/annotations_test.go
+++ b/internal/manifests/collector/annotations_test.go
@@ -79,3 +79,64 @@ func TestAnnotationsPropagateDown(t *testing.T) {
 	assert.Equal(t, "mycomponent", podAnnotations["myapp"])
 	assert.Equal(t, "pod_annotation_value", podAnnotations["pod_annotation"])
 }
+
+func promConfig(t *testing.T, replacement string) v1alpha1.PrometheusConfig {
+	t.Helper()
+	cfg := map[string]interface{}{
+		"scrape_configs": []interface{}{
+			map[string]interface{}{
+				"job_name": "kubernetes-pods-annotated",
+				"relabel_configs": []interface{}{
+					map[string]interface{}{
+						"target_label": "bug2probe",
+						"replacement":  replacement,
+					},
+				},
+			},
+		},
+	}
+	return v1alpha1.PrometheusConfig{
+		Config: &v1alpha1.AnyConfig{Object: cfg},
+	}
+}
+
+// TestPrometheusConfigChangeBumpsHash verifies a Prometheus-only change bumps the pod-template hash.
+func TestPrometheusConfigChangeBumpsHash(t *testing.T) {
+	base := v1alpha1.AmazonCloudWatchAgent{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"},
+		Spec: v1alpha1.AmazonCloudWatchAgentSpec{
+			Config:     "agent-config",
+			Prometheus: promConfig(t, "value2"),
+		},
+	}
+
+	// same spec twice -> stable hash
+	h1 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"]
+	h2 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"]
+	assert.Equal(t, h1, h2, "hash must be stable when nothing changes")
+
+	// change ONLY the prometheus config -> hash must change
+	changed := base
+	changed.Spec.Prometheus = promConfig(t, "value3")
+	h3 := PodAnnotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"]
+	assert.NotEqual(t, h1, h3, "pod annotation hash must change when only Spec.Prometheus changes")
+
+	// metadata annotations hash must also reflect the prometheus change
+	a1 := Annotations(base)["amazon-cloudwatch-agent-operator-config/sha256"]
+	a3 := Annotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"]
+	assert.NotEqual(t, a1, a3, "metadata annotation hash must change when only Spec.Prometheus changes")
+}
+
+// TestEmptyPrometheusHashUnchanged verifies non-Prometheus agents keep a stable hash.
+func TestEmptyPrometheusHashUnchanged(t *testing.T) {
+	otelcol := v1alpha1.AmazonCloudWatchAgent{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"},
+		Spec:       v1alpha1.AmazonCloudWatchAgentSpec{Config: "test"},
+	}
+
+	// sha256("test") == 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08
+	assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
+		Annotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"])
+	assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
+		PodAnnotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"])
+}