diff --git a/Cargo.lock b/Cargo.lock
index 02573c11a..db3144c69 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5215,9 +5215,9 @@ dependencies = [
 
 [[package]]
 name = "yoke-derive"
-version = "0.8.3"
+version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "33811428bee40dbceb6d545e95754741d17a6aef9a4849f0fd62e2ba4f412a78"
+checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71"
 dependencies = [
  "proc-macro2",
  "quote",
diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md
index ee60b2114..c1249063c 100644
--- a/crates/stackable-operator/CHANGELOG.md
+++ b/crates/stackable-operator/CHANGELOG.md
@@ -4,6 +4,18 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Add utility for product agents ([#1290]):
+  - `TlsClientCredential`, enum for TLS-based platform access used for adding volumes and mounts.
+  - `Anonymous`, for platform access to products that don't authenticate the agent.
+  - `ManagementPolicy`, whether an agent adopts existing resources and deletes resources in the product.
+  - `Ready`/`Synced` conditions of resources managed by agents.
+  - Annotation marking resources owned by an agent.
+  - Labels and selectors for agents.
+
+[#1290]: https://github.com/stackabletech/operator-rs/pull/1290
+
 ## [0.119.0] - 2026-09-23
 
 ### Removed
diff --git a/crates/stackable-operator/src/commons/mod.rs b/crates/stackable-operator/src/commons/mod.rs
index cc4ab30e3..8da787e4d 100644
--- a/crates/stackable-operator/src/commons/mod.rs
+++ b/crates/stackable-operator/src/commons/mod.rs
@@ -6,6 +6,7 @@ pub mod cluster_operation;
 pub mod networking;
 pub mod opa;
 pub mod pdb;
+pub mod platform_access;
 pub mod product_image_selection;
 pub mod random_secret_creation;
 pub mod rbac;
diff --git a/crates/stackable-operator/src/commons/platform_access/condition.rs b/crates/stackable-operator/src/commons/platform_access/condition.rs
new file mode 100644
index 000000000..5c32c2e42
--- /dev/null
+++ b/crates/stackable-operator/src/commons/platform_access/condition.rs
@@ -0,0 +1,144 @@
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
+use strum::Display;
+
+/// Condition types of resources an agent manages inside a product.
+#[derive(Clone, Copy, Debug, Display, Eq, PartialEq)]
+pub enum ManagedConditionType {
+    /// The resource exists in the product and is usable.
+    Ready,
+
+    /// The last reconciliation applied the spec to the resource in the product.
+    Synced,
+}
+
+/// The desired state of one condition, without the bookkeeping fields.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedCondition {
+    pub type_: ManagedConditionType,
+
+    pub status: bool,
+
+    /// A CamelCase reason like `Created` or `UnsupportedChange`.
+    pub reason: String,
+
+    pub message: String,
+}
+
+/// Merges the given conditions into the current ones.
+///
+/// `lastTransitionTime` only changes if the status of a condition changes. Conditions of other
+/// types are kept.
+pub fn update_conditions(
+    current: &[Condition],
+    updates: Vec<ManagedCondition>,
+    now: &Time,
+) -> Vec<Condition> {
+    let mut conditions = current.to_vec();
+
+    for update in updates {
+        let type_ = update.type_.to_string();
+        let status = if update.status { "True" } else { "False" }.to_owned();
+        let existing = conditions
+            .iter()
+            .position(|condition| condition.type_ == type_);
+
+        let last_transition_time = match existing {
+            Some(index) if conditions[index].status == status => {
+                conditions[index].last_transition_time.clone()
+            }
+            _ => now.clone(),
+        };
+
+        let condition = Condition {
+            last_transition_time,
+            message: update.message,
+            observed_generation: None,
+            reason: update.reason,
+            status,
+            type_,
+        };
+
+        match existing {
+            Some(index) => conditions[index] = condition,
+            None => conditions.push(condition),
+        }
+    }
+
+    conditions
+}
+
+#[cfg(test)]
+mod tests {
+    use k8s_openapi::jiff::Timestamp;
+
+    use super::*;
+
+    fn time(seconds: i64) -> Time {
+        Time(Timestamp::from_second(seconds).expect("the timestamp should be valid"))
+    }
+
+    fn ready(status: bool, reason: &str) -> ManagedCondition {
+        ManagedCondition {
+            type_: ManagedConditionType::Ready,
+            status,
+            reason: reason.to_owned(),
+            message: String::new(),
+        }
+    }
+
+    #[test]
+    fn add_missing_condition() {
+        let conditions = update_conditions(&[], vec![ready(true, "Created")], &time(10));
+
+        assert_eq!(1, conditions.len());
+        assert_eq!("Ready", conditions[0].type_);
+        assert_eq!("True", conditions[0].status);
+        assert_eq!("Created", conditions[0].reason);
+        assert_eq!(time(10), conditions[0].last_transition_time);
+    }
+
+    #[test]
+    fn keep_transition_time_if_status_is_unchanged() {
+        let current = update_conditions(&[], vec![ready(true, "Created")], &time(10));
+
+        let conditions = update_conditions(&current, vec![ready(true, "Updated")], &time(20));
+
+        assert_eq!("Updated", conditions[0].reason);
+        assert_eq!(time(10), conditions[0].last_transition_time);
+    }
+
+    #[test]
+    fn update_transition_time_if_status_changes() {
+        let current = update_conditions(&[], vec![ready(true, "Created")], &time(10));
+
+        let conditions =
+            update_conditions(&current, vec![ready(false, "KafkaRejected")], &time(20));
+
+        assert_eq!("False", conditions[0].status);
+        assert_eq!(time(20), conditions[0].last_transition_time);
+    }
+
+    #[test]
+    fn keep_conditions_of_other_types() {
+        let other = Condition {
+            last_transition_time: time(5),
+            message: String::new(),
+            observed_generation: None,
+            reason: "Other".to_owned(),
+            status: "True".to_owned(),
+            type_: "Other".to_owned(),
+        };
+
+        let conditions = update_conditions(
+            std::slice::from_ref(&other),
+            vec![ready(true, "Created")],
+            &time(10),
+        );
+
+        let types: Vec<_> = conditions
+            .iter()
+            .map(|condition| condition.type_.as_str())
+            .collect();
+        assert_eq!(vec!["Other", "Ready"], types);
+    }
+}
diff --git a/crates/stackable-operator/src/commons/platform_access/mod.rs b/crates/stackable-operator/src/commons/platform_access/mod.rs
new file mode 100644
index 000000000..d557ad887
--- /dev/null
+++ b/crates/stackable-operator/src/commons/platform_access/mod.rs
@@ -0,0 +1,5 @@
+//! Helpers for agents managing resources inside a product. The CRD fragments live in
+//! [`crate::crd::platform_access`].
+
+pub mod condition;
+pub mod ownership;
diff --git a/crates/stackable-operator/src/commons/platform_access/ownership.rs b/crates/stackable-operator/src/commons/platform_access/ownership.rs
new file mode 100644
index 000000000..9bb48d1eb
--- /dev/null
+++ b/crates/stackable-operator/src/commons/platform_access/ownership.rs
@@ -0,0 +1,37 @@
+use kube::api::ObjectMeta;
+
+/// Annotation an agent sets on an object once it created or adopted the resource in the product.
+///
+/// The value is the RFC 3339 timestamp of that moment. Agents only delete resources they own.
+pub const OWNED_SINCE_ANNOTATION: &str = "stackable.tech/owned-since";
+
+/// Whether the agent owns the resource in the product that the object with this metadata manages.
+pub fn is_owned(metadata: &ObjectMeta) -> bool {
+    metadata
+        .annotations
+        .as_ref()
+        .is_some_and(|annotations| annotations.contains_key(OWNED_SINCE_ANNOTATION))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use rstest::rstest;
+
+    use super::*;
+
+    #[rstest]
+    #[case::no_annotations(None, false)]
+    #[case::other_annotation(Some(("stackable.tech/other", "x")), false)]
+    #[case::owned(Some((OWNED_SINCE_ANNOTATION, "2026-10-06T09:12:44Z")), true)]
+    fn ownership(#[case] annotation: Option<(&str, &str)>, #[case] expected: bool) {
+        let metadata = ObjectMeta {
+            annotations: annotation
+                .map(|(key, value)| BTreeMap::from([(key.to_owned(), value.to_owned())])),
+            ..ObjectMeta::default()
+        };
+
+        assert_eq!(expected, is_owned(&metadata));
+    }
+}
diff --git a/crates/stackable-operator/src/crd/mod.rs b/crates/stackable-operator/src/crd/mod.rs
index 28f40f997..0c299cf9b 100644
--- a/crates/stackable-operator/src/crd/mod.rs
+++ b/crates/stackable-operator/src/crd/mod.rs
@@ -7,6 +7,7 @@ use serde::{Deserialize, Serialize};
 pub mod authentication;
 pub mod git_sync;
 pub mod listener;
+pub mod platform_access;
 pub mod s3;
 pub mod scaler;
 
diff --git a/crates/stackable-operator/src/crd/platform_access/anonymous/mod.rs b/crates/stackable-operator/src/crd/platform_access/anonymous/mod.rs
new file mode 100644
index 000000000..04b52eae6
--- /dev/null
+++ b/crates/stackable-operator/src/crd/platform_access/anonymous/mod.rs
@@ -0,0 +1,32 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::versioned::versioned;
+
+#[versioned(version(name = "v1alpha1"))]
+pub mod versioned {
+    /// No credential: the product does not authenticate the agent.
+    #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    pub struct Anonymous {}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::v1alpha1::Anonymous;
+
+    #[test]
+    fn deserialize_empty_object() {
+        let anonymous: Anonymous =
+            serde_json::from_str("{}").expect("an empty object should be valid");
+
+        assert_eq!(Anonymous {}, anonymous);
+    }
+
+    #[test]
+    fn schema_is_an_object() {
+        let schema = serde_json::to_value(schemars::schema_for!(Anonymous))
+            .expect("the schema should be serializable");
+
+        assert_eq!("object", schema["type"]);
+    }
+}
diff --git a/crates/stackable-operator/src/crd/platform_access/management/mod.rs b/crates/stackable-operator/src/crd/platform_access/management/mod.rs
new file mode 100644
index 000000000..fa904584e
--- /dev/null
+++ b/crates/stackable-operator/src/crd/platform_access/management/mod.rs
@@ -0,0 +1,44 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::versioned::versioned;
+
+mod v1alpha1_impl;
+
+#[versioned(version(name = "v1alpha1"))]
+pub mod versioned {
+    /// What an agent may do with a resource inside the product beyond applying its spec.
+    #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    #[serde(default, rename_all = "camelCase")]
+    pub struct ManagementPolicy {
+        /// Whether a resource that already exists in the product, but was not created by the
+        /// agent, is taken over. `Refuse` by default.
+        pub adoption: AdoptionPolicy,
+
+        /// What happens to the resource in the product when this object is deleted. `Retain` by
+        /// default.
+        pub deletion: DeletionPolicy,
+    }
+
+    /// Whether an agent takes over a resource that already exists in the product.
+    #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    pub enum AdoptionPolicy {
+        /// Report the existing resource as a conflict and leave it untouched.
+        #[default]
+        Refuse,
+
+        /// Take over the existing resource and apply the spec to it.
+        Adopt,
+    }
+
+    /// What happens to a resource in the product when the object managing it is deleted.
+    #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    pub enum DeletionPolicy {
+        /// Keep the resource in the product.
+        #[default]
+        Retain,
+
+        /// Delete the resource in the product, if the agent created or adopted it.
+        Delete,
+    }
+}
diff --git a/crates/stackable-operator/src/crd/platform_access/management/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/platform_access/management/v1alpha1_impl.rs
new file mode 100644
index 000000000..2686b588e
--- /dev/null
+++ b/crates/stackable-operator/src/crd/platform_access/management/v1alpha1_impl.rs
@@ -0,0 +1,44 @@
+use crate::crd::platform_access::management::v1alpha1::{
+    AdoptionPolicy, DeletionPolicy, ManagementPolicy,
+};
+
+impl ManagementPolicy {
+    /// Whether the agent takes over a resource that already exists in the product.
+    pub fn adopts_existing(&self) -> bool {
+        self.adoption == AdoptionPolicy::Adopt
+    }
+
+    /// Whether the agent deletes the resource in the product when this object is deleted.
+    pub fn deletes_on_cleanup(&self) -> bool {
+        self.deletion == DeletionPolicy::Delete
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use rstest::rstest;
+
+    use super::*;
+
+    #[rstest]
+    #[case::defaults("{}", false, false)]
+    #[case::adopt(r#"{"adoption": "Adopt"}"#, true, false)]
+    #[case::delete(r#"{"deletion": "Delete"}"#, false, true)]
+    #[case::explicit_defaults(r#"{"adoption": "Refuse", "deletion": "Retain"}"#, false, false)]
+    fn deserialize_policy(
+        #[case] input: &str,
+        #[case] adopts_existing: bool,
+        #[case] deletes_on_cleanup: bool,
+    ) {
+        let policy: ManagementPolicy =
+            serde_json::from_str(input).expect("the policy should be valid");
+
+        assert_eq!(adopts_existing, policy.adopts_existing());
+        assert_eq!(deletes_on_cleanup, policy.deletes_on_cleanup());
+    }
+
+    #[test]
+    fn reject_unknown_policy() {
+        assert!(serde_json::from_str::<ManagementPolicy>(r#"{"deletion": "Orphan"}"#).is_err());
+    }
+}
diff --git a/crates/stackable-operator/src/crd/platform_access/mod.rs b/crates/stackable-operator/src/crd/platform_access/mod.rs
new file mode 100644
index 000000000..e116b7317
--- /dev/null
+++ b/crates/stackable-operator/src/crd/platform_access/mod.rs
@@ -0,0 +1,9 @@
+//! CRD fragments for resources managed by product agents, and for the credentials the agents use.
+
+pub mod anonymous;
+pub mod management;
+pub mod tls;
+
+pub mod v1alpha1 {
+    pub use super::{anonymous::v1alpha1::*, management::v1alpha1::*, tls::v1alpha1::*};
+}
diff --git a/crates/stackable-operator/src/crd/platform_access/tls/mod.rs b/crates/stackable-operator/src/crd/platform_access/tls/mod.rs
new file mode 100644
index 000000000..92c1222b1
--- /dev/null
+++ b/crates/stackable-operator/src/crd/platform_access/tls/mod.rs
@@ -0,0 +1,32 @@
+use const_format::concatcp;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    constants::secret::SECRET_BASE_PATH,
+    v2::types::kubernetes::{SecretClassName, SecretName},
+    versioned::versioned,
+};
+
+mod v1alpha1_impl;
+
+/// Name of the volume holding the TLS client certificate.
+pub const VOLUME_NAME: &str = "tls-client-cert";
+
+/// Mount path for the TLS client certificate.
+pub const MOUNT_PATH: &str = concatcp!(SECRET_BASE_PATH, "/", VOLUME_NAME);
+
+#[versioned(version(name = "v1alpha1"))]
+pub mod versioned {
+    /// Source of a TLS client certificate: a secret-operator SecretClass or a static Secret.
+    #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub enum TlsClientCredential {
+        /// An AutoTLS SecretClass used to provision the certificate.
+        SecretClass(SecretClassName),
+
+        /// A static Secret holding the certificate in the keys `tls.crt` and `tls.key` (PEM), e.g. as
+        /// a Secret of type `kubernetes.io/tls`.
+        Secret(SecretName),
+    }
+}
diff --git a/crates/stackable-operator/src/crd/platform_access/tls/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/platform_access/tls/v1alpha1_impl.rs
new file mode 100644
index 000000000..936cebf16
--- /dev/null
+++ b/crates/stackable-operator/src/crd/platform_access/tls/v1alpha1_impl.rs
@@ -0,0 +1,106 @@
+use k8s_openapi::api::core::v1::{SecretVolumeSource, Volume, VolumeMount};
+
+use crate::{
+    builder::pod::{
+        PodBuilder,
+        container::ContainerBuilder,
+        volume::{
+            SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder, VolumeMountBuilder,
+        },
+    },
+    commons::secret_class::SecretClassVolumeProvisionParts,
+    crd::platform_access::tls::{MOUNT_PATH, VOLUME_NAME, v1alpha1::TlsClientCredential},
+};
+
+impl TlsClientCredential {
+    /// Adds the certificate volume to the Pod and mounts it into all given containers.
+    /// - TlsClientCredential::Secret mounts the Secret
+    /// - TlsClientCredential::SecretClass adds a secret-operator volume
+    pub fn add_volumes_and_mounts(
+        &self,
+        pod_builder: &mut PodBuilder,
+        container_builders: Vec<&mut ContainerBuilder>,
+    ) {
+        let (volumes, mounts) = self.volumes_and_mounts();
+        pod_builder
+            .add_volumes(volumes)
+            .expect("Only a single platform access authentication variant can be chosen.");
+        for container_builder in container_builders {
+            container_builder
+                .add_volume_mounts(mounts.clone())
+                .expect("Only a single platform access authentication variant can be chosen.");
+        }
+    }
+
+    fn volumes_and_mounts(&self) -> (Vec<Volume>, Vec<VolumeMount>) {
+        let volume = match self {
+            Self::SecretClass(secret_class) => VolumeBuilder::new(VOLUME_NAME)
+                .ephemeral(
+                    SecretOperatorVolumeSourceBuilder::new(
+                        secret_class,
+                        SecretClassVolumeProvisionParts::PublicPrivate,
+                    )
+                    .with_pod_scope()
+                    .with_format(SecretFormat::TlsPem)
+                    .build()
+                    .expect("the annotations are built from a valid SecretClass name"),
+                )
+                .build(),
+            Self::Secret(secret) => VolumeBuilder::new(VOLUME_NAME)
+                .secret(SecretVolumeSource {
+                    secret_name: Some(secret.to_string()),
+                    ..SecretVolumeSource::default()
+                })
+                .build(),
+        };
+        let mount = VolumeMountBuilder::new(VOLUME_NAME, MOUNT_PATH).build();
+        (vec![volume], vec![mount])
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::*;
+    use crate::v2::types::kubernetes::{SecretClassName, SecretName};
+
+    #[test]
+    fn secret_class_credential_is_provisioned_by_secret_operator() {
+        let credential = TlsClientCredential::SecretClass(SecretClassName::from_str_unsafe("tls"));
+
+        let (volumes, mounts) = credential.volumes_and_mounts();
+        let volumes = serde_json::to_value(volumes).expect("serializable");
+
+        assert_eq!(volumes[0]["name"], "tls-client-cert");
+        assert_eq!(
+            volumes[0]["ephemeral"]["volumeClaimTemplate"]["metadata"]["annotations"],
+            json!({
+                "secrets.stackable.tech/class": "tls",
+                "secrets.stackable.tech/format": "tls-pem",
+                "secrets.stackable.tech/provision-parts": "public-private",
+                "secrets.stackable.tech/scope": "pod"
+            })
+        );
+        assert_eq!(
+            serde_json::to_value(mounts).expect("serializable"),
+            json!([{"mountPath": "/stackable/secrets/tls-client-cert", "name": "tls-client-cert"}])
+        );
+    }
+
+    #[test]
+    fn static_secret_credential_is_mounted_directly() {
+        let credential = TlsClientCredential::Secret(SecretName::from_str_unsafe("my-cert"));
+
+        let (volumes, mounts) = credential.volumes_and_mounts();
+
+        assert_eq!(
+            serde_json::to_value(volumes).expect("serializable"),
+            json!([{"name": "tls-client-cert", "secret": {"secretName": "my-cert"}}])
+        );
+        assert_eq!(
+            serde_json::to_value(mounts).expect("serializable"),
+            json!([{"mountPath": "/stackable/secrets/tls-client-cert", "name": "tls-client-cert"}])
+        );
+    }
+}
diff --git a/crates/stackable-operator/src/v2/kvp/label.rs b/crates/stackable-operator/src/v2/kvp/label.rs
index a0d806557..383ae5401 100644
--- a/crates/stackable-operator/src/v2/kvp/label.rs
+++ b/crates/stackable-operator/src/v2/kvp/label.rs
@@ -28,6 +28,35 @@ pub fn recommended_labels_for_cluster_resources(
     ])
 }
 
+/// Creates the recommended labels for agent resources, like the agent Deployment.
+pub fn recommended_labels_for_agent_resources(
+    cluster_name: &ClusterName,
+    product_name: &ProductName,
+    product_version: &ProductVersion,
+    operator_name: &OperatorName,
+    controller_name: &ControllerName,
+) -> Labels {
+    Labels::from_iter([
+        label_app_kubernetes_io_instance(cluster_name),
+        label_app_kubernetes_io_name(product_name),
+        label_app_kubernetes_io_version(product_version),
+        label_app_kubernetes_io_component_agent(),
+        label_app_kubernetes_io_managed_by(operator_name, controller_name),
+        label_stackable_tech_vendor(),
+    ])
+}
+
+/// Creates the agent selector.
+///
+/// The returned labels are a subset of the recommended labels for agent resources.
+pub fn agent_selector(cluster_name: &ClusterName, product_name: &ProductName) -> Labels {
+    Labels::from_iter([
+        label_app_kubernetes_io_instance(cluster_name),
+        label_app_kubernetes_io_name(product_name),
+        label_app_kubernetes_io_component_agent(),
+    ])
+}
+
 /// Creates the recommended labels for role resources, like discovery ConfigMaps.
 pub fn recommended_labels_for_role_resources(
     cluster_name: &ClusterName,
@@ -155,6 +184,11 @@ pub fn label_app_kubernetes_io_component(role_name: &RoleName) -> Label {
         .expect("the value implements NameIsValidLabelValue and is therefore a valid label value")
 }
 
+/// Creates the `app.kubernetes.io/component` label with the value `agent`.
+pub fn label_app_kubernetes_io_component_agent() -> Label {
+    Label::component("agent").expect("\"agent\" is a valid label value")
+}
+
 /// Creates the `app.kubernetes.io/role-group` label with the given role group as value.
 pub fn label_app_kubernetes_io_role_group(role_group_name: &RoleGroupName) -> Label {
     Label::role_group(&role_group_name.to_label_value())
@@ -214,6 +248,52 @@ mod tests {
         assert_eq!(expected_labels, actual_labels.into());
     }
 
+    #[test]
+    fn recommended_labels_for_agent_resources_produces_expected_labels() {
+        let actual_labels = recommended_labels_for_agent_resources(
+            &ClusterName::from_str_unsafe("cluster-name"),
+            &ProductName::from_str_unsafe("my-product"),
+            &ProductVersion::from_str_unsafe("1.0.0"),
+            &OperatorName::from_str_unsafe("my-operator"),
+            &ControllerName::from_str_unsafe("my-controller"),
+        );
+
+        let expected_labels: BTreeMap<_, _> = [
+            ("app.kubernetes.io/component", "agent"),
+            ("app.kubernetes.io/instance", "cluster-name"),
+            ("app.kubernetes.io/managed-by", "my-operator_my-controller"),
+            ("app.kubernetes.io/name", "my-product"),
+            ("app.kubernetes.io/version", "1.0.0"),
+            ("stackable.tech/vendor", "Stackable"),
+        ]
+        .map(|(k, v)| (k.to_owned(), v.to_owned()))
+        .into();
+
+        assert_eq!(expected_labels, actual_labels.into());
+    }
+
+    #[test]
+    fn agent_selector_is_subset_of_recommended_agent_labels() {
+        let cluster_name = ClusterName::from_str_unsafe("cluster-name");
+        let product_name = ProductName::from_str_unsafe("my-product");
+
+        let agent_labels = recommended_labels_for_agent_resources(
+            &cluster_name,
+            &product_name,
+            &ProductVersion::from_str_unsafe("1.0.0"),
+            &OperatorName::from_str_unsafe("my-operator"),
+            &ControllerName::from_str_unsafe("my-controller"),
+        );
+
+        let agent_selector = agent_selector(&cluster_name, &product_name);
+
+        assert!(
+            agent_selector
+                .iter()
+                .all(|selector| agent_labels.contains(selector))
+        );
+    }
+
     #[test]
     fn recommended_labels_for_role_resources_produces_expected_labels() {
         let actual_labels = recommended_labels_for_role_resources(